/* copied from lov/lov_ea.c, just for debugging, will be removed later */
void mdt_dump_lmm(int level, const struct lov_mds_md *lmm, __u64 valid)
{
	const struct lov_ost_data_v1	*lod;
	int				 i;
	__u16				 count;

	if (likely(!cfs_cdebug_show(level, DEBUG_SUBSYSTEM)))
		return;

	count = le16_to_cpu(((struct lov_user_md *)lmm)->lmm_stripe_count);

	CDEBUG(level, "objid "DOSTID", magic 0x%08X, pattern %#X\n",
	       POSTID(&lmm->lmm_oi), le32_to_cpu(lmm->lmm_magic),
	       le32_to_cpu(lmm->lmm_pattern));
	CDEBUG(level, "stripe_size=0x%x, stripe_count=0x%x\n",
	       le32_to_cpu(lmm->lmm_stripe_size), count);

	/* If it's a directory or a released file, then there are
	 * no actual objects to print, so bail out. */
	if (valid & OBD_MD_FLDIREA ||
	    le32_to_cpu(lmm->lmm_pattern) & LOV_PATTERN_F_RELEASED)
		return;

	LASSERT(count <= LOV_MAX_STRIPE_COUNT);
	for (i = 0, lod = lmm->lmm_objects; i < count; i++, lod++) {
		struct ost_id oi;

		ostid_le_to_cpu(&lod->l_ost_oi, &oi);
		CDEBUG(level, "stripe %u idx %u subobj "DOSTID"\n",
		       i, le32_to_cpu(lod->l_ost_idx), POSTID(&oi));
	}
}

/* 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_log_hdr *llh = loghandle->lgh_hdr;
        int rc = 0;
        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);
        }

	spin_lock(&loghandle->lgh_hdr_lock);
	if (!ext2_clear_bit(index, llh->llh_bitmap)) {
		spin_unlock(&loghandle->lgh_hdr_lock);
		CDEBUG(D_RPCTRACE, "Catalog index %u already clear?\n", index);
		RETURN(-ENOENT);
	}

	llh->llh_count--;

	if ((llh->llh_flags & LLOG_F_ZAP_WHEN_EMPTY) &&
	    (llh->llh_count == 1) &&
	    (loghandle->lgh_last_idx == (LLOG_BITMAP_BYTES * 8) - 1)) {
		spin_unlock(&loghandle->lgh_hdr_lock);
		rc = llog_destroy(env, loghandle);
		if (rc < 0) {
			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_err, rc);
		}
		RETURN(1);
	}
	spin_unlock(&loghandle->lgh_hdr_lock);

	rc = llog_write(env, loghandle, &llh->llh_hdr, NULL, 0, NULL, 0);
	if (rc < 0) {
		CERROR("%s: fail to write header for 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_err, rc);
	}
	RETURN(0);
out_err:
	spin_lock(&loghandle->lgh_hdr_lock);
	ext2_set_bit(index, llh->llh_bitmap);
	llh->llh_count++;
	spin_unlock(&loghandle->lgh_hdr_lock);
	return rc;
}

static int llog_remove_log(const struct lu_env *env, struct llog_handle *cat,
			   struct llog_logid *logid)
{
	struct llog_handle	*log;
	int			 rc;

	ENTRY;

	rc = llog_cat_id2handle(env, cat, &log, logid);
	if (rc) {
		CDEBUG(D_IOCTL, "cannot find log #"DOSTID"#%08x\n",
		       POSTID(&logid->lgl_oi), logid->lgl_ogen);
		RETURN(-ENOENT);
	}

	rc = llog_destroy(env, log);
	if (rc) {
		CDEBUG(D_IOCTL, "cannot destroy log\n");
		GOTO(out, rc);
	}
	llog_cat_cleanup(env, cat, log, log->u.phd.phd_cookie.lgc_index);
out:
	llog_handle_put(log);
	RETURN(rc);

}

void dump_lsm(unsigned int level, const struct lov_stripe_md *lsm)
{
	CDEBUG(level, "lsm %p, objid " DOSTID ", maxbytes %#llx, magic 0x%08X, stripe_size %u, stripe_count %u, refc: %d, layout_gen %u, pool [" LOV_POOLNAMEF "]\n",
	       lsm,
	       POSTID(&lsm->lsm_oi), lsm->lsm_maxbytes, lsm->lsm_magic,
	       lsm->lsm_stripe_size, lsm->lsm_stripe_count,
	       atomic_read(&lsm->lsm_refc), lsm->lsm_layout_gen,
	       lsm->lsm_pool_name);
}

void obdo_cpy_md(struct obdo *dst, struct obdo *src, obd_flag valid)
{
	CDEBUG(D_INODE, "src obdo "DOSTID" valid "LPX64", dst obdo "DOSTID"\n",
	       POSTID(&src->o_oi), src->o_valid, POSTID(&dst->o_oi));
	if (valid & OBD_MD_FLATIME)
		dst->o_atime = src->o_atime;
	if (valid & OBD_MD_FLMTIME)
		dst->o_mtime = src->o_mtime;
	if (valid & OBD_MD_FLCTIME)
		dst->o_ctime = src->o_ctime;
	if (valid & OBD_MD_FLSIZE)
		dst->o_size = src->o_size;
	if (valid & OBD_MD_FLBLOCKS) /* allocation of space */
		dst->o_blocks = src->o_blocks;
	if (valid & OBD_MD_FLBLKSZ)
		dst->o_blksize = src->o_blksize;
	if (valid & OBD_MD_FLTYPE)
		dst->o_mode = (dst->o_mode & ~S_IFMT) | (src->o_mode & S_IFMT);
	if (valid & OBD_MD_FLMODE)
		dst->o_mode = (dst->o_mode & S_IFMT) | (src->o_mode & ~S_IFMT);
	if (valid & OBD_MD_FLUID)
		dst->o_uid = src->o_uid;
	if (valid & OBD_MD_FLGID)
		dst->o_gid = src->o_gid;
	if (valid & OBD_MD_FLFLAGS)
		dst->o_flags = src->o_flags;
	if (valid & OBD_MD_FLFID) {
		dst->o_parent_seq = src->o_parent_seq;
		dst->o_parent_ver = src->o_parent_ver;
	}
	if (valid & OBD_MD_FLGENER)
		dst->o_parent_oid = src->o_parent_oid;
	if (valid & OBD_MD_FLHANDLE)
		dst->o_handle = src->o_handle;
	if (valid & OBD_MD_FLCOOKIE)
		dst->o_lcookie = src->o_lcookie;

	dst->o_valid |= valid;
}

static void print_llogd_body(struct llogd_body *d)
{
	CDEBUG(D_OTHER, "llogd body: %p\n", d);
	CDEBUG(D_OTHER, "\tlgd_logid.lgl_oi: "DOSTID"\n",
	       POSTID(&d->lgd_logid.lgl_oi));
	CDEBUG(D_OTHER, "\tlgd_logid.lgl_ogen: %#x\n", d->lgd_logid.lgl_ogen);
	CDEBUG(D_OTHER, "\tlgd_ctxt_idx: %#x\n", d->lgd_ctxt_idx);
	CDEBUG(D_OTHER, "\tlgd_llh_flags: %#x\n", d->lgd_llh_flags);
	CDEBUG(D_OTHER, "\tlgd_index: %#x\n", d->lgd_index);
	CDEBUG(D_OTHER, "\tlgd_saved_index: %#x\n", d->lgd_saved_index);
	CDEBUG(D_OTHER, "\tlgd_len: %#x\n", d->lgd_len);
	CDEBUG(D_OTHER, "\tlgd_cur_offset: %#llx\n", d->lgd_cur_offset);
}

/* the replicators commit callback */
static int lov_llog_repl_cancel(const struct lu_env *env,
				struct llog_ctxt *ctxt,
				struct lov_stripe_md *lsm,
				int count, struct llog_cookie *cookies,
				int flags)
{
        struct lov_obd *lov;
        struct obd_device *obd = ctxt->loc_obd;
        int rc = 0, i;
        ENTRY;

        LASSERT(lsm != NULL);
        LASSERT(count == lsm->lsm_stripe_count);

        lov = &obd->u.lov;
        obd_getref(obd);
        for (i = 0; i < count; i++, cookies++) {
                struct lov_oinfo *loi = lsm->lsm_oinfo[i];
                struct obd_device *child =
                        lov->lov_tgts[loi->loi_ost_idx]->ltd_exp->exp_obd;
                struct llog_ctxt *cctxt =
                        llog_get_context(child, ctxt->loc_idx);
                int err;

		err = llog_cancel(env, cctxt, NULL, 1, cookies, flags);
		llog_ctxt_put(cctxt);
		if (err && lov->lov_tgts[loi->loi_ost_idx]->ltd_active) {
			CERROR("%s: objid "DOSTID" subobj "DOSTID
			       " on OST idx %d: rc = %d\n",
			       obd->obd_name, POSTID(&lsm->lsm_oi),
			       POSTID(&loi->loi_oi), loi->loi_ost_idx, err);
			if (!rc)
				rc = err;
		}
	}
	obd_putref(obd);
	RETURN(rc);
}

static int osc_object_print(const struct lu_env *env, void *cookie,
			    lu_printer_t p, const struct lu_object *obj)
{
	struct osc_object   *osc   = lu2osc(obj);
	struct lov_oinfo    *oinfo = osc->oo_oinfo;
	struct osc_async_rc *ar    = &oinfo->loi_ar;

	(*p)(env, cookie, "id: " DOSTID " idx: %d gen: %d kms_valid: %u kms %llu rc: %d force_sync: %d min_xid: %llu ",
	     POSTID(&oinfo->loi_oi), oinfo->loi_ost_idx,
	     oinfo->loi_ost_gen, oinfo->loi_kms_valid, oinfo->loi_kms,
	     ar->ar_rc, ar->ar_force_sync, ar->ar_min_xid);
	osc_lvb_print(env, cookie, p, &oinfo->loi_lvb);
	return 0;
}

void lov_dump_lmm_common(int level, void *lmmp)
{
	struct lov_mds_md *lmm = lmmp;
	struct ost_id	oi;

	lmm_oi_le_to_cpu(&oi, &lmm->lmm_oi);
	CDEBUG(level, "objid "DOSTID", magic 0x%08x, pattern %#x\n",
	       POSTID(&oi), le32_to_cpu(lmm->lmm_magic),
	       le32_to_cpu(lmm->lmm_pattern));
	CDEBUG(level, "stripe_size %u, stripe_count %u, layout_gen %u\n",
	       le32_to_cpu(lmm->lmm_stripe_size),
	       le16_to_cpu(lmm->lmm_stripe_count),
	       le16_to_cpu(lmm->lmm_layout_gen));
}

/* copied from lov/lov_ea.c, just for debugging, will be removed later */
void mdt_dump_lmm(int level, const struct lov_mds_md *lmm)
{
        const struct lov_ost_data_v1 *lod;
        int                           i;
        __u16                         count;

        count = le16_to_cpu(((struct lov_user_md*)lmm)->lmm_stripe_count);

	CDEBUG(level, "objid "DOSTID", magic 0x%08X, pattern %#X\n",
	       POSTID(&lmm->lmm_oi), le32_to_cpu(lmm->lmm_magic),
	       le32_to_cpu(lmm->lmm_pattern));
        CDEBUG(level,"stripe_size=0x%x, stripe_count=0x%x\n",
               le32_to_cpu(lmm->lmm_stripe_size), count);
        if (count == LOV_ALL_STRIPES)
                return;

	LASSERT(count <= LOV_MAX_STRIPE_COUNT);
	for (i = 0, lod = lmm->lmm_objects; i < count; i++, lod++) {
		struct ost_id	oi;
		ostid_le_to_cpu((struct ost_id *)&lod->l_ost_oi, &oi);
		CDEBUG(level, "stripe %u idx %u subobj "DOSTID"\n",
		       i, le32_to_cpu(lod->l_ost_idx), POSTID(&oi));
	}
}

int llog_catalog_list(const struct lu_env *env, struct dt_device *d,
		      int count, struct obd_ioctl_data *data,
		      const struct lu_fid *fid)
{
	int			 size, i;
	struct llog_catid	*idarray;
	struct llog_logid	*id;
	char			*out;
	int			 l, remains, rc = 0;

	ENTRY;

	if (count == 0) { /* get total number of logs */
		rc = llog_osd_get_cat_list(env, d, 0, 0, NULL, fid);
		if (rc < 0)
			RETURN(rc);
		count = rc;
	}

	size = sizeof(*idarray) * count;

	OBD_ALLOC_LARGE(idarray, size);
	if (!idarray)
		RETURN(-ENOMEM);

	rc = llog_osd_get_cat_list(env, d, 0, count, idarray, fid);
	if (rc)
		GOTO(out, rc);

	out = data->ioc_bulk;
	remains = data->ioc_inllen1;
	for (i = 0; i < count; i++) {
		id = &idarray[i].lci_logid;
		l = snprintf(out, remains,
			     "catalog log: #"DOSTID"#%08x\n",
			     POSTID(&id->lgl_oi),
			     id->lgl_ogen);
		out += l;
		remains -= l;
		if (remains <= 0)
			break;
	}
out:
	OBD_FREE_LARGE(idarray, size);
	RETURN(rc);
}

static int changelog_init_cb(const struct lu_env *env, struct llog_handle *llh,
			     struct llog_rec_hdr *hdr, void *data)
{
	struct mdd_device *mdd = (struct mdd_device *)data;
	struct llog_changelog_rec *rec = (struct llog_changelog_rec *)hdr;

	LASSERT(llh->lgh_hdr->llh_flags & LLOG_F_IS_PLAIN);
	LASSERT(rec->cr_hdr.lrh_type == CHANGELOG_REC);

	CDEBUG(D_INFO,
	       "seeing record at index %d/%d/"LPU64" t=%x %.*s in log"
	       DOSTID"\n", hdr->lrh_index, rec->cr_hdr.lrh_index,
	       rec->cr.cr_index, rec->cr.cr_type, rec->cr.cr_namelen,
	       changelog_rec_name(&rec->cr), POSTID(&llh->lgh_id.lgl_oi));

	mdd->mdd_cl.mc_index = rec->cr.cr_index;
	return LLOG_PROC_BREAK;
}

/** Merge the lock value block(&lvb) attributes from each of the stripes in a
 * file into a single lvb. It is expected that the caller initializes the
 * current atime, mtime, ctime to avoid regressing a more uptodate time on
 * the local client.
 *
 * If \a kms_only is set then we do not consider the recently seen size (rss)
 * when updating the known minimum size (kms).  Even when merging RSS, we will
 * take the KMS value if it's larger.  This prevents getattr from stomping on
 * dirty cached pages which extend the file size. */
int lov_merge_lvb(struct obd_export *exp,
		  struct lov_stripe_md *lsm, struct ost_lvb *lvb, int kms_only)
{
	int   rc;
	__u64 kms;

	ENTRY;
	lov_stripe_lock(lsm);
	rc = lov_merge_lvb_kms(lsm, lvb, &kms);
	lov_stripe_unlock(lsm);
	if (kms_only)
		lvb->lvb_size = kms;

	CDEBUG(D_INODE, "merged for ID "DOSTID" s="LPU64" m="LPU64" a="LPU64
	       " c="LPU64" b="LPU64"\n", POSTID(&lsm->lsm_oi), lvb->lvb_size,
	       lvb->lvb_mtime, lvb->lvb_atime, lvb->lvb_ctime, lvb->lvb_blocks);
	RETURN(rc);
}

int lov_setea(struct obd_export *exp, struct lov_stripe_md **lsmp,
              struct lov_user_md *lump)
{
        int i;
        int rc;
        struct obd_export *oexp;
        struct lov_obd *lov = &exp->exp_obd->u.lov;
        obd_id last_id = 0;
        struct lov_user_ost_data_v1 *lmm_objects;

        ENTRY;

        if (lump->lmm_magic == LOV_USER_MAGIC_V3)
                lmm_objects = ((struct lov_user_md_v3 *)lump)->lmm_objects;
        else
                lmm_objects = lump->lmm_objects;

        for (i = 0; i < lump->lmm_stripe_count; i++) {
                __u32 len = sizeof(last_id);
                oexp = lov->lov_tgts[lmm_objects[i].l_ost_idx]->ltd_exp;
                rc = obd_get_info(NULL, oexp, sizeof(KEY_LAST_ID), KEY_LAST_ID,
                                  &len, &last_id, NULL);
                if (rc)
                        RETURN(rc);
		if (ostid_id(&lmm_objects[i].l_ost_oi) > last_id) {
			CERROR("Setting EA for object > than last id on"
			       " ost idx %d "DOSTID" > "LPD64" \n",
			       lmm_objects[i].l_ost_idx,
			       POSTID(&lmm_objects[i].l_ost_oi), last_id);
			RETURN(-EINVAL);
		}
        }

        rc = lov_setstripe(exp, 0, lsmp, lump);
        if (rc)
                RETURN(rc);

        for (i = 0; i < lump->lmm_stripe_count; i++) {
                (*lsmp)->lsm_oinfo[i]->loi_ost_idx =
                        lmm_objects[i].l_ost_idx;
		(*lsmp)->lsm_oinfo[i]->loi_oi = lmm_objects[i].l_ost_oi;
        }
        RETURN(0);
}

static void lov_dump_lmm_objects(int level, struct lov_ost_data *lod,
				 int stripe_count)
{
	int i;

	if (stripe_count > LOV_V1_INSANE_STRIPE_COUNT) {
		CDEBUG(level, "bad stripe_count %u > max_stripe_count %u\n",
		       stripe_count, LOV_V1_INSANE_STRIPE_COUNT);
		return;
	}

	for (i = 0; i < stripe_count; ++i, ++lod) {
		struct ost_id oi;

		ostid_le_to_cpu(&lod->l_ost_oi, &oi);
		CDEBUG(level, "stripe %u idx %u subobj "DOSTID"\n", i,
		       le32_to_cpu(lod->l_ost_idx), POSTID(&oi));
	}
}

static int llog_lvfs_close(const struct lu_env *env,
			   struct llog_handle *handle)
{
	int rc;

	if (handle->lgh_file == NULL)
		return 0;
	rc = filp_close(handle->lgh_file, 0);
	if (rc)
		CERROR("%s: error closing llog #"DOSTID"#%08x: "
		       "rc = %d\n", handle->lgh_ctxt->loc_obd->obd_name,
		       POSTID(&handle->lgh_id.lgl_oi),
		       handle->lgh_id.lgl_ogen, rc);
	handle->lgh_file = NULL;
	if (handle->lgh_name) {
		OBD_FREE(handle->lgh_name, strlen(handle->lgh_name) + 1);
		handle->lgh_name = NULL;
	}
	return rc;
}

void obdo_to_inode(struct inode *dst, struct obdo *src, obd_flag valid)
{
	valid &= src->o_valid;

	LASSERTF(!(valid & (OBD_MD_FLTYPE | OBD_MD_FLGENER | OBD_MD_FLFID |
			    OBD_MD_FLID | OBD_MD_FLGROUP)),
		 "object "DOSTID", valid %x\n", POSTID(&src->o_oi), valid);

	if (valid & (OBD_MD_FLCTIME | OBD_MD_FLMTIME))
		CDEBUG(D_INODE,
		       "valid %#llx, cur time %lu/%lu, new %llu/%llu\n",
		       src->o_valid, LTIME_S(dst->i_mtime),
		       LTIME_S(dst->i_ctime), src->o_mtime, src->o_ctime);

	if (valid & OBD_MD_FLATIME)
		LTIME_S(dst->i_atime) = src->o_atime;
	if (valid & OBD_MD_FLMTIME)
		LTIME_S(dst->i_mtime) = src->o_mtime;
	if (valid & OBD_MD_FLCTIME && src->o_ctime > LTIME_S(dst->i_ctime))
		LTIME_S(dst->i_ctime) = src->o_ctime;
	if (valid & OBD_MD_FLSIZE)
		i_size_write(dst, src->o_size);
	if (valid & OBD_MD_FLBLOCKS) { /* allocation of space */
		dst->i_blocks = src->o_blocks;
		if (dst->i_blocks < src->o_blocks) /* overflow */
			dst->i_blocks = -1;

	}
	if (valid & OBD_MD_FLBLKSZ)
		dst->i_blkbits = ffs(src->o_blksize)-1;
	if (valid & OBD_MD_FLMODE)
		dst->i_mode = (dst->i_mode & S_IFMT) | (src->o_mode & ~S_IFMT);
	if (valid & OBD_MD_FLUID)
		dst->i_uid = make_kuid(&init_user_ns, src->o_uid);
	if (valid & OBD_MD_FLGID)
		dst->i_gid = make_kgid(&init_user_ns, src->o_gid);
	if (valid & OBD_MD_FLFLAGS)
		dst->i_flags = src->o_flags;
}

static int changelog_user_init_cb(const struct lu_env *env,
				  struct llog_handle *llh,
				  struct llog_rec_hdr *hdr, void *data)
{
        struct mdd_device *mdd = (struct mdd_device *)data;
        struct llog_changelog_user_rec *rec =
                (struct llog_changelog_user_rec *)hdr;

        LASSERT(llh->lgh_hdr->llh_flags & LLOG_F_IS_PLAIN);
        LASSERT(rec->cur_hdr.lrh_type == CHANGELOG_USER_REC);

        CDEBUG(D_INFO, "seeing user at index %d/%d id=%d endrec="LPU64
               " in log "DOSTID"\n", hdr->lrh_index, rec->cur_hdr.lrh_index,
               rec->cur_id, rec->cur_endrec, POSTID(&llh->lgh_id.lgl_oi));

	spin_lock(&mdd->mdd_cl.mc_user_lock);
	mdd->mdd_cl.mc_lastuser = rec->cur_id;
	if (rec->cur_endrec > mdd->mdd_cl.mc_index)
		mdd->mdd_cl.mc_index = rec->cur_endrec;
	spin_unlock(&mdd->mdd_cl.mc_user_lock);

	return LLOG_PROC_BREAK;
}

static int llog_changelog_cancel(const struct lu_env *env,
				 struct llog_ctxt *ctxt,
				 struct llog_cookie *cookies, int flags)
{
	struct llog_handle	*cathandle = ctxt->loc_handle;
	int			 rc;

	ENTRY;

	/* This should only be called with the catalog handle */
	LASSERT(cathandle->lgh_hdr->llh_flags & LLOG_F_IS_CAT);

	rc = llog_cat_process(env, cathandle, llog_changelog_cancel_cb,
			      (void *)cookies, 0, 0);
	if (rc >= 0)
		/* 0 or 1 means we're done */
		rc = 0;
	else
		CERROR("%s: cancel idx %u of catalog "DOSTID" rc=%d\n",
		       ctxt->loc_obd->obd_name, cathandle->lgh_last_idx,
		       POSTID(&cathandle->lgh_id.lgl_oi), rc);

	RETURN(rc);
}

/**
 * Implementation of the llog_operations::lop_prev_block
 *
 * This function finds the llog block to return which contains
 * record with required index but in reverse order - from end of llog
 * to the beginning.
 * It is main part of reverse llog processing.
 *
 * \param[in] env	execution environment
 * \param[in] loghandle	llog handle of the current llog
 * \param[in] prev_idx	target index to find
 * \param[in] buf	pointer to data buffer to fill
 * \param[in] len	required len to read, it is LLOG_CHUNK_SIZE usually.
 *
 * \retval		0 on successful buffer read
 * \retval		negative value on error
 */
static int llog_osd_prev_block(const struct lu_env *env,
			       struct llog_handle *loghandle,
			       int prev_idx, void *buf, int len)
{
	struct llog_thread_info	*lgi = llog_info(env);
	struct dt_object	*o;
	struct dt_device	*dt;
	loff_t			 cur_offset;
	int			 rc;

	ENTRY;

	if (len == 0 || len & (LLOG_CHUNK_SIZE - 1))
		RETURN(-EINVAL);

	CDEBUG(D_OTHER, "looking for log index %u\n", prev_idx);

	LASSERT(loghandle);
	LASSERT(loghandle->lgh_ctxt);

	o = loghandle->lgh_obj;
	LASSERT(o);
	LASSERT(dt_object_exists(o));
	dt = lu2dt_dev(o->do_lu.lo_dev);
	LASSERT(dt);

	cur_offset = LLOG_CHUNK_SIZE;
	llog_skip_over(&cur_offset, 0, prev_idx);

	rc = dt_attr_get(env, o, &lgi->lgi_attr, BYPASS_CAPA);
	if (rc)
		GOTO(out, rc);

	while (cur_offset < lgi->lgi_attr.la_size) {
		struct llog_rec_hdr	*rec, *last_rec;
		struct llog_rec_tail	*tail;

		lgi->lgi_buf.lb_len = len;
		lgi->lgi_buf.lb_buf = buf;
		rc = dt_read(env, o, &lgi->lgi_buf, &cur_offset);
		if (rc < 0) {
			CERROR("%s: can't read llog block from log "DFID
			       " offset "LPU64": rc = %d\n",
			       o->do_lu.lo_dev->ld_obd->obd_name,
			       PFID(lu_object_fid(&o->do_lu)), cur_offset, rc);
			GOTO(out, rc);
		}

		if (rc == 0) /* end of file, nothing to do */
			GOTO(out, rc);

		if (rc < sizeof(*tail)) {
			CERROR("%s: invalid llog block at log id "DOSTID"/%u "
			       "offset "LPU64"\n",
			       o->do_lu.lo_dev->ld_obd->obd_name,
			       POSTID(&loghandle->lgh_id.lgl_oi),
			       loghandle->lgh_id.lgl_ogen, cur_offset);
			GOTO(out, rc = -EINVAL);
		}

		rec = buf;
		if (LLOG_REC_HDR_NEEDS_SWABBING(rec))
			lustre_swab_llog_rec(rec);

		tail = (struct llog_rec_tail *)((char *)buf + rc -
						sizeof(struct llog_rec_tail));
		/* get the last record in block */
		last_rec = (struct llog_rec_hdr *)((char *)buf + rc -
						   le32_to_cpu(tail->lrt_len));

		if (LLOG_REC_HDR_NEEDS_SWABBING(last_rec))
			lustre_swab_llog_rec(last_rec);
		LASSERT(last_rec->lrh_index == tail->lrt_index);

		/* this shouldn't happen */
		if (tail->lrt_index == 0) {
			CERROR("%s: invalid llog tail at log id "DOSTID"/%u "
			       "offset "LPU64"\n",
			       o->do_lu.lo_dev->ld_obd->obd_name,
			       POSTID(&loghandle->lgh_id.lgl_oi),
			       loghandle->lgh_id.lgl_ogen, cur_offset);
			GOTO(out, rc = -EINVAL);
		}
//.........这里部分代码省略.........