LU-3534 osp: send updates by separate thread

[fs/lustre-release.git] / lustre / osp / osp_object.c

diff --git a/lustre/osp/osp_object.c b/lustre/osp/osp_object.c
index 2898da6..ff6107e 100644 (file)
--- a/lustre/osp/osp_object.c
+++ b/lustre/osp/osp_object.c
@@ -168,7 +168,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 +319,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.
@@ -541,7 +541,7 @@ int osp_attr_get(const struct lu_env *env, struct dt_object *dt,
        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;
@@ -561,12 +561,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,
@@ -607,7 +607,7 @@ out:
        if (req != NULL)
                ptlrpc_req_finished(req);
 
-       dt_update_request_destroy(update);
+       osp_update_request_destroy(update);
 
        return rc;
 }
@@ -857,7 +857,7 @@ static int osp_declare_xattr_get(const struct lu_env *env, struct dt_object *dt,
                mutex_unlock(&osp->opd_async_requests_mutex);
                osp_oac_xattr_put(oxe);
        } else {
-               struct dt_update_request *update;
+               struct osp_update_request *update;
 
                /* XXX: Currently, we trigger the batched async OUT
                 *      RPC via dt_declare_xattr_get(). It is not
@@ -865,8 +865,8 @@ static int osp_declare_xattr_get(const struct lu_env *env, struct dt_object *dt,
                 *
                 *      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) {
+               if (update != NULL && update->our_req != NULL &&
+                   update->our_req->ourq_count > 0) {
                        osp->opd_async_requests = NULL;
                        mutex_unlock(&osp->opd_async_requests_mutex);
                        rc = osp_unplug_async_request(env, osp, update);
@@ -908,7 +908,7 @@ int osp_xattr_get(const struct lu_env *env, struct dt_object *dt,
        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;
@@ -959,12 +959,12 @@ unlock:
                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);
        if (rc != 0) {
                CERROR("%s: Insert update error "DFID": rc = %d\n",
                       dname, PFID(lu_object_fid(&dt->do_lu)), rc);
@@ -1077,7 +1077,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);
@@ -1141,23 +1141,21 @@ int osp_xattr_set(const struct lu_env *env, struct dt_object *dt,
                  struct thandle *th)
 {
        struct osp_object       *o = dt2osp_obj(dt);
-       struct dt_update_request *update;
+       struct osp_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);
+       update = thandle_to_osp_update_request(th);
        LASSERT(update != NULL);
 
        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);
+       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)
-               return rc;
+               RETURN(rc);
 
        oxe = osp_oac_xattr_find_or_add(o, name, buf->lb_len);
        if (oxe == NULL) {
@@ -1247,17 +1245,17 @@ 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)
 {
-       struct dt_update_request *update;
+       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_object        *o     = dt2osp_obj(dt);
        struct osp_xattr_entry   *oxe;
        int                       rc;
 
-       update = thandle_to_dt_update_request(th);
+       update = thandle_to_osp_update_request(th);
        LASSERT(update != NULL);
 
-       rc = out_xattr_del_pack(env, &update->dur_buf, fid, name,
-                               update->dur_batchid);
+       rc = osp_update_rpc_pack(env, xattr_del, update, OUT_XATTR_DEL,
+                                fid, name);
        if (rc != 0 || o->opo_ooa == NULL)
                return rc;
 
@@ -1554,6 +1552,7 @@ static int osp_object_destroy(const struct lu_env *env, struct dt_object *dt,
        int                      rc = 0;
 
        ENTRY;
+
        o->opo_non_exist = 1;
 
        LASSERT(!osp->opd_connect_mdt);
@@ -1670,7 +1669,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