cluster/dht: Unwind with proper op_ret

1. Expected behavior of get_real_filename feature.

A getxattr on a existing dir with glusterfs.get_real_filename:<filename>
as key should result in one of the following things.

a. A value returned for that key having the real filename (a file whose
match is a case insensitive match to the filename passed in key).
b. op_ret = -1 and errno set to ENOENT meaning that no such file exists
under the specified dir in any case.
c. op_ret = -1 and errno set to ENODATA. This is a case assuming no
xlator interprets the glusterfs.get_real_filename key and it get
passed down to the posix xlator. Naturally, posix xlator would not
find any xattr with this key and would return ENODATA. This will be
interpreted specially by the caller as the feature not being supported
by underlying glusterfs.

2. What assumptions are wrong?
Initially the key used to be user.glusterfs.get_real_filename.
In that case, when posix xlator did a getxattr call it would have
received ENODATA as error. However, the key has now changed to
glusterfs.get_real_filename. This leads to a EOPNOTSUPP error instead.

Considering the above information, this is a rewrite of
get_real_filename logic in dht.

Change-Id: I012e9150047fc8563be91b0d112a368ac1cbf598
BUG: 1204140
Signed-off-by: Raghavendra Talur <rtalur@redhat.com>
Reviewed-on: http://review.gluster.org/9956
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: N Balachandran <nbalacha@redhat.com>
Reviewed-by: Shyamsundar Ranganathan <srangana@redhat.com>

1 files changed, 86 insertions, 12 deletions

diff --git a/xlators/cluster/dht/src/dht-common.c b/xlators/cluster/dht/src/dht-common.c
index 6d36fdbe284..c91c7eeec48 100644
--- a/xlators/cluster/dht/src/dht-common.c
+++ b/xlators/cluster/dht/src/dht-common.c
@@ -2742,20 +2742,94 @@ dht_getxattr_get_real_filename_cbk (call_frame_t *frame, void *cookie,
 
         local = frame->local;
 
-	if (op_ret != -1) {
-		if (local->xattr)
-			dict_unref (local->xattr);
-		local->xattr = dict_ref (xattr);
-
-		if (local->xattr_req)
-			dict_unref (local->xattr_req);
-		local->xattr_req = dict_ref (xdata);
-	}
+        LOCK (&frame->lock);
+        {
+                if (local->op_errno == ENODATA ||
+                    local->op_errno == EOPNOTSUPP) {
+                        /* Nothing to do here, we have already found
+                         * a subvol which does not have the get_real_filename
+                         * optimization. If condition is for simple logic.
+                         */
+                        goto unlock;
+                }
+
+                if (op_ret == -1) {
+
+                        if (op_errno == ENODATA || op_errno == EOPNOTSUPP) {
+                                /* This subvol does not have the optimization.
+                                 * Better let the user know we don't support it.
+                                 * Remove previous results if any.
+                                 */
+
+                                if (local->xattr) {
+                                        dict_unref (local->xattr);
+                                        local->xattr = NULL;
+                                }
+
+                                if (local->xattr_req) {
+                                        dict_unref (local->xattr_req);
+                                        local->xattr_req = NULL;
+                                }
+
+                                local->op_ret = op_ret;
+                                local->op_errno = op_errno;
+                                gf_log (this->name, GF_LOG_WARNING, "At least "
+                                        "one of the bricks does not support "
+                                        "this operation. Please upgrade all "
+                                        "bricks.");
+                                goto unlock;
+                        }
+
+                        if (op_errno == ENOENT) {
+                                /* Do nothing, our defaults are set to this.
+                                 */
+                                goto unlock;
+                        }
+
+                        /* This is a place holder for every other error
+                         * case. I am not sure of how to interpret
+                         * ENOTCONN etc. As of now, choosing to ignore
+                         * down subvol and return a good result(if any)
+                         * from other subvol.
+                         */
+                         gf_log (this->name, GF_LOG_WARNING,
+                                 "Failed to get real filename. "
+                                 "error:%s", strerror (op_errno));
+                        goto unlock;
+
+                }
+
+
+                /* This subvol has the required file.
+                 * There could be other subvols which have returned
+                 * success already, choosing to return the latest good
+                 * result.
+                 */
+                if (local->xattr)
+                        dict_unref (local->xattr);
+                local->xattr = dict_ref (xattr);
+
+                if (local->xattr_req) {
+                        dict_unref (local->xattr_req);
+                        local->xattr_req = NULL;
+                }
+                if (xdata)
+                        local->xattr_req = dict_ref (xdata);
+
+                local->op_ret = op_ret;
+                local->op_errno = 0;
+                gf_log (this->name, GF_LOG_DEBUG, "Found a matching "
+                        "file.");
+        }
+unlock:
+        UNLOCK (&frame->lock);
+
 
 	this_call_cnt = dht_frame_return (frame);
 	if (is_last_call (this_call_cnt)) {
-		DHT_STACK_UNWIND (getxattr, frame, local->op_ret, op_errno,
-				  local->xattr, local->xattr_req);
+                DHT_STACK_UNWIND (getxattr, frame, local->op_ret,
+                                  local->op_errno, local->xattr,
+                                  local->xattr_req);
 	}
 
 	return 0;
@@ -2779,7 +2853,7 @@ dht_getxattr_get_real_filename (call_frame_t *frame, xlator_t *this,
 	cnt = local->call_cnt = layout->cnt;
 
 	local->op_ret = -1;
-	local->op_errno = ENODATA;
+	local->op_errno = ENOENT;
 
 	for (i = 0; i < cnt; i++) {
 		subvol = layout->list[i].xlator;