Whamcloud - gitweb
LU-3534 osp: move RPC pack from declare to execution phase
[fs/lustre-release.git] / lustre / osp / osp_object.c
index c984151..2898da6 100644 (file)
@@ -575,7 +575,7 @@ int osp_attr_get(const struct lu_env *env, struct dt_object *dt,
                GOTO(out, rc);
        }
 
-       rc = osp_remote_sync(env, osp, update, &req, false);
+       rc = osp_remote_sync(env, osp, update, &req);
        if (rc != 0) {
                if (rc == -ENOENT) {
                        osp2lu_obj(obj)->lo_header->loh_attr &= ~LOHA_EXISTS;
@@ -612,15 +612,29 @@ out:
        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
@@ -649,71 +663,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;
 }
 
 /**
@@ -744,24 +704,39 @@ static int osp_attr_set(const struct lu_env *env, struct dt_object *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);
 }
@@ -996,7 +971,7 @@ unlock:
                GOTO(out, rc);
        }
 
-       rc = osp_remote_sync(env, osp, update, &req, false);
+       rc = osp_remote_sync(env, osp, update, &req);
        if (rc != 0) {
                if (rc == -ENOENT) {
                        dt->do_lu.lo_header->loh_attr &= ~LOHA_EXISTS;
@@ -1110,68 +1085,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 = thandle_to_dt_update_request(th);
-       LASSERT(update != NULL);
-
-       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.
  *
@@ -1199,16 +1112,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);
 }
 
 /**
@@ -1216,11 +1120,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
@@ -1237,44 +1140,65 @@ 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)
 {
-       int rc = 0;
-
-       if (is_only_remote_trans(th)) {
-               rc = __osp_xattr_set(env, dt, buf, name, fl, th);
-
-               CDEBUG(D_INFO, "xattr %s set object "DFID": rc = %d\n",
-                      name, PFID(&dt->do_lu.lo_header->loh_fid), rc);
-       }
-
-       return rc;
-}
-
-static int __osp_xattr_del(const struct lu_env *env, struct dt_object *dt,
-                          const char *name, struct thandle *th)
-{
+       struct osp_object       *o = dt2osp_obj(dt);
        struct dt_update_request *update;
-       const struct lu_fid      *fid;
-       struct osp_object        *o     = dt2osp_obj(dt);
-       struct osp_xattr_entry   *oxe;
-       int                       rc;
+       struct osp_xattr_entry  *oxe;
+       int                     rc;
+       ENTRY;
 
+       LASSERT(buf->lb_len > 0 && buf->lb_buf != NULL);
        update = thandle_to_dt_update_request(th);
        LASSERT(update != NULL);
 
-       fid = lu_object_fid(&dt->do_lu);
-
-       rc = out_xattr_del_pack(env, &update->dur_buf, fid, name,
-                               update->dur_batchid);
+       CDEBUG(D_INODE, DFID" set xattr '%s' with size %zd\n",
+              PFID(lu_object_fid(&dt->do_lu)), name, buf->lb_len);
 
+       rc = out_xattr_set_pack(env, &update->dur_buf,
+                               lu_object_fid(&dt->do_lu),
+                               buf, name, fl, update->dur_batchid);
        if (rc != 0 || o->opo_ooa == NULL)
                return rc;
 
-       oxe = osp_oac_xattr_find(o, name, true);
-       if (oxe != NULL)
-               /* Drop the ref for entry on list. */
+       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;
+                       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);
 }
 
 /**
@@ -1299,16 +1223,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);
 }
 
 /**
@@ -1332,16 +1247,26 @@ int osp_declare_xattr_del(const struct lu_env *env, struct dt_object *dt,
 int osp_xattr_del(const struct lu_env *env, struct dt_object *dt,
                  const char *name, struct thandle *th)
 {
-       int rc = 0;
+       struct dt_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_dt_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 = out_xattr_del_pack(env, &update->dur_buf, fid, name,
+                               update->dur_batchid);
+       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;
 }
 
 /**
@@ -1593,13 +1518,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);
@@ -1622,20 +1546,22 @@ 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);