1 files changed, 278 insertions, 176 deletions

diff --git a/xlators/cluster/dht/src/dht-linkfile.c b/xlators/cluster/dht/src/dht-linkfile.c
index 77e7818ef85..89ec6cca56e 100644
--- a/xlators/cluster/dht/src/dht-linkfile.c
+++ b/xlators/cluster/dht/src/dht-linkfile.c
@@ -1,226 +1,328 @@
 /*
-   Copyright (c) 2008-2009 Gluster, Inc. <http://www.gluster.com>
-   This file is part of GlusterFS.
-
-   GlusterFS is free software; you can redistribute it and/or modify
-   it under the terms of the GNU General Public License as published
-   by the Free Software Foundation; either version 3 of the License,
-   or (at your option) any later version.
-
-   GlusterFS is distributed in the hope that it will be useful, but
-   WITHOUT ANY WARRANTY; without even the implied warranty of
-   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
-   General Public License for more details.
-
-   You should have received a copy of the GNU General Public License
-   along with this program.  If not, see
-   <http://www.gnu.org/licenses/>.
+  Copyright (c) 2008-2012 Red Hat, Inc. <http://www.redhat.com>
+  This file is part of GlusterFS.
+
+  This file is licensed to you under your choice of the GNU Lesser
+  General Public License, version 3 or any later version (LGPLv3 or
+  later), or the GNU General Public License, version 2 (GPLv2), in all
+  cases as published by the Free Software Foundation.
 */
 
-#ifndef _CONFIG_H
-#define _CONFIG_H
-#include "config.h"
-#endif
+#include <glusterfs/compat.h>
+#include "dht-common.h"
 
+static int
+dht_linkfile_lookup_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
+                        int op_ret, int op_errno, inode_t *inode,
+                        struct iatt *stbuf, dict_t *xattr,
+                        struct iatt *postparent)
+{
+    char is_linkfile = 0;
+    dht_conf_t *conf = NULL;
+    dht_local_t *local = NULL;
+    xlator_t *prev = NULL;
+    char gfid[GF_UUID_BUF_SIZE] = {0};
+
+    local = frame->local;
+    prev = cookie;
+    conf = this->private;
+
+    if (op_ret)
+        goto out;
+
+    gf_uuid_unparse(local->loc.gfid, gfid);
+
+    is_linkfile = check_is_linkfile(inode, stbuf, xattr, conf->link_xattr_name);
+    if (!is_linkfile)
+        gf_smsg(this->name, GF_LOG_WARNING, 0, DHT_MSG_NOT_LINK_FILE_ERROR,
+                "name=%s", prev->name, "path=%s", local->loc.path, "gfid=%s",
+                gfid, NULL);
+out:
+    local->linkfile.linkfile_cbk(frame, cookie, this, op_ret, op_errno, inode,
+                                 stbuf, postparent, postparent, xattr);
+    return 0;
+}
 
-#include "glusterfs.h"
-#include "xlator.h"
-#include "compat.h"
-#include "dht-common.h"
+static int
+dht_linkfile_create_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
+                        int op_ret, int op_errno, inode_t *inode,
+                        struct iatt *stbuf, struct iatt *preparent,
+                        struct iatt *postparent, dict_t *xdata)
+{
+    dht_local_t *local = NULL;
+    xlator_t *subvol = NULL;
+    dict_t *xattrs = NULL;
+    dht_conf_t *conf = NULL;
+    int ret = -1;
+
+    local = frame->local;
+
+    if (!op_ret)
+        local->linked = _gf_true;
+
+    FRAME_SU_UNDO(frame, dht_local_t);
+
+    if (op_ret && (op_errno == EEXIST)) {
+        conf = this->private;
+        subvol = cookie;
+        if (!subvol)
+            goto out;
+        xattrs = dict_new();
+        if (!xattrs)
+            goto out;
+        ret = dict_set_uint32(xattrs, conf->link_xattr_name, 256);
+        if (ret) {
+            gf_smsg(this->name, GF_LOG_ERROR, 0, DHT_MSG_DICT_SET_FAILED,
+                    "mame=%s", conf->link_xattr_name, NULL);
+            goto out;
+        }
+
+        STACK_WIND_COOKIE(frame, dht_linkfile_lookup_cbk, subvol, subvol,
+                          subvol->fops->lookup, &local->linkfile.loc, xattrs);
+        if (xattrs)
+            dict_unref(xattrs);
+        return 0;
+    }
+out:
+    local->linkfile.linkfile_cbk(frame, cookie, this, op_ret, op_errno, inode,
+                                 stbuf, preparent, postparent, xdata);
+    if (xattrs)
+        dict_unref(xattrs);
+    return 0;
+}
 
+int
+dht_linkfile_create(call_frame_t *frame, fop_mknod_cbk_t linkfile_cbk,
+                    xlator_t *this, xlator_t *tovol, xlator_t *fromvol,
+                    loc_t *loc)
+{
+    dht_local_t *local = NULL;
+    dict_t *dict = NULL;
+    int need_unref = 0;
+    int ret = 0;
+    dht_conf_t *conf = this->private;
+    char gfid[GF_UUID_BUF_SIZE] = {0};
+
+    local = frame->local;
+    local->linkfile.linkfile_cbk = linkfile_cbk;
+    local->linkfile.srcvol = tovol;
+    loc_copy(&local->linkfile.loc, loc);
+
+    local->linked = _gf_false;
+
+    dict = local->params;
+    if (!dict) {
+        dict = dict_new();
+        if (!dict)
+            goto out;
+        need_unref = 1;
+    }
+
+    if (!gf_uuid_is_null(local->gfid)) {
+        gf_uuid_unparse(local->gfid, gfid);
+
+        ret = dict_set_gfuuid(dict, "gfid-req", local->gfid, true);
+        if (ret)
+            gf_smsg("dht-linkfile", GF_LOG_INFO, 0, DHT_MSG_DICT_SET_FAILED,
+                    "path=%s", loc->path, "gfid=%s", gfid, NULL);
+    } else {
+        gf_uuid_unparse(loc->gfid, gfid);
+    }
+
+    ret = dict_set_str(dict, GLUSTERFS_INTERNAL_FOP_KEY, "yes");
+    if (ret)
+        gf_smsg("dht-linkfile", GF_LOG_INFO, 0, DHT_MSG_DICT_SET_FAILED,
+                "path=%s", loc->path, "key=%s", GLUSTERFS_INTERNAL_FOP_KEY,
+                "gfid=%s", gfid, NULL);
+
+    ret = dict_set_str(dict, conf->link_xattr_name, tovol->name);
+
+    if (ret < 0) {
+        gf_smsg(frame->this->name, GF_LOG_INFO, 0, DHT_MSG_CREATE_LINK_FAILED,
+                "path=%s", loc->path, "gfid=%s", gfid, NULL);
+        goto out;
+    }
+
+    local->link_subvol = fromvol;
+    /* Always create as root:root. dht_linkfile_attr_heal fixes the
+     * ownsership */
+    FRAME_SU_DO(frame, dht_local_t);
+    STACK_WIND_COOKIE(frame, dht_linkfile_create_cbk, fromvol, fromvol,
+                      fromvol->fops->mknod, loc, S_IFREG | DHT_LINKFILE_MODE, 0,
+                      0, dict);
+
+    if (need_unref && dict)
+        dict_unref(dict);
+
+    return 0;
+out:
+    local->linkfile.linkfile_cbk(frame, frame->this, frame->this, -1, ENOMEM,
+                                 loc->inode, NULL, NULL, NULL, NULL);
 
+    if (need_unref && dict)
+        dict_unref(dict);
+
+    return 0;
+}
 
 int
-dht_linkfile_xattr_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
-			int op_ret, int op_errno)
+dht_linkfile_unlink_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
+                        int32_t op_ret, int32_t op_errno,
+                        struct iatt *preparent, struct iatt *postparent,
+                        dict_t *xdata)
 {
-	dht_local_t *local = NULL;
+    dht_local_t *local = NULL;
+    xlator_t *subvol = NULL;
+    char gfid[GF_UUID_BUF_SIZE] = {0};
 
+    local = frame->local;
+    subvol = cookie;
 
-	local = frame->local;
-	local->linkfile.linkfile_cbk (frame, cookie, this, op_ret, op_errno,
-				      local->linkfile.inode,
-				      &local->linkfile.stbuf, NULL, NULL);
+    if (op_ret == -1) {
+        gf_uuid_unparse(local->loc.gfid, gfid);
+        gf_smsg(this->name, GF_LOG_INFO, op_errno, DHT_MSG_UNLINK_FAILED,
+                "path=%s", local->loc.path, "gfid=%s", gfid, "subvolume=%s",
+                subvol->name, NULL);
+    }
 
-	return 0;
-}
+    DHT_STACK_DESTROY(frame);
 
+    return 0;
+}
 
 int
-dht_linkfile_create_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
-                         int op_ret, int op_errno, inode_t *inode,
-                         struct stat *stbuf, struct stat *preparent,
-                         struct stat *postparent)
+dht_linkfile_unlink(call_frame_t *frame, xlator_t *this, xlator_t *subvol,
+                    loc_t *loc)
 {
-	dht_local_t  *local = NULL;
-	call_frame_t *prev = NULL;
-	dict_t       *xattr = NULL;
-	data_t       *str_data = NULL;
-	int           ret = -1;
-
-	local = frame->local;
-	prev  = cookie;
-
-	if (op_ret == -1)
-		goto err;
-
-	xattr = get_new_dict ();
-	if (!xattr) {
-		gf_log (this->name, GF_LOG_ERROR,
-			"Out of memory");
-		op_errno = ENOMEM;
-		goto err;
-	}
-
-	local->linkfile.xattr = dict_ref (xattr);
-	local->linkfile.inode = inode_ref (inode);
-
-	str_data = str_to_data (local->linkfile.srcvol->name);
-	if (!str_data) {
-		gf_log (this->name, GF_LOG_ERROR,
-			"Out of memory");
-		op_errno = ENOMEM;
-		goto err;
-	}
-
-	ret = dict_set (xattr, "trusted.glusterfs.dht.linkto", str_data);
-	if (ret < 0) {
-		gf_log (this->name, GF_LOG_DEBUG,
-			"failed to initialize linkfile data");
-		op_errno = EINVAL;
-	}
-	str_data = NULL;
-
-	local->linkfile.stbuf = *stbuf;
-
-	STACK_WIND (frame, dht_linkfile_xattr_cbk,
-		    prev->this, prev->this->fops->setxattr,
-		    &local->linkfile.loc, local->linkfile.xattr, 0);
-
-	return 0;
+    call_frame_t *unlink_frame = NULL;
+    dht_local_t *unlink_local = NULL;
+
+    unlink_frame = copy_frame(frame);
+    if (!unlink_frame) {
+        goto err;
+    }
+
+    /* Using non-fop value here, as anyways, 'local->fop' is not used in
+       this particular case */
+    unlink_local = dht_local_init(unlink_frame, loc, NULL, GF_FOP_MAXVALUE);
+    if (!unlink_local) {
+        goto err;
+    }
+
+    STACK_WIND_COOKIE(unlink_frame, dht_linkfile_unlink_cbk, subvol, subvol,
+                      subvol->fops->unlink, &unlink_local->loc, 0, NULL);
 
+    return 0;
 err:
-	if (str_data) {
-		data_destroy (str_data);
-		str_data = NULL;
-	}
-
-        local->linkfile.linkfile_cbk (frame, cookie, this, op_ret, op_errno,
-                                      inode, stbuf, preparent, postparent);
-	return 0;
-}
+    if (unlink_frame)
+        DHT_STACK_DESTROY(unlink_frame);
 
+    return -1;
+}
 
-int
-dht_linkfile_create (call_frame_t *frame, fop_mknod_cbk_t linkfile_cbk,
-		     xlator_t *tovol, xlator_t *fromvol, loc_t *loc)
+xlator_t *
+dht_linkfile_subvol(xlator_t *this, inode_t *inode, struct iatt *stbuf,
+                    dict_t *xattr)
 {
-	dht_local_t *local = NULL;
+    dht_conf_t *conf = NULL;
+    xlator_t *subvol = NULL;
+    void *volname = NULL;
+    int i = 0, ret = 0;
 
+    conf = this->private;
 
-	local = frame->local;
-	local->linkfile.linkfile_cbk = linkfile_cbk;
-	local->linkfile.srcvol = tovol;
-	loc_copy (&local->linkfile.loc, loc);
+    if (!xattr)
+        goto out;
 
-	STACK_WIND (frame, dht_linkfile_create_cbk,
-		    fromvol, fromvol->fops->mknod, loc,
-		    S_IFREG | DHT_LINKFILE_MODE, 0);
+    ret = dict_get_ptr(xattr, conf->link_xattr_name, &volname);
 
-	return 0;
-}
+    if ((-1 == ret) || !volname)
+        goto out;
 
+    for (i = 0; i < conf->subvolume_cnt; i++) {
+        if (strcmp(conf->subvolumes[i]->name, (char *)volname) == 0) {
+            subvol = conf->subvolumes[i];
+            break;
+        }
+    }
 
-int
-dht_linkfile_unlink_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
-			 int32_t op_ret, int32_t op_errno,
-                         struct stat *preparent, struct stat *postparent)
+out:
+    return subvol;
+}
+
+static int
+dht_linkfile_setattr_cbk(call_frame_t *frame, void *cookie, xlator_t *this,
+                         int op_ret, int op_errno, struct iatt *statpre,
+                         struct iatt *statpost, dict_t *xdata)
 {
-	dht_local_t   *local = NULL;
-	call_frame_t  *prev = NULL;
-	xlator_t      *subvol = NULL;
+    dht_local_t *local = NULL;
+    loc_t *loc = NULL;
 
-	local = frame->local;
-	prev = cookie;
-	subvol = prev->this;
+    local = frame->local;
+    loc = &local->loc;
 
-	if (op_ret == -1) {
-		gf_log (this->name, GF_LOG_DEBUG,
-			"unlinking linkfile %s on %s failed (%s)",
-			local->loc.path, subvol->name, strerror (op_errno));
-	}
+    if (op_ret)
+        gf_smsg(this->name, GF_LOG_ERROR, op_errno, DHT_MSG_SETATTR_FAILED,
+                "path=%s", (loc->path ? loc->path : "NULL"), "gfid=%s",
+                uuid_utoa(local->gfid), NULL);
 
-	DHT_STACK_DESTROY (frame);
+    DHT_STACK_DESTROY(frame);
 
-	return 0;
+    return 0;
 }
 
-
 int
-dht_linkfile_unlink (call_frame_t *frame, xlator_t *this,
-		     xlator_t *subvol, loc_t *loc)
+dht_linkfile_attr_heal(call_frame_t *frame, xlator_t *this)
 {
-	call_frame_t *unlink_frame = NULL;
-	dht_local_t  *unlink_local = NULL;
-
-	unlink_frame = copy_frame (frame);
-	if (!unlink_frame) {
-		gf_log (this->name, GF_LOG_ERROR,
-			"Out of memory");
-		goto err;
-	}
-
-	unlink_local = dht_local_init (unlink_frame);
-	if (!unlink_local) {
-		gf_log (this->name, GF_LOG_ERROR,
-			"Out of memory");
-		goto err;
-	}
-
-	loc_copy (&unlink_local->loc, loc);
-
-	STACK_WIND (unlink_frame, dht_linkfile_unlink_cbk,
-		    subvol, subvol->fops->unlink,
-		    &unlink_local->loc);
-
-	return 0;
-err:
-	if (unlink_frame)
-		DHT_STACK_DESTROY (unlink_frame);
+    int ret = -1;
+    call_frame_t *copy = NULL;
+    dht_local_t *local = NULL;
+    dht_local_t *copy_local = NULL;
+    xlator_t *subvol = NULL;
+    struct iatt stbuf = {
+        0,
+    };
+    dict_t *xattr = NULL;
 
-	return -1;
-}
+    local = frame->local;
 
+    GF_VALIDATE_OR_GOTO("dht", local, out);
+    GF_VALIDATE_OR_GOTO("dht", local->link_subvol, out);
 
-xlator_t *
-dht_linkfile_subvol (xlator_t *this, inode_t *inode, struct stat *stbuf,
-		     dict_t *xattr)
-{
-	dht_conf_t *conf = NULL;
-	xlator_t   *subvol = NULL;
-	void       *volname = NULL;
-	int         i = 0, ret = 0;
+    if (local->stbuf.ia_type == IA_INVAL)
+        return 0;
+
+    DHT_MARK_FOP_INTERNAL(xattr);
 
+    gf_uuid_copy(local->loc.gfid, local->stbuf.ia_gfid);
 
-	conf = this->private;
+    copy = copy_frame(frame);
 
-	if (!xattr)
-		goto out;
+    if (!copy)
+        goto out;
 
-	ret = dict_get_ptr (xattr, "trusted.glusterfs.dht.linkto", &volname);
+    copy_local = dht_local_init(copy, &local->loc, NULL, 0);
 
-	if ((-1 == ret) || !volname)
-		goto out;
+    if (!copy_local)
+        goto out;
 
-	for (i = 0; i < conf->subvolume_cnt; i++) {
-		if (strcmp (conf->subvolumes[i]->name, (char *)volname) == 0) {
-			subvol = conf->subvolumes[i];
-			break;
-		}
-	}
+    stbuf = local->stbuf;
+    subvol = local->link_subvol;
 
+    copy->local = copy_local;
+
+    FRAME_SU_DO(copy, dht_local_t);
+
+    STACK_WIND(copy, dht_linkfile_setattr_cbk, subvol, subvol->fops->setattr,
+               &copy_local->loc, &stbuf, (GF_SET_ATTR_UID | GF_SET_ATTR_GID),
+               xattr);
+    ret = 0;
 out:
-	return subvol;
-}
+    if ((ret < 0) && (copy))
+        DHT_STACK_DESTROY(copy);
 
+    if (xattr)
+        dict_unref(xattr);
 
+    return ret;
+}