/** get (name+parent_id) for an entry
 * \param linkno hardlink index
 * \retval -ENODATA after last link
 * \retval -ERANGE if namelen is too small
 */
int Lustre_GetNameParent(const char *path, int linkno,
                         lustre_fid *pfid, char *name,
                         int namelen)
{
    int rc, i, len;
    char buf[4096];
    struct linkea_data     ldata      = { 0 };
    struct lu_buf          lb = { 0 };

    rc = lgetxattr(path, XATTR_NAME_LINK, buf, sizeof(buf));
    if (rc < 0)
        return -errno;

    lb.lb_buf = buf;
    lb.lb_len = sizeof(buf);
    ldata.ld_buf = &lb;
    ldata.ld_leh = (struct link_ea_header *)buf;

    ldata.ld_lee = LINKEA_FIRST_ENTRY(ldata);
    ldata.ld_reclen = (ldata.ld_lee->lee_reclen[0] << 8)
               | ldata.ld_lee->lee_reclen[1];

    if (linkno >= ldata.ld_leh->leh_reccount)
        /* beyond last link */
        return -ENODATA;

    for (i = 0; i < linkno; i++) {
        ldata.ld_lee = LINKEA_NEXT_ENTRY(ldata);
        ldata.ld_reclen = (ldata.ld_lee->lee_reclen[0] << 8)
                   | ldata.ld_lee->lee_reclen[1];
    }

    memcpy(pfid, &ldata.ld_lee->lee_parent_fid, sizeof(*pfid));
    fid_be_to_cpu(pfid, pfid);

    if (!fid_is_sane(pfid))
    {
        DisplayLog(LVL_MAJOR, __func__, "insane fid: "DFID, PFID(pfid));
        return -EPROTO;
    }

    len = ldata.ld_reclen - sizeof(struct link_ea_entry);
    if (len >= namelen)
        return -ERANGE;

    strncpy(name, ldata.ld_lee->lee_name, len);
    name[len] = '\0';
    return 0;
}

/*
 * Concurrency: no concurrent access is possible that early in object
 * life-cycle.
 */
static int osd_object_init(const struct lu_env *env, struct lu_object *l,
			   const struct lu_object_conf *conf)
{
	struct osd_object	*obj = osd_obj(l);
	struct osd_device	*osd = osd_obj2dev(obj);
	uint64_t		 oid;
	int			 rc;
	ENTRY;

	LASSERT(osd_invariant(obj));

	if (fid_is_otable_it(&l->lo_header->loh_fid)) {
		obj->oo_dt.do_ops = &osd_obj_otable_it_ops;
		l->lo_header->loh_attr |= LOHA_EXISTS;
		RETURN(0);
	}

	if (conf != NULL && conf->loc_flags & LOC_F_NEW)
		GOTO(out, rc = 0);

	rc = osd_fid_lookup(env, osd, lu_object_fid(l), &oid);
	if (rc == 0) {
		LASSERT(obj->oo_db == NULL);
		rc = __osd_obj2dbuf(env, osd->od_os, oid, &obj->oo_db);
		/* EEXIST will be returned if object is being deleted in ZFS */
		if (rc == -EEXIST) {
			rc = 0;
			GOTO(out, rc);
		}
		if (rc != 0) {
			CERROR("%s: lookup "DFID"/"LPX64" failed: rc = %d\n",
			       osd->od_svname, PFID(lu_object_fid(l)), oid, rc);
			GOTO(out, rc);
		}
		LASSERT(obj->oo_db);
		rc = osd_object_init0(env, obj);
		if (rc != 0)
			GOTO(out, rc);

		rc = osd_check_lma(env, obj);
		if (rc != 0)
			GOTO(out, rc);
	} else if (rc == -ENOENT) {
		rc = 0;
	}
	LASSERT(osd_invariant(obj));
out:
	RETURN(rc);
}

/* Get POSIX path from fid (fid2path wrapper) */
int Lustre_GetFullPath( const entry_id_t * p_id, char *fullpath, unsigned int len )
{
    char          *curr = fullpath;
    int            rc;
    long long      recno = -1;
    int            linkno = 0;
    char           fid[256];
    const char    *mpath = NULL;
    unsigned int   mlen = 0;

    mpath = get_mount_point(&mlen);

    /* set mountpoint at the beginning of the path */
    strcpy( fullpath, mpath );
    curr += mlen;

/* add the slash only if fid2path doesn't */
#ifndef _FID2PATH_LEADING_SLASH
    /* add slash */
    *curr = '/';
    curr ++;
#endif
    /* just in case fid2path returns nothing */
    *curr = '\0';

    /* fid string */
    sprintf( fid, DFID, PFID(p_id) );

    /* MDT device */

    /* ask the path to lustre */
    rc = llapi_fid2path( mpath, fid, curr, len - mlen - 2, &recno,
                         &linkno );

    if ( (rc != 0) && (rc != -ENOENT) && (rc != -ESTALE) )
        DisplayLog( LVL_CRIT, "Fid2Path",
                    "Error %d calling llapi_fid2path(%s,%s,%lld,%d), errno=%d."
                    " Cannot retrieve full path for %s",
                    rc, mpath, fid, recno, linkno, errno, fid );
    /* curr == fullpath => fullpath is root: '/'
     * so don't remove final slash */
    else if (curr != fullpath)
    {
        while (FINAL_SLASH(fullpath))
            REMOVE_FINAL_SLASH(fullpath);
    }

    return rc;
}

/* Partial request to create object only */
static int mdt_md_mkobj(struct mdt_thread_info *info)
{
        struct mdt_device      *mdt = info->mti_mdt;
        struct mdt_object      *o;
        struct mdt_body        *repbody;
        struct md_attr         *ma = &info->mti_attr;
        int rc;
        ENTRY;

        DEBUG_REQ(D_INODE, mdt_info_req(info), "Partial create "DFID"",
                  PFID(info->mti_rr.rr_fid2));

        repbody = req_capsule_server_get(info->mti_pill, &RMF_MDT_BODY);

        o = mdt_object_find(info->mti_env, mdt, info->mti_rr.rr_fid2,
                            MDT_OBJ_MAY_NOT_EXIST);
        if (!IS_ERR(o)) {
                struct md_object *next = mdt_object_child(o);

                ma->ma_need = MA_INODE;
                ma->ma_valid = 0;

                /*
                 * Cross-ref create can encounter already created obj in case of
                 * recovery, just get attr in that case.
                 */
                if (mdt_object_exists(o) == 1) {
                        rc = mo_attr_get(info->mti_env, next, ma);
                } else {
                        /*
                         * Here, NO permission check for object_create,
                         * such check has been done on the original MDS.
                         */
                        rc = mo_object_create(info->mti_env, next,
                                              &info->mti_spec, ma);
                }
                if (rc == 0) {
                        /* Return fid & attr to client. */
                        if (ma->ma_valid & MA_INODE)
                                mdt_pack_attr2body(info, repbody, &ma->ma_attr,
                                                   mdt_object_fid(o));
                }
                mdt_object_put(info->mti_env, o);
        } else
                rc = PTR_ERR(o);

        mdt_create_pack_capa(info, rc, o, repbody);
        RETURN(rc);
}

static int
lprocfs_fid_fid_seq_show(struct seq_file *m, void *unused)
{
	struct lu_client_seq *seq = (struct lu_client_seq *)m->private;
	int rc;
	ENTRY;

	LASSERT(seq != NULL);

	mutex_lock(&seq->lcs_mutex);
	rc = seq_printf(m, DFID"\n", PFID(&seq->lcs_fid));
	mutex_unlock(&seq->lcs_mutex);

	RETURN(rc);
}

static int lov_layout_wait(const struct lu_env *env, struct lov_object *lov)
{
	struct l_wait_info lwi = { 0 };
	ENTRY;

	while (cfs_atomic_read(&lov->lo_active_ios) > 0) {
		CDEBUG(D_INODE, "file:"DFID" wait for active IO, now: %d.\n",
			PFID(lu_object_fid(lov2lu(lov))),
			cfs_atomic_read(&lov->lo_active_ios));

		l_wait_event(lov->lo_waitq,
			     cfs_atomic_read(&lov->lo_active_ios) == 0, &lwi);
	}
	RETURN(0);
}

/**
 * Get the first operation for a given id.
 * @return an operation to be processed when it is possible.
 *         NULL else. 
 *        
 */
entry_proc_op_t *id_constraint_get_first_op( entry_id_t * p_id )
{
    unsigned int   hash_index;
    id_constraint_item_t *p_curr;
    entry_proc_op_t *p_op = NULL;

    /* compute id hash value */
    hash_index = hash_id( p_id, ID_HASH_SIZE );

    P( id_hash[hash_index].lock );

    for ( p_curr = id_hash[hash_index].id_list_first; p_curr != NULL; p_curr = p_curr->p_next )
    {
        if ( entry_id_equal( p_id, &p_curr->op_ptr->entry_id ) )
        {
            p_op = p_curr->op_ptr;
            break;
        }
    }
#ifdef _DEBUG_ID_CONSTRAINT
    if ( p_op )
       printf( "first op on id "DFID" at stage %u (list %u)\n",
               PFID(&p_op->entry_id), p_op->pipeline_stage, hash_index );
    else
    {

        printf( "no registered operation on "DFID"?\n", PFID(p_id));
        printf( "etat de la file %u:\n", hash_index );
        for ( p_curr = id_hash[hash_index].id_list_first; p_curr != NULL; p_curr = p_curr->p_next )
            printf( DFID"\n", PFID(&p_curr->op_ptr->entry_id) );
    }
#endif
    V( id_hash[hash_index].lock );
    return p_op;

}

static int vvp_prune(const struct lu_env *env, struct cl_object *obj)
{
	struct inode *inode = vvp_object_inode(obj);
	int rc;

	rc = cl_sync_file_range(inode, 0, OBD_OBJECT_EOF, CL_FSYNC_LOCAL, 1);
	if (rc < 0) {
		CDEBUG(D_VFSTRACE, DFID ": writeback failed: %d\n",
		       PFID(lu_object_fid(&obj->co_lu)), rc);
		return rc;
	}

	truncate_inode_pages(inode->i_mapping, 0);
	return 0;
}

static int out_xattr_get(struct tgt_session_info *tsi)
{
	const struct lu_env	*env = tsi->tsi_env;
	struct tgt_thread_info	*tti = tgt_th_info(env);
	struct update		*update = tti->tti_u.update.tti_update;
	struct lu_buf		*lbuf = &tti->tti_buf;
	struct update_reply     *reply = tti->tti_u.update.tti_update_reply;
	struct dt_object        *obj = tti->tti_u.update.tti_dt_object;
	char			*name;
	void			*ptr;
	int			 rc;

	ENTRY;

	name = (char *)update_param_buf(update, 0, NULL);
	if (name == NULL) {
		CERROR("%s: empty name for xattr get: rc = %d\n",
		       tgt_name(tsi->tsi_tgt), -EPROTO);
		RETURN(err_serious(-EPROTO));
	}

	ptr = update_get_buf_internal(reply, 0, NULL);
	LASSERT(ptr != NULL);

	/* The first 4 bytes(int) are used to store the result */
	lbuf->lb_buf = (char *)ptr + sizeof(int);
	lbuf->lb_len = UPDATE_BUFFER_SIZE - sizeof(struct update_reply);
	dt_read_lock(env, obj, MOR_TGT_CHILD);
	rc = dt_xattr_get(env, obj, lbuf, name, NULL);
	dt_read_unlock(env, obj);
	if (rc < 0) {
		lbuf->lb_len = 0;
		GOTO(out, rc);
	}
	if (rc == 0) {
		lbuf->lb_len = 0;
		GOTO(out, rc = -ENOENT);
	}
	lbuf->lb_len = rc;
	rc = 0;
	CDEBUG(D_INFO, "%s: "DFID" get xattr %s len %d\n",
	       tgt_name(tsi->tsi_tgt), PFID(lu_object_fid(&obj->do_lu)),
	       name, (int)lbuf->lb_len);
out:
	*(int *)ptr = rc;
	reply->ur_lens[0] = lbuf->lb_len + sizeof(int);
	RETURN(rc);
}

static int ll_page_mkwrite(struct vm_area_struct *vma, struct vm_fault *vmf)
{
	int count = 0;
	bool printed = false;
	bool retry;
	int result;

	ll_stats_ops_tally(ll_i2sbi(file_inode(vma->vm_file)),
			   LPROC_LL_MKWRITE, 1);

	file_update_time(vma->vm_file);
        do {
                retry = false;
                result = ll_page_mkwrite0(vma, vmf->page, &retry);

                if (!printed && ++count > 16) {
			const struct dentry *de = file_dentry(vma->vm_file);

			CWARN("app(%s): the page %lu of file "DFID" is under"
			      " heavy contention\n",
			      current->comm, vmf->pgoff,
			      PFID(ll_inode2fid(de->d_inode)));
                        printed = true;
                }
        } while (retry);

        switch(result) {
        case 0:
                LASSERT(PageLocked(vmf->page));
                result = VM_FAULT_LOCKED;
                break;
        case -ENODATA:
        case -EFAULT:
                result = VM_FAULT_NOPAGE;
                break;
        case -ENOMEM:
                result = VM_FAULT_OOM;
                break;
        case -EAGAIN:
                result = VM_FAULT_RETRY;
                break;
        default:
                result = VM_FAULT_SIGBUS;
                break;
        }

        return result;
}

/**
 * Cliens updates SOM attributes on MDS (including llog cookies):
 * obd_getattr with no lock and md_setattr.
 */
int ll_som_update(struct inode *inode, struct md_op_data *op_data)
{
	struct ll_inode_info *lli = ll_i2info(inode);
	struct ptlrpc_request *request = NULL;
	__u32 old_flags;
	struct obdo *oa;
	int rc;

	LASSERT(op_data != NULL);
	if (lli->lli_flags & LLIF_MDS_SIZE_LOCK)
		CERROR("ino %lu/%u(flags %u) som valid it just after recovery\n",
		       inode->i_ino, inode->i_generation,
		       lli->lli_flags);

	oa = kmem_cache_alloc(obdo_cachep, GFP_NOFS | __GFP_ZERO);
	if (!oa) {
		CERROR("can't allocate memory for Size-on-MDS update.\n");
		return -ENOMEM;
	}

	old_flags = op_data->op_flags;
	op_data->op_flags = MF_SOM_CHANGE;

	/* If inode is already in another epoch, skip getattr from OSTs. */
	if (lli->lli_ioepoch == op_data->op_ioepoch) {
		rc = ll_inode_getattr(inode, oa, op_data->op_ioepoch,
				      old_flags & MF_GETATTR_LOCK);
		if (rc) {
			oa->o_valid = 0;
			if (rc != -ENOENT)
				CERROR("inode_getattr failed (%d): unable to send a Size-on-MDS attribute update for inode %lu/%u\n",
				       rc, inode->i_ino,
				       inode->i_generation);
		} else {
			CDEBUG(D_INODE, "Size-on-MDS update on "DFID"\n",
			       PFID(&lli->lli_fid));
		}
		/* Install attributes into op_data. */
		md_from_obdo(op_data, oa, oa->o_valid);
	}

	rc = md_setattr(ll_i2sbi(inode)->ll_md_exp, op_data,
			NULL, 0, NULL, 0, &request, NULL);
	ptlrpc_req_finished(request);

	kmem_cache_free(obdo_cachep, oa);
	return rc;
}

static int
seq_client_proc_read_fid(char *page, char **start, off_t off,
                         int count, int *eof, void *data)
{
        struct lu_client_seq *seq = (struct lu_client_seq *)data;
	int rc;
	ENTRY;

        LASSERT(seq != NULL);

        cfs_down(&seq->lcs_sem);
        rc = snprintf(page, count, DFID"\n", PFID(&seq->lcs_fid));
        cfs_up(&seq->lcs_sem);

	RETURN(rc);
}

static int osc_object_prune(const struct lu_env *env, struct cl_object *obj)
{
	struct osc_object       *osc = cl2osc(obj);
	struct ldlm_res_id      *resname = &osc_env_info(env)->oti_resname;

	LASSERTF(osc->oo_npages == 0,
		 DFID "still have %lu pages, obj: %p, osc: %p\n",
		 PFID(lu_object_fid(&obj->co_lu)), osc->oo_npages, obj, osc);

	/* DLM locks don't hold a reference of osc_object so we have to
	 * clear it before the object is being destroyed. */
	ostid_build_res_name(&osc->oo_oinfo->loi_oi, resname);
	ldlm_resource_iterate(osc_export(osc)->exp_obd->obd_namespace, resname,
			      osc_object_ast_clear, osc);
	return 0;
}

/**
 * Lookup method for "fid" object. Only filenames with correct SEQ:OID format
 * are valid. We also check if object with passed fid exists or not.
 */
static int obf_lookup(const struct lu_env *env, struct md_object *p,
                      const struct lu_name *lname, struct lu_fid *f,
                      struct md_op_spec *spec)
{
        char *name = (char *)lname->ln_name;
        struct mdd_device *mdd = mdo2mdd(p);
        struct mdd_object *child;
        int rc = 0;

        while (*name == '[')
                name++;

        sscanf(name, SFID, RFID(f));
        if (!fid_is_sane(f)) {
		CWARN("%s: Trying to lookup invalid FID [%s] in %s/%s, FID "
		      "format should be "DFID"\n", mdd2obd_dev(mdd)->obd_name,
		      lname->ln_name, dot_lustre_name, mdd_obf_dir_name,
		      (__u64)FID_SEQ_NORMAL, 1, 0);
                GOTO(out, rc = -EINVAL);
        }

	if (!fid_is_norm(f) && !fid_is_igif(f) && !fid_is_root(f) &&
	    !fid_seq_is_dot(f->f_seq)) {
		CWARN("%s: Trying to lookup invalid FID "DFID" in %s/%s, "
		      "sequence should be >= "LPX64" or within ["LPX64","
		      ""LPX64"].\n", mdd2obd_dev(mdd)->obd_name, PFID(f),
		      dot_lustre_name, mdd_obf_dir_name, (__u64)FID_SEQ_NORMAL,
		      (__u64)FID_SEQ_IGIF, (__u64)FID_SEQ_IGIF_MAX);
		GOTO(out, rc = -EINVAL);
	}

        /* Check if object with this fid exists */
        child = mdd_object_find(env, mdd, f);
        if (child == NULL)
                GOTO(out, rc = 0);
        if (IS_ERR(child))
                GOTO(out, rc = PTR_ERR(child));

        if (mdd_object_exists(child) == 0)
                rc = -ENOENT;

        mdd_object_put(env, child);

out:
        return rc;
}

/**
 *  delete an orphan \a obj from orphan index.
 *  \param obj file or directory.
 *  \param th  transaction for index deletion and object destruction.
 *
 *  \pre obj->mod_count == 0 && ORPHAN_OBJ is set for obj.
 *
 *  \retval 0  success
 *  \retval  -ve index operation error.
 */
int mdd_orphan_delete(const struct lu_env *env, struct mdd_object *obj,
		      struct thandle *th)
{
	struct mdd_device *mdd = mdo2mdd(&obj->mod_obj);
	struct dt_object *dor = mdd->mdd_orphans;
	struct dt_key *key;
	int rc = 0;

	ENTRY;

	LASSERT(mdd_write_locked(env, obj) != 0);
	LASSERT(obj->mod_flags & ORPHAN_OBJ);
	LASSERT(obj->mod_count == 0);

	LASSERT(dor);

	key = mdd_orphan_key_fill(env, mdo2fid(obj));
	dt_write_lock(env, mdd->mdd_orphans, MOR_TGT_ORPHAN);

	if (OBD_FAIL_CHECK(OBD_FAIL_MDS_ORPHAN_DELETE))
		goto ref_del;

	rc = dt_delete(env, mdd->mdd_orphans, key, th);
	if (rc == -ENOENT) {
		key = mdd_orphan_key_fill_20(env, mdo2fid(obj));
		rc = dt_delete(env, mdd->mdd_orphans, key, th);
	}

ref_del:
	if (!rc) {
		/* lov objects will be destroyed by caller */
		mdo_ref_del(env, obj, th);
		if (S_ISDIR(mdd_object_type(obj))) {
			mdo_ref_del(env, obj, th);
			dt_ref_del(env, mdd->mdd_orphans, th);
		}
		obj->mod_flags &= ~ORPHAN_OBJ;
	} else {
		CERROR("%s: could not delete orphan object "DFID": rc = %d\n",
		       mdd2obd_dev(mdd)->obd_name, PFID(mdo2fid(obj)), rc);
	}

	dt_write_unlock(env, mdd->mdd_orphans);
	RETURN(rc);
}

/**
 * \a connectable - is nfsd will connect himself or this should be done
 *		  at lustre
 *
 * The return value is file handle type:
 * 1 -- contains child file handle;
 * 2 -- contains child file handle and parent file handle;
 * 255 -- error.
 */
static int ll_encode_fh(struct inode *inode, __u32 *fh, int *plen,
			struct inode *parent)
{
	struct lustre_nfs_fid *nfs_fid = (void *)fh;

	CDEBUG(D_INFO, "encoding for (%lu,"DFID") maxlen=%d minlen=%d\n",
	      inode->i_ino, PFID(ll_inode2fid(inode)), *plen,
	      (int)sizeof(struct lustre_nfs_fid));

	if (*plen < sizeof(struct lustre_nfs_fid) / 4)
		return 255;

	nfs_fid->lnf_child = *ll_inode2fid(inode);
	nfs_fid->lnf_parent = *ll_inode2fid(parent);
	*plen = sizeof(struct lustre_nfs_fid) / 4;

	return LUSTRE_NFS_FID;
}

/**
 * FLR: verify the layout version of object.
 *
 * \param[in] env	execution environment
 * \param[in] fo	OFD object
 * \param[in] oa	OBDO structure with layout version
 *
 * \retval		0 on successful verification
 * \retval		-EINPROGRESS layout version is in transfer
 * \retval		-ESTALE the layout version on client is stale
 */
int ofd_verify_layout_version(const struct lu_env *env,
			      struct ofd_object *fo, const struct obdo *oa)
{
	__u32 layout_version;
	int rc;
	ENTRY;

	if (unlikely(OBD_FAIL_CHECK(OBD_FAIL_OST_SKIP_LV_CHECK)))
		GOTO(out, rc = 0);

	rc = ofd_object_ff_load(env, fo);
	if (rc < 0) {
		if (rc == -ENODATA)
			rc = -EINPROGRESS;
		GOTO(out, rc);
	}

	layout_version = fo->ofo_ff.ff_layout_version;
	if (oa->o_layout_version >= layout_version &&
	    oa->o_layout_version <= layout_version + fo->ofo_ff.ff_range)
		GOTO(out, rc = 0);

	/* normal traffic, decide if to return ESTALE or EINPROGRESS */
	layout_version &= ~LU_LAYOUT_RESYNC;

	/* this update is not legitimate */
	if ((oa->o_layout_version & ~LU_LAYOUT_RESYNC) <= layout_version)
		GOTO(out, rc = -ESTALE);

	/* layout version may not be transmitted yet */
	if ((oa->o_layout_version & ~LU_LAYOUT_RESYNC) > layout_version)
		GOTO(out, rc = -EINPROGRESS);

	EXIT;

out:
	CDEBUG(D_INODE, DFID " verify layout version: %u vs. %u/%u, rc: %d\n",
	       PFID(lu_object_fid(&fo->ofo_obj.do_lu)),
	       oa->o_layout_version, fo->ofo_ff.ff_layout_version,
	       fo->ofo_ff.ff_range, rc);
	return rc;

}

/**
 * Mark the linkEA as overflow with current timestamp,
 * and remove the last linkEA entry.
 *
 * Return the new linkEA size.
 */
int linkea_overflow_shrink(struct linkea_data *ldata)
{
	struct link_ea_header *leh;
	struct lu_name tname;
	struct lu_fid tfid;
	int count;

	leh = ldata->ld_leh = ldata->ld_buf->lb_buf;
	if (leh->leh_magic == __swab32(LINK_EA_MAGIC)) {
		leh->leh_magic = LINK_EA_MAGIC;
		leh->leh_reccount = __swab32(leh->leh_reccount);
		leh->leh_overflow_time = __swab32(leh->leh_overflow_time);
		leh->leh_padding = __swab32(leh->leh_padding);
	}

	LASSERT(leh->leh_reccount > 0);

	leh->leh_len = sizeof(struct link_ea_header);
	leh->leh_reccount--;
	if (unlikely(leh->leh_reccount == 0))
		return 0;

	leh->leh_overflow_time = cfs_time_current_sec();
	if (unlikely(leh->leh_overflow_time == 0))
		leh->leh_overflow_time++;
	ldata->ld_reclen = 0;
	ldata->ld_lee = (struct link_ea_entry *)(leh + 1);
	for (count = 0; count < leh->leh_reccount; count++) {
		linkea_entry_unpack(ldata->ld_lee, &ldata->ld_reclen,
				    &tname, &tfid);
		leh->leh_len += ldata->ld_reclen;
		ldata->ld_lee = (struct link_ea_entry *)((char *)ldata->ld_lee +
							 ldata->ld_reclen);
	}

	linkea_entry_unpack(ldata->ld_lee, &ldata->ld_reclen, &tname, &tfid);
	CDEBUG(D_INODE, "No enough space to hold the last linkea entry '"
	       DFID": %.*s', shrink it, left %d linkea entries, size %llu\n",
	       PFID(&tfid), tname.ln_namelen, tname.ln_name,
	       leh->leh_reccount, leh->leh_len);

	return leh->leh_len;
}

static int plain_print_cb(const struct lu_env *env, struct llog_handle *llh,
			  struct llog_rec_hdr *rec, void *data)
{
	struct lu_fid fid = {0};

	if (!(llh->lgh_hdr->llh_flags & LLOG_F_IS_PLAIN)) {
		CERROR("log is not plain\n");
		return -EINVAL;
	}

	logid_to_fid(&llh->lgh_id, &fid);

	CDEBUG(D_INFO, "seeing record at index %d in log "DFID"\n",
	       rec->lrh_index, PFID(&fid));

	plain_counter++;

	return 0;
}

/** Cliens updates SOM attributes on MDS: obd_getattr and md_setattr. */
int llu_som_update(struct inode *inode, struct md_op_data *op_data)
{
        struct llu_inode_info *lli = llu_i2info(inode);
        struct llu_sb_info *sbi = llu_i2sbi(inode);
        struct obdo oa = { 0 };
        __u32 old_flags;
        int rc;
        ENTRY;

        LASSERT(!(lli->lli_flags & LLIF_MDS_SIZE_LOCK));
        LASSERT(sbi->ll_lco.lco_flags & OBD_CONNECT_SOM);

        old_flags = op_data->op_flags;
        op_data->op_flags = MF_SOM_CHANGE;

        /* If inode is already in another epoch, skip getattr from OSTs. */
        if (lli->lli_ioepoch == op_data->op_ioepoch) {
                rc = llu_inode_getattr(inode, &oa, op_data->op_ioepoch,
                                       old_flags & MF_GETATTR_LOCK);
                if (rc) {
                        oa.o_valid = 0;
                        if (rc == -ENOENT)
                                CDEBUG(D_INODE, "objid "LPX64" is destroyed\n",
                                       lli->lli_smd->lsm_object_id);
                        else
                                CERROR("inode_getattr failed (%d): unable to "
                                       "send a Size-on-MDS attribute update "
                                       "for inode %llu/%lu\n", rc,
                                       (long long)llu_i2stat(inode)->st_ino,
                                       lli->lli_st_generation);
                }  else {
                        CDEBUG(D_INODE, "Size-on-MDS update on "DFID"\n",
                               PFID(&lli->lli_fid));
                }

                /* Install attributes into op_data. */
                md_from_obdo(op_data, &oa, oa.o_valid);
        }

        rc = llu_md_setattr(inode, op_data, NULL);
        RETURN(rc);
}