dht: fix use-after-free introduced by 70e6ee2

Change-Id: I97e73c0aae74fc5d80c975f56f2f7a64e3e1ae95
Updates: #2169
Signed-off-by: Xavi Hernandez <xhernandez@redhat.com>

2 files changed, 39 insertions, 3 deletions

diff --git a/xlators/cluster/dht/src/dht-common.c b/xlators/cluster/dht/src/dht-common.c
index b98d9a8918..4a7cb96ab0 100644
--- a/xlators/cluster/dht/src/dht-common.c
+++ b/xlators/cluster/dht/src/dht-common.c
@@ -6648,6 +6648,7 @@ dht_queue_readdir(call_frame_t *frame, xlator_t *xl, off_t offset,
                   fop_readdir_cbk_t cbk)
 {
     dht_local_t *local;
+    int32_t queue;
 
     local = frame->local;
 
@@ -6670,7 +6671,14 @@ dht_queue_readdir(call_frame_t *frame, xlator_t *xl, off_t offset,
 
             /* If a new readdirp request has been added before returning
              * from winding, we process it. */
-        } while (uatomic_sub_return(&local->queue, 1) != 0);
+        } while ((queue = uatomic_sub_return(&local->queue, 1)) > 0);
+
+        if (queue < 0) {
+            /* A negative value means that an unwind has been called before
+             * returning from the previous wind. This means that 'local' is
+             * not needed anymore and must be destroyed. */
+            dht_local_wipe(frame->this, local);
+        }
     }
 }
 
@@ -6681,6 +6689,7 @@ dht_queue_readdirp(call_frame_t *frame, xlator_t *xl, off_t offset,
                    fop_readdirp_cbk_t cbk)
 {
     dht_local_t *local;
+    int32_t queue;
 
     local = frame->local;
 
@@ -6693,7 +6702,11 @@ dht_queue_readdirp(call_frame_t *frame, xlator_t *xl, off_t offset,
             STACK_WIND_COOKIE(frame, cbk, local->queue_xl, local->queue_xl,
                               local->queue_xl->fops->readdirp, local->fd,
                               local->size, local->queue_offset, local->xattr);
-        } while (uatomic_sub_return(&local->queue, 1) != 0);
+        } while ((queue = uatomic_sub_return(&local->queue, 1)) > 0);
+
+        if (queue < 0) {
+            dht_local_wipe(frame->this, local);
+        }
     }
 }
 
@@ -6983,6 +6996,17 @@ unwind:
     if (prev != dht_last_up_subvol(this))
         op_errno = 0;
 
+    /* If we are inside a recursive call (or not inside a recursive call but
+     * the cbk is completed before the wind returns), local->queue will be 1.
+     * In this case we cannot destroy 'local' because it will be needed by
+     * the caller of STACK_WIND. In this case, we decrease the value to let
+     * the caller know that the operation has terminated and it must destroy
+     * 'local'. If local->queue 0, we can destroy it here because there are
+     * no other users. */
+    if (uatomic_sub_return(&local->queue, 1) >= 0) {
+        frame->local = NULL;
+    }
+
     DHT_STACK_UNWIND(readdirp, frame, op_ret, op_errno, &entries, NULL);
 
     gf_dirent_free(&entries);
@@ -7101,6 +7125,17 @@ unwind:
     if (prev != dht_last_up_subvol(this))
         op_errno = 0;
 
+    /* If we are inside a recursive call (or not inside a recursive call but
+     * the cbk is completed before the wind returns), local->queue will be 1.
+     * In this case we cannot destroy 'local' because it will be needed by
+     * the caller of STACK_WIND. In this case, we decrease the value to let
+     * the caller know that the operation has terminated and it must destroy
+     * 'local'. If local->queue 0, we can destroy it here because there are
+     * no other users. */
+    if (uatomic_sub_return(&local->queue, 1) >= 0) {
+        frame->local = NULL;
+    }
+
     if (!skip_hashed_check) {
         DHT_STACK_UNWIND(readdir, frame, op_ret, op_errno, &entries, NULL);
         gf_dirent_free(&entries);
@@ -7108,6 +7143,7 @@ unwind:
     } else {
         DHT_STACK_UNWIND(readdir, frame, op_ret, op_errno, orig_entries, NULL);
     }
+
     return 0;
 }
 
diff --git a/xlators/cluster/dht/src/dht-common.h b/xlators/cluster/dht/src/dht-common.h
index 9d5c7e15e5..712a2e98c4 100644
--- a/xlators/cluster/dht/src/dht-common.h
+++ b/xlators/cluster/dht/src/dht-common.h
@@ -336,7 +336,7 @@ struct dht_local {
     /* for nested readdirs */
     xlator_t *queue_xl;
     off_t queue_offset;
-    uint32_t queue;
+    int32_t queue;
 
     /* inodelks during filerename for backward compatibility */
     dht_lock_t **rename_inodelk_backward_compatible;