LU-15727 lod: honor append_pool with default composite layouts

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

diff --git a/lustre/osp/osp_md_object.c b/lustre/osp/osp_md_object.c
index 37ce566..6ff56f4 100644 (file)
--- a/lustre/osp/osp_md_object.c
+++ b/lustre/osp/osp_md_object.c
@@ -164,7 +164,12 @@ int osp_md_create(const struct lu_env *env, struct dt_object *dt,
        update = thandle_to_osp_update_request(th);
        LASSERT(update != NULL);
 
-       LASSERT(attr->la_valid & LA_TYPE);
+       if (!(attr->la_valid & LA_TYPE)) {
+               CERROR("%s: create type not specified: valid %llx\n",
+                      dt->do_lu.lo_dev->ld_obd->obd_name, attr->la_valid);
+               GOTO(out, rc = -EINVAL);
+       }
+
        rc = OSP_UPDATE_RPC_PACK(env, out_create_pack, update,
                                 lu_object_fid(&dt->do_lu), attr, hint, dof);
        if (rc != 0)
@@ -300,7 +305,6 @@ static void osp_md_ah_init(const struct lu_env *env,
        LASSERT(ah);
 
        ah->dah_parent = parent;
-       ah->dah_mode = child_mode;
 }
 
 /**
@@ -513,7 +517,7 @@ static int osp_md_index_lookup(const struct lu_env *env, struct dt_object *dt,
        }
 
        fid = lbuf->lb_buf;
-       if (ptlrpc_rep_need_swab(req))
+       if (req_capsule_rep_need_swab(&req->rq_pill))
                lustre_swab_lu_fid(fid);
        if (!fid_is_sane(fid)) {
                CERROR("%s: lookup "DFID" %s invalid fid "DFID"\n",
@@ -1019,6 +1023,15 @@ int osp_md_declare_destroy(const struct lu_env *env, struct dt_object *dt,
        return osp_trans_update_request_create(th);
 }
 
+static int osp_destroy_interpreter(const struct lu_env *env,
+                                  struct object_update_reply *reply,
+                                  struct ptlrpc_request *req,
+                                  struct osp_object *obj,
+                                  void *data, int index, int rc)
+{
+       return 0;
+}
+
 /**
  * Implement OSP layer dt_object_operations::do_destroy() interface.
  *
@@ -1041,9 +1054,10 @@ int osp_md_destroy(const struct lu_env *env, struct dt_object *dt,
        struct osp_device *osp = lu2osp_dev(dt->do_lu.lo_dev);
        struct osp_update_request *update;
        int rc = 0;
-
        ENTRY;
+
        o->opo_non_exist = 1;
+       o->opo_destroyed = 1;
 
        LASSERT(osp->opd_connect_mdt);
        update = thandle_to_osp_update_request(th);
@@ -1054,6 +1068,12 @@ int osp_md_destroy(const struct lu_env *env, struct dt_object *dt,
        if (rc != 0)
                RETURN(rc);
 
+       /* retain the object and it's status until it's destroyed on remote */
+       rc = osp_insert_update_callback(env, update, o, NULL,
+                                       osp_destroy_interpreter);
+       if (rc != 0)
+               RETURN(rc);
+
        set_bit(LU_OBJECT_HEARD_BANSHEE, &dt->do_lu.lo_header->loh_flags);
        rc = osp_insert_update_callback(env, update, dt2osp_obj(dt), NULL,
                                        NULL);
@@ -1061,7 +1081,7 @@ int osp_md_destroy(const struct lu_env *env, struct dt_object *dt,
        RETURN(rc);
 }
 
-struct dt_object_operations osp_md_obj_ops = {
+const struct dt_object_operations osp_md_obj_ops = {
        .do_read_lock         = osp_md_read_lock,
        .do_write_lock        = osp_md_write_lock,
        .do_read_unlock       = osp_md_read_unlock,
@@ -1089,6 +1109,7 @@ struct dt_object_operations osp_md_obj_ops = {
        .do_object_lock       = osp_md_object_lock,
        .do_object_unlock     = osp_md_object_unlock,
        .do_invalidate        = osp_invalidate,
+       .do_check_stale       = osp_check_stale,
 };
 
 /**
@@ -1114,6 +1135,9 @@ static ssize_t osp_md_declare_write(const struct lu_env *env,
        struct osp_device *osp = dt2osp_dev(th->th_dev);
        int rc;
 
+       if (dt2osp_obj(dt)->opo_destroyed)
+               return -ENOENT;
+
        rc = osp_trans_update_request_create(th);
        if (rc != 0)
                return rc;
@@ -1127,6 +1151,24 @@ static ssize_t osp_md_declare_write(const struct lu_env *env,
        return 0;
 }
 
+static int osp_write_interpreter(const struct lu_env *env,
+                                 struct object_update_reply *reply,
+                                 struct ptlrpc_request *req,
+                                 struct osp_object *obj,
+                                 void *data, int index, int rc)
+{
+       if (rc) {
+               CDEBUG(D_HA, "error "DFID": rc = %d\n",
+                      PFID(lu_object_fid(&obj->opo_obj.do_lu)), rc);
+               OBD_RACE(OBD_FAIL_OUT_OBJECT_MISS);
+               spin_lock(&obj->opo_lock);
+               obj->opo_attr.la_valid = 0;
+               obj->opo_stale = 1;
+               spin_unlock(&obj->opo_lock);
+       }
+       return 0;
+}
+
 /**
  * Implementation of dt_body_operations::dbo_write
  *
@@ -1152,6 +1194,9 @@ static ssize_t osp_md_write(const struct lu_env *env, struct dt_object *dt,
        ssize_t                   rc;
        ENTRY;
 
+       if (obj->opo_destroyed)
+               RETURN(-ENOENT);
+
        update = thandle_to_osp_update_request(th);
        LASSERT(update != NULL);
 
@@ -1167,6 +1212,12 @@ static ssize_t osp_md_write(const struct lu_env *env, struct dt_object *dt,
        if (rc < 0)
                RETURN(rc);
 
+       /* to be able to invalidate object's state in case of an error */
+       rc = osp_insert_update_callback(env, update, obj, NULL,
+                       osp_write_interpreter);
+       if (rc < 0)
+               RETURN(rc);
+
        /* XXX: how about the write error happened later? */
        *pos += buf->lb_len;
 
@@ -1211,6 +1262,9 @@ static ssize_t osp_md_read(const struct lu_env *env, struct dt_object *dt,
        int rc;
        ENTRY;
 
+       if (dt2osp_obj(dt)->opo_destroyed)
+               RETURN(-ENOENT);
+
        /* Because it needs send the update buffer right away,
         * just create an update buffer, instead of attaching the
         * update_remote list of the thandle.  */
@@ -1240,7 +1294,7 @@ static ssize_t osp_md_read(const struct lu_env *env, struct dt_object *dt,
 
        /* allocate bulk descriptor */
        desc = ptlrpc_prep_bulk_imp(req, pages, 1,
-                                   PTLRPC_BULK_PUT_SINK | PTLRPC_BULK_BUF_KIOV,
+                                   PTLRPC_BULK_PUT_SINK,
                                    MDS_BULK_PORTAL,
                                    &ptlrpc_bulk_kiov_nopin_ops);
        if (desc == NULL)
@@ -1292,7 +1346,7 @@ out_update:
 }
 
 /* These body operation will be used to write symlinks during migration etc */
-struct dt_body_operations osp_md_body_ops = {
+const struct dt_body_operations osp_md_body_ops = {
        .dbo_declare_write      = osp_md_declare_write,
        .dbo_write              = osp_md_write,
        .dbo_read               = osp_md_read,