自学教程：C++ GOTO函数代码示例

bool_mongoc_client_recv_gle (mongoc_client_t  *client,                         uint32_t          server_id,                         bson_t          **gle_doc,                         bson_error_t     *error){   mongoc_buffer_t buffer;   mongoc_rpc_t rpc;   bson_iter_t iter;   bool ret = false;   bson_t b;   ENTRY;   BSON_ASSERT (client);   BSON_ASSERT (server_id);   if (gle_doc) {      *gle_doc = NULL;   }   _mongoc_buffer_init (&buffer, NULL, 0, NULL, NULL);   if (!mongoc_cluster_try_recv (&client->cluster, &rpc, &buffer,                                 server_id, error)) {      mongoc_topology_invalidate_server(client->topology, server_id);      GOTO (cleanup);   }   if (rpc.header.opcode != MONGOC_OPCODE_REPLY) {      bson_set_error (error,                      MONGOC_ERROR_PROTOCOL,                      MONGOC_ERROR_PROTOCOL_INVALID_REPLY,                      "Received message other than OP_REPLY.");      GOTO (cleanup);   }   if (_mongoc_rpc_reply_get_first (&rpc.reply, &b)) {      if ((rpc.reply.flags & MONGOC_REPLY_QUERY_FAILURE)) {         _bson_to_error (&b, error);         bson_destroy (&b);         GOTO (cleanup);      }      if (gle_doc) {         *gle_doc = bson_copy (&b);      }      if (!bson_iter_init_find (&iter, &b, "ok") ||          BSON_ITER_HOLDS_DOUBLE (&iter)) {        if (bson_iter_double (&iter) == 0.0) {          _bson_to_error (&b, error);        }      }      bson_destroy (&b);      ret = true;   }cleanup:   _mongoc_buffer_destroy (&buffer);   RETURN (ret);}

static voidmongo_cursor_foreach_dispatch (MongoConnection    *connection,                               MongoMessageReply  *reply,                               GSimpleAsyncResult *simple){   MongoCursorCallback func;   MongoCursorPrivate *priv;   GCancellable *cancellable;   MongoCursor *cursor;   MongoBson *bson;   gpointer func_data;   guint64 cursor_id;   gchar *db_and_collection;   GList *iter;   GList *list;   guint offset;   guint i;   ENTRY;   g_assert(MONGO_IS_CONNECTION(connection));   g_assert(reply);   g_assert(G_IS_SIMPLE_ASYNC_RESULT(simple));   func = g_object_get_data(G_OBJECT(simple), "foreach-func");   func_data = g_object_get_data(G_OBJECT(simple), "foreach-data");   g_assert(func);   cursor = MONGO_CURSOR(g_async_result_get_source_object(G_ASYNC_RESULT(simple)));   g_assert(MONGO_IS_CURSOR(cursor));   cancellable = g_object_get_data(G_OBJECT(simple), "cancellable");   g_assert(!cancellable || G_IS_CANCELLABLE(cancellable));   priv = cursor->priv;   if (!(list = mongo_message_reply_get_documents(reply))) {      GOTO(stop);   }   offset = mongo_message_reply_get_offset(reply);   for (iter = list, i = 0; iter; iter = iter->next, i++) {      bson = iter->data;      if (priv->limit && (offset + i) >= priv->limit) {         GOTO(stop);      }      if (!func(cursor, bson, func_data)) {         GOTO(stop);      }   }   offset = mongo_message_reply_get_offset(reply);   cursor_id = mongo_message_reply_get_cursor_id(reply);   if (!cursor_id || ((offset + g_list_length(list)) >= priv->limit)) {      GOTO(stop);   }   /*    * TODO: How do we know if we are finished if EXHAUST is set?    */   if (!(cursor->priv->flags & MONGO_QUERY_EXHAUST)) {      db_and_collection = g_strdup_printf("%s.%s",                                          cursor->priv->database,                                          cursor->priv->collection);      mongo_connection_getmore_async(connection,                                     db_and_collection,                                     cursor->priv->batch_size,                                     cursor_id,                                     cancellable,                                     mongo_cursor_foreach_getmore_cb,                                     simple);      g_free(db_and_collection);   }   g_object_unref(cursor);   EXIT;stop:   if (mongo_message_reply_get_cursor_id(reply)) {      mongo_connection_kill_cursors_async(connection,                                          &cursor_id,                                          1,                                          cancellable,                                          mongo_cursor_kill_cursors_cb,                                          NULL);   }   g_simple_async_result_set_op_res_gboolean(simple, TRUE);   mongo_simple_async_result_complete_in_idle(simple);   g_object_unref(simple);   g_object_unref(cursor);   EXIT;}

/* returns negative on error; 0 if success; 1 if success & log destroyed */int llog_cancel_rec(const struct lu_env *env, struct llog_handle *loghandle,		    int index){	struct llog_thread_info *lgi = llog_info(env);	struct dt_device	*dt;	struct llog_log_hdr	*llh = loghandle->lgh_hdr;	struct thandle		*th;	int			 rc;	int rc1;	bool subtract_count = false;	ENTRY;	CDEBUG(D_RPCTRACE, "Canceling %d in log "DOSTID"/n", index,	       POSTID(&loghandle->lgh_id.lgl_oi));	if (index == 0) {		CERROR("Can't cancel index 0 which is header/n");		RETURN(-EINVAL);	}	LASSERT(loghandle != NULL);	LASSERT(loghandle->lgh_ctxt != NULL);	LASSERT(loghandle->lgh_obj != NULL);	dt = lu2dt_dev(loghandle->lgh_obj->do_lu.lo_dev);	th = dt_trans_create(env, dt);	if (IS_ERR(th))		RETURN(PTR_ERR(th));	rc = llog_declare_write_rec(env, loghandle, &llh->llh_hdr, index, th);	if (rc < 0)		GOTO(out_trans, rc);	if ((llh->llh_flags & LLOG_F_ZAP_WHEN_EMPTY))		rc = llog_declare_destroy(env, loghandle, th);	th->th_wait_submit = 1;	rc = dt_trans_start_local(env, dt, th);	if (rc < 0)		GOTO(out_trans, rc);	down_write(&loghandle->lgh_lock);	/* clear bitmap */	mutex_lock(&loghandle->lgh_hdr_mutex);	if (!ext2_clear_bit(index, LLOG_HDR_BITMAP(llh))) {		CDEBUG(D_RPCTRACE, "Catalog index %u already clear?/n", index);		GOTO(out_unlock, rc);	}	loghandle->lgh_hdr->llh_count--;	subtract_count = true;	/* Pass this index to llog_osd_write_rec(), which will use the index	 * to only update the necesary bitmap. */	lgi->lgi_cookie.lgc_index = index;	/* update header */	rc = llog_write_rec(env, loghandle, &llh->llh_hdr, &lgi->lgi_cookie,			    LLOG_HEADER_IDX, th);	if (rc != 0)		GOTO(out_unlock, rc);	if ((llh->llh_flags & LLOG_F_ZAP_WHEN_EMPTY) &&	    (llh->llh_count == 1) &&	    ((loghandle->lgh_last_idx == LLOG_HDR_BITMAP_SIZE(llh) - 1) ||	     (loghandle->u.phd.phd_cat_handle != NULL &&	      loghandle->u.phd.phd_cat_handle->u.chd.chd_current_log !=		loghandle))) {		/* never try to destroy it again */		llh->llh_flags &= ~LLOG_F_ZAP_WHEN_EMPTY;		rc = llog_trans_destroy(env, loghandle, th);		if (rc < 0) {			/* Sigh, can not destroy the final plain llog, but			 * the bitmap has been clearly, so the record can not			 * be accessed anymore, let's return 0 for now, and			 * the orphan will be handled by LFSCK. */			CERROR("%s: can't destroy empty llog #"DOSTID			       "#%08x: rc = %d/n",			       loghandle->lgh_ctxt->loc_obd->obd_name,			       POSTID(&loghandle->lgh_id.lgl_oi),			       loghandle->lgh_id.lgl_ogen, rc);			GOTO(out_unlock, rc);		}		rc = LLOG_DEL_PLAIN;	}out_unlock:	mutex_unlock(&loghandle->lgh_hdr_mutex);	up_write(&loghandle->lgh_lock);out_trans:	rc1 = dt_trans_stop(env, dt, th);	if (rc == 0)		rc = rc1;	if (rc < 0 && subtract_count) {		mutex_lock(&loghandle->lgh_hdr_mutex);		loghandle->lgh_hdr->llh_count++;		ext2_set_bit(index, LLOG_HDR_BITMAP(llh));		mutex_unlock(&loghandle->lgh_hdr_mutex);	}//.........这里部分代码省略.........

static int llog_process_thread(void *arg){	struct llog_process_info	*lpi = arg;	struct llog_handle		*loghandle = lpi->lpi_loghandle;	struct llog_log_hdr		*llh = loghandle->lgh_hdr;	struct llog_process_cat_data	*cd  = lpi->lpi_catdata;	char				*buf;	size_t				 chunk_size;	__u64				 cur_offset, tmp_offset;	int				 rc = 0, index = 1, last_index;	int				 saved_index = 0;	int				 last_called_index = 0;	ENTRY;	if (llh == NULL)		RETURN(-EINVAL);	cur_offset = chunk_size = llh->llh_hdr.lrh_len;	/* expect chunk_size to be power of two */	LASSERT(is_power_of_2(chunk_size));	OBD_ALLOC_LARGE(buf, chunk_size);	if (buf == NULL) {		lpi->lpi_rc = -ENOMEM;		RETURN(0);	}	if (cd != NULL) {		last_called_index = cd->lpcd_first_idx;		index = cd->lpcd_first_idx + 1;	}	if (cd != NULL && cd->lpcd_last_idx)		last_index = cd->lpcd_last_idx;	else		last_index = LLOG_HDR_BITMAP_SIZE(llh) - 1;	while (rc == 0) {		struct llog_rec_hdr *rec;		off_t chunk_offset;		unsigned int buf_offset = 0;		bool partial_chunk;		/* skip records not set in bitmap */		while (index <= last_index &&		       !ext2_test_bit(index, LLOG_HDR_BITMAP(llh)))			++index;		/* There are no indices prior the last_index */		if (index > last_index)			break;		CDEBUG(D_OTHER, "index: %d last_index %d/n", index,		       last_index);repeat:		/* get the buf with our target record; avoid old garbage */		memset(buf, 0, chunk_size);		rc = llog_next_block(lpi->lpi_env, loghandle, &saved_index,				     index, &cur_offset, buf, chunk_size);		if (rc != 0)			GOTO(out, rc);		/* NB: after llog_next_block() call the cur_offset is the		 * offset of the next block after read one.		 * The absolute offset of the current chunk is calculated		 * from cur_offset value and stored in chunk_offset variable.		 */		tmp_offset = cur_offset;		if (do_div(tmp_offset, chunk_size) != 0) {			partial_chunk = true;			chunk_offset = cur_offset & ~(chunk_size - 1);		} else {			partial_chunk = false;			chunk_offset = cur_offset - chunk_size;		}		/* NB: when rec->lrh_len is accessed it is already swabbed		 * since it is used at the "end" of the loop and the rec		 * swabbing is done at the beginning of the loop. */		for (rec = (struct llog_rec_hdr *)(buf + buf_offset);		     (char *)rec < buf + chunk_size;		     rec = llog_rec_hdr_next(rec)) {			CDEBUG(D_OTHER, "processing rec 0x%p type %#x/n",			       rec, rec->lrh_type);			if (LLOG_REC_HDR_NEEDS_SWABBING(rec))				lustre_swab_llog_rec(rec);			CDEBUG(D_OTHER, "after swabbing, type=%#x idx=%d/n",			       rec->lrh_type, rec->lrh_index);			/* for partial chunk the end of it is zeroed, check			 * for index 0 to distinguish it. */			if (partial_chunk && rec->lrh_index == 0) {				/* concurrent llog_add() might add new records				 * while llog_processing, check this is not				 * the case and re-read the current chunk				 * otherwise. *///.........这里部分代码省略.........

static int osd_object_destroy(const struct lu_env *env,			      struct dt_object *dt, struct thandle *th){	struct osd_thread_info	*info = osd_oti_get(env);	char			*buf = info->oti_str;	struct osd_object	*obj = osd_dt_obj(dt);	struct osd_device	*osd = osd_obj2dev(obj);	const struct lu_fid	*fid = lu_object_fid(&dt->do_lu);	struct osd_thandle	*oh;	int			 rc;	uint64_t		 oid, zapid;	ENTRY;	down_write(&obj->oo_guard);	if (unlikely(!dt_object_exists(dt) || obj->oo_destroyed))		GOTO(out, rc = -ENOENT);	LASSERT(obj->oo_db != NULL);	oh = container_of0(th, struct osd_thandle, ot_super);	LASSERT(oh != NULL);	LASSERT(oh->ot_tx != NULL);	/* remove obj ref from index dir (it depends) */	zapid = osd_get_name_n_idx(env, osd, fid, buf, sizeof(info->oti_str));	rc = -zap_remove(osd->od_os, zapid, buf, oh->ot_tx);	if (rc) {		CERROR("%s: zap_remove(%s) failed: rc = %d/n",		       osd->od_svname, buf, rc);		GOTO(out, rc);	}	rc = osd_xattrs_destroy(env, obj, oh);	if (rc) {		CERROR("%s: cannot destroy xattrs for %s: rc = %d/n",		       osd->od_svname, buf, rc);		GOTO(out, rc);	}	/* Remove object from inode accounting. It is not fatal for the destroy	 * operation if something goes wrong while updating accounting, but we	 * still log an error message to notify the administrator */	rc = -zap_increment_int(osd->od_os, osd->od_iusr_oid,				obj->oo_attr.la_uid, -1, oh->ot_tx);	if (rc)		CERROR("%s: failed to remove "DFID" from accounting ZAP for usr"		       " %d: rc = %d/n", osd->od_svname, PFID(fid),		       obj->oo_attr.la_uid, rc);	rc = -zap_increment_int(osd->od_os, osd->od_igrp_oid,				obj->oo_attr.la_gid, -1, oh->ot_tx);	if (rc)		CERROR("%s: failed to remove "DFID" from accounting ZAP for grp"		       " %d: rc = %d/n", osd->od_svname, PFID(fid),		       obj->oo_attr.la_gid, rc);	oid = obj->oo_db->db_object;	if (unlikely(obj->oo_destroy == OSD_DESTROY_NONE)) {		/* this may happen if the destroy wasn't declared		 * e.g. when the object is created and then destroyed		 * in the same transaction - we don't need additional		 * space for destroy specifically */		LASSERT(obj->oo_attr.la_size <= osd_sync_destroy_max_size);		rc = -dmu_object_free(osd->od_os, oid, oh->ot_tx);		if (rc)			CERROR("%s: failed to free %s %llu: rc = %d/n",			       osd->od_svname, buf, oid, rc);	} else if (obj->oo_destroy == OSD_DESTROY_SYNC) {		rc = -dmu_object_free(osd->od_os, oid, oh->ot_tx);		if (rc)			CERROR("%s: failed to free %s %llu: rc = %d/n",			       osd->od_svname, buf, oid, rc);	} else { /* asynchronous destroy */		rc = osd_object_unlinked_add(obj, oh);		if (rc)			GOTO(out, rc);		rc = -zap_add_int(osd->od_os, osd->od_unlinkedid,				  oid, oh->ot_tx);		if (rc)			CERROR("%s: zap_add_int() failed %s %llu: rc = %d/n",			       osd->od_svname, buf, oid, rc);	}out:	/* not needed in the cache anymore */	set_bit(LU_OBJECT_HEARD_BANSHEE, &dt->do_lu.lo_header->loh_flags);	if (rc == 0)		obj->oo_destroyed = 1;	up_write(&obj->oo_guard);	RETURN (0);}

/* * Set the attributes of an object * * The transaction passed to this routine must have * dmu_tx_hold_bonus(tx, oid) called and then assigned * to a transaction group. */static int osd_attr_set(const struct lu_env *env, struct dt_object *dt,			const struct lu_attr *la, struct thandle *handle){	struct osd_thread_info	*info = osd_oti_get(env);	sa_bulk_attr_t		*bulk = osd_oti_get(env)->oti_attr_bulk;	struct osd_object	*obj = osd_dt_obj(dt);	struct osd_device	*osd = osd_obj2dev(obj);	struct osd_thandle	*oh;	struct osa_attr		*osa = &info->oti_osa;	__u64			 valid = la->la_valid;	int			 cnt;	int			 rc = 0;	ENTRY;	down_read(&obj->oo_guard);	if (unlikely(!dt_object_exists(dt) || obj->oo_destroyed))		GOTO(out, rc = -ENOENT);	LASSERT(handle != NULL);	LASSERT(osd_invariant(obj));	LASSERT(obj->oo_sa_hdl);	oh = container_of0(handle, struct osd_thandle, ot_super);	/* Assert that the transaction has been assigned to a	   transaction group. */	LASSERT(oh->ot_tx->tx_txg != 0);	/* Only allow set size for regular file */	if (!S_ISREG(dt->do_lu.lo_header->loh_attr))		valid &= ~(LA_SIZE | LA_BLOCKS);	if (valid & LA_CTIME && la->la_ctime == obj->oo_attr.la_ctime)		valid &= ~LA_CTIME;	if (valid & LA_MTIME && la->la_mtime == obj->oo_attr.la_mtime)		valid &= ~LA_MTIME;	if (valid & LA_ATIME && la->la_atime == obj->oo_attr.la_atime)		valid &= ~LA_ATIME;	if (valid == 0)		GOTO(out, rc = 0);	if (valid & LA_FLAGS) {		struct lustre_mdt_attrs *lma;		struct lu_buf buf;		if (la->la_flags & LUSTRE_LMA_FL_MASKS) {			CLASSERT(sizeof(info->oti_buf) >= sizeof(*lma));			lma = (struct lustre_mdt_attrs *)&info->oti_buf;			buf.lb_buf = lma;			buf.lb_len = sizeof(info->oti_buf);			rc = osd_xattr_get(env, &obj->oo_dt, &buf,					   XATTR_NAME_LMA);			if (rc > 0) {				lma->lma_incompat =					le32_to_cpu(lma->lma_incompat);				lma->lma_incompat |=					lustre_to_lma_flags(la->la_flags);				lma->lma_incompat =					cpu_to_le32(lma->lma_incompat);				buf.lb_buf = lma;				buf.lb_len = sizeof(*lma);				rc = osd_xattr_set_internal(env, obj, &buf,							    XATTR_NAME_LMA,							    LU_XATTR_REPLACE,							    oh);			}			if (rc < 0) {				CWARN("%s: failed to set LMA flags: rc = %d/n",				       osd->od_svname, rc);				RETURN(rc);			}		}	}	/* do both accounting updates outside oo_attr_lock below */	if ((valid & LA_UID) && (la->la_uid != obj->oo_attr.la_uid)) {		/* Update user accounting. Failure isn't fatal, but we still		 * log an error message */		rc = -zap_increment_int(osd->od_os, osd->od_iusr_oid,					la->la_uid, 1, oh->ot_tx);		if (rc)			CERROR("%s: failed to update accounting ZAP for user "				"%d (%d)/n", osd->od_svname, la->la_uid, rc);		rc = -zap_increment_int(osd->od_os, osd->od_iusr_oid,					obj->oo_attr.la_uid, -1, oh->ot_tx);		if (rc)			CERROR("%s: failed to update accounting ZAP for user "				"%d (%d)/n", osd->od_svname,				obj->oo_attr.la_uid, rc);	}//.........这里部分代码省略.........

/* Test catalogue additions */static int llog_test_4(const struct lu_env *env, struct obd_device *obd){	struct llog_handle	*cath;	char			 name[10];	int			 rc, rc2, i, buflen;	struct llog_mini_rec	 lmr;	struct llog_cookie	 cookie;	struct llog_ctxt	*ctxt;	int			 num_recs = 0;	char			*buf;	struct llog_rec_hdr	 rec;	ENTRY;	ctxt = llog_get_context(obd, LLOG_TEST_ORIG_CTXT);	LASSERT(ctxt);	lmr.lmr_hdr.lrh_len = lmr.lmr_tail.lrt_len = LLOG_MIN_REC_SIZE;	lmr.lmr_hdr.lrh_type = 0xf00f00;	sprintf(name, "%x", llog_test_rand + 1);	CWARN("4a: create a catalog log with name: %s/n", name);	rc = llog_open_create(env, ctxt, &cath, NULL, name);	if (rc) {		CERROR("4a: llog_create with name %s failed: %d/n", name, rc);		GOTO(ctxt_release, rc);        }	rc = llog_init_handle(env, cath, LLOG_F_IS_CAT, &uuid);	if (rc) {		CERROR("4a: can't init llog handle: %d/n", rc);		GOTO(out, rc);	}	num_recs++;	cat_logid = cath->lgh_id;	CWARN("4b: write 1 record into the catalog/n");	rc = llog_cat_add(env, cath, &lmr.lmr_hdr, &cookie, NULL);	if (rc != 1) {		CERROR("4b: write 1 catalog record failed at: %d/n", rc);		GOTO(out, rc);	}	num_recs++;	rc = verify_handle("4b", cath, 2);	if (rc)		GOTO(out, rc);	rc = verify_handle("4b", cath->u.chd.chd_current_log, num_recs);	if (rc)		GOTO(out, rc);	CWARN("4c: cancel 1 log record/n");	rc = llog_cat_cancel_records(env, cath, 1, &cookie);	if (rc) {		CERROR("4c: cancel 1 catalog based record failed: %d/n", rc);		GOTO(out, rc);	}	num_recs--;	rc = verify_handle("4c", cath->u.chd.chd_current_log, num_recs);	if (rc)		GOTO(out, rc);	CWARN("4d: write %d more log records/n", LLOG_TEST_RECNUM);	for (i = 0; i < LLOG_TEST_RECNUM; i++) {		rc = llog_cat_add(env, cath, &lmr.lmr_hdr, NULL, NULL);		if (rc) {			CERROR("4d: write %d records failed at #%d: %d/n",			       LLOG_TEST_RECNUM, i + 1, rc);			GOTO(out, rc);		}		num_recs++;	}	/* make sure new plain llog appears */	rc = verify_handle("4d", cath, 3);	if (rc)		GOTO(out, rc);	CWARN("4e: add 5 large records, one record per block/n");	buflen = LLOG_CHUNK_SIZE - sizeof(struct llog_rec_hdr) -		 sizeof(struct llog_rec_tail);	OBD_ALLOC(buf, buflen);	if (buf == NULL)		GOTO(out, rc = -ENOMEM);	for (i = 0; i < 5; i++) {		rec.lrh_len = buflen;		rec.lrh_type = OBD_CFG_REC;		rc = llog_cat_add(env, cath, &rec, NULL, buf);		if (rc) {			CERROR("4e: write 5 records failed at #%d: %d/n",			       i + 1, rc);			GOTO(out_free, rc);		}		num_recs++;	}out_free:	OBD_FREE(buf, buflen);out://.........这里部分代码省略.........

/* Test log and catalogue processing */static int llog_test_5(const struct lu_env *env, struct obd_device *obd){	struct llog_handle	*llh = NULL;	char			 name[10];	int			 rc, rc2;	struct llog_mini_rec	 lmr;	struct llog_ctxt	*ctxt;	ENTRY;	ctxt = llog_get_context(obd, LLOG_TEST_ORIG_CTXT);	LASSERT(ctxt);	lmr.lmr_hdr.lrh_len = lmr.lmr_tail.lrt_len = LLOG_MIN_REC_SIZE;	lmr.lmr_hdr.lrh_type = 0xf00f00;	CWARN("5a: re-open catalog by id/n");	rc = llog_open(env, ctxt, &llh, &cat_logid, NULL, LLOG_OPEN_EXISTS);	if (rc) {		CERROR("5a: llog_create with logid failed: %d/n", rc);		GOTO(out_put, rc);	}	rc = llog_init_handle(env, llh, LLOG_F_IS_CAT, &uuid);	if (rc) {		CERROR("5a: can't init llog handle: %d/n", rc);		GOTO(out, rc);	}	CWARN("5b: print the catalog entries.. we expect 2/n");	cat_counter = 0;	rc = llog_process(env, llh, cat_print_cb, "test 5", NULL);	if (rc) {		CERROR("5b: process with cat_print_cb failed: %d/n", rc);		GOTO(out, rc);	}	if (cat_counter != 2) {		CERROR("5b: %d entries in catalog/n", cat_counter);		GOTO(out, rc = -EINVAL);	}	CWARN("5c: Cancel %d records, see one log zapped/n", LLOG_TEST_RECNUM);	cancel_count = 0;	rc = llog_cat_process(env, llh, llog_cancel_rec_cb, "foobar", 0, 0);	if (rc != -LLOG_EEMPTY) {		CERROR("5c: process with cat_cancel_cb failed: %d/n", rc);		GOTO(out, rc);	}	CWARN("5c: print the catalog entries.. we expect 1/n");	cat_counter = 0;	rc = llog_process(env, llh, cat_print_cb, "test 5", NULL);	if (rc) {		CERROR("5c: process with cat_print_cb failed: %d/n", rc);		GOTO(out, rc);	}	if (cat_counter != 1) {		CERROR("5c: %d entries in catalog/n", cat_counter);		GOTO(out, rc = -EINVAL);	}	CWARN("5d: add 1 record to the log with many canceled empty pages/n");	rc = llog_cat_add(env, llh, &lmr.lmr_hdr, NULL, NULL);	if (rc) {		CERROR("5d: add record to the log with many canceled empty "		       "pages failed/n");		GOTO(out, rc);	}	CWARN("5e: print plain log entries.. expect 6/n");	plain_counter = 0;	rc = llog_cat_process(env, llh, plain_print_cb, "foobar", 0, 0);	if (rc) {		CERROR("5e: process with plain_print_cb failed: %d/n", rc);		GOTO(out, rc);	}	if (plain_counter != 6) {		CERROR("5e: found %d records/n", plain_counter);		GOTO(out, rc = -EINVAL);	}	CWARN("5f: print plain log entries reversely.. expect 6/n");	plain_counter = 0;	rc = llog_cat_reverse_process(env, llh, plain_print_cb, "foobar");	if (rc) {		CERROR("5f: reversely process with plain_print_cb failed:"		       "%d/n", rc);		GOTO(out, rc);	}	if (plain_counter != 6) {		CERROR("5f: found %d records/n", plain_counter);		GOTO(out, rc = -EINVAL);	}out:	CWARN("5g: close re-opened catalog/n");	rc2 = llog_cat_close(env, llh);	if (rc2) {		CERROR("5g: close log %s failed: %d/n", name, rc2);//.........这里部分代码省略.........

int liblustre_process_log(struct config_llog_instance *cfg,                          char *mgsnid, char *profile,                          int allow_recov){        struct lustre_cfg_bufs bufs;        struct lustre_cfg *lcfg;        char  *peer = "MGS_UUID";        struct obd_device *obd;        struct obd_export *exp;        char  *name = "mgc_dev";        class_uuid_t uuid;        struct obd_uuid mgc_uuid;        struct llog_ctxt *ctxt;        lnet_nid_t nid = 0;        char *mdsnid;        int err, rc = 0;        struct obd_connect_data *ocd = NULL;        ENTRY;        ll_generate_random_uuid(uuid);        class_uuid_unparse(uuid, &mgc_uuid);        nid = libcfs_str2nid(mgsnid);        if (nid == LNET_NID_ANY) {                CERROR("Can't parse NID %s/n", mgsnid);                RETURN(-EINVAL);        }        lustre_cfg_bufs_reset(&bufs, NULL);        lustre_cfg_bufs_set_string(&bufs, 1, peer);        lcfg = lustre_cfg_new(LCFG_ADD_UUID, &bufs);        lcfg->lcfg_nid = nid;        rc = class_process_config(lcfg);        lustre_cfg_free(lcfg);        if (rc < 0)                GOTO(out, rc);        lustre_cfg_bufs_reset(&bufs, name);        lustre_cfg_bufs_set_string(&bufs, 1, LUSTRE_MGC_NAME);        lustre_cfg_bufs_set_string(&bufs, 2, mgc_uuid.uuid);        lcfg = lustre_cfg_new(LCFG_ATTACH, &bufs);        rc = class_process_config(lcfg);        lustre_cfg_free(lcfg);        if (rc < 0)                GOTO(out_del_uuid, rc);        lustre_cfg_bufs_reset(&bufs, name);        lustre_cfg_bufs_set_string(&bufs, 1, LUSTRE_MGS_OBDNAME);        lustre_cfg_bufs_set_string(&bufs, 2, peer);        lcfg = lustre_cfg_new(LCFG_SETUP, &bufs);        rc = class_process_config(lcfg);        lustre_cfg_free(lcfg);        if (rc < 0)                GOTO(out_detach, rc);        while ((mdsnid = strsep(&mgsnid, ","))) {                nid = libcfs_str2nid(mdsnid);                lustre_cfg_bufs_reset(&bufs, NULL);                lustre_cfg_bufs_set_string(&bufs, 1, libcfs_nid2str(nid));                lcfg = lustre_cfg_new(LCFG_ADD_UUID, &bufs);                lcfg->lcfg_nid = nid;                rc = class_process_config(lcfg);                lustre_cfg_free(lcfg);                if (rc) {                        CERROR("Add uuid for %s failed %d/n",                               libcfs_nid2str(nid), rc);                        continue;                }                lustre_cfg_bufs_reset(&bufs, name);                lustre_cfg_bufs_set_string(&bufs, 1, libcfs_nid2str(nid));                lcfg = lustre_cfg_new(LCFG_ADD_CONN, &bufs);                lcfg->lcfg_nid = nid;                rc = class_process_config(lcfg);                lustre_cfg_free(lcfg);                if (rc) {                        CERROR("Add conn for %s failed %d/n",                               libcfs_nid2str(nid), rc);                        continue;                }        }        obd = class_name2obd(name);        if (obd == NULL)                GOTO(out_cleanup, rc = -EINVAL);        OBD_ALLOC(ocd, sizeof(*ocd));        if (ocd == NULL)                GOTO(out_cleanup, rc = -ENOMEM);	ocd->ocd_connect_flags = OBD_CONNECT_VERSION | OBD_CONNECT_AT |				 OBD_CONNECT_FULL20;        ocd->ocd_version = LUSTRE_VERSION_CODE;        rc = obd_connect(NULL, &exp, obd, &mgc_uuid, ocd, NULL);        if (rc) {                CERROR("cannot connect to %s at %s: rc = %d/n",                       LUSTRE_MGS_OBDNAME, mgsnid, rc);                GOTO(out_cleanup, rc);        }//.........这里部分代码省略.........

/* Test named-log reopen; returns opened log on success */static int llog_test_2(const struct lu_env *env, struct obd_device *obd,		       char *name, struct llog_handle **llh){	struct llog_ctxt	*ctxt;	struct llog_handle	*loghandle;	struct llog_logid	 logid;	int			 rc;	ENTRY;	CWARN("2a: re-open a log with name: %s/n", name);	ctxt = llog_get_context(obd, LLOG_TEST_ORIG_CTXT);	LASSERT(ctxt);	rc = llog_open(env, ctxt, llh, NULL, name, LLOG_OPEN_EXISTS);	if (rc) {		CERROR("2a: re-open log with name %s failed: %d/n", name, rc);		GOTO(out_put, rc);	}	rc = llog_init_handle(env, *llh, LLOG_F_IS_PLAIN, &uuid);	if (rc) {		CERROR("2a: can't init llog handle: %d/n", rc);		GOTO(out_close_llh, rc);	}	rc = verify_handle("2", *llh, 1);	if (rc)		GOTO(out_close_llh, rc);	/* XXX: there is known issue with tests 2b, MGS is not able to create	 * anonymous llog, exit now to allow following tests run.	 * It is fixed in upcoming llog over OSD code */	GOTO(out_put, rc);	CWARN("2b: create a log without specified NAME & LOGID/n");	rc = llog_open_create(env, ctxt, &loghandle, NULL, NULL);	if (rc) {		CERROR("2b: create log failed/n");		GOTO(out_close_llh, rc);	}	rc = llog_init_handle(env, loghandle, LLOG_F_IS_PLAIN, &uuid);	if (rc) {		CERROR("2b: can't init llog handle: %d/n", rc);		GOTO(out_close, rc);	}	logid = loghandle->lgh_id;	llog_close(env, loghandle);	CWARN("2c: re-open the log by LOGID/n");	rc = llog_open(env, ctxt, &loghandle, &logid, NULL, LLOG_OPEN_EXISTS);	if (rc) {		CERROR("2c: re-open log by LOGID failed/n");		GOTO(out_close_llh, rc);	}	rc = llog_init_handle(env, loghandle, LLOG_F_IS_PLAIN, &uuid);	if (rc) {		CERROR("2c: can't init llog handle: %d/n", rc);		GOTO(out_close, rc);	}	CWARN("2b: destroy this log/n");	rc = llog_destroy(env, loghandle);	if (rc)		CERROR("2d: destroy log failed/n");out_close:	llog_close(env, loghandle);out_close_llh:	if (rc)		llog_close(env, *llh);out_put:	llog_ctxt_put(ctxt);	RETURN(rc);}

/* This is a callback from the llog_* functions. * Assumes caller has already pushed us into the kernel context. */static int llog_client_open(const struct lu_env *env,			    struct llog_handle *lgh, struct llog_logid *logid,			    char *name, enum llog_open_param open_param){	struct obd_import     *imp;	struct llogd_body     *body;	struct llog_ctxt      *ctxt = lgh->lgh_ctxt;	struct ptlrpc_request *req = NULL;	int		    rc;	LLOG_CLIENT_ENTRY(ctxt, imp);	/* client cannot create llog */	LASSERTF(open_param != LLOG_OPEN_NEW, "%#x/n", open_param);	LASSERT(lgh);	req = ptlrpc_request_alloc(imp, &RQF_LLOG_ORIGIN_HANDLE_CREATE);	if (req == NULL)		GOTO(out, rc = -ENOMEM);	if (name)		req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT,				     strlen(name) + 1);	rc = ptlrpc_request_pack(req, LUSTRE_LOG_VERSION,				 LLOG_ORIGIN_HANDLE_CREATE);	if (rc) {		ptlrpc_request_free(req);		req = NULL;		GOTO(out, rc);	}	ptlrpc_request_set_replen(req);	body = req_capsule_client_get(&req->rq_pill, &RMF_LLOGD_BODY);	if (logid)		body->lgd_logid = *logid;	body->lgd_ctxt_idx = ctxt->loc_idx - 1;	if (name) {		char *tmp;		tmp = req_capsule_client_sized_get(&req->rq_pill, &RMF_NAME,						   strlen(name) + 1);		LASSERT(tmp);		strcpy(tmp, name);	}	rc = ptlrpc_queue_wait(req);	if (rc)		GOTO(out, rc);	body = req_capsule_server_get(&req->rq_pill, &RMF_LLOGD_BODY);	if (body == NULL)		GOTO(out, rc = -EFAULT);	lgh->lgh_id = body->lgd_logid;	lgh->lgh_ctxt = ctxt;out:	LLOG_CLIENT_EXIT(ctxt, imp);	ptlrpc_req_finished(req);	return rc;}

int llog_ioctl(struct llog_ctxt *ctxt, int cmd, struct obd_ioctl_data *data){        struct llog_logid logid;        int err = 0;        struct llog_handle *handle = NULL;        ENTRY;        if (*data->ioc_inlbuf1 == '#') {                err = str2logid(&logid, data->ioc_inlbuf1, data->ioc_inllen1);                if (err)                        GOTO(out, err);                err = llog_create(ctxt, &handle, &logid, NULL);                if (err)                        GOTO(out, err);        } else if (*data->ioc_inlbuf1 == '$') {                char *name = data->ioc_inlbuf1 + 1;                err = llog_create(ctxt, &handle, NULL, name);                if (err)                        GOTO(out, err);        } else {                GOTO(out, err = -EINVAL);        }        err = llog_init_handle(handle, 0, NULL);        if (err)                GOTO(out_close, err = -ENOENT);        switch (cmd) {        case OBD_IOC_LLOG_INFO: {                int l;                int remains = data->ioc_inllen2 +                        cfs_size_round(data->ioc_inllen1);                char *out = data->ioc_bulk;                l = snprintf(out, remains,                             "logid:            #"LPX64"#"LPX64"#%08x/n"                             "flags:            %x (%s)/n"                             "records count:    %d/n"                             "last index:       %d/n",                             handle->lgh_id.lgl_oid, handle->lgh_id.lgl_oseq,                             handle->lgh_id.lgl_ogen,                             handle->lgh_hdr->llh_flags,                             handle->lgh_hdr->llh_flags &                             LLOG_F_IS_CAT ? "cat" : "plain",                             handle->lgh_hdr->llh_count,                             handle->lgh_last_idx);                out += l;                remains -= l;                if (remains <= 0)                        CERROR("not enough space for log header info/n");                GOTO(out_close, err);        }        case OBD_IOC_LLOG_CHECK: {                LASSERT(data->ioc_inllen1);                err = llog_process(handle, llog_check_cb, data, NULL);                if (err == -LLOG_EEMPTY)                        err = 0;                GOTO(out_close, err);        }        case OBD_IOC_LLOG_PRINT: {                LASSERT(data->ioc_inllen1);                err = llog_process(handle, class_config_dump_handler,data,NULL);                if (err == -LLOG_EEMPTY)                        err = 0;                else                        err = llog_process(handle, llog_print_cb, data, NULL);                GOTO(out_close, err);        }        case OBD_IOC_LLOG_CANCEL: {                struct llog_cookie cookie;                struct llog_logid plain;                char *endp;                cookie.lgc_index = simple_strtoul(data->ioc_inlbuf3, &endp, 0);                if (*endp != '/0')                        GOTO(out_close, err = -EINVAL);                if (handle->lgh_hdr->llh_flags & LLOG_F_IS_CAT) {                        cfs_down_write(&handle->lgh_lock);                        err = llog_cancel_rec(handle, cookie.lgc_index);                        cfs_up_write(&handle->lgh_lock);                        GOTO(out_close, err);                }                err = str2logid(&plain, data->ioc_inlbuf2, data->ioc_inllen2);                if (err)                        GOTO(out_close, err);                cookie.lgc_lgl = plain;                if (!(handle->lgh_hdr->llh_flags & LLOG_F_IS_CAT))                        GOTO(out_close, err = -EINVAL);                err = llog_cat_cancel_records(handle, 1, &cookie);                GOTO(out_close, err);        }        case OBD_IOC_LLOG_REMOVE: {                struct llog_logid plain;//.........这里部分代码省略.........

/* * Helper routine to retrieve slave information. * This function converts a quotactl request into quota/accounting object * operations. It is independant of the slave stack which is only accessible * from the OSD layer. * * /param env   - is the environment passed by the caller * /param dev   - is the dt_device this quotactl is executed on * /param oqctl - is the quotactl request */int lquotactl_slv(const struct lu_env *env, struct dt_device *dev,		  struct obd_quotactl *oqctl){	struct lquota_thread_info	*qti = lquota_info(env);	__u64				 key;	struct dt_object		*obj;	struct obd_dqblk		*dqblk = &oqctl->qc_dqblk;	int				 rc;	ENTRY;	if (oqctl->qc_cmd != Q_GETOQUOTA) {		/* as in many other places, dev->dd_lu_dev.ld_obd->obd_name		 * point to an invalid obd_name, to be fixed in LU-1574 */		CERROR("%s: Unsupported quotactl command: %x/n",		       dev->dd_lu_dev.ld_obd->obd_name, oqctl->qc_cmd);		RETURN(-EOPNOTSUPP);	}	if (oqctl->qc_type != USRQUOTA && oqctl->qc_type != GRPQUOTA)		/* no support for directory quota yet */		RETURN(-EOPNOTSUPP);	/* qc_id is a 32-bit field while a key has 64 bits */	key = oqctl->qc_id;	/* Step 1: collect accounting information */	obj = acct_obj_lookup(env, dev, oqctl->qc_type);	if (IS_ERR(obj))		RETURN(-EOPNOTSUPP);	if (obj->do_index_ops == NULL)		GOTO(out, rc = -EINVAL);	/* lookup record storing space accounting information for this ID */	rc = dt_lookup(env, obj, (struct dt_rec *)&qti->qti_acct_rec,		       (struct dt_key *)&key, BYPASS_CAPA);	if (rc < 0)		GOTO(out, rc);	memset(&oqctl->qc_dqblk, 0, sizeof(struct obd_dqblk));	dqblk->dqb_curspace	= qti->qti_acct_rec.bspace;	dqblk->dqb_curinodes	= qti->qti_acct_rec.ispace;	dqblk->dqb_valid	= QIF_USAGE;	lu_object_put(env, &obj->do_lu);	/* Step 2: collect enforcement information */	obj = quota_obj_lookup(env, dev, oqctl->qc_type);	if (IS_ERR(obj))		RETURN(0);	if (obj->do_index_ops == NULL)		GOTO(out, rc = 0);	memset(&qti->qti_slv_rec, 0, sizeof(qti->qti_slv_rec));	/* lookup record storing enforcement information for this ID */	rc = dt_lookup(env, obj, (struct dt_rec *)&qti->qti_slv_rec,		       (struct dt_key *)&key, BYPASS_CAPA);	if (rc < 0 && rc != -ENOENT)		GOTO(out, rc = 0);	if (lu_device_is_md(dev->dd_lu_dev.ld_site->ls_top_dev)) {		dqblk->dqb_ihardlimit = qti->qti_slv_rec.qsr_granted;		dqblk->dqb_bhardlimit = 0;	} else {		dqblk->dqb_ihardlimit = 0;		dqblk->dqb_bhardlimit = qti->qti_slv_rec.qsr_granted;	}	dqblk->dqb_valid |= QIF_LIMITS;	GOTO(out, rc = 0);out:	lu_object_put(env, &obj->do_lu);        return rc;}

/* * Concurrency: @dt is write locked. */static int osd_object_create(const struct lu_env *env, struct dt_object *dt,			     struct lu_attr *attr,			     struct dt_allocation_hint *hint,			     struct dt_object_format *dof,			     struct thandle *th){	struct osd_thread_info	*info = osd_oti_get(env);	struct lustre_mdt_attrs	*lma = &info->oti_mdt_attrs;	struct zpl_direntry	*zde = &info->oti_zde.lzd_reg;	const struct lu_fid	*fid = lu_object_fid(&dt->do_lu);	struct osd_object	*obj = osd_dt_obj(dt);	struct osd_device	*osd = osd_obj2dev(obj);	char			*buf = info->oti_str;	struct osd_thandle	*oh;	dmu_buf_t		*db = NULL;	uint64_t		 zapid, parent = 0;	int			 rc;	ENTRY;	/* concurrent create declarations should not see	 * the object inconsistent (db, attr, etc).	 * in regular cases acquisition should be cheap */	down_write(&obj->oo_guard);	if (unlikely(dt_object_exists(dt)))		GOTO(out, rc = -EEXIST);	LASSERT(osd_invariant(obj));	LASSERT(dof != NULL);	LASSERT(th != NULL);	oh = container_of0(th, struct osd_thandle, ot_super);	/*	 * XXX missing: Quote handling.	 */	LASSERT(obj->oo_db == NULL);	/* to follow ZFS on-disk format we need	 * to initialize parent dnode properly */	if (hint != NULL && hint->dah_parent != NULL &&	    !dt_object_remote(hint->dah_parent))		parent = osd_dt_obj(hint->dah_parent)->oo_db->db_object;	/* we may fix some attributes, better do not change the source */	obj->oo_attr = *attr;	obj->oo_attr.la_valid |= LA_SIZE | LA_NLINK | LA_TYPE;	db = osd_create_type_f(dof->dof_type)(env, obj, &obj->oo_attr, oh);	if (IS_ERR(db)) {		rc = PTR_ERR(db);		db = NULL;		GOTO(out, rc);	}	zde->zde_pad = 0;	zde->zde_dnode = db->db_object;	zde->zde_type = IFTODT(attr->la_mode & S_IFMT);	zapid = osd_get_name_n_idx(env, osd, fid, buf, sizeof(info->oti_str));	rc = -zap_add(osd->od_os, zapid, buf, 8, 1, zde, oh->ot_tx);	if (rc)		GOTO(out, rc);	/* Now add in all of the "SA" attributes */	rc = -sa_handle_get(osd->od_os, db->db_object, NULL,			    SA_HDL_PRIVATE, &obj->oo_sa_hdl);	if (rc)		GOTO(out, rc);	/* configure new osd object */	obj->oo_db = db;	parent = parent != 0 ? parent : zapid;	rc = __osd_attr_init(env, osd, obj->oo_sa_hdl, oh->ot_tx,			     &obj->oo_attr, parent);	if (rc)		GOTO(out, rc);	/* XXX: oo_lma_flags */	obj->oo_dt.do_lu.lo_header->loh_attr |= obj->oo_attr.la_mode & S_IFMT;	smp_mb();	obj->oo_dt.do_lu.lo_header->loh_attr |= LOHA_EXISTS;	if (likely(!fid_is_acct(lu_object_fid(&obj->oo_dt.do_lu))))		/* no body operations for accounting objects */		obj->oo_dt.do_body_ops = &osd_body_ops;	rc = -nvlist_alloc(&obj->oo_sa_xattr, NV_UNIQUE_NAME, KM_SLEEP);	if (rc)		GOTO(out, rc);	/* initialize LMA */	lustre_lma_init(lma, lu_object_fid(&obj->oo_dt.do_lu), 0, 0);	lustre_lma_swab(lma);	rc = -nvlist_add_byte_array(obj->oo_sa_xattr, XATTR_NAME_LMA,//.........这里部分代码省略.........

/* Test client api; open log by name and process */static int llog_test_6(const struct lu_env *env, struct obd_device *obd,		       char *name){	struct obd_device	*mgc_obd;	struct llog_ctxt	*ctxt;	struct obd_uuid		*mgs_uuid;	struct obd_export	*exp;	struct obd_uuid		 uuid = { "LLOG_TEST6_UUID" };	struct llog_handle	*llh = NULL;	struct llog_ctxt	*nctxt;	int			 rc, rc2;	ctxt = llog_get_context(obd, LLOG_TEST_ORIG_CTXT);	LASSERT(ctxt);	mgs_uuid = &ctxt->loc_exp->exp_obd->obd_uuid;	CWARN("6a: re-open log %s using client API/n", name);	mgc_obd = class_find_client_obd(mgs_uuid, LUSTRE_MGC_NAME, NULL);	if (mgc_obd == NULL) {		CERROR("6a: no MGC devices connected to %s found./n",		       mgs_uuid->uuid);		GOTO(ctxt_release, rc = -ENOENT);	}	rc = obd_connect(NULL, &exp, mgc_obd, &uuid,			 NULL /* obd_connect_data */, NULL);	if (rc != -EALREADY) {		CERROR("6a: connect on connected MGC (%s) failed to return"		       " -EALREADY", mgc_obd->obd_name);		if (rc == 0)			obd_disconnect(exp);		GOTO(ctxt_release, rc = -EINVAL);	}	nctxt = llog_get_context(mgc_obd, LLOG_CONFIG_REPL_CTXT);	rc = llog_open(env, nctxt, &llh, NULL, name, LLOG_OPEN_EXISTS);	if (rc) {		CERROR("6a: llog_open failed %d/n", rc);		GOTO(nctxt_put, rc);	}	rc = llog_init_handle(env, llh, LLOG_F_IS_PLAIN, NULL);	if (rc) {		CERROR("6a: llog_init_handle failed %d/n", rc);		GOTO(parse_out, rc);	}	plain_counter = 1; /* llog header is first record */	CWARN("6b: process log %s using client API/n", name);	rc = llog_process(env, llh, plain_print_cb, NULL, NULL);	if (rc)		CERROR("6b: llog_process failed %d/n", rc);	CWARN("6b: processed %d records/n", plain_counter);	rc = verify_handle("6b", llh, plain_counter);	if (rc)		GOTO(parse_out, rc);	plain_counter = 1; /* llog header is first record */	CWARN("6c: process log %s reversely using client API/n", name);	rc = llog_reverse_process(env, llh, plain_print_cb, NULL, NULL);	if (rc)		CERROR("6c: llog_reverse_process failed %d/n", rc);	CWARN("6c: processed %d records/n", plain_counter);	rc = verify_handle("6c", llh, plain_counter);	if (rc)		GOTO(parse_out, rc);parse_out:	rc2 = llog_close(env, llh);	if (rc2) {		CERROR("6: llog_close failed: rc = %d/n", rc2);		if (rc == 0)			rc = rc2;	}nctxt_put:	llog_ctxt_put(nctxt);ctxt_release:	llog_ctxt_put(ctxt);	RETURN(rc);}

/* * Retrieve the attributes of a DMU object */int __osd_object_attr_get(const struct lu_env *env, struct osd_device *o,			  struct osd_object *obj, struct lu_attr *la){	struct osa_attr	*osa = &osd_oti_get(env)->oti_osa;	sa_bulk_attr_t	*bulk = osd_oti_get(env)->oti_attr_bulk;	sa_handle_t	*sa_hdl;	int		 cnt = 0;	int		 rc;	ENTRY;	LASSERT(obj->oo_db != NULL);	rc = -sa_handle_get(o->od_os, obj->oo_db->db_object, NULL,			    SA_HDL_PRIVATE, &sa_hdl);	if (rc)		RETURN(rc);	la->la_valid |= LA_ATIME | LA_MTIME | LA_CTIME | LA_MODE | LA_TYPE |			LA_SIZE | LA_UID | LA_GID | LA_FLAGS | LA_NLINK;	SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_ATIME(o), NULL, osa->atime, 16);	SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_MTIME(o), NULL, osa->mtime, 16);	SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_CTIME(o), NULL, osa->ctime, 16);	SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_MODE(o), NULL, &osa->mode, 8);	SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_SIZE(o), NULL, &osa->size, 8);	SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_LINKS(o), NULL, &osa->nlink, 8);	SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_UID(o), NULL, &osa->uid, 8);	SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_GID(o), NULL, &osa->gid, 8);	SA_ADD_BULK_ATTR(bulk, cnt, SA_ZPL_FLAGS(o), NULL, &osa->flags, 8);	LASSERT(cnt <= ARRAY_SIZE(osd_oti_get(env)->oti_attr_bulk));	rc = -sa_bulk_lookup(sa_hdl, bulk, cnt);	if (rc)		GOTO(out_sa, rc);	la->la_atime = osa->atime[0];	la->la_mtime = osa->mtime[0];	la->la_ctime = osa->ctime[0];	la->la_mode = osa->mode;	la->la_uid = osa->uid;	la->la_gid = osa->gid;	la->la_nlink = osa->nlink;	la->la_flags = attrs_zfs2fs(osa->flags);	la->la_size = osa->size;	/* Try to get extra flag from LMA. Right now, only LMAI_ORPHAN	 * flags is stored in LMA, and it is only for orphan directory */	if (S_ISDIR(la->la_mode) && dt_object_exists(&obj->oo_dt)) {		struct osd_thread_info *info = osd_oti_get(env);		struct lustre_mdt_attrs *lma;		struct lu_buf buf;		lma = (struct lustre_mdt_attrs *)info->oti_buf;		buf.lb_buf = lma;		buf.lb_len = sizeof(info->oti_buf);		rc = osd_xattr_get(env, &obj->oo_dt, &buf, XATTR_NAME_LMA);		if (rc > 0) {			rc = 0;			lma->lma_incompat = le32_to_cpu(lma->lma_incompat);			obj->oo_lma_flags =				lma_to_lustre_flags(lma->lma_incompat);		} else if (rc == -ENODATA) {			rc = 0;		}	}	if (S_ISCHR(la->la_mode) || S_ISBLK(la->la_mode)) {		rc = -sa_lookup(sa_hdl, SA_ZPL_RDEV(o), &osa->rdev, 8);		if (rc)			GOTO(out_sa, rc);		la->la_rdev = osa->rdev;		la->la_valid |= LA_RDEV;	}out_sa:	sa_handle_destroy(sa_hdl);	RETURN(rc);}

static int llog_test_7_sub(const struct lu_env *env, struct llog_ctxt *ctxt){	struct llog_handle	*llh;	int			 rc = 0, i, process_count;	int			 num_recs = 0;	ENTRY;	rc = llog_open_create(env, ctxt, &llh, NULL, NULL);	if (rc) {		CERROR("7_sub: create log failed/n");		RETURN(rc);	}	rc = llog_init_handle(env, llh,			      LLOG_F_IS_PLAIN | LLOG_F_ZAP_WHEN_EMPTY,			      &uuid);	if (rc) {		CERROR("7_sub: can't init llog handle: %d/n", rc);		GOTO(out_close, rc);	}	for (i = 0; i < LLOG_BITMAP_SIZE(llh->lgh_hdr); i++) {		rc = llog_write(env, llh, &llog_records.lrh, NULL, 0,				NULL, -1);		if (rc == -ENOSPC) {			break;		} else if (rc < 0) {			CERROR("7_sub: write recs failed at #%d: %d/n",			       i + 1, rc);			GOTO(out_close, rc);		}		num_recs++;	}	if (rc != -ENOSPC) {		CWARN("7_sub: write record more than BITMAP size!/n");		GOTO(out_close, rc = -EINVAL);	}	rc = verify_handle("7_sub", llh, num_recs + 1);	if (rc) {		CERROR("7_sub: verify handle failed: %d/n", rc);		GOTO(out_close, rc);	}	if (num_recs < LLOG_BITMAP_SIZE(llh->lgh_hdr) - 1)		CWARN("7_sub: records are not aligned, written %d from %u/n",		      num_recs, LLOG_BITMAP_SIZE(llh->lgh_hdr) - 1);	plain_counter = 0;	rc = llog_process(env, llh, test_7_print_cb, "test 7", NULL);	if (rc) {		CERROR("7_sub: llog process failed: %d/n", rc);		GOTO(out_close, rc);	}	process_count = plain_counter;	if (process_count != num_recs) {		CERROR("7_sub: processed %d records from %d total/n",		       process_count, num_recs);		GOTO(out_close, rc = -EINVAL);	}	plain_counter = 0;	rc = llog_reverse_process(env, llh, test_7_cancel_cb, "test 7", NULL);	if (rc) {		CERROR("7_sub: reverse llog process failed: %d/n", rc);		GOTO(out_close, rc);	}	if (process_count != plain_counter) {		CERROR("7_sub: Reverse/direct processing found different"		       "number of records: %d/%d/n",		       plain_counter, process_count);		GOTO(out_close, rc = -EINVAL);	}	if (llog_exist(llh)) {		CERROR("7_sub: llog exists but should be zapped/n");		GOTO(out_close, rc = -EEXIST);	}	rc = verify_handle("7_sub", llh, 1);out_close:	if (rc)		llog_destroy(env, llh);	llog_close(env, llh);	RETURN(rc);}

static int osd_declare_attr_set(const struct lu_env *env,				struct dt_object *dt,				const struct lu_attr *attr,				struct thandle *handle){	struct osd_thread_info	*info = osd_oti_get(env);	struct osd_object	*obj = osd_dt_obj(dt);	struct osd_device	*osd = osd_obj2dev(obj);	dmu_tx_hold_t		*txh;	struct osd_thandle	*oh;	uint64_t		 bspace;	uint32_t		 blksize;	int			 rc = 0;	bool			 found;	ENTRY;	LASSERT(handle != NULL);	LASSERT(osd_invariant(obj));	oh = container_of0(handle, struct osd_thandle, ot_super);	down_read(&obj->oo_guard);	if (unlikely(!dt_object_exists(dt) || obj->oo_destroyed))		GOTO(out, rc = 0);	LASSERT(obj->oo_sa_hdl != NULL);	LASSERT(oh->ot_tx != NULL);	/* regular attributes are part of the bonus buffer */	/* let's check whether this object is already part of	 * transaction.. */	found = false;	for (txh = list_head(&oh->ot_tx->tx_holds); txh;	     txh = list_next(&oh->ot_tx->tx_holds, txh)) {		if (txh->txh_dnode == NULL)			continue;		if (txh->txh_dnode->dn_object != obj->oo_db->db_object)			continue;		/* this object is part of the transaction already		 * we don't need to declare bonus again */		found = true;		break;	}	if (!found)		dmu_tx_hold_bonus(oh->ot_tx, obj->oo_db->db_object);	if (oh->ot_tx->tx_err != 0)		GOTO(out, rc = -oh->ot_tx->tx_err);	if (attr && attr->la_valid & LA_FLAGS) {		/* LMA is usually a part of bonus, no need to declare		 * anything else */	}	if (attr && (attr->la_valid & (LA_UID | LA_GID))) {		sa_object_size(obj->oo_sa_hdl, &blksize, &bspace);		bspace = toqb(bspace * blksize);	}	if (attr && attr->la_valid & LA_UID) {		/* account for user inode tracking ZAP update */		dmu_tx_hold_zap(oh->ot_tx, osd->od_iusr_oid, FALSE, NULL);		/* quota enforcement for user */		if (attr->la_uid != obj->oo_attr.la_uid) {			rc = qsd_transfer(env, osd->od_quota_slave,					  &oh->ot_quota_trans, USRQUOTA,					  obj->oo_attr.la_uid, attr->la_uid,					  bspace, &info->oti_qi);			if (rc)				GOTO(out, rc);		}	}	if (attr && attr->la_valid & LA_GID) {		/* account for user inode tracking ZAP update */		dmu_tx_hold_zap(oh->ot_tx, osd->od_igrp_oid, FALSE, NULL);		/* quota enforcement for group */		if (attr->la_gid != obj->oo_attr.la_gid) {			rc = qsd_transfer(env, osd->od_quota_slave,					  &oh->ot_quota_trans, GRPQUOTA,					  obj->oo_attr.la_gid, attr->la_gid,					  bspace, &info->oti_qi);			if (rc)				GOTO(out, rc);		}	}out:	up_read(&obj->oo_guard);	RETURN(rc);}

/* Test all llog records writing and processing */static int llog_test_7(const struct lu_env *env, struct obd_device *obd){	struct llog_ctxt	*ctxt;	int			 rc;	ENTRY;	ctxt = llog_get_context(obd, LLOG_TEST_ORIG_CTXT);	CWARN("7a: test llog_logid_rec/n");	llog_records.llr.lid_hdr.lrh_len = sizeof(llog_records.llr);	llog_records.llr.lid_tail.lrt_len = sizeof(llog_records.llr);	llog_records.llr.lid_hdr.lrh_type = LLOG_LOGID_MAGIC;	rc = llog_test_7_sub(env, ctxt);	if (rc) {		CERROR("7a: llog_logid_rec test failed/n");		GOTO(out, rc);	}	CWARN("7b: test llog_unlink64_rec/n");	llog_records.lur.lur_hdr.lrh_len = sizeof(llog_records.lur);	llog_records.lur.lur_tail.lrt_len = sizeof(llog_records.lur);	llog_records.lur.lur_hdr.lrh_type = MDS_UNLINK64_REC;	rc = llog_test_7_sub(env, ctxt);	if (rc) {		CERROR("7b: llog_unlink_rec test failed/n");		GOTO(out, rc);	}	CWARN("7c: test llog_setattr64_rec/n");	llog_records.lsr64.lsr_hdr.lrh_len = sizeof(llog_records.lsr64);	llog_records.lsr64.lsr_tail.lrt_len = sizeof(llog_records.lsr64);	llog_records.lsr64.lsr_hdr.lrh_type = MDS_SETATTR64_REC;	rc = llog_test_7_sub(env, ctxt);	if (rc) {		CERROR("7c: llog_setattr64_rec test failed/n");		GOTO(out, rc);	}	CWARN("7d: test llog_size_change_rec/n");	llog_records.lscr.lsc_hdr.lrh_len = sizeof(llog_records.lscr);	llog_records.lscr.lsc_tail.lrt_len = sizeof(llog_records.lscr);	llog_records.lscr.lsc_hdr.lrh_type = OST_SZ_REC;	rc = llog_test_7_sub(env, ctxt);	if (rc) {		CERROR("7d: llog_size_change_rec test failed/n");		GOTO(out, rc);	}	CWARN("7e: test llog_changelog_rec/n");	llog_records.lcr.cr_hdr.lrh_len = sizeof(llog_records.lcr);	llog_records.lcr.cr_tail.lrt_len = sizeof(llog_records.lcr);	llog_records.lcr.cr_hdr.lrh_type = CHANGELOG_REC;	rc = llog_test_7_sub(env, ctxt);	if (rc) {		CERROR("7e: llog_changelog_rec test failed/n");		GOTO(out, rc);	}	CWARN("7f: test llog_changelog_user_rec/n");	llog_records.lcur.cur_hdr.lrh_len = sizeof(llog_records.lcur);	llog_records.lcur.cur_tail.lrt_len = sizeof(llog_records.lcur);	llog_records.lcur.cur_hdr.lrh_type = CHANGELOG_USER_REC;	rc = llog_test_7_sub(env, ctxt);	if (rc) {		CERROR("7f: llog_changelog_user_rec test failed/n");		GOTO(out, rc);	}	CWARN("7g: test llog_gen_rec/n");	llog_records.lgr.lgr_hdr.lrh_len = sizeof(llog_records.lgr);	llog_records.lgr.lgr_tail.lrt_len = sizeof(llog_records.lgr);	llog_records.lgr.lgr_hdr.lrh_type = LLOG_GEN_REC;	rc = llog_test_7_sub(env, ctxt);	if (rc) {		CERROR("7g: llog_size_change_rec test failed/n");		GOTO(out, rc);	}out:	llog_ctxt_put(ctxt);	RETURN(rc);}

/* backup plain llog */int llog_backup(const struct lu_env *env, struct obd_device *obd,		struct llog_ctxt *ctxt, struct llog_ctxt *bctxt,		char *name, char *backup){	struct llog_handle	*llh, *bllh;	int			 rc;	ENTRY;	/* open original log */	rc = llog_open(env, ctxt, &llh, NULL, name, LLOG_OPEN_EXISTS);	if (rc < 0) {		/* the -ENOENT case is also reported to the caller		 * but silently so it should handle that if needed.		 */		if (rc != -ENOENT)			CERROR("%s: failed to open log %s: rc = %d/n",			       obd->obd_name, name, rc);		RETURN(rc);	}	rc = llog_init_handle(env, llh, LLOG_F_IS_PLAIN, NULL);	if (rc)		GOTO(out_close, rc);	/* Make sure there's no old backup log */	rc = llog_erase(env, bctxt, NULL, backup);	if (rc < 0 && rc != -ENOENT)		GOTO(out_close, rc);	/* open backup log */	rc = llog_open_create(env, bctxt, &bllh, NULL, backup);	if (rc) {		CERROR("%s: failed to open backup logfile %s: rc = %d/n",		       obd->obd_name, backup, rc);		GOTO(out_close, rc);	}	/* check that backup llog is not the same object as original one */	if (llh->lgh_obj == bllh->lgh_obj) {		CERROR("%s: backup llog %s to itself (%s), objects %p/%p/n",		       obd->obd_name, name, backup, llh->lgh_obj,		       bllh->lgh_obj);		GOTO(out_backup, rc = -EEXIST);	}	rc = llog_init_handle(env, bllh, LLOG_F_IS_PLAIN, NULL);	if (rc)		GOTO(out_backup, rc);	/* Copy log record by record */	rc = llog_process_or_fork(env, llh, llog_copy_handler, (void *)bllh,				  NULL, false);	if (rc)		CERROR("%s: failed to backup log %s: rc = %d/n",		       obd->obd_name, name, rc);out_backup:	llog_close(env, bllh);out_close:	llog_close(env, llh);	RETURN(rc);}

static int llog_test_setup(struct obd_device *obd, struct lustre_cfg *lcfg){	struct obd_device	*tgt;	struct llog_ctxt	*ctxt;	struct dt_object	*o;	struct lu_env		 env;	struct lu_context	 test_session;	int			 rc;        ENTRY;        if (lcfg->lcfg_bufcount < 2) {                CERROR("requires a TARGET OBD name/n");                RETURN(-EINVAL);        }        if (lcfg->lcfg_buflens[1] < 1) {                CERROR("requires a TARGET OBD name/n");                RETURN(-EINVAL);        }        /* disk obd */        tgt = class_name2obd(lustre_cfg_string(lcfg, 1));        if (!tgt || !tgt->obd_attached || !tgt->obd_set_up) {                CERROR("target device not attached or not set up (%s)/n",                       lustre_cfg_string(lcfg, 1));                RETURN(-EINVAL);        }	rc = lu_env_init(&env, LCT_LOCAL | LCT_MG_THREAD);	if (rc)		RETURN(rc);	rc = lu_context_init(&test_session, LCT_SESSION);	if (rc)		GOTO(cleanup_env, rc);	test_session.lc_thread = (struct ptlrpc_thread *)cfs_current();	lu_context_enter(&test_session);	env.le_ses = &test_session;	CWARN("Setup llog-test device over %s device/n",	      lustre_cfg_string(lcfg, 1));	OBD_SET_CTXT_MAGIC(&obd->obd_lvfs_ctxt);	obd->obd_lvfs_ctxt.dt = lu2dt_dev(tgt->obd_lu_dev);	rc = llog_setup(&env, tgt, &tgt->obd_olg, LLOG_TEST_ORIG_CTXT, tgt,			&llog_osd_ops);	if (rc)		GOTO(cleanup_session, rc);	/* use MGS llog dir for tests */	ctxt = llog_get_context(tgt, LLOG_CONFIG_ORIG_CTXT);	LASSERT(ctxt);	o = ctxt->loc_dir;	llog_ctxt_put(ctxt);	ctxt = llog_get_context(tgt, LLOG_TEST_ORIG_CTXT);	LASSERT(ctxt);	ctxt->loc_dir = o;	llog_ctxt_put(ctxt);	llog_test_rand = cfs_rand();	rc = llog_run_tests(&env, tgt);	if (rc)		llog_test_cleanup(obd);cleanup_session:	lu_context_exit(&test_session);	lu_context_fini(&test_session);cleanup_env:	lu_env_fini(&env);	RETURN(rc);}

int llog_init_handle(const struct lu_env *env, struct llog_handle *handle,		     int flags, struct obd_uuid *uuid){	struct llog_log_hdr	*llh;	enum llog_flag		 fmt = flags & LLOG_F_EXT_MASK;	int			 rc;	int			chunk_size = handle->lgh_ctxt->loc_chunk_size;	ENTRY;	LASSERT(handle->lgh_hdr == NULL);	LASSERT(chunk_size >= LLOG_MIN_CHUNK_SIZE);	OBD_ALLOC_LARGE(llh, chunk_size);	if (llh == NULL)		RETURN(-ENOMEM);	handle->lgh_hdr = llh;	handle->lgh_hdr_size = chunk_size;	/* first assign flags to use llog_client_ops */	llh->llh_flags = flags;	rc = llog_read_header(env, handle, uuid);	if (rc == 0) {		if (unlikely((llh->llh_flags & LLOG_F_IS_PLAIN &&			      flags & LLOG_F_IS_CAT) ||			     (llh->llh_flags & LLOG_F_IS_CAT &&			      flags & LLOG_F_IS_PLAIN))) {			CERROR("%s: llog type is %s but initializing %s/n",			       handle->lgh_ctxt->loc_obd->obd_name,			       llh->llh_flags & LLOG_F_IS_CAT ?			       "catalog" : "plain",			       flags & LLOG_F_IS_CAT ? "catalog" : "plain");			GOTO(out, rc = -EINVAL);		} else if (llh->llh_flags &			   (LLOG_F_IS_PLAIN | LLOG_F_IS_CAT)) {			/*			 * it is possible to open llog without specifying llog			 * type so it is taken from llh_flags			 */			flags = llh->llh_flags;		} else {			/* for some reason the llh_flags has no type set */			CERROR("llog type is not specified!/n");			GOTO(out, rc = -EINVAL);		}		if (unlikely(uuid &&			     !obd_uuid_equals(uuid, &llh->llh_tgtuuid))) {			CERROR("%s: llog uuid mismatch: %s/%s/n",			       handle->lgh_ctxt->loc_obd->obd_name,			       (char *)uuid->uuid,			       (char *)llh->llh_tgtuuid.uuid);			GOTO(out, rc = -EEXIST);		}	}	if (flags & LLOG_F_IS_CAT) {		LASSERT(list_empty(&handle->u.chd.chd_head));		INIT_LIST_HEAD(&handle->u.chd.chd_head);		llh->llh_size = sizeof(struct llog_logid_rec);		llh->llh_flags |= LLOG_F_IS_FIXSIZE;	} else if (!(flags & LLOG_F_IS_PLAIN)) {		CERROR("%s: unknown flags: %#x (expected %#x or %#x)/n",		       handle->lgh_ctxt->loc_obd->obd_name,		       flags, LLOG_F_IS_CAT, LLOG_F_IS_PLAIN);		rc = -EINVAL;	}	llh->llh_flags |= fmt;out:	if (rc) {		OBD_FREE_LARGE(llh, chunk_size);		handle->lgh_hdr = NULL;	}	RETURN(rc);}

static int __init lustre_init(void){	struct proc_dir_entry *entry;	struct lnet_process_id lnet_id;	struct timespec64 ts;	int i, rc, seed[2];	CLASSERT(sizeof(LUSTRE_VOLATILE_HDR) == LUSTRE_VOLATILE_HDR_LEN + 1);	/* print an address of _any_ initialized kernel symbol from this	 * module, to allow debugging with gdb that doesn't support data	 * symbols from modules.*/	CDEBUG(D_INFO, "Lustre client module (%p)./n",	       &lustre_super_operations);	ll_inode_cachep = kmem_cache_create("lustre_inode_cache",					    sizeof(struct ll_inode_info),					    0, SLAB_HWCACHE_ALIGN, NULL);	if (ll_inode_cachep == NULL)		GOTO(out_cache, rc = -ENOMEM);	ll_file_data_slab = kmem_cache_create("ll_file_data",						 sizeof(struct ll_file_data), 0,						 SLAB_HWCACHE_ALIGN, NULL);	if (ll_file_data_slab == NULL)		GOTO(out_cache, rc = -ENOMEM);	entry = lprocfs_register("llite", proc_lustre_root, NULL, NULL);	if (IS_ERR(entry)) {		rc = PTR_ERR(entry);		CERROR("cannot register '/proc/fs/lustre/llite': rc = %d/n",		       rc);		GOTO(out_cache, rc);	}	proc_lustre_fs_root = entry;	cfs_get_random_bytes(seed, sizeof(seed));	/* Nodes with small feet have little entropy. The NID for this	 * node gives the most entropy in the low bits. */	for (i = 0;; i++) {		if (LNetGetId(i, &lnet_id) == -ENOENT)			break;		if (LNET_NETTYP(LNET_NIDNET(lnet_id.nid)) != LOLND)			seed[0] ^= LNET_NIDADDR(lnet_id.nid);	}	ktime_get_ts64(&ts);	cfs_srand(ts.tv_sec ^ seed[0], ts.tv_nsec ^ seed[1]);	rc = vvp_global_init();	if (rc != 0)		GOTO(out_proc, rc);	cl_inode_fini_env = cl_env_alloc(&cl_inode_fini_refcheck,					 LCT_REMEMBER | LCT_NOREF);	if (IS_ERR(cl_inode_fini_env))		GOTO(out_vvp, rc = PTR_ERR(cl_inode_fini_env));	cl_inode_fini_env->le_ctx.lc_cookie = 0x4;	rc = ll_xattr_init();	if (rc != 0)		GOTO(out_inode_fini_env, rc);	lustre_register_client_fill_super(ll_fill_super);	lustre_register_kill_super_cb(ll_kill_super);	lustre_register_client_process_config(ll_process_config);	RETURN(0);out_inode_fini_env:	cl_env_put(cl_inode_fini_env, &cl_inode_fini_refcheck);out_vvp:	vvp_global_fini();out_proc:	lprocfs_remove(&proc_lustre_fs_root);out_cache:	if (ll_inode_cachep != NULL)		kmem_cache_destroy(ll_inode_cachep);	if (ll_file_data_slab != NULL)		kmem_cache_destroy(ll_file_data_slab);	return rc;}

int llog_reverse_process(const struct lu_env *env,			 struct llog_handle *loghandle, llog_cb_t cb,			 void *data, void *catdata){        struct llog_log_hdr *llh = loghandle->lgh_hdr;        struct llog_process_cat_data *cd = catdata;        void *buf;        int rc = 0, first_index = 1, index, idx;	__u32	chunk_size = llh->llh_hdr.lrh_len;        ENTRY;	OBD_ALLOC_LARGE(buf, chunk_size);	if (buf == NULL)		RETURN(-ENOMEM);	if (cd != NULL)		first_index = cd->lpcd_first_idx + 1;	if (cd != NULL && cd->lpcd_last_idx)		index = cd->lpcd_last_idx;	else		index = LLOG_HDR_BITMAP_SIZE(llh) - 1;	while (rc == 0) {		struct llog_rec_hdr *rec;		struct llog_rec_tail *tail;		/* skip records not set in bitmap */		while (index >= first_index &&		       !ext2_test_bit(index, LLOG_HDR_BITMAP(llh)))			--index;		LASSERT(index >= first_index - 1);		if (index == first_index - 1)			break;		/* get the buf with our target record; avoid old garbage */		memset(buf, 0, chunk_size);		rc = llog_prev_block(env, loghandle, index, buf, chunk_size);		if (rc)			GOTO(out, rc);		rec = buf;		idx = rec->lrh_index;		CDEBUG(D_RPCTRACE, "index %u : idx %u/n", index, idx);                while (idx < index) {			rec = (void *)rec + rec->lrh_len;			if (LLOG_REC_HDR_NEEDS_SWABBING(rec))				lustre_swab_llog_rec(rec);                        idx ++;                }		LASSERT(idx == index);		tail = (void *)rec + rec->lrh_len - sizeof(*tail);		/* process records in buffer, starting where we found one */		while ((void *)tail > buf) {			if (tail->lrt_index == 0)				GOTO(out, rc = 0); /* no more records */			/* if set, process the callback on this record */			if (ext2_test_bit(index, LLOG_HDR_BITMAP(llh))) {				rec = (void *)tail - tail->lrt_len +				      sizeof(*tail);				rc = cb(env, loghandle, rec, data);				if (rc == LLOG_PROC_BREAK) {					GOTO(out, rc);				} else if (rc == LLOG_DEL_RECORD) {					rc = llog_cancel_rec(env, loghandle,							     tail->lrt_index);				}                                if (rc)                                        GOTO(out, rc);                        }                        /* previous record, still in buffer? */                        --index;                        if (index < first_index)                                GOTO(out, rc = 0);			tail = (void *)tail - tail->lrt_len;                }        }out:	if (buf != NULL)		OBD_FREE_LARGE(buf, chunk_size);        RETURN(rc);}

/** * Process a granting attempt for plain lock. * Must be called with ns lock held. * * This function looks for any conflicts for /a lock in the granted or * waiting queues. The lock is granted if no conflicts are found in * either queue. * * If /a first_enq is 0 (ie, called from ldlm_reprocess_queue): *   - blocking ASTs have already been sent * * If /a first_enq is 1 (ie, called from ldlm_lock_enqueue): *   - blocking ASTs have not been sent yet, so list of conflicting locks *     would be collected and ASTs sent. */int ldlm_process_plain_lock(struct ldlm_lock *lock, __u64 *flags,			    int first_enq, enum ldlm_error *err,			    struct list_head *work_list){	struct ldlm_resource *res = lock->l_resource;	struct list_head rpc_list;	int rc;	ENTRY;	LASSERT(lock->l_granted_mode != lock->l_req_mode);	check_res_locked(res);	LASSERT(list_empty(&res->lr_converting));	INIT_LIST_HEAD(&rpc_list);        if (!first_enq) {                LASSERT(work_list != NULL);                rc = ldlm_plain_compat_queue(&res->lr_granted, lock, NULL);                if (!rc)                        RETURN(LDLM_ITER_STOP);                rc = ldlm_plain_compat_queue(&res->lr_waiting, lock, NULL);                if (!rc)                        RETURN(LDLM_ITER_STOP);                ldlm_resource_unlink_lock(lock);                ldlm_grant_lock(lock, work_list);                RETURN(LDLM_ITER_CONTINUE);        } restart:        rc = ldlm_plain_compat_queue(&res->lr_granted, lock, &rpc_list);        rc += ldlm_plain_compat_queue(&res->lr_waiting, lock, &rpc_list);        if (rc != 2) {                /* If either of the compat_queue()s returned 0, then we                 * have ASTs to send and must go onto the waiting list.                 *                 * bug 2322: we used to unlink and re-add here, which was a                 * terrible folly -- if we goto restart, we could get                 * re-ordered!  Causes deadlock, because ASTs aren't sent! */		if (list_empty(&lock->l_res_link))                        ldlm_resource_add_lock(res, &res->lr_waiting, lock);                unlock_res(res);                rc = ldlm_run_ast_work(ldlm_res_to_ns(res), &rpc_list,                                       LDLM_WORK_BL_AST);                lock_res(res);		if (rc == -ERESTART) {			/* We were granted while waiting, nothing left to do */			if (lock->l_granted_mode == lock->l_req_mode)				GOTO(out, rc = 0);			/* Lock was destroyed while we were waiting, abort */			if (ldlm_is_destroyed(lock))				GOTO(out, rc = -EAGAIN);			/* Otherwise try again */			GOTO(restart, rc);		}                *flags |= LDLM_FL_BLOCK_GRANTED;        } else {                ldlm_resource_unlink_lock(lock);                ldlm_grant_lock(lock, NULL);        }	rc = 0;out:	*err = rc;	LASSERT(list_empty(&rpc_list));	RETURN(rc);}

voidmongo_cursor_foreach_async (MongoCursor         *cursor,                            MongoCursorCallback  foreach_func,                            gpointer             foreach_data,                            GDestroyNotify       foreach_notify,                            GCancellable        *cancellable,                            GAsyncReadyCallback  callback,                            gpointer             user_data){   MongoCursorPrivate *priv;   GSimpleAsyncResult *simple;   gchar *db_and_collection;   ENTRY;   g_return_if_fail(MONGO_IS_CURSOR(cursor));   g_return_if_fail(foreach_func);   g_return_if_fail(!cancellable || G_IS_CANCELLABLE(cancellable));   g_return_if_fail(callback);   priv = cursor->priv;   if (!priv->connection) {      g_simple_async_report_error_in_idle(G_OBJECT(cursor),                                          callback,                                          user_data,                                          MONGO_CONNECTION_ERROR,                                          MONGO_CONNECTION_ERROR_NOT_CONNECTED,                                          _("Not currently connected."));      GOTO(failure);   }   simple = g_simple_async_result_new(G_OBJECT(cursor),                                      callback,                                      user_data,                                      mongo_cursor_foreach_async);   g_simple_async_result_set_check_cancellable(simple, cancellable);   if (cancellable) {      g_object_set_data_full(G_OBJECT(simple), "cancellable",                             cancellable, (GDestroyNotify)g_object_unref);   }   g_object_set_data(G_OBJECT(simple), "foreach-func", foreach_func);   if (foreach_notify) {      g_object_set_data_full(G_OBJECT(simple), "foreach-data",                             foreach_data, foreach_notify);   } else {      g_object_set_data(G_OBJECT(simple), "foreach-data", foreach_data);   }   db_and_collection = g_strdup_printf("%s.%s",                                       priv->database,                                       priv->collection);   mongo_connection_query_async(priv->connection,                                db_and_collection,                                priv->flags,                                priv->skip,                                priv->limit,                                priv->query,                                priv->fields,                                cancellable,                                mongo_cursor_foreach_query_cb,                                simple);   g_free(db_and_collection);failure:   EXIT;}

static int lov_init_raid0(const struct lu_env *env,			  struct lov_device *dev, struct lov_object *lov,			  const struct cl_object_conf *conf,			  union  lov_layout_state *state){	int result;	int i;	struct cl_object	*stripe;	struct lov_thread_info  *lti     = lov_env_info(env);	struct cl_object_conf   *subconf = &lti->lti_stripe_conf;	struct lov_stripe_md    *lsm     = conf->u.coc_md->lsm;	struct lu_fid	   *ofid    = &lti->lti_fid;	struct lov_layout_raid0 *r0      = &state->raid0;	if (lsm->lsm_magic != LOV_MAGIC_V1 && lsm->lsm_magic != LOV_MAGIC_V3) {		dump_lsm(D_ERROR, lsm);		LASSERTF(0, "magic mismatch, expected %d/%d, actual %d./n",			 LOV_MAGIC_V1, LOV_MAGIC_V3, lsm->lsm_magic);	}	LASSERT(lov->lo_lsm == NULL);	lov->lo_lsm = lsm_addref(lsm);	r0->lo_nr  = lsm->lsm_stripe_count;	LASSERT(r0->lo_nr <= lov_targets_nr(dev));	OBD_ALLOC_LARGE(r0->lo_sub, r0->lo_nr * sizeof r0->lo_sub[0]);	if (r0->lo_sub != NULL) {		result = 0;		subconf->coc_inode = conf->coc_inode;		spin_lock_init(&r0->lo_sub_lock);		/*		 * Create stripe cl_objects.		 */		for (i = 0; i < r0->lo_nr && result == 0; ++i) {			struct cl_device *subdev;			struct lov_oinfo *oinfo = lsm->lsm_oinfo[i];			int ost_idx = oinfo->loi_ost_idx;			result = ostid_to_fid(ofid, &oinfo->loi_oi,					      oinfo->loi_ost_idx);			if (result != 0)				GOTO(out, result);			subdev = lovsub2cl_dev(dev->ld_target[ost_idx]);			subconf->u.coc_oinfo = oinfo;			LASSERTF(subdev != NULL, "not init ost %d/n", ost_idx);			/* In the function below, .hs_keycmp resolves to			 * lu_obj_hop_keycmp() */			/* coverity[overrun-buffer-val] */			stripe = lov_sub_find(env, subdev, ofid, subconf);			if (!IS_ERR(stripe)) {				result = lov_init_sub(env, lov, stripe, r0, i);				if (result == -EAGAIN) { /* try again */					--i;					result = 0;				}			} else {				result = PTR_ERR(stripe);			}		}	} else		result = -ENOMEM;out:	return result;}

static boolmongoc_database_add_user_legacy (mongoc_database_t *database,                                 const char        *username,                                 const char        *password,                                 bson_error_t      *error){    mongoc_collection_t *collection;    mongoc_cursor_t *cursor = NULL;    const bson_t *doc;    bool ret = false;    bson_t query;    bson_t user;    char *input;    char *pwd = NULL;    ENTRY;    bson_return_val_if_fail(database, false);    bson_return_val_if_fail(username, false);    bson_return_val_if_fail(password, false);    /*     * Users are stored in the <dbname>.system.users virtual collection.     * However, this will likely change to a command soon.     */    collection = mongoc_client_get_collection(database->client,                 database->name,                 "system.users");    BSON_ASSERT(collection);    /*     * Hash the users password.     */    input = bson_strdup_printf("%s:mongo:%s", username, password);    pwd = _mongoc_hex_md5(input);    bson_free(input);    /*     * Check to see if the user exists. If so, we will update the     * password instead of inserting a new user.     */    bson_init(&query);    bson_append_utf8(&query, "user", 4, username, -1);    cursor = mongoc_collection_find(collection, MONGOC_QUERY_NONE, 0, 1, 0,                                    &query, NULL, NULL);    if (!mongoc_cursor_next(cursor, &doc)) {        if (mongoc_cursor_error(cursor, error)) {            GOTO (failure);        }        bson_init(&user);        bson_append_utf8(&user, "user", 4, username, -1);        bson_append_bool(&user, "readOnly", 8, false);        bson_append_utf8(&user, "pwd", 3, pwd, -1);    } else {        bson_copy_to_excluding(doc, &user, "pwd", (char *)NULL);        bson_append_utf8(&user, "pwd", 3, pwd, -1);    }    if (!mongoc_collection_save(collection, &user, NULL, error)) {        GOTO (failure_with_user);    }    ret = true;failure_with_user:    bson_destroy(&user);failure:    if (cursor) {        mongoc_cursor_destroy(cursor);    }    mongoc_collection_destroy(collection);    bson_destroy(&query);    bson_free(pwd);    RETURN (ret);}