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 e71484d..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 */
@@ -215,7 +216,6 @@ int out_create_pack(const struct lu_env *env, struct object_update *update,
 
        obdo->o_valid = 0;
        obdo_from_la(obdo, attr, attr->la_valid);
-       lustre_set_wire_obdo(NULL, obdo, obdo);
 
        if (parent_fid != NULL) {
                struct lu_fid *tmp;
@@ -267,7 +267,6 @@ int out_attr_set_pack(const struct lu_env *env, struct object_update *update,
 
        obdo->o_valid = 0;
        obdo_from_la(obdo, attr, attr->la_valid);
-       lustre_set_wire_obdo(NULL, obdo, obdo);
 
        RETURN(0);
 }
@@ -332,14 +331,13 @@ int out_index_delete_pack(const struct lu_env *env,
 }
 EXPORT_SYMBOL(out_index_delete_pack);
 
-int out_object_destroy_pack(const struct lu_env *env,
-                           struct object_update *update,
-                           size_t *max_update_size, const struct lu_fid *fid)
+int out_destroy_pack(const struct lu_env *env, struct object_update *update,
+                    size_t *max_update_size, const struct lu_fid *fid)
 {
        return out_update_pack(env, update, max_update_size, OUT_DESTROY, fid,
                               0, NULL, NULL, 0);
 }
-EXPORT_SYMBOL(out_object_destroy_pack);
+EXPORT_SYMBOL(out_destroy_pack);
 
 int out_write_pack(const struct lu_env *env, struct object_update *update,
                   size_t *max_update_size, const struct lu_fid *fid,
@@ -407,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)
@@ -432,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;
 
@@ -449,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;
@@ -460,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;
 }
@@ -502,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);
 
@@ -540,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);
 
@@ -633,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);
 
@@ -660,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))
@@ -686,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);
@@ -697,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;
 }
@@ -800,8 +810,7 @@ static int out_tx_xattr_set_exec(const struct lu_env *env,
 
                                lu_buf_free(&tbuf);
                                if (update) {
-                                       leh->leh_overflow_time =
-                                                       cfs_time_current_sec();
+                                       leh->leh_overflow_time = ktime_get_real_seconds();
                                        if (unlikely(!leh->leh_overflow_time))
                                                leh->leh_overflow_time++;
                                }
@@ -810,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,
@@ -883,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);
@@ -929,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);
 
@@ -942,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);
 
@@ -1062,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;
@@ -1083,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);