dht: unlink stale linkfiles in rmdir to prevent ENOTEMPTY

Thanks to He Xaobing <allreol@gmail.com>, this patch is inspired by

http://bugs.gluster.com/cgi-bin/bugzilla3/show_bug.cgi?id=188#c2

Signed-off-by: Anand V. Avati <avati@blackhole.gluster.com>
Signed-off-by: Anand V. Avati <avati@dev.gluster.com>

BUG: 188 ([ glusterfs 2.0.6rc2 ] - "Directory not empty" on rm -rf)
URL: http://bugs.gluster.com/cgi-bin/bugzilla3/show_bug.cgi?id=188

3 files changed, 244 insertions, 14 deletions

diff --git a/xlators/cluster/dht/src/dht-common.c b/xlators/cluster/dht/src/dht-common.c
index d677e7ee4..e6b23f783 100644
--- a/xlators/cluster/dht/src/dht-common.c
+++ b/xlators/cluster/dht/src/dht-common.c
@@ -3344,22 +3344,209 @@ err:
 
 
 int
-dht_rmdir_readdir_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
-		       int op_ret, int op_errno, gf_dirent_t *entries)
+dht_rmdir_linkfile_unlink_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                               int op_ret, int op_errno, struct stat *preparent,
+                               struct stat *postparent)
+{
+        dht_local_t    *local = NULL;
+        call_frame_t   *prev = NULL;
+        xlator_t       *src = NULL;
+        call_frame_t   *main_frame = NULL;
+        dht_local_t    *main_local = NULL;
+        int             this_call_cnt = 0;
+
+        local  = frame->local;
+        prev   = cookie;
+        src    = prev->this;
+
+        main_frame = local->main_frame;
+        main_local = main_frame->local;
+
+        if (op_ret == 0) {
+                gf_log (this->name, GF_LOG_TRACE,
+                        "unlinked linkfile %s on %s",
+                        local->loc.path, src->name);
+        } else {
+                main_local->op_ret   = -1;
+                main_local->op_errno = op_errno;
+                gf_log (this->name, GF_LOG_DEBUG,
+                        "unlink of %s on %s failed (%s)",
+                        local->loc.path, src->name, strerror (op_errno));
+        }
+
+        this_call_cnt = dht_frame_return (main_frame);
+        if (is_last_call (this_call_cnt))
+                dht_rmdir_do (main_frame, this);
+
+        DHT_STACK_DESTROY (frame);
+        return 0;
+}
+
+
+int
+dht_rmdir_lookup_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                      int op_ret, int op_errno, inode_t *inode,
+                      struct stat *stbuf, dict_t *xattr, struct stat *parent)
+{
+        dht_local_t    *local = NULL;
+        call_frame_t   *prev = NULL;
+        xlator_t       *src = NULL;
+        call_frame_t   *main_frame = NULL;
+        dht_local_t    *main_local = NULL;
+        int             this_call_cnt = 0;
+
+        local = frame->local;
+        prev  = cookie;
+        src   = prev->this;
+
+        main_frame = local->main_frame;
+        main_local = main_frame->local;
+
+        if (op_ret != 0)
+                goto err;
+
+        if (check_is_linkfile (inode, stbuf, xattr) == 0) {
+                main_local->op_ret  = -1;
+                main_local->op_errno = ENOTEMPTY;
+
+                gf_log (this->name, GF_LOG_WARNING,
+                        "%s on %s found to be not a linkfile (mode=0%o)",
+                        local->loc.path, src->name, stbuf->st_mode);
+                goto err;
+        }
+
+        STACK_WIND (frame, dht_rmdir_linkfile_unlink_cbk,
+                    src, src->fops->unlink, &local->loc);
+        return 0;
+err:
+
+        this_call_cnt = dht_frame_return (main_frame);
+        if (is_last_call (this_call_cnt))
+                dht_rmdir_do (main_frame, this);
+
+        DHT_STACK_DESTROY (frame);
+        return 0;
+}
+
+
+int
+dht_rmdir_is_subvol_empty (call_frame_t *frame, xlator_t *this,
+                           gf_dirent_t *entries, xlator_t *src)
+{
+        int                 ret = 0;
+        int                 build_ret = 0;
+        gf_dirent_t        *trav = NULL;
+        call_frame_t       *lookup_frame = NULL;
+        dht_local_t        *lookup_local = NULL;
+        dht_local_t        *local = NULL;
+
+        local = frame->local;
+
+        list_for_each_entry (trav, &entries->list, list) {
+                if (strcmp (trav->d_name, ".") == 0)
+                        continue;
+                if (strcmp (trav->d_name, "..") == 0)
+                        continue;
+                if (check_is_linkfile (NULL, (&trav->d_stat), NULL) == 1) {
+                        ret++;
+                        continue;
+                }
+
+                /* this entry is either a directory which is neither "." nor "..",
+                   or a non directory which is not a linkfile. the directory is to
+                   be treated as non-empty
+                */
+                return 0;
+        }
+
+        list_for_each_entry (trav, &entries->list, list) {
+                if (strcmp (trav->d_name, ".") == 0)
+                        continue;
+                if (strcmp (trav->d_name, "..") == 0)
+                        continue;
+
+                lookup_frame = NULL;
+                lookup_local = NULL;
+
+                lookup_frame = copy_frame (frame);
+                if (!lookup_frame) {
+                        gf_log (this->name, GF_LOG_ERROR,
+                                "Out of Memory");
+                        /* out of memory, let the rmdir fail
+                           (as non-empty, unfortunately) */
+                        goto err;
+                }
+
+                lookup_local = CALLOC (sizeof (*local), 1);
+                if (!lookup_local) {
+                        gf_log (this->name, GF_LOG_ERROR,
+                                "Out of Memory");
+                        goto err;
+                }
+
+                lookup_frame->local = lookup_local;
+                lookup_local->main_frame = frame;
+
+                build_ret = dht_build_child_loc (this, &lookup_local->loc,
+                                                 &local->loc, trav->d_name);
+                if (build_ret != 0)
+                        goto err;
+
+                gf_log (this->name, GF_LOG_TRACE,
+                        "looking up %s on %s",
+                        lookup_local->loc.path, src->name);
+
+                LOCK (&frame->lock);
+                {
+                        local->call_cnt++;
+                }
+                UNLOCK (&frame->lock);
+
+                STACK_WIND (lookup_frame, dht_rmdir_lookup_cbk,
+                            src, src->fops->lookup,
+                            &lookup_local->loc, NULL);
+                ret++;
+        }
+
+        return ret;
+err:
+        DHT_STACK_DESTROY (lookup_frame);
+        return 0;
+}
+
+
+int
+dht_rmdir_readdirp_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                        int op_ret, int op_errno, gf_dirent_t *entries)
 {
 	dht_local_t  *local = NULL;
 	int           this_call_cnt = -1;
 	call_frame_t *prev = NULL;
+        xlator_t     *src = NULL;
+        int           ret = 0;
 
 	local = frame->local;
 	prev  = cookie;
+        src   = prev->this;
 
 	if (op_ret > 2) {
-		gf_log (this->name, GF_LOG_TRACE,
-			"readdir on %s for %s returned %d entries",
-			prev->this->name, local->loc.path, op_ret);
-		local->op_ret = -1;
-		local->op_errno = ENOTEMPTY;
+                ret = dht_rmdir_is_subvol_empty (frame, this, entries, src);
+
+                switch (ret) {
+                case 0: /* non linkfiles exist */
+                        gf_log (this->name, GF_LOG_TRACE,
+                                "readdir on %s for %s returned %d entries",
+                                prev->this->name, local->loc.path, op_ret);
+                        local->op_ret = -1;
+                        local->op_errno = ENOTEMPTY;
+                        break;
+                default:
+                        /* @ret number of linkfiles are getting unlinked */
+                        gf_log (this->name, GF_LOG_TRACE,
+                                "readdir on %s for %s found %d linkfiles",
+                                prev->this->name, local->loc.path, ret);
+                        break;
+                }
 	}
 
 	this_call_cnt = dht_frame_return (frame);
@@ -3392,8 +3579,8 @@ dht_rmdir_opendir_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 		goto err;
 	}
 
-	STACK_WIND (frame, dht_rmdir_readdir_cbk,
-		    prev->this, prev->this->fops->readdir,
+	STACK_WIND (frame, dht_rmdir_readdirp_cbk,
+		    prev->this, prev->this->fops->readdirp,
 		    local->fd, 4096, 0);
 
 	return 0;
diff --git a/xlators/cluster/dht/src/dht-common.h b/xlators/cluster/dht/src/dht-common.h
index 53aeaa0f7..42a75adfb 100644
--- a/xlators/cluster/dht/src/dht-common.h
+++ b/xlators/cluster/dht/src/dht-common.h
@@ -70,7 +70,7 @@ struct dht_local {
 	struct stat              stbuf;
         struct stat              prebuf;
         struct stat              preoldparent;
-        struct stat              postoldparent; 
+        struct stat              postoldparent;
         struct stat              preparent;
         struct stat              postparent;
 	struct statvfs           statvfs;
@@ -89,6 +89,7 @@ struct dht_local {
 	char                     need_selfheal;
         int                      file_count;
         int                      dir_count;
+        call_frame_t            *main_frame;
 	struct {
 		fop_mknod_cbk_t  linkfile_cbk;
 		struct stat      stbuf;
@@ -154,7 +155,7 @@ struct dht_disk_layout {
 	} list[1];
 };
 typedef struct dht_disk_layout dht_disk_layout_t;
- 
+
 #define ENTRY_MISSING(op_ret, op_errno) (op_ret == -1 && op_errno == ENOENT)
 
 #define is_fs_root(loc) (strcmp (loc->path, "/") == 0)
@@ -269,4 +270,6 @@ void dht_layout_unref (xlator_t *this, dht_layout_t *layout);
 dht_layout_t *dht_layout_ref (xlator_t *this, dht_layout_t *layout);
 xlator_t *dht_first_up_subvol (xlator_t *this);
 
+int dht_build_child_loc (xlator_t *this, loc_t *child, loc_t *parent, char *name);
+
 #endif /* _DHT_H */
diff --git a/xlators/cluster/dht/src/dht-helper.c b/xlators/cluster/dht/src/dht-helper.c
index 9816c1784..5a89ba584 100644
--- a/xlators/cluster/dht/src/dht-helper.c
+++ b/xlators/cluster/dht/src/dht-helper.c
@@ -139,7 +139,7 @@ dht_local_wipe (xlator_t *this, dht_local_t *local)
 		fd_unref (local->fd);
 		local->fd = NULL;
 	}
-	
+
 	if (local->xattr_req)
 		dict_unref (local->xattr_req);
 
@@ -184,6 +184,7 @@ basestr (const char *str)
         return basestr;
 }
 
+
 xlator_t *
 dht_first_up_subvol (xlator_t *this)
 {
@@ -192,7 +193,7 @@ dht_first_up_subvol (xlator_t *this)
 	int         i = 0;
 
 	conf = this->private;
-	
+
 	LOCK (&conf->subvolume_lock);
 	{
 		for (i = 0; i < conf->subvolume_cnt; i++) {
@@ -203,10 +204,11 @@ dht_first_up_subvol (xlator_t *this)
 		}
 	}
 	UNLOCK (&conf->subvolume_lock);
-	
+
 	return child;
 }
 
+
 xlator_t *
 dht_subvol_get_hashed (xlator_t *this, loc_t *loc)
 {
@@ -343,3 +345,41 @@ dht_stat_merge (xlator_t *this, struct stat *to,
 
 	return 0;
 }
+
+
+int
+dht_build_child_loc (xlator_t *this, loc_t *child, loc_t *parent, char *name)
+{
+        if (!child) {
+                goto err;
+        }
+
+        if (strcmp (parent->path, "/") == 0)
+                asprintf ((char **)&child->path, "/%s", name);
+        else
+                asprintf ((char **)&child->path, "%s/%s", parent->path, name);
+
+        if (!child->path) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "Out of memory");
+                goto err;
+        }
+
+        child->name = strrchr (child->path, '/');
+        if (child->name)
+                child->name++;
+
+        child->parent = inode_ref (parent->inode);
+        child->inode = inode_new (parent->inode->table);
+
+        if (!child->inode) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "Out of memory");
+                goto err;
+        }
+
+        return 0;
+err:
+        loc_wipe (child);
+        return -1;
+}