Whamcloud - gitweb
LU-6401 headers: Create a header for obdo related functions
[fs/lustre-release.git] / lustre / osp / osp_object.c
index be29d0c..96e3135 100644 (file)
@@ -27,7 +27,7 @@
  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
  * Use is subject to license terms.
  *
- * Copyright (c) 2014, Intel Corporation.
+ * Copyright (c) 2012, 2015, Intel Corporation.
  */
 /*
  * lustre/osp/osp_object.c
 
 #define DEBUG_SUBSYSTEM S_MDS
 
+#include <lustre_obdo.h>
+#include <lustre_swab.h>
+
 #include "osp_internal.h"
 
 static inline __u32 osp_dev2node(struct osp_device *osp)
@@ -168,7 +171,7 @@ static void osp_object_assign_fid(const struct lu_env *env,
  * \retval             0 for success
  * \retval             negative error number on failure
  */
-static int osp_oac_init(struct osp_object *obj)
+int osp_oac_init(struct osp_object *obj)
 {
        struct osp_object_attr *ooa;
 
@@ -319,7 +322,7 @@ osp_oac_xattr_find_or_add(struct osp_object *obj, const char *name, size_t len)
  *
  * \param[in] obj      pointer to the OSP object
  * \param[in,out] poxe double pointer to the OSP object extended attribute
- *                     entry: the new extended attribute entry is transfered
+ *                     entry: the new extended attribute entry is transferred
  *                     via such pointer target, and if old the extended
  *                     attribute entry exists, then it will be returned back
  *                     via such pointer target.
@@ -493,13 +496,11 @@ static int osp_attr_get_interpterer(const struct lu_env *env,
  *
  * \param[in] env      pointer to the thread context
  * \param[in] dt       pointer to the OSP layer dt_object
- * \param[in] capa     the capability for this operation
  *
  * \retval             0 for success
  * \retval             negative error number on failure
  */
-static int osp_declare_attr_get(const struct lu_env *env, struct dt_object *dt,
-                               struct lustre_capa *capa)
+static int osp_declare_attr_get(const struct lu_env *env, struct dt_object *dt)
 {
        struct osp_object       *obj    = dt2osp_obj(dt);
        struct osp_device       *osp    = lu2osp_dev(dt->do_lu.lo_dev);
@@ -514,6 +515,7 @@ static int osp_declare_attr_get(const struct lu_env *env, struct dt_object *dt,
        mutex_lock(&osp->opd_async_requests_mutex);
        rc = osp_insert_async_request(env, OUT_ATTR_GET, obj, 0, NULL, NULL,
                                      &obj->opo_ooa->ooa_attr,
+                                     sizeof(struct obdo),
                                      osp_attr_get_interpterer);
        mutex_unlock(&osp->opd_async_requests_mutex);
 
@@ -533,18 +535,17 @@ static int osp_declare_attr_get(const struct lu_env *env, struct dt_object *dt,
  * \param[in] env      pointer to the thread context
  * \param[in] dt       pointer to the OSP layer dt_object
  * \param[out] attr    pointer to the buffer to hold the output attribute
- * \param[in] capa     the capability for this operation
  *
  * \retval             0 for success
  * \retval             negative error number on failure
  */
 int osp_attr_get(const struct lu_env *env, struct dt_object *dt,
-                struct lu_attr *attr, struct lustre_capa *capa)
+                struct lu_attr *attr)
 {
        struct osp_device               *osp = lu2osp_dev(dt->do_lu.lo_dev);
        struct osp_object               *obj = dt2osp_obj(dt);
        struct dt_device                *dev = &osp->opd_dt_dev;
-       struct dt_update_request        *update;
+       struct osp_update_request       *update;
        struct object_update_reply      *reply;
        struct ptlrpc_request           *req = NULL;
        int                             rc = 0;
@@ -555,7 +556,7 @@ int osp_attr_get(const struct lu_env *env, struct dt_object *dt,
 
        if (obj->opo_ooa != NULL) {
                spin_lock(&obj->opo_lock);
-               if (obj->opo_ooa->ooa_attr.la_valid != 0) {
+               if (obj->opo_ooa->ooa_attr.la_valid != 0 && !obj->opo_stale) {
                        *attr = obj->opo_ooa->ooa_attr;
                        spin_unlock(&obj->opo_lock);
 
@@ -564,12 +565,12 @@ int osp_attr_get(const struct lu_env *env, struct dt_object *dt,
                spin_unlock(&obj->opo_lock);
        }
 
-       update = dt_update_request_create(dev);
+       update = osp_update_request_create(dev);
        if (IS_ERR(update))
                RETURN(PTR_ERR(update));
 
-       rc = out_attr_get_pack(env, &update->dur_buf,
-                              lu_object_fid(&dt->do_lu));
+       rc = osp_update_rpc_pack(env, attr_get, update, OUT_ATTR_GET,
+                                lu_object_fid(&dt->do_lu));
        if (rc != 0) {
                CERROR("%s: Insert update error "DFID": rc = %d\n",
                       dev->dd_lu_dev.ld_obd->obd_name,
@@ -578,7 +579,7 @@ int osp_attr_get(const struct lu_env *env, struct dt_object *dt,
                GOTO(out, rc);
        }
 
-       rc = out_remote_sync(env, osp->opd_obd->u.cli.cl_import, update, &req);
+       rc = osp_remote_sync(env, osp, update, &req);
        if (rc != 0) {
                if (rc == -ENOENT) {
                        osp2lu_obj(obj)->lo_header->loh_attr &= ~LOHA_EXISTS;
@@ -604,26 +605,45 @@ int osp_attr_get(const struct lu_env *env, struct dt_object *dt,
        if (rc != 0)
                GOTO(out, rc);
 
-       GOTO(out, rc = 0);
+       spin_lock(&obj->opo_lock);
+       if (obj->opo_stale)
+               obj->opo_stale = 0;
+       spin_unlock(&obj->opo_lock);
+
+       GOTO(out, rc);
 
 out:
        if (req != NULL)
                ptlrpc_req_finished(req);
 
-       dt_update_request_destroy(update);
+       osp_update_request_destroy(update);
 
        return rc;
 }
 
-static int __osp_attr_set(const struct lu_env *env, struct dt_object *dt,
-                         const struct lu_attr *attr, struct thandle *th)
+/**
+ * Implement OSP layer dt_object_operations::do_declare_attr_set() interface.
+ *
+ * If the transaction is not remote one, then declare the credits that will
+ * be used for the subsequent llog record for the object's attributes.
+ *
+ * \param[in] env      pointer to the thread context
+ * \param[in] dt       pointer to the OSP layer dt_object
+ * \param[in] attr     pointer to the attribute to be set
+ * \param[in] th       pointer to the transaction handler
+ *
+ * \retval             0 for success
+ * \retval             negative error number on failure
+ */
+static int osp_declare_attr_set(const struct lu_env *env, struct dt_object *dt,
+                               const struct lu_attr *attr, struct thandle *th)
 {
        struct osp_device       *d = lu2osp_dev(dt->do_lu.lo_dev);
        struct osp_object       *o = dt2osp_obj(dt);
-       struct lu_attr          *la;
-       int                      rc = 0;
-       ENTRY;
+       int                     rc;
 
+       if (is_only_remote_trans(th))
+               return osp_md_declare_attr_set(env, dt, attr, th);
        /*
         * Usually we don't allow server stack to manipulate size
         * but there is a special case when striping is created
@@ -652,71 +672,17 @@ static int __osp_attr_set(const struct lu_env *env, struct dt_object *dt,
                LASSERT(!dt_object_exists(dt));
                osp_object_assign_fid(env, d, o);
                rc = osp_object_truncate(env, dt, attr->la_size);
-               if (rc)
+               if (rc != 0)
                        RETURN(rc);
        }
 
        if (!(attr->la_valid & (LA_UID | LA_GID)))
                RETURN(0);
 
-       if (!is_only_remote_trans(th)) {
-               /*
-                * track all UID/GID changes via llog
-                */
-               rc = osp_sync_declare_add(env, o, MDS_SETATTR64_REC, th);
-       } else {
-               /* It is for OST-object attr_set directly without updating
-                * local MDT-object attribute. It is usually used by LFSCK. */
-               rc = __osp_md_attr_set(env, dt, attr, th);
-       }
-
-       if (rc != 0 || o->opo_ooa == NULL)
-               RETURN(rc);
-
-       /* Update the OSP object attributes cache. */
-       la = &o->opo_ooa->ooa_attr;
-       spin_lock(&o->opo_lock);
-       if (attr->la_valid & LA_UID) {
-               la->la_uid = attr->la_uid;
-               la->la_valid |= LA_UID;
-       }
-
-       if (attr->la_valid & LA_GID) {
-               la->la_gid = attr->la_gid;
-               la->la_valid |= LA_GID;
-       }
-       spin_unlock(&o->opo_lock);
-
-       RETURN(0);
-}
-
-/**
- * Implement OSP layer dt_object_operations::do_declare_attr_set() interface.
- *
- * If the transaction is not remote one, then declare the credits that will
- * be used for the subsequent llog record for the object's attributes.
- *
- * \param[in] env      pointer to the thread context
- * \param[in] dt       pointer to the OSP layer dt_object
- * \param[in] attr     pointer to the attribute to be set
- * \param[in] th       pointer to the transaction handler
- *
- * \retval             0 for success
- * \retval             negative error number on failure
- */
-static int osp_declare_attr_set(const struct lu_env *env, struct dt_object *dt,
-                               const struct lu_attr *attr, struct thandle *th)
-{
-       int rc = 0;
-
-       if (!is_only_remote_trans(th)) {
-               rc = __osp_attr_set(env, dt, attr, th);
-
-               CDEBUG(D_INFO, "declare set attr "DFID": rc = %d\n",
-                      PFID(&dt->do_lu.lo_header->loh_fid), rc);
-       }
+       /* track all UID/GID changes via llog */
+       rc = osp_sync_declare_add(env, o, MDS_SETATTR64_REC, th);
 
-       return rc;
+       return 0;
 }
 
 /**
@@ -736,37 +702,50 @@ static int osp_declare_attr_set(const struct lu_env *env, struct dt_object *dt,
  * \param[in] dt       pointer to the OSP layer dt_object
  * \param[in] attr     pointer to the attribute to be set
  * \param[in] th       pointer to the transaction handler
- * \param[in] capa     the capability for this operation
  *
  * \retval             0 for success
  * \retval             negative error number on failure
  */
 static int osp_attr_set(const struct lu_env *env, struct dt_object *dt,
-                       const struct lu_attr *attr, struct thandle *th,
-                       struct lustre_capa *capa)
+                       const struct lu_attr *attr, struct thandle *th)
 {
        struct osp_object       *o = dt2osp_obj(dt);
        int                      rc = 0;
        ENTRY;
 
-       if (is_only_remote_trans(th)) {
-               rc = __osp_attr_set(env, dt, attr, th);
+       /* we're interested in uid/gid changes only */
+       if (!(attr->la_valid & (LA_UID | LA_GID)))
+               RETURN(0);
+
+       if (!is_only_remote_trans(th)) {
+               rc = osp_sync_add(env, o, MDS_SETATTR64_REC, th, attr);
+               /* XXX: send new uid/gid to OST ASAP? */
+       } else {
+               struct lu_attr  *la;
 
+               /* It is for OST-object attr_set directly without updating
+                * local MDT-object attribute. It is usually used by LFSCK. */
+               rc = osp_md_attr_set(env, dt, attr, th);
                CDEBUG(D_INFO, "(1) set attr "DFID": rc = %d\n",
                       PFID(&dt->do_lu.lo_header->loh_fid), rc);
 
-               RETURN(rc);
-       }
-
-       /* we're interested in uid/gid changes only */
-       if (!(attr->la_valid & (LA_UID | LA_GID)))
-               RETURN(0);
+               if (rc != 0 || o->opo_ooa == NULL)
+                       RETURN(rc);
 
-       rc = osp_sync_add(env, o, MDS_SETATTR64_REC, th, attr);
-       /* XXX: send new uid/gid to OST ASAP? */
+               /* Update the OSP object attributes cache. */
+               la = &o->opo_ooa->ooa_attr;
+               spin_lock(&o->opo_lock);
+               if (attr->la_valid & LA_UID) {
+                       la->la_uid = attr->la_uid;
+                       la->la_valid |= LA_UID;
+               }
 
-       CDEBUG(D_INFO, "(2) set attr "DFID": rc = %d\n",
-              PFID(&dt->do_lu.lo_header->loh_fid), rc);
+               if (attr->la_valid & LA_GID) {
+                       la->la_gid = attr->la_gid;
+                       la->la_valid |= LA_GID;
+               }
+               spin_unlock(&o->opo_lock);
+       }
 
        RETURN(rc);
 }
@@ -849,14 +828,12 @@ static int osp_xattr_get_interpterer(const struct lu_env *env,
  * \param[in] dt       pointer to the OSP layer dt_object
  * \param[out] buf     pointer to the lu_buf to hold the extended attribute
  * \param[in] name     the name for the expected extended attribute
- * \param[in] capa     the capability for this operation
  *
  * \retval             0 for success
  * \retval             negative error number on failure
  */
 static int osp_declare_xattr_get(const struct lu_env *env, struct dt_object *dt,
-                                struct lu_buf *buf, const char *name,
-                                struct lustre_capa *capa)
+                                struct lu_buf *buf, const char *name)
 {
        struct osp_object       *obj     = dt2osp_obj(dt);
        struct osp_device       *osp     = lu2osp_dev(dt->do_lu.lo_dev);
@@ -883,25 +860,28 @@ static int osp_declare_xattr_get(const struct lu_env *env, struct dt_object *dt,
 
        mutex_lock(&osp->opd_async_requests_mutex);
        rc = osp_insert_async_request(env, OUT_XATTR_GET, obj, 1,
-                                     &namelen, (const void **)&name, oxe,
+                                     &namelen, (const void **)&name,
+                                     oxe, buf->lb_len,
                                      osp_xattr_get_interpterer);
        if (rc != 0) {
                mutex_unlock(&osp->opd_async_requests_mutex);
                osp_oac_xattr_put(oxe);
        } else {
-               struct dt_update_request *update;
+               struct osp_update_request *our;
+               struct osp_update_request_sub *ours;
 
                /* XXX: Currently, we trigger the batched async OUT
                 *      RPC via dt_declare_xattr_get(). It is not
                 *      perfect solution, but works well now.
                 *
                 *      We will improve it in the future. */
-               update = osp->opd_async_requests;
-               if (update != NULL && update->dur_buf.ub_req != NULL &&
-                   update->dur_buf.ub_req->ourq_count > 0) {
+               our = osp->opd_async_requests;
+               ours = osp_current_object_update_request(our);
+               if (ours != NULL && ours->ours_req != NULL &&
+                   ours->ours_req->ourq_count > 0) {
                        osp->opd_async_requests = NULL;
                        mutex_unlock(&osp->opd_async_requests_mutex);
-                       rc = osp_unplug_async_request(env, osp, update);
+                       rc = osp_unplug_async_request(env, osp, our);
                } else {
                        mutex_unlock(&osp->opd_async_requests_mutex);
                }
@@ -929,20 +909,18 @@ static int osp_declare_xattr_get(const struct lu_env *env, struct dt_object *dt,
  * \param[in] dt       pointer to the OSP layer dt_object
  * \param[out] buf     pointer to the lu_buf to hold the extended attribute
  * \param[in] name     the name for the expected extended attribute
- * \param[in] capa     the capability for this operation
  *
  * \retval             0 for success
  * \retval             negative error number on failure
  */
 int osp_xattr_get(const struct lu_env *env, struct dt_object *dt,
-                 struct lu_buf *buf, const char *name,
-                 struct lustre_capa *capa)
+                 struct lu_buf *buf, const char *name)
 {
        struct osp_device       *osp    = lu2osp_dev(dt->do_lu.lo_dev);
        struct osp_object       *obj    = dt2osp_obj(dt);
        struct dt_device        *dev    = &osp->opd_dt_dev;
        struct lu_buf           *rbuf   = &osp_env_info(env)->osi_lb2;
-       struct dt_update_request *update = NULL;
+       struct osp_update_request *update = NULL;
        struct ptlrpc_request   *req    = NULL;
        struct object_update_reply *reply;
        struct osp_xattr_entry  *oxe    = NULL;
@@ -967,46 +945,49 @@ int osp_xattr_get(const struct lu_env *env, struct dt_object *dt,
        if (unlikely(obj->opo_non_exist))
                RETURN(-ENOENT);
 
-       oxe = osp_oac_xattr_find(obj, name, false);
-       if (oxe != NULL) {
-               spin_lock(&obj->opo_lock);
-               if (oxe->oxe_ready) {
-                       if (!oxe->oxe_exist)
-                               GOTO(unlock, rc = -ENODATA);
+       /* Only cache xattr for OST object */
+       if (!osp->opd_connect_mdt) {
+               oxe = osp_oac_xattr_find(obj, name, false);
+               if (oxe != NULL) {
+                       spin_lock(&obj->opo_lock);
+                       if (oxe->oxe_ready) {
+                               if (!oxe->oxe_exist)
+                                       GOTO(unlock, rc = -ENODATA);
 
-                       if (buf->lb_buf == NULL)
-                               GOTO(unlock, rc = oxe->oxe_vallen);
+                               if (buf->lb_buf == NULL)
+                                       GOTO(unlock, rc = oxe->oxe_vallen);
 
-                       if (buf->lb_len < oxe->oxe_vallen)
-                               GOTO(unlock, rc = -ERANGE);
+                               if (buf->lb_len < oxe->oxe_vallen)
+                                       GOTO(unlock, rc = -ERANGE);
 
-                       memcpy(buf->lb_buf, oxe->oxe_value, oxe->oxe_vallen);
+                               memcpy(buf->lb_buf, oxe->oxe_value,
+                                      oxe->oxe_vallen);
 
-                       GOTO(unlock, rc = oxe->oxe_vallen);
+                               GOTO(unlock, rc = oxe->oxe_vallen);
 
 unlock:
-                       spin_unlock(&obj->opo_lock);
-                       osp_oac_xattr_put(oxe);
+                               spin_unlock(&obj->opo_lock);
+                               osp_oac_xattr_put(oxe);
 
-                       return rc;
+                               return rc;
+                       }
+                       spin_unlock(&obj->opo_lock);
                }
-               spin_unlock(&obj->opo_lock);
        }
-
-       update = dt_update_request_create(dev);
+       update = osp_update_request_create(dev);
        if (IS_ERR(update))
                GOTO(out, rc = PTR_ERR(update));
 
-       rc = out_xattr_get_pack(env, &update->dur_buf,
-                               lu_object_fid(&dt->do_lu), name);
+       rc = osp_update_rpc_pack(env, xattr_get, update, OUT_XATTR_GET,
+                                lu_object_fid(&dt->do_lu), name, buf->lb_len);
        if (rc != 0) {
                CERROR("%s: Insert update error "DFID": rc = %d\n",
                       dname, PFID(lu_object_fid(&dt->do_lu)), rc);
                GOTO(out, rc);
        }
 
-       rc = out_remote_sync(env, osp->opd_obd->u.cli.cl_import, update, &req);
-       if (rc != 0) {
+       rc = osp_remote_sync(env, osp, update, &req);
+       if (rc < 0) {
                if (rc == -ENOENT) {
                        dt->do_lu.lo_header->loh_attr &= ~LOHA_EXISTS;
                        obj->opo_non_exist = 1;
@@ -1054,14 +1035,13 @@ unlock:
                GOTO(out, rc);
 
        if (buf->lb_buf == NULL)
-               GOTO(out, rc = rbuf->lb_len);
+               GOTO(out, rc);
 
        if (unlikely(buf->lb_len < rbuf->lb_len))
                GOTO(out, rc = -ERANGE);
 
        memcpy(buf->lb_buf, rbuf->lb_buf, rbuf->lb_len);
-       rc = rbuf->lb_len;
-       if (obj->opo_ooa == NULL)
+       if (obj->opo_ooa == NULL || osp->opd_connect_mdt)
                GOTO(out, rc);
 
        if (oxe == NULL) {
@@ -1111,7 +1091,7 @@ out:
                ptlrpc_req_finished(req);
 
        if (update != NULL && !IS_ERR(update))
-               dt_update_request_destroy(update);
+               osp_update_request_destroy(update);
 
        if (oxe != NULL)
                osp_oac_xattr_put(oxe);
@@ -1119,76 +1099,6 @@ out:
        return rc;
 }
 
-static int __osp_xattr_set(const struct lu_env *env, struct dt_object *dt,
-                          const struct lu_buf *buf, const char *name,
-                          int flag, struct thandle *th)
-{
-       struct osp_object       *o = dt2osp_obj(dt);
-       struct dt_update_request *update;
-       struct osp_xattr_entry  *oxe;
-       int                     rc;
-       ENTRY;
-
-       LASSERT(buf->lb_len > 0 && buf->lb_buf != NULL);
-
-       update = dt_update_request_find_or_create(th, dt);
-       if (IS_ERR(update)) {
-               CERROR("%s: Get OSP update buf failed "DFID": rc = %d\n",
-                      dt->do_lu.lo_dev->ld_obd->obd_name,
-                      PFID(lu_object_fid(&dt->do_lu)),
-                      (int)PTR_ERR(update));
-
-               RETURN(PTR_ERR(update));
-       }
-
-       rc = out_xattr_set_pack(env, &update->dur_buf,
-                               lu_object_fid(&dt->do_lu),
-                               buf, name, flag, update->dur_batchid);
-       if (rc != 0 || o->opo_ooa == NULL)
-               RETURN(rc);
-
-       oxe = osp_oac_xattr_find_or_add(o, name, buf->lb_len);
-       if (oxe == NULL) {
-               CWARN("%s: cannot cache xattr '%s' of "DFID"\n",
-                     dt->do_lu.lo_dev->ld_obd->obd_name,
-                     name, PFID(lu_object_fid(&dt->do_lu)));
-
-               RETURN(0);
-       }
-
-       if (oxe->oxe_buflen - oxe->oxe_namelen - 1 < buf->lb_len) {
-               struct osp_xattr_entry *old = oxe;
-               struct osp_xattr_entry *tmp;
-
-               tmp = osp_oac_xattr_replace(o, &old, buf->lb_len);
-               osp_oac_xattr_put(oxe);
-               oxe = tmp;
-               if (tmp == NULL) {
-                       CWARN("%s: cannot update cached xattr '%s' of "DFID"\n",
-                             dt->do_lu.lo_dev->ld_obd->obd_name,
-                             name, PFID(lu_object_fid(&dt->do_lu)));
-                       spin_lock(&o->opo_lock);
-                       old->oxe_ready = 0;
-                       spin_unlock(&o->opo_lock);
-
-                       RETURN(0);
-               }
-
-               /* Drop the ref for entry on list. */
-               osp_oac_xattr_put(old);
-       }
-
-       spin_lock(&o->opo_lock);
-       oxe->oxe_vallen = buf->lb_len;
-       memcpy(oxe->oxe_value, buf->lb_buf, buf->lb_len);
-       oxe->oxe_exist = 1;
-       oxe->oxe_ready = 1;
-       spin_unlock(&o->opo_lock);
-       osp_oac_xattr_put(oxe);
-
-       RETURN(0);
-}
-
 /**
  * Implement OSP layer dt_object_operations::do_declare_xattr_set() interface.
  *
@@ -1216,16 +1126,7 @@ int osp_declare_xattr_set(const struct lu_env *env, struct dt_object *dt,
                          const struct lu_buf *buf, const char *name,
                          int flag, struct thandle *th)
 {
-       int rc = 0;
-
-       if (!is_only_remote_trans(th)) {
-               rc = __osp_xattr_set(env, dt, buf, name, flag, th);
-
-               CDEBUG(D_INFO, "declare xattr %s set object "DFID": rc = %d\n",
-                      name, PFID(&dt->do_lu.lo_header->loh_fid), rc);
-       }
-
-       return rc;
+       return osp_trans_update_request_create(th);
 }
 
 /**
@@ -1233,11 +1134,10 @@ int osp_declare_xattr_set(const struct lu_env *env, struct dt_object *dt,
  *
  * Set extended attribute to the specified MDT/OST object.
  *
- * If it is remote transaction, it will add an OUT_XATTR_SET sub-request into
- * the OUT RPC that will be flushed when the transaction stop. And if the OSP
- * attributes cache is initialized, then check whether the name extended
- * attribute entry exists in the cache or not. If yes, replace it; otherwise,
- * add the extended attribute to the cache.
+ * Add an OUT_XATTR_SET sub-request into the OUT RPC that will be flushed in
+ * the transaction stop. And if the OSP attributes cache is initialized, then
+ * check whether the name extended attribute entry exists in the cache or not.
+ * If yes, replace it; otherwise, add the extended attribute to the cache.
  *
  * \param[in] env      pointer to the thread context
  * \param[in] dt       pointer to the OSP layer dt_object
@@ -1246,54 +1146,72 @@ int osp_declare_xattr_set(const struct lu_env *env, struct dt_object *dt,
  * \param[in] fl       to indicate the detailed set operation: LU_XATTR_CREATE
  *                     or LU_XATTR_REPLACE or others
  * \param[in] th       pointer to the transaction handler
- * \param[in] capa     the capability for this operation
  *
  * \retval             0 for success
  * \retval             negative error number on failure
  */
 int osp_xattr_set(const struct lu_env *env, struct dt_object *dt,
                  const struct lu_buf *buf, const char *name, int fl,
-                 struct thandle *th, struct lustre_capa *capa)
+                 struct thandle *th)
 {
-       int rc = 0;
+       struct osp_object       *o = dt2osp_obj(dt);
+       struct osp_device       *osp = lu2osp_dev(dt->do_lu.lo_dev);
+       struct osp_update_request *update;
+       struct osp_xattr_entry  *oxe;
+       int                     rc;
+       ENTRY;
 
-       if (is_only_remote_trans(th)) {
-               rc = __osp_xattr_set(env, dt, buf, name, fl, th);
+       update = thandle_to_osp_update_request(th);
+       LASSERT(update != NULL);
 
-               CDEBUG(D_INFO, "xattr %s set object "DFID": rc = %d\n",
-                      name, PFID(&dt->do_lu.lo_header->loh_fid), rc);
-       }
+       CDEBUG(D_INODE, DFID" set xattr '%s' with size %zd\n",
+              PFID(lu_object_fid(&dt->do_lu)), name, buf->lb_len);
 
-       return rc;
-}
+       rc = osp_update_rpc_pack(env, xattr_set, update, OUT_XATTR_SET,
+                                lu_object_fid(&dt->do_lu), buf, name, fl);
+       if (rc != 0 || o->opo_ooa == NULL || osp->opd_connect_mdt)
+               RETURN(rc);
 
-static int __osp_xattr_del(const struct lu_env *env, struct dt_object *dt,
-                          const char *name, struct thandle *th)
-{
-       struct dt_update_request *update;
-       const struct lu_fid      *fid;
-       struct osp_object        *o     = dt2osp_obj(dt);
-       struct osp_xattr_entry   *oxe;
-       int                       rc;
+       oxe = osp_oac_xattr_find_or_add(o, name, buf->lb_len);
+       if (oxe == NULL) {
+               CWARN("%s: cannot cache xattr '%s' of "DFID"\n",
+                     dt->do_lu.lo_dev->ld_obd->obd_name,
+                     name, PFID(lu_object_fid(&dt->do_lu)));
 
-       update = dt_update_request_find_or_create(th, dt);
-       if (IS_ERR(update))
-               return PTR_ERR(update);
+               RETURN(0);
+       }
 
-       fid = lu_object_fid(&dt->do_lu);
+       if (oxe->oxe_buflen - oxe->oxe_namelen - 1 < buf->lb_len) {
+               struct osp_xattr_entry *old = oxe;
+               struct osp_xattr_entry *tmp;
 
-       rc = out_xattr_del_pack(env, &update->dur_buf, fid, name,
-                               update->dur_batchid);
+               tmp = osp_oac_xattr_replace(o, &old, buf->lb_len);
+               osp_oac_xattr_put(oxe);
+               oxe = tmp;
+               if (tmp == NULL) {
+                       CWARN("%s: cannot update cached xattr '%s' of "DFID"\n",
+                             dt->do_lu.lo_dev->ld_obd->obd_name,
+                             name, PFID(lu_object_fid(&dt->do_lu)));
+                       spin_lock(&o->opo_lock);
+                       old->oxe_ready = 0;
+                       spin_unlock(&o->opo_lock);
 
-       if (rc != 0 || o->opo_ooa == NULL)
-               return rc;
+                       RETURN(0);
+               }
 
-       oxe = osp_oac_xattr_find(o, name, true);
-       if (oxe != NULL)
                /* Drop the ref for entry on list. */
-               osp_oac_xattr_put(oxe);
+               osp_oac_xattr_put(old);
+       }
 
-       return 0;
+       spin_lock(&o->opo_lock);
+       oxe->oxe_vallen = buf->lb_len;
+       memcpy(oxe->oxe_value, buf->lb_buf, buf->lb_len);
+       oxe->oxe_exist = 1;
+       oxe->oxe_ready = 1;
+       spin_unlock(&o->opo_lock);
+       osp_oac_xattr_put(oxe);
+
+       RETURN(0);
 }
 
 /**
@@ -1318,16 +1236,7 @@ static int __osp_xattr_del(const struct lu_env *env, struct dt_object *dt,
 int osp_declare_xattr_del(const struct lu_env *env, struct dt_object *dt,
                          const char *name, struct thandle *th)
 {
-       int rc = 0;
-
-       if (!is_only_remote_trans(th)) {
-               rc = __osp_xattr_del(env, dt, name, th);
-
-               CDEBUG(D_INFO, "declare xattr %s del object "DFID": rc = %d\n",
-                      name, PFID(&dt->do_lu.lo_header->loh_fid), rc);
-       }
-
-       return rc;
+       return osp_trans_update_request_create(th);
 }
 
 /**
@@ -1344,25 +1253,33 @@ int osp_declare_xattr_del(const struct lu_env *env, struct dt_object *dt,
  * \param[in] dt       pointer to the OSP layer dt_object
  * \param[in] name     the name of the extended attribute to be set
  * \param[in] th       pointer to the transaction handler
- * \param[in] capa     the capability for this operation
  *
  * \retval             0 for success
  * \retval             negative error number on failure
  */
 int osp_xattr_del(const struct lu_env *env, struct dt_object *dt,
-                 const char *name, struct thandle *th,
-                 struct lustre_capa *capa)
+                 const char *name, struct thandle *th)
 {
-       int rc = 0;
+       struct osp_update_request *update;
+       const struct lu_fid      *fid = lu_object_fid(&dt->do_lu);
+       struct osp_object        *o     = dt2osp_obj(dt);
+       struct osp_xattr_entry   *oxe;
+       int                       rc;
 
-       if (is_only_remote_trans(th)) {
-               rc = __osp_xattr_del(env, dt, name, th);
+       update = thandle_to_osp_update_request(th);
+       LASSERT(update != NULL);
 
-               CDEBUG(D_INFO, "xattr %s del object "DFID": rc = %d\n",
-                      name, PFID(&dt->do_lu.lo_header->loh_fid), rc);
-       }
+       rc = osp_update_rpc_pack(env, xattr_del, update, OUT_XATTR_DEL,
+                                fid, name);
+       if (rc != 0 || o->opo_ooa == NULL)
+               return rc;
 
-       return rc;
+       oxe = osp_oac_xattr_find(o, name, true);
+       if (oxe != NULL)
+               /* Drop the ref for entry on list. */
+               osp_oac_xattr_put(oxe);
+
+       return 0;
 }
 
 /**
@@ -1400,6 +1317,7 @@ static int osp_declare_object_create(const struct lu_env *env,
        struct osp_device       *d = lu2osp_dev(dt->do_lu.lo_dev);
        struct osp_object       *o = dt2osp_obj(dt);
        const struct lu_fid     *fid = lu_object_fid(&dt->do_lu);
+       struct thandle          *local_th;
        int                      rc = 0;
 
        ENTRY;
@@ -1428,13 +1346,19 @@ static int osp_declare_object_create(const struct lu_env *env,
         */
        /* rc = osp_sync_declare_add(env, o, MDS_UNLINK64_REC, th); */
 
+       local_th = osp_get_storage_thandle(env, th, d);
+       if (IS_ERR(local_th))
+               RETURN(PTR_ERR(local_th));
+
        if (unlikely(!fid_is_zero(fid))) {
                /* replay case: caller knows fid */
                osi->osi_off = sizeof(osi->osi_id) * d->opd_index;
                osi->osi_lb.lb_len = sizeof(osi->osi_id);
                osi->osi_lb.lb_buf = NULL;
+
                rc = dt_declare_record_write(env, d->opd_last_used_oid_file,
-                                            &osi->osi_lb, osi->osi_off, th);
+                                            &osi->osi_lb, osi->osi_off,
+                                            local_th);
                RETURN(rc);
        }
 
@@ -1459,7 +1383,8 @@ static int osp_declare_object_create(const struct lu_env *env,
                osi->osi_lb.lb_len = sizeof(osi->osi_id);
                osi->osi_lb.lb_buf = NULL;
                rc = dt_declare_record_write(env, d->opd_last_used_oid_file,
-                                            &osi->osi_lb, osi->osi_off, th);
+                                            &osi->osi_lb, osi->osi_off,
+                                            local_th);
        } else {
                /* not needed in the cache anymore */
                set_bit(LU_OBJECT_HEARD_BANSHEE,
@@ -1499,6 +1424,7 @@ static int osp_object_create(const struct lu_env *env, struct dt_object *dt,
        struct osp_object       *o = dt2osp_obj(dt);
        int                     rc = 0;
        struct lu_fid           *fid = &osi->osi_fid;
+       struct thandle          *local_th;
        ENTRY;
 
        if (is_only_remote_trans(th) &&
@@ -1541,6 +1467,9 @@ static int osp_object_create(const struct lu_env *env, struct dt_object *dt,
        if (osp_precreate_end_seq(env, d) && osp_is_fid_client(d))
                th->th_sync = 1;
 
+       local_th = osp_get_storage_thandle(env, th, d);
+       if (IS_ERR(local_th))
+               RETURN(PTR_ERR(local_th));
        /*
         * it's OK if the import is inactive by this moment - id was created
         * by OST earlier, we just need to maintain it consistently on the disk
@@ -1575,7 +1504,7 @@ static int osp_object_create(const struct lu_env *env, struct dt_object *dt,
                           &d->opd_last_used_fid.f_oid, d->opd_index);
 
        rc = dt_record_write(env, d->opd_last_used_oid_file, &osi->osi_lb,
-                            &osi->osi_off, th);
+                            &osi->osi_off, local_th);
 
        CDEBUG(D_HA, "%s: Wrote last used FID: "DFID", index %d: %d\n",
               d->opd_obd->obd_name, PFID(fid), d->opd_index, rc);
@@ -1602,13 +1531,12 @@ int osp_declare_object_destroy(const struct lu_env *env,
                               struct dt_object *dt, struct thandle *th)
 {
        struct osp_object       *o = dt2osp_obj(dt);
+       struct osp_device       *osp = lu2osp_dev(dt->do_lu.lo_dev);
        int                      rc = 0;
 
        ENTRY;
 
-       /*
-        * track objects to be destroyed via llog
-        */
+       LASSERT(!osp->opd_connect_mdt);
        rc = osp_sync_declare_add(env, o, MDS_UNLINK64_REC, th);
 
        RETURN(rc);
@@ -1631,20 +1559,23 @@ int osp_declare_object_destroy(const struct lu_env *env,
  * \retval             0 for success
  * \retval             negative error number on failure
  */
-int osp_object_destroy(const struct lu_env *env, struct dt_object *dt,
-                      struct thandle *th)
+static int osp_object_destroy(const struct lu_env *env, struct dt_object *dt,
+                             struct thandle *th)
 {
        struct osp_object       *o = dt2osp_obj(dt);
+       struct osp_device       *osp = lu2osp_dev(dt->do_lu.lo_dev);
        int                      rc = 0;
 
        ENTRY;
 
        o->opo_non_exist = 1;
-       /*
-        * once transaction is committed put proper command on
-        * the queue going to our OST
-        */
+
+       LASSERT(!osp->opd_connect_mdt);
+       /* once transaction is committed put proper command on
+        * the queue going to our OST. */
        rc = osp_sync_add(env, o, MDS_UNLINK64_REC, th, NULL);
+       if (rc < 0)
+               RETURN(rc);
 
        /* not needed in cache any more */
        set_bit(LU_OBJECT_HEARD_BANSHEE, &dt->do_lu.lo_header->loh_flags);
@@ -1655,8 +1586,7 @@ int osp_object_destroy(const struct lu_env *env, struct dt_object *dt,
 static int osp_orphan_index_lookup(const struct lu_env *env,
                                   struct dt_object *dt,
                                   struct dt_rec *rec,
-                                  const struct dt_key *key,
-                                  struct lustre_capa *capa)
+                                  const struct dt_key *key)
 {
        return -EOPNOTSUPP;
 }
@@ -1675,7 +1605,6 @@ static int osp_orphan_index_insert(const struct lu_env *env,
                                   const struct dt_rec *rec,
                                   const struct dt_key *key,
                                   struct thandle *handle,
-                                  struct lustre_capa *capa,
                                   int ignore_quota)
 {
        return -EOPNOTSUPP;
@@ -1692,8 +1621,7 @@ static int osp_orphan_index_declare_delete(const struct lu_env *env,
 static int osp_orphan_index_delete(const struct lu_env *env,
                                   struct dt_object *dt,
                                   const struct dt_key *key,
-                                  struct thandle *handle,
-                                  struct lustre_capa *capa)
+                                  struct thandle *handle)
 {
        return -EOPNOTSUPP;
 }
@@ -1704,13 +1632,12 @@ static int osp_orphan_index_delete(const struct lu_env *env,
  * \param[in] env      pointer to the thread context
  * \param[in] dt       pointer to the index object to be iterated
  * \param[in] attr     unused
- * \param[in] capa     the capability for this operation
  *
  * \retval             pointer to the iteration structure
  * \retval             negative error number on failure
  */
 struct dt_it *osp_it_init(const struct lu_env *env, struct dt_object *dt,
-                         __u32 attr, struct lustre_capa *capa)
+                         __u32 attr)
 {
        struct osp_it *it;
 
@@ -1757,7 +1684,7 @@ void osp_it_fini(const struct lu_env *env, struct dt_it *di)
  * Get more records for the iteration from peer.
  *
  * The new records will be filled in an array of pages. The OSP side
- * allows 1MB bulk data to be transfered.
+ * allows 1MB bulk data to be transferred.
  *
  * \param[in] env      pointer to the thread context
  * \param[in] it       pointer to the iteration structure
@@ -1770,6 +1697,7 @@ static int osp_it_fetch(const struct lu_env *env, struct osp_it *it)
        struct lu_device         *dev   = it->ooi_obj->do_lu.lo_dev;
        struct osp_device        *osp   = lu2osp_dev(dev);
        struct page             **pages;
+       struct lu_device *top_device;
        struct ptlrpc_request    *req   = NULL;
        struct ptlrpc_bulk_desc  *desc;
        struct idx_info          *ii;
@@ -1805,6 +1733,13 @@ static int osp_it_fetch(const struct lu_env *env, struct osp_it *it)
                RETURN(rc);
        }
 
+       /* Let's allow this request during recovery, otherwise
+        * if the remote target is also in recovery status,
+        * it might cause deadlock */
+       top_device = dev->ld_site->ls_top_dev;
+       if (top_device->ld_obd->obd_recovering)
+               req->rq_allow_replay = 1;
+
        req->rq_request_portal = OUT_PORTAL;
        ii = req_capsule_client_get(&req->rq_pill, &RMF_IDX_INFO);
        memset(ii, 0, sizeof(*ii));
@@ -1829,15 +1764,18 @@ static int osp_it_fetch(const struct lu_env *env, struct osp_it *it)
 
        ptlrpc_at_set_req_timeout(req);
 
-       desc = ptlrpc_prep_bulk_imp(req, npages, 1, BULK_PUT_SINK,
-                                   MDS_BULK_PORTAL);
+       desc = ptlrpc_prep_bulk_imp(req, npages, 1,
+                                   PTLRPC_BULK_PUT_SINK | PTLRPC_BULK_BUF_KIOV,
+                                   MDS_BULK_PORTAL,
+                                   &ptlrpc_bulk_kiov_pin_ops);
        if (desc == NULL) {
                ptlrpc_request_free(req);
                RETURN(-ENOMEM);
        }
 
        for (i = 0; i < npages; i++)
-               ptlrpc_prep_bulk_page_pin(desc, pages[i], 0, PAGE_CACHE_SIZE);
+               desc->bd_frag_ops->add_kiov_frag(desc, pages[i], 0,
+                                                PAGE_CACHE_SIZE);
 
        ptlrpc_request_set_replen(req);
        rc = ptlrpc_queue_wait(req);
@@ -2203,14 +2141,18 @@ static int osp_object_init(const struct lu_env *env, struct lu_object *o,
 
                po->opo_obj.do_ops = &osp_md_obj_ops;
                po->opo_obj.do_body_ops = &osp_md_body_ops;
-               rc = po->opo_obj.do_ops->do_attr_get(env, lu2dt_obj(o),
-                                                    la, NULL);
-               if (rc == 0)
-                       o->lo_header->loh_attr |=
-                               LOHA_EXISTS | (la->la_mode & S_IFMT);
-               if (rc == -ENOENT) {
+               if (conf != NULL && conf->loc_flags & LOC_F_NEW) {
                        po->opo_non_exist = 1;
-                       rc = 0;
+               } else {
+                       rc = po->opo_obj.do_ops->do_attr_get(env, lu2dt_obj(o),
+                                                            la);
+                       if (rc == 0)
+                               o->lo_header->loh_attr |=
+                                       LOHA_EXISTS | (la->la_mode & S_IFMT);
+                       if (rc == -ENOENT) {
+                               po->opo_non_exist = 1;
+                               rc = 0;
+                       }
                }
                init_rwsem(&po->opo_sem);
        }