LU-14531 osd: serialize access to object vs object destroy

in osd-zfs as ZFS doesn't provide an internal mechanism for this.

Signed-off-by: Alex Zhuravlev <bzzz@whamcloud.com>
Change-Id: I5f25710a5cf1568f124733a15e77a37ffcb55434
Reviewed-on: https://review.whamcloud.com/43233
Reviewed-by: Andreas Dilger <adilger@whamcloud.com>
Reviewed-by: Mike Pershin <mpershin@whamcloud.com>
Tested-by: jenkins <devops@whamcloud.com>
Tested-by: Maloo <maloo@whamcloud.com>
Reviewed-by: Oleg Drokin <green@whamcloud.com>

diff --git a/lustre/osd-zfs/osd_index.c b/lustre/osd-zfs/osd_index.c
index 5a83262..6dedc24 100644 (file)
--- a/lustre/osd-zfs/osd_index.c
+++ b/lustre/osd-zfs/osd_index.c
@@ -662,22 +662,29 @@ static int osd_dir_lookup(const struct lu_env *env, struct dt_object *dt,
        }
 
        memset(&oti->oti_zde.lzd_fid, 0, sizeof(struct lu_fid));
+
+       down_read(&obj->oo_guard);
+       if (obj->oo_destroyed)
+               GOTO(out_unlock, rc = -ENOENT);
+
        rc = osd_zap_lookup(osd, obj->oo_dn->dn_object, obj->oo_dn,
                            (char *)key, 8, sizeof(oti->oti_zde) / 8,
                            (void *)&oti->oti_zde);
-       if (rc != 0)
+       if (rc != 0) {
+               up_read(&obj->oo_guard);
                RETURN(rc);
+       }
 
        oid = oti->oti_zde.lzd_reg.zde_dnode;
        if (likely(fid_is_sane(&oti->oti_zde.lzd_fid))) {
                memcpy(rec, &oti->oti_zde.lzd_fid, sizeof(struct lu_fid));
-               GOTO(out, rc = 0);
+               GOTO(out_unlock, rc = 0);
        }
 
        rc = osd_get_fid_by_oid(env, osd, oti->oti_zde.lzd_reg.zde_dnode, fid);
 
-       GOTO(out, rc);
-
+out_unlock:
+       up_read(&obj->oo_guard);
 out:
        if (!rc && !osd_remote_fid(env, osd, fid)) {
                /*

diff --git a/lustre/osd-zfs/osd_io.c b/lustre/osd-zfs/osd_io.c
index f42eb5f..a5dbac4 100644 (file)
--- a/lustre/osd-zfs/osd_io.c
+++ b/lustre/osd-zfs/osd_io.c
@@ -219,6 +219,10 @@ static ssize_t osd_write(const struct lu_env *env, struct dt_object *dt,
        LASSERT(th != NULL);
        oh = container_of(th, struct osd_thandle, ot_super);
 
+       down_read(&obj->oo_guard);
+       if (obj->oo_destroyed)
+               GOTO(out, rc = -ENOENT);
+
        osd_dmu_write(osd, obj->oo_dn, offset, (uint64_t)buf->lb_len,
                      buf->lb_buf, oh->ot_tx);
        write_lock(&obj->oo_attr_lock);
@@ -240,6 +244,7 @@ static ssize_t osd_write(const struct lu_env *env, struct dt_object *dt,
        rc = buf->lb_len;
 
 out:
+       up_read(&obj->oo_guard);
        RETURN(rc);
 }
 
@@ -876,6 +881,11 @@ static int osd_write_commit(const struct lu_env *env, struct dt_object *dt,
         * changing the block size.
         */
        down_read(&obj->oo_guard);
+       if (obj->oo_destroyed) {
+               up_read(&obj->oo_guard);
+               RETURN(-ENOENT);
+       }
+
        for (i = 0; i < npages; i++) {
                CDEBUG(D_INODE, "write %u bytes at %u\n",
                        (unsigned) lnb[i].lnb_len,
@@ -941,13 +951,13 @@ static int osd_write_commit(const struct lu_env *env, struct dt_object *dt,
                osd_evict_dbufs_after_write(obj, lnb[i].lnb_file_offset,
                                            abufsz);
        }
-       up_read(&obj->oo_guard);
 
        if (unlikely(new_size == 0)) {
                /* no pages to write, no transno is needed */
                th->th_local = 1;
                /* it is important to return 0 even when all lnb_rc == -ENOSPC
                 * since ofd_commitrw_write() retries several times on ENOSPC */
+               up_read(&obj->oo_guard);
                record_end_io(osd, WRITE, 0, 0, 0);
                RETURN(0);
        }
@@ -968,6 +978,8 @@ static int osd_write_commit(const struct lu_env *env, struct dt_object *dt,
                write_unlock(&obj->oo_attr_lock);
        }
 
+       up_read(&obj->oo_guard);
+
        record_end_io(osd, WRITE, 0, iosize, npages);
 
        RETURN(rc);
@@ -1082,6 +1094,10 @@ static int osd_punch(const struct lu_env *env, struct dt_object *dt,
                len = end - start;
        write_unlock(&obj->oo_attr_lock);
 
+       down_read(&obj->oo_guard);
+       if (obj->oo_destroyed)
+               GOTO(out, rc = -ENOENT);
+
        rc = __osd_object_punch(obj, osd->od_os, oh->ot_tx, start, len);
 
        /* set new size */
@@ -1092,6 +1108,8 @@ static int osd_punch(const struct lu_env *env, struct dt_object *dt,
                rc = osd_object_sa_update(obj, SA_ZPL_SIZE(osd),
                                          &obj->oo_attr.la_size, 8, oh);
        }
+out:
+       up_read(&obj->oo_guard);
        RETURN(rc);
 }
 

diff --git a/lustre/osd-zfs/osd_object.c b/lustre/osd-zfs/osd_object.c
index ffec3d3..a81358b 100644 (file)
--- a/lustre/osd-zfs/osd_object.c
+++ b/lustre/osd-zfs/osd_object.c
@@ -127,17 +127,19 @@ void osd_object_sa_dirty_rele(const struct lu_env *env, struct osd_thandle *oh)
                write_lock(&obj->oo_attr_lock);
                list_del_init(&obj->oo_sa_linkage);
                write_unlock(&obj->oo_attr_lock);
-               if (obj->oo_late_xattr) {
+               if (obj->oo_late_xattr && obj->oo_destroyed == 0) {
                        /*
                         * take oo_guard to protect oo_sa_xattr buffer
                         * from concurrent update by osd_xattr_set()
                         */
                        LASSERT(oh->ot_assigned != 0);
                        down_write(&obj->oo_guard);
-                       if (obj->oo_late_attr_set)
-                               __osd_sa_attr_init(env, obj, oh);
-                       else if (obj->oo_late_xattr)
-                               __osd_sa_xattr_update(env, obj, oh);
+                       if (obj->oo_destroyed == 0) {
+                               if (obj->oo_late_attr_set)
+                                       __osd_sa_attr_init(env, obj, oh);
+                               else if (obj->oo_late_xattr)
+                                       __osd_sa_xattr_update(env, obj, oh);
+                       }
                        up_write(&obj->oo_guard);
                }
                sa_spill_rele(obj->oo_sa_hdl);