Revert "LU-16046 revert: "LU-9964 llite: prevent mulitple group locks""

This reverts commit bc37f89a81ea0a2fae8668e21247552e8894bfd8.

unreverting the revert since the fix that replaced it was bad and
ther are better ideas on how to amend this fix now rather than
full-on revert

Change-Id: I1ef28c13715e7ea98021e1f83331e5533c2a8868
Signed-off-by: Oleg Drokin <green@whamcloud.com>

diff --git a/lustre/ldlm/ldlm_request.c b/lustre/ldlm/ldlm_request.c
index 8d609d1..62595ac 100644 (file)
--- a/lustre/ldlm/ldlm_request.c
+++ b/lustre/ldlm/ldlm_request.c
@@ -1032,7 +1032,8 @@ int ldlm_cli_enqueue(struct obd_export *exp, struct ptlrpc_request **reqp,
        lock->l_conn_export = exp;
        lock->l_export = NULL;
        lock->l_blocking_ast = einfo->ei_cb_bl;
-       lock->l_flags |= (*flags & (LDLM_FL_NO_LRU | LDLM_FL_EXCL));
+       lock->l_flags |= (*flags & (LDLM_FL_NO_LRU | LDLM_FL_EXCL |
+                                   LDLM_FL_ATOMIC_CB));
        lock->l_activity = ktime_get_real_seconds();
 
        /* lock not sent to server yet */

diff --git a/lustre/llite/file.c b/lustre/llite/file.c
index 86e55f4..a454902 100644 (file)
--- a/lustre/llite/file.c
+++ b/lustre/llite/file.c
@@ -2632,6 +2632,7 @@ out:
        RETURN(rc);
 }
 
+
 static int
 ll_get_grouplock(struct inode *inode, struct file *file, unsigned long arg)
 {
@@ -2647,18 +2648,28 @@ ll_get_grouplock(struct inode *inode, struct file *file, unsigned long arg)
                RETURN(-EINVAL);
        }
 
-        if (ll_file_nolock(file))
-                RETURN(-EOPNOTSUPP);
+       if (ll_file_nolock(file))
+               RETURN(-EOPNOTSUPP);
+retry:
+       if (file->f_flags & O_NONBLOCK) {
+               if (!mutex_trylock(&lli->lli_group_mutex))
+                       RETURN(-EAGAIN);
+       } else
+               mutex_lock(&lli->lli_group_mutex);
 
-       read_lock(&lli->lli_lock);
        if (fd->fd_flags & LL_FILE_GROUP_LOCKED) {
                CWARN("group lock already existed with gid %lu\n",
                      fd->fd_grouplock.lg_gid);
-               read_unlock(&lli->lli_lock);
-               RETURN(-EINVAL);
+               GOTO(out, rc = -EINVAL);
+       }
+       if (arg != lli->lli_group_gid && lli->lli_group_users != 0) {
+               if (file->f_flags & O_NONBLOCK)
+                       GOTO(out, rc = -EAGAIN);
+               mutex_unlock(&lli->lli_group_mutex);
+               wait_var_event(&lli->lli_group_users, !lli->lli_group_users);
+               GOTO(retry, rc = 0);
        }
        LASSERT(fd->fd_grouplock.lg_lock == NULL);
-       read_unlock(&lli->lli_lock);
 
        /**
         * XXX: group lock needs to protect all OST objects while PFL
@@ -2678,7 +2689,7 @@ ll_get_grouplock(struct inode *inode, struct file *file, unsigned long arg)
 
                env = cl_env_get(&refcheck);
                if (IS_ERR(env))
-                       RETURN(PTR_ERR(env));
+                       GOTO(out, rc = PTR_ERR(env));
 
                rc = cl_object_layout_get(env, obj, &cl);
                if (rc >= 0 && cl.cl_is_composite)
@@ -2687,28 +2698,26 @@ ll_get_grouplock(struct inode *inode, struct file *file, unsigned long arg)
 
                cl_env_put(env, &refcheck);
                if (rc < 0)
-                       RETURN(rc);
+                       GOTO(out, rc);
        }
 
        rc = cl_get_grouplock(ll_i2info(inode)->lli_clob,
                              arg, (file->f_flags & O_NONBLOCK), &grouplock);
-       if (rc)
-               RETURN(rc);
 
-       write_lock(&lli->lli_lock);
-       if (fd->fd_flags & LL_FILE_GROUP_LOCKED) {
-               write_unlock(&lli->lli_lock);
-               CERROR("another thread just won the race\n");
-               cl_put_grouplock(&grouplock);
-               RETURN(-EINVAL);
-       }
+       if (rc)
+               GOTO(out, rc);
 
        fd->fd_flags |= LL_FILE_GROUP_LOCKED;
        fd->fd_grouplock = grouplock;
-       write_unlock(&lli->lli_lock);
+       if (lli->lli_group_users == 0)
+               lli->lli_group_gid = grouplock.lg_gid;
+       lli->lli_group_users++;
 
        CDEBUG(D_INFO, "group lock %lu obtained\n", arg);
-       RETURN(0);
+out:
+       mutex_unlock(&lli->lli_group_mutex);
+
+       RETURN(rc);
 }
 
 static int ll_put_grouplock(struct inode *inode, struct file *file,
@@ -2717,32 +2726,40 @@ static int ll_put_grouplock(struct inode *inode, struct file *file,
        struct ll_inode_info   *lli = ll_i2info(inode);
        struct ll_file_data    *fd = file->private_data;
        struct ll_grouplock     grouplock;
+       int                     rc;
        ENTRY;
 
-       write_lock(&lli->lli_lock);
+       mutex_lock(&lli->lli_group_mutex);
        if (!(fd->fd_flags & LL_FILE_GROUP_LOCKED)) {
-               write_unlock(&lli->lli_lock);
-                CWARN("no group lock held\n");
-                RETURN(-EINVAL);
-        }
+               CWARN("no group lock held\n");
+               GOTO(out, rc = -EINVAL);
+       }
 
        LASSERT(fd->fd_grouplock.lg_lock != NULL);
 
        if (fd->fd_grouplock.lg_gid != arg) {
                CWARN("group lock %lu doesn't match current id %lu\n",
                      arg, fd->fd_grouplock.lg_gid);
-               write_unlock(&lli->lli_lock);
-               RETURN(-EINVAL);
+               GOTO(out, rc = -EINVAL);
        }
 
        grouplock = fd->fd_grouplock;
        memset(&fd->fd_grouplock, 0, sizeof(fd->fd_grouplock));
        fd->fd_flags &= ~LL_FILE_GROUP_LOCKED;
-       write_unlock(&lli->lli_lock);
 
        cl_put_grouplock(&grouplock);
+
+       lli->lli_group_users--;
+       if (lli->lli_group_users == 0) {
+               lli->lli_group_gid = 0;
+               wake_up_var(&lli->lli_group_users);
+       }
        CDEBUG(D_INFO, "group lock %lu released\n", arg);
-       RETURN(0);
+       GOTO(out, rc = 0);
+out:
+       mutex_unlock(&lli->lli_group_mutex);
+
+       RETURN(rc);
 }
 
 /**

diff --git a/lustre/llite/llite_internal.h b/lustre/llite/llite_internal.h
index a1c60fe..da36467 100644 (file)
--- a/lustre/llite/llite_internal.h
+++ b/lustre/llite/llite_internal.h
@@ -241,6 +241,10 @@ struct ll_inode_info {
                        enum pcc_dataset_flags   lli_pcc_dsflags;
                        struct pcc_inode        *lli_pcc_inode;
 
+                       struct mutex             lli_group_mutex;
+                       __u64                    lli_group_users;
+                       unsigned long            lli_group_gid;
+
                        __u64                    lli_attr_valid;
                        __u64                    lli_lazysize;
                        __u64                    lli_lazyblocks;

diff --git a/lustre/llite/llite_lib.c b/lustre/llite/llite_lib.c
index 1bc27da..e097c33 100644 (file)
--- a/lustre/llite/llite_lib.c
+++ b/lustre/llite/llite_lib.c
@@ -1236,6 +1236,9 @@ void ll_lli_init(struct ll_inode_info *lli)
                lli->lli_pcc_inode = NULL;
                lli->lli_pcc_dsflags = PCC_DATASET_INVALID;
                lli->lli_pcc_generation = 0;
+               mutex_init(&lli->lli_group_mutex);
+               lli->lli_group_users = 0;
+               lli->lli_group_gid = 0;
        }
        mutex_init(&lli->lli_layout_mutex);
        memset(lli->lli_jobid, 0, sizeof(lli->lli_jobid));

diff --git a/lustre/osc/osc_lock.c b/lustre/osc/osc_lock.c
index 6453b31..1ebd516 100644 (file)
--- a/lustre/osc/osc_lock.c
+++ b/lustre/osc/osc_lock.c
@@ -1233,6 +1233,8 @@ int osc_lock_init(const struct lu_env *env,
 
        oscl->ols_flags = osc_enq2ldlm_flags(enqflags);
        oscl->ols_speculative = !!(enqflags & CEF_SPECULATIVE);
+       if (lock->cll_descr.cld_mode == CLM_GROUP)
+               oscl->ols_flags |= LDLM_FL_ATOMIC_CB;
 
        if (oscl->ols_flags & LDLM_FL_HAS_INTENT) {
                oscl->ols_flags |= LDLM_FL_BLOCK_GRANTED;

diff --git a/lustre/tests/sanity.sh b/lustre/tests/sanity.sh
index 0e5640f..f90d408 100755 (executable)
--- a/lustre/tests/sanity.sh
+++ b/lustre/tests/sanity.sh
@@ -43,7 +43,6 @@ always_except LU-6493  42b
 always_except LU-14541 277
 always_except LU-9054  312
 always_except LU-8411  407
-always_except LU-16046 244b
 
 if $SHARED_KEY; then
        always_except LU-14181 64e 64f