1 files changed, 342 insertions, 26 deletions

diff --git a/xlators/features/locks/src/common.c b/xlators/features/locks/src/common.c
index 6e7fb4b2f63..a2c6be93e03 100644
--- a/xlators/features/locks/src/common.c
+++ b/xlators/features/locks/src/common.c
@@ -14,9 +14,7 @@
 
 #include <glusterfs/glusterfs.h>
 #include <glusterfs/compat.h>
-#include <glusterfs/xlator.h>
 #include <glusterfs/logging.h>
-#include <glusterfs/common-utils.h>
 #include <glusterfs/syncop.h>
 
 #include "locks.h"
@@ -462,11 +460,16 @@ pl_inode_get(xlator_t *this, inode_t *inode, pl_local_t *local)
         INIT_LIST_HEAD(&pl_inode->blocked_calls);
         INIT_LIST_HEAD(&pl_inode->metalk_list);
         INIT_LIST_HEAD(&pl_inode->queued_locks);
+        INIT_LIST_HEAD(&pl_inode->waiting);
         gf_uuid_copy(pl_inode->gfid, inode->gfid);
 
         pl_inode->check_mlock_info = _gf_true;
         pl_inode->mlock_enforced = _gf_false;
 
+        /* -2 means never looked up. -1 means something went wrong and link
+         * tracking is disabled. */
+        pl_inode->links = -2;
+
         ret = __inode_ctx_put(inode, this, (uint64_t)(long)(pl_inode));
         if (ret) {
             pthread_mutex_destroy(&pl_inode->mutex);
@@ -478,7 +481,7 @@ pl_inode_get(xlator_t *this, inode_t *inode, pl_local_t *local)
 unlock:
     UNLOCK(&inode->lock);
 
-    if (pl_is_mandatory_locking_enabled(pl_inode) &&
+    if ((pl_inode != NULL) && pl_is_mandatory_locking_enabled(pl_inode) &&
         pl_inode->check_mlock_info && local) {
         /* Note: The lock enforcement information per file can be stored in the
            attribute flag of stat(x) in posix. With that there won't be a need
@@ -493,7 +496,8 @@ unlock:
 /* Create a new posix_lock_t */
 posix_lock_t *
 new_posix_lock(struct gf_flock *flock, client_t *client, pid_t client_pid,
-               gf_lkowner_t *owner, fd_t *fd, uint32_t lk_flags, int blocking)
+               gf_lkowner_t *owner, fd_t *fd, uint32_t lk_flags, int blocking,
+               int32_t *op_errno)
 {
     posix_lock_t *lock = NULL;
 
@@ -501,8 +505,14 @@ new_posix_lock(struct gf_flock *flock, client_t *client, pid_t client_pid,
     GF_VALIDATE_OR_GOTO("posix-locks", client, out);
     GF_VALIDATE_OR_GOTO("posix-locks", fd, out);
 
+    if (!pl_is_lk_owner_valid(owner, client)) {
+        *op_errno = EINVAL;
+        goto out;
+    }
+
     lock = GF_CALLOC(1, sizeof(posix_lock_t), gf_locks_mt_posix_lock_t);
     if (!lock) {
+        *op_errno = ENOMEM;
         goto out;
     }
 
@@ -520,6 +530,7 @@ new_posix_lock(struct gf_flock *flock, client_t *client, pid_t client_pid,
     if (lock->client_uid == NULL) {
         GF_FREE(lock);
         lock = NULL;
+        *op_errno = ENOMEM;
         goto out;
     }
 
@@ -594,13 +605,11 @@ static void
 __insert_lock(pl_inode_t *pl_inode, posix_lock_t *lock)
 {
     if (lock->blocked)
-        gettimeofday(&lock->blkd_time, NULL);
+        lock->blkd_time = gf_time();
     else
-        gettimeofday(&lock->granted_time, NULL);
+        lock->granted_time = gf_time();
 
     list_add_tail(&lock->list, &pl_inode->ext_list);
-
-    return;
 }
 
 /* Return true if the locks overlap, false otherwise */
@@ -956,7 +965,7 @@ grant_blocked_locks(xlator_t *this, pl_inode_t *pl_inode)
     struct list_head granted_list;
     posix_lock_t *tmp = NULL;
     posix_lock_t *lock = NULL;
-
+    pl_local_t *local = NULL;
     INIT_LIST_HEAD(&granted_list);
 
     pthread_mutex_lock(&pl_inode->mutex);
@@ -971,9 +980,9 @@ grant_blocked_locks(xlator_t *this, pl_inode_t *pl_inode)
 
         pl_trace_out(this, lock->frame, NULL, NULL, F_SETLKW, &lock->user_flock,
                      0, 0, NULL);
-
-        STACK_UNWIND_STRICT(lk, lock->frame, 0, 0, &lock->user_flock, NULL);
-
+        local = lock->frame->local;
+        PL_STACK_UNWIND_AND_FREE(local, lk, lock->frame, 0, 0,
+                                 &lock->user_flock, NULL);
         __destroy_lock(lock);
     }
 
@@ -988,10 +997,12 @@ pl_send_prelock_unlock(xlator_t *this, pl_inode_t *pl_inode,
         0,
     };
     posix_lock_t *unlock_lock = NULL;
+    int32_t op_errno = 0;
 
     struct list_head granted_list;
     posix_lock_t *tmp = NULL;
     posix_lock_t *lock = NULL;
+    pl_local_t *local = NULL;
 
     int ret = -1;
 
@@ -1005,7 +1016,7 @@ pl_send_prelock_unlock(xlator_t *this, pl_inode_t *pl_inode,
 
     unlock_lock = new_posix_lock(&flock, old_lock->client, old_lock->client_pid,
                                  &old_lock->owner, old_lock->fd,
-                                 old_lock->lk_flags, 0);
+                                 old_lock->lk_flags, 0, &op_errno);
     GF_VALIDATE_OR_GOTO(this->name, unlock_lock, out);
     ret = 0;
 
@@ -1019,9 +1030,9 @@ pl_send_prelock_unlock(xlator_t *this, pl_inode_t *pl_inode,
 
         pl_trace_out(this, lock->frame, NULL, NULL, F_SETLKW, &lock->user_flock,
                      0, 0, NULL);
-
-        STACK_UNWIND_STRICT(lk, lock->frame, 0, 0, &lock->user_flock, NULL);
-
+        local = lock->frame->local;
+        PL_STACK_UNWIND_AND_FREE(local, lk, lock->frame, 0, 0,
+                                 &lock->user_flock, NULL);
         __destroy_lock(lock);
     }
 
@@ -1080,6 +1091,10 @@ pl_setlk(xlator_t *this, pl_inode_t *pl_inode, posix_lock_t *lock,
                    lock->fl_type == F_UNLCK ? "Unlock" : "Lock",
                    lock->client_pid, lkowner_utoa(&lock->owner),
                    lock->user_flock.l_start, lock->user_flock.l_len);
+
+            pl_trace_block(this, lock->frame, NULL, NULL, F_SETLKW,
+                           &lock->user_flock, NULL);
+
             lock->blocked = 1;
             __insert_lock(pl_inode, lock);
             ret = -1;
@@ -1182,13 +1197,6 @@ pl_lock_preempt(pl_inode_t *pl_inode, posix_lock_t *reqlock)
             list_del_init(&rw->list);
             list_add(&rw->list, &unwind_rw_list);
         }
-
-        while (pl_inode->fop_wind_count != 0) {
-            gf_msg(THIS->name, GF_LOG_TRACE, 0, 0,
-                   "waiting for fops to be drained");
-            pthread_cond_wait(&pl_inode->check_fop_wind_count,
-                              &pl_inode->mutex);
-        }
     }
     pthread_mutex_unlock(&pl_inode->mutex);
 
@@ -1206,7 +1214,6 @@ pl_lock_preempt(pl_inode_t *pl_inode, posix_lock_t *reqlock)
     {
         pl_clean_local(rw->stub->frame->local);
         call_unwind_error(rw->stub, -1, EBUSY);
-        GF_FREE(lock);
     }
 
     return ret;
@@ -1266,10 +1273,319 @@ pl_local_init(call_frame_t *frame, xlator_t *this, loc_t *loc, fd_t *fd)
             return -1;
         }
 
-        local->inode = (fd ? inode_ref(fd->inode) : inode_ref(loc->inode));
+        local->inode = (loc ? inode_ref(loc->inode) : inode_ref(fd->inode));
 
         frame->local = local;
     }
 
     return 0;
-}
\ No newline at end of file
+}
+
+gf_boolean_t
+pl_is_lk_owner_valid(gf_lkowner_t *owner, client_t *client)
+{
+    if (client && (client->opversion < GD_OP_VERSION_7_0)) {
+        return _gf_true;
+    }
+
+    if (is_lk_owner_null(owner)) {
+        return _gf_false;
+    }
+    return _gf_true;
+}
+
+static int32_t
+pl_inode_from_loc(loc_t *loc, inode_t **pinode)
+{
+    inode_t *inode = NULL;
+    int32_t error = 0;
+
+    if (loc->inode != NULL) {
+        inode = inode_ref(loc->inode);
+        goto done;
+    }
+
+    if (loc->parent == NULL) {
+        error = EINVAL;
+        goto done;
+    }
+
+    if (!gf_uuid_is_null(loc->gfid)) {
+        inode = inode_find(loc->parent->table, loc->gfid);
+        if (inode != NULL) {
+            goto done;
+        }
+    }
+
+    if (loc->name == NULL) {
+        error = EINVAL;
+        goto done;
+    }
+
+    inode = inode_grep(loc->parent->table, loc->parent, loc->name);
+    if (inode == NULL) {
+        /* We haven't found any inode. This means that the file doesn't exist
+         * or that even if it exists, we don't have any knowledge about it, so
+         * we don't have locks on it either, which is fine for our purposes. */
+        goto done;
+    }
+
+done:
+    *pinode = inode;
+
+    return error;
+}
+
+static gf_boolean_t
+pl_inode_has_owners(xlator_t *xl, client_t *client, pl_inode_t *pl_inode,
+                    struct timespec *now, struct list_head *contend)
+{
+    pl_dom_list_t *dom;
+    pl_inode_lock_t *lock;
+    gf_boolean_t has_owners = _gf_false;
+
+    list_for_each_entry(dom, &pl_inode->dom_list, inode_list)
+    {
+        list_for_each_entry(lock, &dom->inodelk_list, list)
+        {
+            /* If the lock belongs to the same client, we assume it's related
+             * to the same operation, so we allow the removal to continue. */
+            if (lock->client == client) {
+                continue;
+            }
+            /* If the lock belongs to an internal process, we don't block the
+             * removal. */
+            if (lock->client_pid < 0) {
+                continue;
+            }
+            if (contend == NULL) {
+                return _gf_true;
+            }
+            has_owners = _gf_true;
+            inodelk_contention_notify_check(xl, lock, now, contend);
+        }
+    }
+
+    return has_owners;
+}
+
+int32_t
+pl_inode_remove_prepare(xlator_t *xl, call_frame_t *frame, loc_t *loc,
+                        pl_inode_t **ppl_inode, struct list_head *contend)
+{
+    struct timespec now;
+    inode_t *inode;
+    pl_inode_t *pl_inode;
+    int32_t error;
+
+    pl_inode = NULL;
+
+    error = pl_inode_from_loc(loc, &inode);
+    if ((error != 0) || (inode == NULL)) {
+        goto done;
+    }
+
+    pl_inode = pl_inode_get(xl, inode, NULL);
+    if (pl_inode == NULL) {
+        inode_unref(inode);
+        error = ENOMEM;
+        goto done;
+    }
+
+    /* pl_inode_from_loc() already increments ref count for inode, so
+     * we only assign here our reference. */
+    pl_inode->inode = inode;
+
+    timespec_now(&now);
+
+    pthread_mutex_lock(&pl_inode->mutex);
+
+    if (pl_inode->removed) {
+        error = ESTALE;
+        goto unlock;
+    }
+
+    if (pl_inode_has_owners(xl, frame->root->client, pl_inode, &now, contend)) {
+        error = -1;
+        /* We skip the unlock here because the caller must create a stub when
+         * we return -1 and do a call to pl_inode_remove_complete(), which
+         * assumes the lock is still acquired and will release it once
+         * everything else is prepared. */
+        goto done;
+    }
+
+    pl_inode->is_locked = _gf_true;
+    pl_inode->remove_running++;
+
+unlock:
+    pthread_mutex_unlock(&pl_inode->mutex);
+
+done:
+    *ppl_inode = pl_inode;
+
+    return error;
+}
+
+int32_t
+pl_inode_remove_complete(xlator_t *xl, pl_inode_t *pl_inode, call_stub_t *stub,
+                         struct list_head *contend)
+{
+    pl_inode_lock_t *lock;
+    int32_t error = -1;
+
+    if (stub != NULL) {
+        list_add_tail(&stub->list, &pl_inode->waiting);
+        pl_inode->is_locked = _gf_true;
+    } else {
+        error = ENOMEM;
+
+        while (!list_empty(contend)) {
+            lock = list_first_entry(contend, pl_inode_lock_t, list);
+            list_del_init(&lock->list);
+            __pl_inodelk_unref(lock);
+        }
+    }
+
+    pthread_mutex_unlock(&pl_inode->mutex);
+
+    if (error < 0) {
+        inodelk_contention_notify(xl, contend);
+    }
+
+    inode_unref(pl_inode->inode);
+
+    return error;
+}
+
+void
+pl_inode_remove_wake(struct list_head *list)
+{
+    call_stub_t *stub;
+
+    while (!list_empty(list)) {
+        stub = list_first_entry(list, call_stub_t, list);
+        list_del_init(&stub->list);
+
+        call_resume(stub);
+    }
+}
+
+void
+pl_inode_remove_cbk(xlator_t *xl, pl_inode_t *pl_inode, int32_t error)
+{
+    struct list_head contend, granted;
+    struct timespec now;
+    pl_dom_list_t *dom;
+
+    if (pl_inode == NULL) {
+        return;
+    }
+
+    INIT_LIST_HEAD(&contend);
+    INIT_LIST_HEAD(&granted);
+    timespec_now(&now);
+
+    pthread_mutex_lock(&pl_inode->mutex);
+
+    if (error == 0) {
+        if (pl_inode->links >= 0) {
+            pl_inode->links--;
+        }
+        if (pl_inode->links == 0) {
+            pl_inode->removed = _gf_true;
+        }
+    }
+
+    pl_inode->remove_running--;
+
+    if ((pl_inode->remove_running == 0) && list_empty(&pl_inode->waiting)) {
+        pl_inode->is_locked = _gf_false;
+
+        list_for_each_entry(dom, &pl_inode->dom_list, inode_list)
+        {
+            __grant_blocked_inode_locks(xl, pl_inode, &granted, dom, &now,
+                                        &contend);
+        }
+    }
+
+    pthread_mutex_unlock(&pl_inode->mutex);
+
+    unwind_granted_inodes(xl, pl_inode, &granted);
+
+    inodelk_contention_notify(xl, &contend);
+
+    inode_unref(pl_inode->inode);
+}
+
+void
+pl_inode_remove_unlocked(xlator_t *xl, pl_inode_t *pl_inode,
+                         struct list_head *list)
+{
+    call_stub_t *stub, *tmp;
+
+    if (!pl_inode->is_locked) {
+        return;
+    }
+
+    list_for_each_entry_safe(stub, tmp, &pl_inode->waiting, list)
+    {
+        if (!pl_inode_has_owners(xl, stub->frame->root->client, pl_inode, NULL,
+                                 NULL)) {
+            list_move_tail(&stub->list, list);
+        }
+    }
+}
+
+/* This function determines if an inodelk attempt can be done now or it needs
+ * to wait.
+ *
+ * Possible return values:
+ *   < 0: An error occurred. Currently only -ESTALE can be returned if the
+ *        inode has been deleted previously by unlink/rmdir/rename
+ *   = 0: The lock can be attempted.
+ *   > 0: The lock needs to wait because a conflicting remove operation is
+ *        ongoing.
+ */
+int32_t
+pl_inode_remove_inodelk(pl_inode_t *pl_inode, pl_inode_lock_t *lock)
+{
+    pl_dom_list_t *dom;
+    pl_inode_lock_t *ilock;
+
+    /* If the inode has been deleted, we won't allow any lock. */
+    if (pl_inode->removed) {
+        return -ESTALE;
+    }
+
+    /* We only synchronize with locks made for regular operations coming from
+     * the user. Locks done for internal purposes are hard to control and could
+     * lead to long delays or deadlocks quite easily. */
+    if (lock->client_pid < 0) {
+        return 0;
+    }
+    if (!pl_inode->is_locked) {
+        return 0;
+    }
+    if (pl_inode->remove_running > 0) {
+        return 1;
+    }
+
+    list_for_each_entry(dom, &pl_inode->dom_list, inode_list)
+    {
+        list_for_each_entry(ilock, &dom->inodelk_list, list)
+        {
+            /* If a lock from the same client is already granted, we allow this
+             * one to continue. This is necessary to prevent deadlocks when
+             * multiple locks are taken for the same operation.
+             *
+             * On the other side it's unlikely that the same client sends
+             * completely unrelated locks for the same inode.
+             */
+            if (ilock->client == lock->client) {
+                return 0;
+            }
+        }
+    }
+
+    return 1;
+}