LU-13195 osp: invalidate object on write error

[fs/lustre-release.git] / lustre / target / out_lib.c

diff --git a/lustre/target/out_lib.c b/lustre/target/out_lib.c
index c486f3f..63a1eb9 100644 (file)
--- a/lustre/target/out_lib.c
+++ b/lustre/target/out_lib.c
@@ -20,7 +20,7 @@
  * GPL HEADER END
  */
 /*
- * Copyright (c) 2014, 2015, Intel Corporation.
+ * Copyright (c) 2014, 2017, Intel Corporation.
  */
 /*
  * lustre/target/out_lib.c
@@ -53,6 +53,7 @@ const char *update_op_str(__u16 opc)
                [OUT_ATTR_GET] = "attr_get",
                [OUT_XATTR_SET] = "xattr_set",
                [OUT_XATTR_GET] = "xattr_get",
+               [OUT_XATTR_LIST] = "xattr_list",
                [OUT_INDEX_LOOKUP] = "lookup",
                [OUT_INDEX_INSERT] = "insert",
                [OUT_INDEX_DELETE] = "delete",
@@ -102,7 +103,7 @@ int out_update_header_pack(const struct lu_env *env,
        unsigned int                    i;
        size_t                          update_size;
 
-       if (((reply_size + 7) >> 3) >= 1ULL << 16)
+       if (reply_size  >= LNET_MTU)
                return -EINVAL;
 
        /* Check whether the packing exceeding the maxima update length */
@@ -404,6 +405,15 @@ int out_xattr_get_pack(const struct lu_env *env, struct object_update *update,
 }
 EXPORT_SYMBOL(out_xattr_get_pack);
 
+int out_xattr_list_pack(const struct lu_env *env, struct object_update *update,
+                      size_t *max_update_size, const struct lu_fid *fid,
+                      const int bufsize)
+{
+       return out_update_pack(env, update, max_update_size, OUT_XATTR_LIST,
+                              fid, 0, NULL, NULL, bufsize);
+}
+EXPORT_SYMBOL(out_xattr_list_pack);
+
 int out_read_pack(const struct lu_env *env, struct object_update *update,
                  size_t *max_update_size, const struct lu_fid *fid,
                  size_t size, loff_t pos)
@@ -429,7 +439,7 @@ static int tx_extend_args(struct thandle_exec_args *ta, int new_alloc_ta)
        if (ta->ta_alloc_args >= new_alloc_ta)
                return 0;
 
-       OBD_ALLOC(new_ta, sizeof(*new_ta) * new_alloc_ta);
+       OBD_ALLOC_PTR_ARRAY(new_ta, new_alloc_ta);
        if (new_ta == NULL)
                return -ENOMEM;
 
@@ -446,8 +456,7 @@ static int tx_extend_args(struct thandle_exec_args *ta, int new_alloc_ta)
 
        /* free the old args */
        if (ta->ta_args != NULL)
-               OBD_FREE(ta->ta_args, sizeof(ta->ta_args[0]) *
-                                     ta->ta_alloc_args);
+               OBD_FREE_PTR_ARRAY(ta->ta_args, ta->ta_alloc_args);
 
        ta->ta_args = new_ta;
        ta->ta_alloc_args = new_alloc_ta;
@@ -457,7 +466,7 @@ out:
                        if (new_ta[i] != NULL)
                                OBD_FREE_PTR(new_ta[i]);
                }
-               OBD_FREE(new_ta, sizeof(*new_ta) * new_alloc_ta);
+               OBD_FREE_PTR_ARRAY(new_ta, new_alloc_ta);
        }
        return rc;
 }
@@ -499,7 +508,7 @@ static int out_obj_destroy(const struct lu_env *env, struct dt_object *dt_obj,
        CDEBUG(D_INFO, "%s: destroy "DFID"\n", dt_obd_name(th->th_dev),
               PFID(lu_object_fid(&dt_obj->do_lu)));
 
-       dt_write_lock(env, dt_obj, MOR_TGT_CHILD);
+       dt_write_lock(env, dt_obj, DT_TGT_CHILD);
        rc = dt_destroy(env, dt_obj, th);
        dt_write_unlock(env, dt_obj);
 
@@ -537,7 +546,7 @@ int out_tx_create_exec(const struct lu_env *env, struct thandle *th,
               arg->u.create.dof.dof_type,
               arg->u.create.attr.la_mode & S_IFMT);
 
-       dt_write_lock(env, dt_obj, MOR_TGT_CHILD);
+       dt_write_lock(env, dt_obj, DT_TGT_CHILD);
        rc = dt_create(env, dt_obj, &arg->u.create.attr,
                       &arg->u.create.hint, &arg->u.create.dof, th);
 
@@ -630,7 +639,7 @@ static int out_tx_attr_set_exec(const struct lu_env *env, struct thandle *th,
        CDEBUG(D_OTHER, "%s: attr set "DFID"\n", dt_obd_name(th->th_dev),
               PFID(lu_object_fid(&dt_obj->do_lu)));
 
-       dt_write_lock(env, dt_obj, MOR_TGT_CHILD);
+       dt_write_lock(env, dt_obj, DT_TGT_CHILD);
        rc = dt_attr_set(env, dt_obj, &arg->u.attr_set.attr, th);
        dt_write_unlock(env, dt_obj);
 
@@ -657,6 +666,10 @@ int out_attr_set_add_exec(const struct lu_env *env, struct dt_object *dt_obj,
        if (rc != 0)
                return rc;
 
+       if (attr->la_valid & LA_FLAGS &&
+           attr->la_flags & LUSTRE_SET_SYNC_FL)
+               th->th_sync |= 1;
+
        arg = tx_add_exec(ta, out_tx_attr_set_exec, out_tx_attr_set_undo,
                          file, line);
        if (IS_ERR(arg))
@@ -683,7 +696,7 @@ static int out_tx_write_exec(const struct lu_env *env, struct thandle *th,
        if (OBD_FAIL_CHECK(OBD_FAIL_OUT_ENOSPC)) {
                rc = -ENOSPC;
        } else {
-               dt_write_lock(env, dt_obj, MOR_TGT_CHILD);
+               dt_write_lock(env, dt_obj, DT_TGT_CHILD);
                rc = dt_record_write(env, dt_obj, &arg->u.write.buf,
                                     &arg->u.write.pos, th);
                dt_write_unlock(env, dt_obj);
@@ -694,7 +707,7 @@ static int out_tx_write_exec(const struct lu_env *env, struct thandle *th,
 
        if (arg->reply != NULL)
                object_update_result_insert(arg->reply, NULL, 0, arg->index,
-                                           rc);
+                                           rc < 0 ? rc : 0);
 
        return rc > 0 ? 0 : rc;
 }
@@ -806,7 +819,7 @@ static int out_tx_xattr_set_exec(const struct lu_env *env,
                        linkea = false;
                }
 
-               dt_write_lock(env, dt_obj, MOR_TGT_CHILD);
+               dt_write_lock(env, dt_obj, DT_TGT_CHILD);
 
 again:
                rc = dt_xattr_set(env, dt_obj, ldata.ld_buf,
@@ -879,7 +892,7 @@ static int out_tx_xattr_del_exec(const struct lu_env *env, struct thandle *th,
        if (!lu_object_exists(&dt_obj->do_lu))
                GOTO(out, rc = -ENOENT);
 
-       dt_write_lock(env, dt_obj, MOR_TGT_CHILD);
+       dt_write_lock(env, dt_obj, DT_TGT_CHILD);
        rc = dt_xattr_del(env, dt_obj, arg->u.xattr_set.name,
                          th);
        dt_write_unlock(env, dt_obj);
@@ -925,7 +938,7 @@ static int out_obj_ref_add(const struct lu_env *env,
 {
        int rc;
 
-       dt_write_lock(env, dt_obj, MOR_TGT_CHILD);
+       dt_write_lock(env, dt_obj, DT_TGT_CHILD);
        rc = dt_ref_add(env, dt_obj, th);
        dt_write_unlock(env, dt_obj);
 
@@ -938,7 +951,7 @@ static int out_obj_ref_del(const struct lu_env *env,
 {
        int rc;
 
-       dt_write_lock(env, dt_obj, MOR_TGT_CHILD);
+       dt_write_lock(env, dt_obj, DT_TGT_CHILD);
        rc = dt_ref_del(env, dt_obj, th);
        dt_write_unlock(env, dt_obj);
 
@@ -1058,8 +1071,8 @@ static int out_obj_index_insert(const struct lu_env *env,
        if (dt_try_as_dir(env, dt_obj) == 0)
                return -ENOTDIR;
 
-       dt_write_lock(env, dt_obj, MOR_TGT_CHILD);
-       rc = dt_insert(env, dt_obj, rec, key, th, 0);
+       dt_write_lock(env, dt_obj, DT_TGT_CHILD);
+       rc = dt_insert(env, dt_obj, rec, key, th);
        dt_write_unlock(env, dt_obj);
 
        return rc;
@@ -1079,7 +1092,7 @@ static int out_obj_index_delete(const struct lu_env *env,
        if (dt_try_as_dir(env, dt_obj) == 0)
                return -ENOTDIR;
 
-       dt_write_lock(env, dt_obj, MOR_TGT_CHILD);
+       dt_write_lock(env, dt_obj, DT_TGT_CHILD);
        rc = dt_delete(env, dt_obj, key, th);
        dt_write_unlock(env, dt_obj);