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 void
mongo_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(&q