summaryrefslogtreecommitdiffstats
path: root/xlators/protocol/client/src/client-handshake.c
diff options
context:
space:
mode:
Diffstat (limited to 'xlators/protocol/client/src/client-handshake.c')
-rw-r--r--xlators/protocol/client/src/client-handshake.c1960
1 files changed, 1960 insertions, 0 deletions
diff --git a/xlators/protocol/client/src/client-handshake.c b/xlators/protocol/client/src/client-handshake.c
new file mode 100644
index 000000000..5668fea53
--- /dev/null
+++ b/xlators/protocol/client/src/client-handshake.c
@@ -0,0 +1,1960 @@
+/*
+ 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 "fd-lk.h"
+#include "client.h"
+#include "xlator.h"
+#include "defaults.h"
+#include "glusterfs.h"
+#include "statedump.h"
+#include "compat-errno.h"
+
+#include "glusterfs3.h"
+#include "portmap-xdr.h"
+#include "rpc-common-xdr.h"
+
+#define CLIENT_REOPEN_MAX_ATTEMPTS 1024
+extern rpc_clnt_prog_t clnt3_3_fop_prog;
+extern rpc_clnt_prog_t clnt_pmap_prog;
+
+int client_ping_cbk (struct rpc_req *req, struct iovec *iov, int count,
+ void *myframe);
+
+int client_set_lk_version_cbk (struct rpc_req *req, struct iovec *iov,
+ int count, void *myframe);
+
+int client_set_lk_version (xlator_t *this);
+
+typedef struct client_fd_lk_local {
+ int ref;
+ gf_boolean_t error;
+ gf_lock_t lock;
+ clnt_fd_ctx_t *fdctx;
+}clnt_fd_lk_local_t;
+
+/* Handshake */
+
+void
+rpc_client_ping_timer_expired (void *data)
+{
+ rpc_transport_t *trans = NULL;
+ rpc_clnt_connection_t *conn = NULL;
+ int disconnect = 0;
+ int transport_activity = 0;
+ struct timespec timeout = {0, };
+ struct timeval current = {0, };
+ struct rpc_clnt *clnt = NULL;
+ xlator_t *this = NULL;
+ clnt_conf_t *conf = NULL;
+
+ this = data;
+
+ if (!this || !this->private) {
+ gf_log (THIS->name, GF_LOG_WARNING, "xlator initialization not done");
+ goto out;
+ }
+
+ conf = this->private;
+
+ clnt = conf->rpc;
+ if (!clnt) {
+ gf_log (this->name, GF_LOG_WARNING, "rpc not initialized");
+ goto out;
+ }
+
+ conn = &clnt->conn;
+ trans = conn->trans;
+
+ if (!trans) {
+ gf_log (this->name, GF_LOG_WARNING, "transport not initialized");
+ goto out;
+ }
+
+ pthread_mutex_lock (&conn->lock);
+ {
+ if (conn->ping_timer)
+ gf_timer_call_cancel (this->ctx,
+ conn->ping_timer);
+ gettimeofday (&current, NULL);
+
+ if (((current.tv_sec - conn->last_received.tv_sec) <
+ conf->opt.ping_timeout)
+ || ((current.tv_sec - conn->last_sent.tv_sec) <
+ conf->opt.ping_timeout)) {
+ transport_activity = 1;
+ }
+
+ if (transport_activity) {
+ gf_log (trans->name, GF_LOG_TRACE,
+ "ping timer expired but transport activity "
+ "detected - not bailing transport");
+ timeout.tv_sec = conf->opt.ping_timeout;
+ timeout.tv_nsec = 0;
+
+ conn->ping_timer =
+ gf_timer_call_after (this->ctx, timeout,
+ rpc_client_ping_timer_expired,
+ (void *) this);
+ if (conn->ping_timer == NULL)
+ gf_log (trans->name, GF_LOG_WARNING,
+ "unable to setup ping timer");
+
+ } else {
+ conn->ping_started = 0;
+ conn->ping_timer = NULL;
+ disconnect = 1;
+ }
+ }
+ pthread_mutex_unlock (&conn->lock);
+
+ if (disconnect) {
+ gf_log (trans->name, GF_LOG_CRITICAL,
+ "server %s has not responded in the last %d "
+ "seconds, disconnecting.",
+ conn->trans->peerinfo.identifier,
+ conf->opt.ping_timeout);
+
+ rpc_transport_disconnect (conn->trans);
+ }
+
+out:
+ return;
+}
+
+void
+client_start_ping (void *data)
+{
+ xlator_t *this = NULL;
+ clnt_conf_t *conf = NULL;
+ rpc_clnt_connection_t *conn = NULL;
+ int32_t ret = -1;
+ struct timespec timeout = {0, };
+ call_frame_t *frame = NULL;
+ int frame_count = 0;
+
+ this = data;
+ if (!this || !this->private) {
+ gf_log (THIS->name, GF_LOG_WARNING, "xlator not initialized");
+ goto fail;
+ }
+
+ conf = this->private;
+ if (!conf->rpc) {
+ gf_log (this->name, GF_LOG_WARNING, "rpc not initialized");
+ goto fail;
+ }
+ conn = &conf->rpc->conn;
+
+ if (conf->opt.ping_timeout == 0) {
+ gf_log (this->name, GF_LOG_INFO, "ping timeout is 0, returning");
+ return;
+ }
+
+ pthread_mutex_lock (&conn->lock);
+ {
+ if (conn->ping_timer)
+ gf_timer_call_cancel (this->ctx, conn->ping_timer);
+
+ conn->ping_timer = NULL;
+ conn->ping_started = 0;
+
+ if (conn->saved_frames)
+ /* treat the case where conn->saved_frames is NULL
+ as no pending frames */
+ frame_count = conn->saved_frames->count;
+
+ if ((frame_count == 0) || !conn->connected) {
+ /* using goto looked ugly here,
+ * hence getting out this way */
+ /* unlock */
+ gf_log (this->name, GF_LOG_DEBUG,
+ "returning as transport is already disconnected"
+ " OR there are no frames (%d || %d)",
+ frame_count, !conn->connected);
+
+ pthread_mutex_unlock (&conn->lock);
+ return;
+ }
+
+ if (frame_count < 0) {
+ gf_log (this->name, GF_LOG_WARNING,
+ "saved_frames->count is %"PRId64,
+ conn->saved_frames->count);
+ conn->saved_frames->count = 0;
+ }
+
+ timeout.tv_sec = conf->opt.ping_timeout;
+ timeout.tv_nsec = 0;
+
+ conn->ping_timer =
+ gf_timer_call_after (this->ctx, timeout,
+ rpc_client_ping_timer_expired,
+ (void *) this);
+
+ if (conn->ping_timer == NULL) {
+ gf_log (this->name, GF_LOG_WARNING,
+ "unable to setup ping timer");
+ } else {
+ conn->ping_started = 1;
+ }
+ }
+ pthread_mutex_unlock (&conn->lock);
+
+ frame = create_frame (this, this->ctx->pool);
+ if (!frame)
+ goto fail;
+
+ ret = client_submit_request (this, NULL, frame, conf->handshake,
+ GF_HNDSK_PING, client_ping_cbk, NULL,
+ NULL, 0, NULL, 0, NULL, (xdrproc_t)NULL);
+ if (ret) {
+ gf_log (THIS->name, GF_LOG_ERROR,
+ "failed to start ping timer");
+ }
+
+ return;
+
+fail:
+ if (frame) {
+ STACK_DESTROY (frame->root);
+ }
+
+ return;
+}
+
+
+int
+client_ping_cbk (struct rpc_req *req, struct iovec *iov, int count,
+ void *myframe)
+{
+ xlator_t *this = NULL;
+ rpc_clnt_connection_t *conn = NULL;
+ struct timespec timeout = {0, };
+ call_frame_t *frame = NULL;
+ clnt_conf_t *conf = NULL;
+
+ if (!myframe) {
+ gf_log (THIS->name, GF_LOG_WARNING,
+ "frame with the request is NULL");
+ goto out;
+ }
+ frame = myframe;
+ this = frame->this;
+ if (!this || !this->private) {
+ gf_log (THIS->name, GF_LOG_WARNING,
+ "xlator private is not set");
+ goto out;
+ }
+
+ conf = this->private;
+ conn = &conf->rpc->conn;
+
+ pthread_mutex_lock (&conn->lock);
+ {
+ if (req->rpc_status == -1) {
+ if (conn->ping_timer != NULL) {
+ gf_log (this->name, GF_LOG_WARNING,
+ "socket or ib related error");
+ gf_timer_call_cancel (this->ctx,
+ conn->ping_timer);
+ conn->ping_timer = NULL;
+ } else {
+ /* timer expired and transport bailed out */
+ gf_log (this->name, GF_LOG_WARNING,
+ "timer must have expired");
+ }
+
+ goto unlock;
+ }
+
+
+ timeout.tv_sec = conf->opt.ping_timeout;
+ timeout.tv_nsec = 0;
+
+ gf_timer_call_cancel (this->ctx,
+ conn->ping_timer);
+
+ conn->ping_timer =
+ gf_timer_call_after (this->ctx, timeout,
+ client_start_ping, (void *)this);
+
+ if (conn->ping_timer == NULL)
+ gf_log (this->name, GF_LOG_WARNING,
+ "failed to set the ping timer");
+ }
+unlock:
+ pthread_mutex_unlock (&conn->lock);
+out:
+ if (frame)
+ STACK_DESTROY (frame->root);
+ return 0;
+}
+
+
+int
+client3_getspec_cbk (struct rpc_req *req, struct iovec *iov, int count,
+ void *myframe)
+{
+ gf_getspec_rsp rsp = {0,};
+ call_frame_t *frame = NULL;
+ int ret = 0;
+
+ frame = myframe;
+
+ if (!frame || !frame->this) {
+ gf_log (THIS->name, GF_LOG_ERROR, "frame not found with the request, "
+ "returning EINVAL");
+ rsp.op_ret = -1;
+ rsp.op_errno = EINVAL;
+ goto out;
+ }
+ if (-1 == req->rpc_status) {
+ gf_log (frame->this->name, GF_LOG_WARNING,
+ "received RPC status error, returning ENOTCONN");
+ rsp.op_ret = -1;
+ rsp.op_errno = ENOTCONN;
+ goto out;
+ }
+
+ ret = xdr_to_generic (*iov, &rsp, (xdrproc_t)xdr_gf_getspec_rsp);
+ if (ret < 0) {
+ gf_log (frame->this->name, GF_LOG_ERROR,
+ "XDR decoding failed, returning EINVAL");
+ rsp.op_ret = -1;
+ rsp.op_errno = EINVAL;
+ goto out;
+ }
+
+ if (-1 == rsp.op_ret) {
+ gf_log (frame->this->name, GF_LOG_WARNING,
+ "failed to get the 'volume file' from server");
+ goto out;
+ }
+
+out:
+ CLIENT_STACK_UNWIND (getspec, frame, rsp.op_ret, rsp.op_errno,
+ rsp.spec);
+
+ /* Don't use 'GF_FREE', this is allocated by libc */
+ free (rsp.spec);
+
+ return 0;
+}
+
+int32_t client3_getspec (call_frame_t *frame, xlator_t *this, void *data)
+{
+ clnt_conf_t *conf = NULL;
+ clnt_args_t *args = NULL;
+ gf_getspec_req req = {0,};
+ int op_errno = ESTALE;
+ int ret = 0;
+
+ if (!frame || !this || !data)
+ goto unwind;
+
+ args = data;
+ conf = this->private;
+ req.flags = args->flags;
+ req.key = (char *)args->name;
+
+ ret = client_submit_request (this, &req, frame, conf->handshake,
+ GF_HNDSK_GETSPEC, client3_getspec_cbk,
+ NULL, NULL, 0, NULL, 0, NULL,
+ (xdrproc_t)xdr_gf_getspec_req);
+
+ if (ret) {
+ gf_log (this->name, GF_LOG_WARNING,
+ "failed to send the request");
+ }
+
+ return 0;
+unwind:
+ CLIENT_STACK_UNWIND (getspec, frame, -1, op_errno, NULL);
+ return 0;
+
+}
+
+int
+client_notify_parents_child_up (xlator_t *this)
+{
+ clnt_conf_t *conf = NULL;
+ int ret = 0;
+
+ conf = this->private;
+ ret = default_notify (this, GF_EVENT_CHILD_UP, NULL);
+ if (ret)
+ gf_log (this->name, GF_LOG_INFO,
+ "notify of CHILD_UP failed");
+
+ conf->last_sent_event = GF_EVENT_CHILD_UP;
+ return 0;
+}
+
+int
+clnt_fd_lk_reacquire_failed (xlator_t *this, clnt_fd_ctx_t *fdctx,
+ clnt_conf_t *conf)
+{
+ int ret = -1;
+
+ GF_VALIDATE_OR_GOTO ("client", this, out);
+ GF_VALIDATE_OR_GOTO (this->name, conf, out);
+ GF_VALIDATE_OR_GOTO (this->name, fdctx, out);
+
+ pthread_mutex_lock (&conf->lock);
+ {
+ fdctx->remote_fd = -1;
+ fdctx->lk_heal_state = GF_LK_HEAL_DONE;
+ }
+ pthread_mutex_unlock (&conf->lock);
+
+ ret = 0;
+out:
+ return ret;
+}
+
+int
+client_set_lk_version_cbk (struct rpc_req *req, struct iovec *iov,
+ int count, void *myframe)
+{
+ int32_t ret = -1;
+ call_frame_t *fr = NULL;
+ gf_set_lk_ver_rsp rsp = {0,};
+
+ fr = (call_frame_t *) myframe;
+ GF_VALIDATE_OR_GOTO ("client", fr, out);
+
+ if (req->rpc_status == -1) {
+ gf_log (fr->this->name, GF_LOG_WARNING,
+ "received RPC status error");
+ goto out;
+ }
+
+ ret = xdr_to_generic (*iov, &rsp, (xdrproc_t)xdr_gf_set_lk_ver_rsp);
+ if (ret < 0)
+ gf_log (fr->this->name, GF_LOG_WARNING,
+ "xdr decoding failed");
+ else
+ gf_log (fr->this->name, GF_LOG_INFO,
+ "Server lk version = %d", rsp.lk_ver);
+
+ ret = 0;
+out:
+ if (fr)
+ STACK_DESTROY (fr->root);
+
+ return ret;
+}
+
+//TODO: Check for all released fdctx and destroy them
+int
+client_set_lk_version (xlator_t *this)
+{
+ int ret = -1;
+ clnt_conf_t *conf = NULL;
+ call_frame_t *frame = NULL;
+ gf_set_lk_ver_req req = {0, };
+
+ GF_VALIDATE_OR_GOTO ("client", this, err);
+
+ conf = (clnt_conf_t *) this->private;
+
+ req.lk_ver = client_get_lk_ver (conf);
+ ret = gf_asprintf (&req.uid, "%s-%s-%d",
+ this->ctx->process_uuid, this->name,
+ this->graph->id);
+ if (ret == -1)
+ goto err;
+
+ frame = create_frame (this, this->ctx->pool);
+ if (!frame) {
+ ret = -1;
+ goto out;
+ }
+
+ gf_log (this->name, GF_LOG_DEBUG, "Sending SET_LK_VERSION");
+
+ ret = client_submit_request (this, &req, frame,
+ conf->handshake,
+ GF_HNDSK_SET_LK_VER,
+ client_set_lk_version_cbk,
+ NULL, NULL, 0, NULL, 0, NULL,
+ (xdrproc_t)xdr_gf_set_lk_ver_req);
+out:
+ GF_FREE (req.uid);
+ return ret;
+err:
+ gf_log (this->name, GF_LOG_WARNING,
+ "Failed to send SET_LK_VERSION to server");
+
+ return ret;
+}
+
+int
+client_fd_lk_count (fd_lk_ctx_t *lk_ctx)
+{
+ int count = 0;
+ fd_lk_ctx_node_t *fd_lk = NULL;
+
+ GF_VALIDATE_OR_GOTO ("client", lk_ctx, err);
+
+ LOCK (&lk_ctx->lock);
+ {
+ list_for_each_entry (fd_lk, &lk_ctx->lk_list, next)
+ count++;
+ }
+ UNLOCK (&lk_ctx->lock);
+
+ return count;
+err:
+ return -1;
+}
+
+clnt_fd_lk_local_t *
+clnt_fd_lk_local_ref (xlator_t *this, clnt_fd_lk_local_t *local)
+{
+ GF_VALIDATE_OR_GOTO (this->name, local, out);
+
+ LOCK (&local->lock);
+ {
+ local->ref++;
+ }
+ UNLOCK (&local->lock);
+out:
+ return local;
+}
+
+int
+clnt_fd_lk_local_unref (xlator_t *this, clnt_fd_lk_local_t *local)
+{
+ int ref = -1;
+
+ GF_VALIDATE_OR_GOTO (this->name, local, out);
+
+ LOCK (&local->lock);
+ {
+ ref = --local->ref;
+ }
+ UNLOCK (&local->lock);
+
+ if (ref == 0) {
+ LOCK_DESTROY (&local->lock);
+ GF_FREE (local);
+ }
+out:
+ return ref;
+}
+
+clnt_fd_lk_local_t *
+clnt_fd_lk_local_create (clnt_fd_ctx_t *fdctx)
+{
+ clnt_fd_lk_local_t *local = NULL;
+
+ local = GF_CALLOC (1, sizeof (clnt_fd_lk_local_t),
+ gf_client_mt_clnt_fd_lk_local_t);
+ if (!local)
+ goto out;
+
+ local->ref = 1;
+ local->error = _gf_false;
+ local->fdctx = fdctx;
+
+ LOCK_INIT (&local->lock);
+out:
+ return local;
+}
+
+void
+clnt_mark_fd_bad (clnt_conf_t *conf, clnt_fd_ctx_t *fdctx)
+{
+ pthread_mutex_lock (&conf->lock);
+ {
+ fdctx->remote_fd = -1;
+ }
+ pthread_mutex_unlock (&conf->lock);
+}
+
+int
+clnt_release_reopen_fd_cbk (struct rpc_req *req, struct iovec *iov,
+ int count, void *myframe)
+{
+ xlator_t *this = NULL;
+ call_frame_t *frame = NULL;
+ clnt_conf_t *conf = NULL;
+ clnt_fd_ctx_t *fdctx = NULL;
+
+ frame = myframe;
+ this = frame->this;
+ fdctx = (clnt_fd_ctx_t *) frame->local;
+ conf = (clnt_conf_t *) this->private;
+
+ clnt_fd_lk_reacquire_failed (this, fdctx, conf);
+
+ fdctx->reopen_done (fdctx, this);
+
+ frame->local = NULL;
+ STACK_DESTROY (frame->root);
+
+ return 0;
+}
+
+int
+clnt_release_reopen_fd (xlator_t *this, clnt_fd_ctx_t *fdctx)
+{
+ int ret = -1;
+ clnt_conf_t *conf = NULL;
+ call_frame_t *frame = NULL;
+ gfs3_release_req req = {{0,},};
+
+ conf = (clnt_conf_t *) this->private;
+
+ frame = create_frame (this, this->ctx->pool);
+ if (!frame)
+ goto out;
+
+ frame->local = (void *) fdctx;
+ req.fd = fdctx->remote_fd;
+
+ ret = client_submit_request (this, &req, frame, conf->fops,
+ GFS3_OP_RELEASE,
+ clnt_release_reopen_fd_cbk, NULL,
+ NULL, 0, NULL, 0, NULL,
+ (xdrproc_t)xdr_gfs3_releasedir_req);
+ return 0;
+ out:
+ if (ret) {
+ clnt_fd_lk_reacquire_failed (this, fdctx, conf);
+ fdctx->reopen_done (fdctx, this);
+ if (frame) {
+ frame->local = NULL;
+ STACK_DESTROY (frame->root);
+ }
+ }
+ return 0;
+}
+
+int
+clnt_reacquire_lock_error (xlator_t *this, clnt_fd_ctx_t *fdctx,
+ clnt_conf_t *conf)
+{
+ int32_t ret = -1;
+
+ GF_VALIDATE_OR_GOTO ("client", this, out);
+ GF_VALIDATE_OR_GOTO (this->name, fdctx, out);
+ GF_VALIDATE_OR_GOTO (this->name, conf, out);
+
+ clnt_release_reopen_fd (this, fdctx);
+
+ ret = 0;
+out:
+ return ret;
+}
+
+gf_boolean_t
+clnt_fd_lk_local_error_status (xlator_t *this,
+ clnt_fd_lk_local_t *local)
+{
+ gf_boolean_t error = _gf_false;
+
+ LOCK (&local->lock);
+ {
+ error = local->error;
+ }
+ UNLOCK (&local->lock);
+
+ return error;
+}
+
+int
+clnt_fd_lk_local_mark_error (xlator_t *this,
+ clnt_fd_lk_local_t *local)
+{
+ int32_t ret = -1;
+ clnt_conf_t *conf = NULL;
+ gf_boolean_t error = _gf_false;
+
+ GF_VALIDATE_OR_GOTO ("client", this, out);
+ GF_VALIDATE_OR_GOTO (this->name, local, out);
+
+ conf = (clnt_conf_t *) this->private;
+
+ LOCK (&local->lock);
+ {
+ error = local->error;
+ local->error = _gf_true;
+ }
+ UNLOCK (&local->lock);
+
+ if (!error)
+ clnt_reacquire_lock_error (this, local->fdctx, conf);
+ ret = 0;
+out:
+ return ret;
+}
+
+int
+client_reacquire_lock_cbk (struct rpc_req *req, struct iovec *iov,
+ int count, void *myframe)
+{
+ int32_t ret = -1;
+ xlator_t *this = NULL;
+ gfs3_lk_rsp rsp = {0,};
+ call_frame_t *frame = NULL;
+ clnt_conf_t *conf = NULL;
+ clnt_fd_ctx_t *fdctx = NULL;
+ clnt_fd_lk_local_t *local = NULL;
+ struct gf_flock lock = {0,};
+
+ frame = (call_frame_t *) myframe;
+ this = frame->this;
+ local = (clnt_fd_lk_local_t *) frame->local;
+ conf = (clnt_conf_t *) this->private;
+
+ if (req->rpc_status == -1) {
+ gf_log ("client", GF_LOG_WARNING,
+ "request failed at rpc");
+ goto out;
+ }
+
+ ret = xdr_to_generic (*iov, &rsp, (xdrproc_t)xdr_gfs3_lk_rsp);
+ if (ret < 0) {
+ gf_log (this->name, GF_LOG_ERROR, "XDR decoding failed");
+ goto out;
+ }
+
+ if (rsp.op_ret == -1) {
+ gf_log (this->name, GF_LOG_ERROR, "lock request failed");
+ ret = -1;
+ goto out;
+ }
+
+ fdctx = local->fdctx;
+
+ gf_proto_flock_to_flock (&rsp.flock, &lock);
+
+ gf_log (this->name, GF_LOG_DEBUG, "%s type lock reacquired on file "
+ "with gfid %s from %"PRIu64 " to %"PRIu64,
+ get_lk_type (lock.l_type), uuid_utoa (fdctx->gfid),
+ lock.l_start, lock.l_start + lock.l_len);
+
+ if (!clnt_fd_lk_local_error_status (this, local) &&
+ clnt_fd_lk_local_unref (this, local) == 0) {
+ pthread_mutex_lock (&conf->lock);
+ {
+ fdctx->lk_heal_state = GF_LK_HEAL_DONE;
+ }
+ pthread_mutex_unlock (&conf->lock);
+
+ fdctx->reopen_done (fdctx, this);
+ }
+
+ ret = 0;
+out:
+ if (ret < 0) {
+ clnt_fd_lk_local_mark_error (this, local);
+
+ clnt_fd_lk_local_unref (this, local);
+ }
+
+ frame->local = NULL;
+ STACK_DESTROY (frame->root);
+
+ return ret;
+}
+
+int
+_client_reacquire_lock (xlator_t *this, clnt_fd_ctx_t *fdctx)
+{
+ int32_t ret = -1;
+ int32_t gf_cmd = 0;
+ int32_t gf_type = 0;
+ gfs3_lk_req req = {{0,},};
+ struct gf_flock flock = {0,};
+ fd_lk_ctx_t *lk_ctx = NULL;
+ clnt_fd_lk_local_t *local = NULL;
+ fd_lk_ctx_node_t *fd_lk = NULL;
+ call_frame_t *frame = NULL;
+ clnt_conf_t *conf = NULL;
+
+ conf = (clnt_conf_t *) this->private;
+ lk_ctx = fdctx->lk_ctx;
+
+ local = clnt_fd_lk_local_create (fdctx);
+ if (!local) {
+ gf_log (this->name, GF_LOG_WARNING, "clnt_fd_lk_local_create "
+ "failed, aborting reacquring of locks on %s.",
+ uuid_utoa (fdctx->gfid));
+ clnt_reacquire_lock_error (this, fdctx, conf);
+ goto out;
+ }
+
+ list_for_each_entry (fd_lk, &lk_ctx->lk_list, next) {
+ memcpy (&flock, &fd_lk->user_flock,
+ sizeof (struct gf_flock));
+
+ /* Always send F_SETLK even if the cmd was F_SETLKW */
+ /* to avoid frame being blocked if lock cannot be granted. */
+ ret = client_cmd_to_gf_cmd (F_SETLK, &gf_cmd);
+ if (ret) {
+ gf_log (this->name, GF_LOG_WARNING,
+ "client_cmd_to_gf_cmd failed, "
+ "aborting reacquiring of locks");
+ break;
+ }
+
+ gf_type = client_type_to_gf_type (flock.l_type);
+ req.fd = fdctx->remote_fd;
+ req.cmd = gf_cmd;
+ req.type = gf_type;
+ (void) gf_proto_flock_from_flock (&req.flock,
+ &flock);
+
+ memcpy (req.gfid, fdctx->gfid, 16);
+
+ frame = create_frame (this, this->ctx->pool);
+ if (!frame) {
+ ret = -1;
+ break;
+ }
+
+ frame->local = clnt_fd_lk_local_ref (this, local);
+ frame->root->lk_owner = fd_lk->user_flock.l_owner;
+
+ ret = client_submit_request (this, &req, frame,
+ conf->fops, GFS3_OP_LK,
+ client_reacquire_lock_cbk,
+ NULL, NULL, 0, NULL, 0, NULL,
+ (xdrproc_t)xdr_gfs3_lk_req);
+ if (ret) {
+ gf_log (this->name, GF_LOG_WARNING,
+ "reacquiring locks failed on file with gfid %s",
+ uuid_utoa (fdctx->gfid));
+ break;
+ }
+
+ ret = 0;
+ frame = NULL;
+ }
+
+ if (local)
+ (void) clnt_fd_lk_local_unref (this, local);
+out:
+ return ret;
+}
+
+int
+client_reacquire_lock (xlator_t *this, clnt_fd_ctx_t *fdctx)
+{
+ int32_t ret = -1;
+ fd_lk_ctx_t *lk_ctx = NULL;
+
+ GF_VALIDATE_OR_GOTO ("client", this, out);
+ GF_VALIDATE_OR_GOTO (this->name, fdctx, out);
+
+ if (client_fd_lk_list_empty (fdctx->lk_ctx, _gf_false)) {
+ gf_log (this->name, GF_LOG_DEBUG,
+ "fd lock list is empty");
+ fdctx->reopen_done (fdctx, this);
+ } else {
+ lk_ctx = fdctx->lk_ctx;
+
+ LOCK (&lk_ctx->lock);
+ {
+ (void) _client_reacquire_lock (this, fdctx);
+ }
+ UNLOCK (&lk_ctx->lock);
+ }
+ ret = 0;
+out:
+ return ret;
+}
+
+void
+client_default_reopen_done (clnt_fd_ctx_t *fdctx, xlator_t *this)
+{
+ gf_log_callingfn (this->name, GF_LOG_WARNING,
+ "This function should never be called");
+}
+
+void
+client_reopen_done (clnt_fd_ctx_t *fdctx, xlator_t *this)
+{
+ clnt_conf_t *conf = NULL;
+ gf_boolean_t destroy = _gf_false;
+
+ conf = this->private;
+
+ pthread_mutex_lock (&conf->lock);
+ {
+ fdctx->reopen_attempts = 0;
+ if (!fdctx->released)
+ list_add_tail (&fdctx->sfd_pos, &conf->saved_fds);
+ else
+ destroy = _gf_true;
+ fdctx->reopen_done = client_default_reopen_done;
+ }
+ pthread_mutex_unlock (&conf->lock);
+
+ if (destroy)
+ client_fdctx_destroy (this, fdctx);
+}
+
+void
+client_child_up_reopen_done (clnt_fd_ctx_t *fdctx, xlator_t *this)
+{
+ clnt_conf_t *conf = NULL;
+ uint64_t fd_count = 0;
+
+ conf = this->private;
+
+ LOCK (&conf->rec_lock);
+ {
+ fd_count = --(conf->reopen_fd_count);
+ }
+ UNLOCK (&conf->rec_lock);
+
+ client_reopen_done (fdctx, this);
+ if (fd_count == 0) {
+ gf_log (this->name, GF_LOG_INFO,
+ "last fd open'd/lock-self-heal'd - notifying CHILD-UP");
+ client_set_lk_version (this);
+ client_notify_parents_child_up (this);
+ }
+}
+
+int
+client3_3_reopen_cbk (struct rpc_req *req, struct iovec *iov, int count,
+ void *myframe)
+{
+ int32_t ret = -1;
+ gfs3_open_rsp rsp = {0,};
+ gf_boolean_t attempt_lock_recovery = _gf_false;
+ clnt_local_t *local = NULL;
+ clnt_conf_t *conf = NULL;
+ clnt_fd_ctx_t *fdctx = NULL;
+ call_frame_t *frame = NULL;
+ xlator_t *this = NULL;
+
+ frame = myframe;
+ this = frame->this;
+ conf = this->private;
+ local = frame->local;
+ fdctx = local->fdctx;
+
+ if (-1 == req->rpc_status) {
+ gf_log (frame->this->name, GF_LOG_WARNING,
+ "received RPC status error, returning ENOTCONN");
+ rsp.op_ret = -1;
+ rsp.op_errno = ENOTCONN;
+ goto out;
+ }
+
+ ret = xdr_to_generic (*iov, &rsp, (xdrproc_t)xdr_gfs3_open_rsp);
+ if (ret < 0) {
+ gf_log (frame->this->name, GF_LOG_ERROR, "XDR decoding failed");
+ rsp.op_ret = -1;
+ rsp.op_errno = EINVAL;
+ goto out;
+ }
+
+ if (rsp.op_ret < 0) {
+ gf_log (frame->this->name, GF_LOG_WARNING,
+ "reopen on %s failed (%s)",
+ local->loc.path, strerror (rsp.op_errno));
+ } else {
+ gf_log (frame->this->name, GF_LOG_DEBUG,
+ "reopen on %s succeeded (remote-fd = %"PRId64")",
+ local->loc.path, rsp.fd);
+ }
+
+ if (rsp.op_ret == -1) {
+ ret = -1;
+ goto out;
+ }
+
+ pthread_mutex_lock (&conf->lock);
+ {
+ fdctx->remote_fd = rsp.fd;
+ if (!fdctx->released) {
+ if (conf->lk_heal &&
+ !client_fd_lk_list_empty (fdctx->lk_ctx,
+ _gf_false)) {
+ attempt_lock_recovery = _gf_true;
+ fdctx->lk_heal_state = GF_LK_HEAL_IN_PROGRESS;
+ }
+ }
+ }
+ pthread_mutex_unlock (&conf->lock);
+
+ ret = 0;
+
+ if (attempt_lock_recovery) {
+ /* Delay decrementing the reopen fd count untill all the
+ locks corresponding to this fd are acquired.*/
+ gf_log (this->name, GF_LOG_DEBUG, "acquiring locks "
+ "on %s", local->loc.path);
+ ret = client_reacquire_lock (frame->this, local->fdctx);
+ if (ret) {
+ clnt_reacquire_lock_error (this, local->fdctx, conf);
+ gf_log (this->name, GF_LOG_WARNING, "acquiring locks "
+ "failed on %s", local->loc.path);
+ }
+ }
+
+out:
+ if (!attempt_lock_recovery)
+ fdctx->reopen_done (fdctx, this);
+
+ frame->local = NULL;
+ STACK_DESTROY (frame->root);
+
+ client_local_wipe (local);
+
+ return 0;
+}
+
+int
+client3_3_reopendir_cbk (struct rpc_req *req, struct iovec *iov, int count,
+ void *myframe)
+{
+ int32_t ret = -1;
+ gfs3_open_rsp rsp = {0,};
+ clnt_local_t *local = NULL;
+ clnt_conf_t *conf = NULL;
+ clnt_fd_ctx_t *fdctx = NULL;
+ call_frame_t *frame = NULL;
+
+ frame = myframe;
+ local = frame->local;
+ fdctx = local->fdctx;
+ conf = frame->this->private;
+
+
+ if (-1 == req->rpc_status) {
+ gf_log (frame->this->name, GF_LOG_WARNING,
+ "received RPC status error, returning ENOTCONN");
+ rsp.op_ret = -1;
+ rsp.op_errno = ENOTCONN;
+ goto out;
+ }
+
+ ret = xdr_to_generic (*iov, &rsp, (xdrproc_t)xdr_gfs3_opendir_rsp);
+ if (ret < 0) {
+ gf_log (frame->this->name, GF_LOG_ERROR, "XDR decoding failed");
+ rsp.op_ret = -1;
+ rsp.op_errno = EINVAL;
+ goto out;
+ }
+
+ if (rsp.op_ret < 0) {
+ gf_log (frame->this->name, GF_LOG_WARNING,
+ "reopendir on %s failed (%s)",
+ local->loc.path, strerror (rsp.op_errno));
+ } else {
+ gf_log (frame->this->name, GF_LOG_INFO,
+ "reopendir on %s succeeded (fd = %"PRId64")",
+ local->loc.path, rsp.fd);
+ }
+
+ if (-1 == rsp.op_ret) {
+ ret = -1;
+ goto out;
+ }
+
+ pthread_mutex_lock (&conf->lock);
+ {
+ fdctx->remote_fd = rsp.fd;
+ }
+ pthread_mutex_unlock (&conf->lock);
+
+out:
+ fdctx->reopen_done (fdctx, frame->this);
+
+ frame->local = NULL;
+ STACK_DESTROY (frame->root);
+ client_local_wipe (local);
+
+ return 0;
+}
+
+static int
+protocol_client_reopendir (clnt_fd_ctx_t *fdctx, xlator_t *this)
+{
+ int ret = -1;
+ gfs3_opendir_req req = {{0,},};
+ clnt_local_t *local = NULL;
+ call_frame_t *frame = NULL;
+ clnt_conf_t *conf = NULL;
+
+ conf = this->private;
+
+ local = mem_get0 (this->local_pool);
+ if (!local) {
+ ret = -1;
+ goto out;
+ }
+ local->fdctx = fdctx;
+
+ uuid_copy (local->loc.gfid, fdctx->gfid);
+ ret = loc_path (&local->loc, NULL);
+ if (ret < 0)
+ goto out;
+
+ frame = create_frame (this, this->ctx->pool);
+ if (!frame) {
+ ret = -1;
+ goto out;
+ }
+
+ memcpy (req.gfid, fdctx->gfid, 16);
+
+ gf_log (frame->this->name, GF_LOG_DEBUG,
+ "attempting reopen on %s", local->loc.path);
+
+ frame->local = local;
+
+ ret = client_submit_request (this, &req, frame, conf->fops,
+ GFS3_OP_OPENDIR,
+ client3_3_reopendir_cbk, NULL,
+ NULL, 0, NULL, 0, NULL,
+ (xdrproc_t)xdr_gfs3_opendir_req);
+ if (ret) {
+ gf_log (this->name, GF_LOG_ERROR,
+ "failed to send the re-opendir request");
+ }
+
+ return 0;
+
+out:
+ if (frame) {
+ frame->local = NULL;
+ STACK_DESTROY (frame->root);
+ }
+
+ if (local)
+ client_local_wipe (local);
+
+ fdctx->reopen_done (fdctx, this);
+
+ return 0;
+
+}
+
+static int
+protocol_client_reopenfile (clnt_fd_ctx_t *fdctx, xlator_t *this)
+{
+ int ret = -1;
+ gfs3_open_req req = {{0,},};
+ clnt_local_t *local = NULL;
+ call_frame_t *frame = NULL;
+ clnt_conf_t *conf = NULL;
+
+ conf = this->private;
+
+ frame = create_frame (this, this->ctx->pool);
+ if (!frame) {
+ ret = -1;
+ goto out;
+ }
+
+ local = mem_get0 (this->local_pool);
+ if (!local) {
+ ret = -1;
+ goto out;
+ }
+
+ local->fdctx = fdctx;
+ uuid_copy (local->loc.gfid, fdctx->gfid);
+ ret = loc_path (&local->loc, NULL);
+ if (ret < 0)
+ goto out;
+
+ frame->local = local;
+
+ memcpy (req.gfid, fdctx->gfid, 16);
+ req.flags = gf_flags_from_flags (fdctx->flags);
+ req.flags = req.flags & (~(O_TRUNC|O_CREAT|O_EXCL));
+
+ gf_log (frame->this->name, GF_LOG_DEBUG,
+ "attempting reopen on %s", local->loc.path);
+
+ ret = client_submit_request (this, &req, frame, conf->fops,
+ GFS3_OP_OPEN, client3_3_reopen_cbk, NULL,
+ NULL, 0, NULL, 0, NULL,
+ (xdrproc_t)xdr_gfs3_open_req);
+ if (ret) {
+ gf_log (this->name, GF_LOG_ERROR,
+ "failed to send the re-open request");
+ }
+
+ return 0;
+
+out:
+ if (frame) {
+ frame->local = NULL;
+ STACK_DESTROY (frame->root);
+ }
+
+ if (local)
+ client_local_wipe (local);
+
+ fdctx->reopen_done (fdctx, this);
+
+ return 0;
+
+}
+
+static void
+protocol_client_reopen (clnt_fd_ctx_t *fdctx, xlator_t *this)
+{
+ if (fdctx->is_dir)
+ protocol_client_reopendir (fdctx, this);
+ else
+ protocol_client_reopenfile (fdctx, this);
+}
+
+gf_boolean_t
+__is_fd_reopen_in_progress (clnt_fd_ctx_t *fdctx)
+{
+ if (fdctx->reopen_done == client_default_reopen_done)
+ return _gf_false;
+ return _gf_true;
+}
+
+void
+client_attempt_reopen (fd_t *fd, xlator_t *this)
+{
+ clnt_conf_t *conf = NULL;
+ clnt_fd_ctx_t *fdctx = NULL;
+ gf_boolean_t reopen = _gf_false;
+
+ if (!fd || !this)
+ goto out;
+
+ conf = this->private;
+ pthread_mutex_lock (&conf->lock);
+ {
+ fdctx = this_fd_get_ctx (fd, this);
+ if (!fdctx)
+ goto unlock;
+ if (__is_fd_reopen_in_progress (fdctx))
+ goto unlock;
+ if (fdctx->remote_fd != -1)
+ goto unlock;
+
+ if (fdctx->reopen_attempts == CLIENT_REOPEN_MAX_ATTEMPTS) {
+ reopen = _gf_true;
+ fdctx->reopen_done = client_reopen_done;
+ list_del_init (&fdctx->sfd_pos);
+ } else {
+ fdctx->reopen_attempts++;
+ }
+ }
+unlock:
+ pthread_mutex_unlock (&conf->lock);
+ if (reopen)
+ protocol_client_reopen (fdctx, this);
+out:
+ return;
+}
+
+int
+client_post_handshake (call_frame_t *frame, xlator_t *this)
+{
+ clnt_conf_t *conf = NULL;
+ clnt_fd_ctx_t *tmp = NULL;
+ clnt_fd_ctx_t *fdctx = NULL;
+ struct list_head reopen_head;
+
+ int count = 0;
+
+ if (!this || !this->private)
+ goto out;
+
+ conf = this->private;
+ INIT_LIST_HEAD (&reopen_head);
+
+ pthread_mutex_lock (&conf->lock);
+ {
+ list_for_each_entry_safe (fdctx, tmp, &conf->saved_fds,
+ sfd_pos) {
+ if (fdctx->remote_fd != -1)
+ continue;
+
+ fdctx->reopen_done = client_child_up_reopen_done;
+ list_del_init (&fdctx->sfd_pos);
+ list_add_tail (&fdctx->sfd_pos, &reopen_head);
+ count++;
+ }
+ }
+ pthread_mutex_unlock (&conf->lock);
+
+ /* Delay notifying CHILD_UP to parents
+ until all locks are recovered */
+ if (count > 0) {
+ gf_log (this->name, GF_LOG_INFO,
+ "%d fds open - Delaying child_up until they are re-opened",
+ count);
+ client_save_number_fds (conf, count);
+
+ list_for_each_entry_safe (fdctx, tmp, &reopen_head, sfd_pos) {
+ list_del_init (&fdctx->sfd_pos);
+
+ protocol_client_reopen (fdctx, this);
+ }
+ } else {
+ gf_log (this->name, GF_LOG_DEBUG,
+ "No fds to open - notifying all parents child up");
+ client_set_lk_version (this);
+ client_notify_parents_child_up (this);
+ }
+out:
+ return 0;
+}
+
+int
+client_setvolume_cbk (struct rpc_req *req, struct iovec *iov, int count, void *myframe)
+{
+ call_frame_t *frame = NULL;
+ clnt_conf_t *conf = NULL;
+ xlator_t *this = NULL;
+ dict_t *reply = NULL;
+ char *process_uuid = NULL;
+ char *remote_error = NULL;
+ char *remote_subvol = NULL;
+ gf_setvolume_rsp rsp = {0,};
+ int ret = 0;
+ int32_t op_ret = 0;
+ int32_t op_errno = 0;
+ gf_boolean_t auth_fail = _gf_false;
+ uint32_t lk_ver = 0;
+
+ frame = myframe;
+ this = frame->this;
+ conf = this->private;
+
+ if (-1 == req->rpc_status) {
+ gf_log (frame->this->name, GF_LOG_WARNING,
+ "received RPC status error");
+ op_ret = -1;
+ goto out;
+ }
+
+ ret = xdr_to_generic (*iov, &rsp, (xdrproc_t)xdr_gf_setvolume_rsp);
+ if (ret < 0) {
+ gf_log (this->name, GF_LOG_ERROR, "XDR decoding failed");
+ op_ret = -1;
+ goto out;
+ }
+ op_ret = rsp.op_ret;
+ op_errno = gf_error_to_errno (rsp.op_errno);
+ if (-1 == rsp.op_ret) {
+ gf_log (frame->this->name, GF_LOG_WARNING,
+ "failed to set the volume (%s)",
+ (op_errno)? strerror (op_errno) : "--");
+ }
+
+ reply = dict_new ();
+ if (!reply)
+ goto out;
+
+ if (rsp.dict.dict_len) {
+ ret = dict_unserialize (rsp.dict.dict_val,
+ rsp.dict.dict_len, &reply);
+ if (ret < 0) {
+ gf_log (frame->this->name, GF_LOG_WARNING,
+ "failed to unserialize buffer to dict");
+ goto out;
+ }
+ }
+
+ ret = dict_get_str (reply, "ERROR", &remote_error);
+ if (ret < 0) {
+ gf_log (this->name, GF_LOG_WARNING,
+ "failed to get ERROR string from reply dict");
+ }
+
+ ret = dict_get_str (reply, "process-uuid", &process_uuid);
+ if (ret < 0) {
+ gf_log (this->name, GF_LOG_WARNING,
+ "failed to get 'process-uuid' from reply dict");
+ }
+
+ if (op_ret < 0) {
+ gf_log (this->name, GF_LOG_ERROR,
+ "SETVOLUME on remote-host failed: %s",
+ remote_error ? remote_error : strerror (op_errno));
+ errno = op_errno;
+ if (remote_error &&
+ (strcmp ("Authentication failed", remote_error) == 0)) {
+ auth_fail = _gf_true;
+ op_ret = 0;
+ }
+ if (op_errno == ESTALE) {
+ ret = default_notify (this, GF_EVENT_VOLFILE_MODIFIED, NULL);
+ if (ret)
+ gf_log (this->name, GF_LOG_INFO,
+ "notify of VOLFILE_MODIFIED failed");
+ conf->last_sent_event = GF_EVENT_VOLFILE_MODIFIED;
+ }
+ goto out;
+ }
+
+ ret = dict_get_str (this->options, "remote-subvolume",
+ &remote_subvol);
+ if (ret || !remote_subvol) {
+ gf_log (this->name, GF_LOG_WARNING,
+ "failed to find key 'remote-subvolume' in the options");
+ goto out;
+ }
+
+ ret = dict_get_uint32 (reply, "clnt-lk-version", &lk_ver);
+ if (ret) {
+ gf_log (this->name, GF_LOG_WARNING,
+ "failed to find key 'clnt-lk-version' in the options");
+ goto out;
+ }
+
+ gf_log (this->name, GF_LOG_DEBUG, "clnt-lk-version = %d, "
+ "server-lk-version = %d", client_get_lk_ver (conf), lk_ver);
+ /* TODO: currently setpeer path is broken */
+ /*
+ if (process_uuid && req->conn &&
+ !strcmp (this->ctx->process_uuid, process_uuid)) {
+ rpc_transport_t *peer_trans = NULL;
+ uint64_t peertrans_int = 0;
+
+ ret = dict_get_uint64 (reply, "transport-ptr",
+ &peertrans_int);
+ if (ret)
+ goto out;
+
+ gf_log (this->name, GF_LOG_WARNING,
+ "attaching to the local volume '%s'",
+ remote_subvol);
+
+ peer_trans = (void *) (long) (peertrans_int);
+
+ rpc_transport_setpeer (req->conn->trans, peer_trans);
+ }
+ */
+
+ gf_log (this->name, GF_LOG_INFO,
+ "Connected to %s, attached to remote volume '%s'.",
+ conf->rpc->conn.trans->peerinfo.identifier,
+ remote_subvol);
+
+ rpc_clnt_set_connected (&conf->rpc->conn);
+
+ op_ret = 0;
+ conf->connecting = 0;
+ conf->connected = 1;
+
+ conf->need_different_port = 0;
+
+ if (lk_ver != client_get_lk_ver (conf)) {
+ gf_log (this->name, GF_LOG_INFO, "Server and Client "
+ "lk-version numbers are not same, reopening the fds");
+ client_mark_fd_bad (this);
+ client_post_handshake (frame, frame->this);
+ } else {
+ /*TODO: Traverse the saved fd list, and send
+ release to the server on fd's that were closed
+ during grace period */
+ gf_log (this->name, GF_LOG_INFO, "Server and Client "
+ "lk-version numbers are same, no need to "
+ "reopen the fds");
+ }
+
+out:
+ if (auth_fail) {
+ gf_log (this->name, GF_LOG_INFO, "sending AUTH_FAILED event");
+ ret = default_notify (this, GF_EVENT_AUTH_FAILED, NULL);
+ if (ret)
+ gf_log (this->name, GF_LOG_INFO,
+ "notify of AUTH_FAILED failed");
+ conf->connecting = 0;
+ conf->connected = 0;
+ conf->last_sent_event = GF_EVENT_AUTH_FAILED;
+ ret = -1;
+ }
+ if (-1 == op_ret) {
+ /* Let the connection/re-connection happen in
+ * background, for now, don't hang here,
+ * tell the parents that i am all ok..
+ */
+ gf_log (this->name, GF_LOG_INFO, "sending CHILD_CONNECTING event");
+ ret = default_notify (this, GF_EVENT_CHILD_CONNECTING, NULL);
+ if (ret)
+ gf_log (this->name, GF_LOG_INFO,
+ "notify of CHILD_CONNECTING failed");
+ conf->last_sent_event = GF_EVENT_CHILD_CONNECTING;
+ conf->connecting= 1;
+ ret = 0;
+ }
+
+ free (rsp.dict.dict_val);
+
+ STACK_DESTROY (frame->root);
+
+ if (reply)
+ dict_unref (reply);
+
+ return ret;
+}
+
+int
+client_setvolume (xlator_t *this, struct rpc_clnt *rpc)
+{
+ int ret = 0;
+ gf_setvolume_req req = {{0,},};
+ call_frame_t *fr = NULL;
+ char *process_uuid_xl = NULL;
+ clnt_conf_t *conf = NULL;
+ dict_t *options = NULL;
+
+ options = this->options;
+ conf = this->private;
+
+ if (conf->fops) {
+ ret = dict_set_int32 (options, "fops-version",
+ conf->fops->prognum);
+ if (ret < 0) {
+ gf_log (this->name, GF_LOG_ERROR,
+ "failed to set version-fops(%d) in handshake msg",
+ conf->fops->prognum);
+ goto fail;
+ }
+ }
+
+ if (conf->mgmt) {
+ ret = dict_set_int32 (options, "mgmt-version", conf->mgmt->prognum);
+ if (ret < 0) {
+ gf_log (this->name, GF_LOG_ERROR,
+ "failed to set version-mgmt(%d) in handshake msg",
+ conf->mgmt->prognum);
+ goto fail;
+ }
+ }
+
+ /* With multiple graphs possible in the same process, we need a
+ field to bring the uniqueness. Graph-ID should be enough to get the
+ job done
+ */
+ ret = gf_asprintf (&process_uuid_xl, "%s-%s-%d",
+ this->ctx->process_uuid, this->name,
+ this->graph->id);
+ if (-1 == ret) {
+ gf_log (this->name, GF_LOG_ERROR,
+ "asprintf failed while setting process_uuid");
+ goto fail;
+ }
+
+ ret = dict_set_dynstr (options, "process-uuid", process_uuid_xl);
+ if (ret < 0) {
+ gf_log (this->name, GF_LOG_ERROR,
+ "failed to set process-uuid(%s) in handshake msg",
+ process_uuid_xl);
+ goto fail;
+ }
+
+ ret = dict_set_str (options, "client-version", PACKAGE_VERSION);
+ if (ret < 0) {
+ gf_log (this->name, GF_LOG_WARNING,
+ "failed to set client-version(%s) in handshake msg",
+ PACKAGE_VERSION);
+ }
+
+ if (this->ctx->cmd_args.volfile_server) {
+ if (this->ctx->cmd_args.volfile_id) {
+ ret = dict_set_str (options, "volfile-key",
+ this->ctx->cmd_args.volfile_id);
+ if (ret)
+ gf_log (this->name, GF_LOG_ERROR,
+ "failed to set 'volfile-key'");
+ }
+ ret = dict_set_uint32 (options, "volfile-checksum",
+ this->graph->volfile_checksum);
+ if (ret)
+ gf_log (this->name, GF_LOG_ERROR,
+ "failed to set 'volfile-checksum'");
+ }
+
+ ret = dict_set_int16 (options, "clnt-lk-version",
+ client_get_lk_ver (conf));
+ if (ret < 0) {
+ gf_log (this->name, GF_LOG_WARNING,
+ "failed to set clnt-lk-version(%"PRIu32") in handshake msg",
+ client_get_lk_ver (conf));
+ }
+
+ ret = dict_serialized_length (options);
+ if (ret < 0) {
+ gf_log (this->name, GF_LOG_ERROR,
+ "failed to get serialized length of dict");
+ ret = -1;
+ goto fail;
+ }
+ req.dict.dict_len = ret;
+ req.dict.dict_val = GF_CALLOC (1, req.dict.dict_len,
+ gf_client_mt_clnt_req_buf_t);
+ ret = dict_serialize (options, req.dict.dict_val);
+ if (ret < 0) {
+ gf_log (this->name, GF_LOG_ERROR,
+ "failed to serialize dictionary");
+ goto fail;
+ }
+
+ fr = create_frame (this, this->ctx->pool);
+ if (!fr)
+ goto fail;
+
+ ret = client_submit_request (this, &req, fr, conf->handshake,
+ GF_HNDSK_SETVOLUME, client_setvolume_cbk,
+ NULL, NULL, 0, NULL, 0, NULL,
+ (xdrproc_t)xdr_gf_setvolume_req);
+
+fail:
+ GF_FREE (req.dict.dict_val);
+
+ return ret;
+}
+
+int
+select_server_supported_programs (xlator_t *this, gf_prog_detail *prog)
+{
+ gf_prog_detail *trav = NULL;
+ clnt_conf_t *conf = NULL;
+ int ret = -1;
+
+ if (!this || !prog) {
+ gf_log (THIS->name, GF_LOG_WARNING,
+ "xlator not found OR RPC program not found");
+ goto out;
+ }
+
+ conf = this->private;
+ trav = prog;
+
+ while (trav) {
+ /* Select 'programs' */
+ if ((clnt3_3_fop_prog.prognum == trav->prognum) &&
+ (clnt3_3_fop_prog.progver == trav->progver)) {
+ conf->fops = &clnt3_3_fop_prog;
+ gf_log (this->name, GF_LOG_INFO,
+ "Using Program %s, Num (%"PRId64"), "
+ "Version (%"PRId64")",
+ trav->progname, trav->prognum, trav->progver);
+ ret = 0;
+ }
+ if (ret) {
+ gf_log (this->name, GF_LOG_TRACE,
+ "%s (%"PRId64") not supported", trav->progname,
+ trav->progver);
+ }
+ trav = trav->next;
+ }
+
+out:
+ return ret;
+}
+
+
+int
+server_has_portmap (xlator_t *this, gf_prog_detail *prog)
+{
+ gf_prog_detail *trav = NULL;
+ int ret = -1;
+
+ if (!this || !prog) {
+ gf_log (THIS->name, GF_LOG_WARNING,
+ "xlator not found OR RPC program not found");
+ goto out;
+ }
+
+ trav = prog;
+
+ while (trav) {
+ if ((trav->prognum == GLUSTER_PMAP_PROGRAM) &&
+ (trav->progver == GLUSTER_PMAP_VERSION)) {
+ gf_log (this->name, GF_LOG_DEBUG,
+ "detected portmapper on server");
+ ret = 0;
+ break;
+ }
+ trav = trav->next;
+ }
+
+out:
+ return ret;
+}
+
+
+int
+client_query_portmap_cbk (struct rpc_req *req, struct iovec *iov, int count, void *myframe)
+{
+ struct pmap_port_by_brick_rsp rsp = {0,};
+ call_frame_t *frame = NULL;
+ clnt_conf_t *conf = NULL;
+ int ret = -1;
+ struct rpc_clnt_config config = {0, };
+ xlator_t *this = NULL;
+
+ frame = myframe;
+ if (!frame || !frame->this || !frame->this->private) {
+ gf_log (THIS->name, GF_LOG_WARNING,
+ "frame not found with rpc request");
+ goto out;
+ }
+ this = frame->this;
+ conf = frame->this->private;
+
+ if (-1 == req->rpc_status) {
+ gf_log (this->name, GF_LOG_WARNING,
+ "received RPC status error, try again later");
+ goto out;
+ }
+
+ ret = xdr_to_generic (*iov, &rsp, (xdrproc_t)xdr_pmap_port_by_brick_rsp);
+ if (ret < 0) {
+ gf_log (this->name, GF_LOG_ERROR, "XDR decoding failed");
+ goto out;
+ }
+
+ if (-1 == rsp.op_ret) {
+ ret = -1;
+ gf_log (this->name, ((!conf->portmap_err_logged) ?
+ GF_LOG_ERROR : GF_LOG_DEBUG),
+ "failed to get the port number for remote subvolume. "
+ "Please run 'gluster volume status' on server to see "
+ "if brick process is running.");
+ conf->portmap_err_logged = 1;
+ goto out;
+ }
+
+ conf->portmap_err_logged = 0;
+ conf->disconnect_err_logged = 0;
+
+ config.remote_port = rsp.port;
+ rpc_clnt_reconfig (conf->rpc, &config);
+
+ conf->skip_notify = 1;
+ conf->quick_reconnect = 1;
+
+out:
+ if (frame)
+ STACK_DESTROY (frame->root);
+
+ if (conf) {
+ /* Need this to connect the same transport on different port */
+ /* ie, glusterd to glusterfsd */
+ rpc_transport_disconnect (conf->rpc->conn.trans);
+ }
+
+ return ret;
+}
+
+
+int
+client_query_portmap (xlator_t *this, struct rpc_clnt *rpc)
+{
+ int ret = -1;
+ pmap_port_by_brick_req req = {0,};
+ call_frame_t *fr = NULL;
+ clnt_conf_t *conf = NULL;
+ dict_t *options = NULL;
+ char *remote_subvol = NULL;
+ char *xprt = NULL;
+ char brick_name[PATH_MAX] = {0,};
+
+ options = this->options;
+ conf = this->private;
+
+ ret = dict_get_str (options, "remote-subvolume", &remote_subvol);
+ if (ret < 0) {
+ gf_log (this->name, GF_LOG_ERROR,
+ "remote-subvolume not set in volfile");
+ goto fail;
+ }
+
+ req.brick = remote_subvol;
+
+ /* FIXME: Dirty work around */
+ if (!dict_get_str (options, "transport-type", &xprt)) {
+ /* This logic is required only in case of 'rdma' client
+ transport-type and the volume is of 'tcp,rdma'
+ transport type. */
+ if (!strcmp (xprt, "rdma")) {
+ if (!conf->need_different_port) {
+ snprintf (brick_name, PATH_MAX, "%s.rdma",
+ remote_subvol);
+ req.brick = brick_name;
+ conf->need_different_port = 1;
+ conf->skip_notify = 1;
+ } else {
+ conf->need_different_port = 0;
+ conf->skip_notify = 0;
+ }
+ }
+ }
+
+ fr = create_frame (this, this->ctx->pool);
+ if (!fr) {
+ ret = -1;
+ goto fail;
+ }
+
+ ret = client_submit_request (this, &req, fr, &clnt_pmap_prog,
+ GF_PMAP_PORTBYBRICK,
+ client_query_portmap_cbk,
+ NULL, NULL, 0, NULL, 0, NULL,
+ (xdrproc_t)xdr_pmap_port_by_brick_req);
+
+fail:
+ return ret;
+}
+
+
+int
+client_dump_version_cbk (struct rpc_req *req, struct iovec *iov, int count,
+ void *myframe)
+{
+ gf_dump_rsp rsp = {0,};
+ gf_prog_detail *trav = NULL;
+ gf_prog_detail *next = NULL;
+ call_frame_t *frame = NULL;
+ clnt_conf_t *conf = NULL;
+ int ret = 0;
+
+ frame = myframe;
+ conf = frame->this->private;
+
+ if (-1 == req->rpc_status) {
+ gf_log (frame->this->name, GF_LOG_WARNING,
+ "received RPC status error");
+ goto out;
+ }
+
+ ret = xdr_to_generic (*iov, &rsp, (xdrproc_t)xdr_gf_dump_rsp);
+ if (ret < 0) {
+ gf_log (frame->this->name, GF_LOG_ERROR, "XDR decoding failed");
+ goto out;
+ }
+ if (-1 == rsp.op_ret) {
+ gf_log (frame->this->name, GF_LOG_WARNING,
+ "failed to get the 'versions' from server");
+ goto out;
+ }
+
+ if (server_has_portmap (frame->this, rsp.prog) == 0) {
+ ret = client_query_portmap (frame->this, conf->rpc);
+ goto out;
+ }
+
+ /* Check for the proper version string */
+ /* Reply in "Name:Program-Number:Program-Version,..." format */
+ ret = select_server_supported_programs (frame->this, rsp.prog);
+ if (ret) {
+ gf_log (frame->this->name, GF_LOG_ERROR,
+ "server doesn't support the version");
+ goto out;
+ }
+
+ client_setvolume (frame->this, conf->rpc);
+
+out:
+ /* don't use GF_FREE, buffer was allocated by libc */
+ if (rsp.prog) {
+ trav = rsp.prog;
+ while (trav) {
+ next = trav->next;
+ free (trav->progname);
+ free (trav);
+ trav = next;
+ }
+ }
+
+ STACK_DESTROY (frame->root);
+
+ if (ret != 0)
+ rpc_transport_disconnect (conf->rpc->conn.trans);
+
+ return ret;
+}
+
+int
+client_handshake (xlator_t *this, struct rpc_clnt *rpc)
+{
+ call_frame_t *frame = NULL;
+ clnt_conf_t *conf = NULL;
+ gf_dump_req req = {0,};
+ int ret = 0;
+
+ conf = this->private;
+ if (!conf->handshake) {
+ gf_log (this->name, GF_LOG_WARNING, "handshake program not found");
+ goto out;
+ }
+
+ frame = create_frame (this, this->ctx->pool);
+ if (!frame)
+ goto out;
+
+ req.gfs_id = 0xbabe;
+ ret = client_submit_request (this, &req, frame, conf->dump,
+ GF_DUMP_DUMP, client_dump_version_cbk,
+ NULL, NULL, 0, NULL, 0,
+ NULL, (xdrproc_t)xdr_gf_dump_req);
+
+out:
+ return ret;
+}
+
+char *clnt_handshake_procs[GF_HNDSK_MAXVALUE] = {
+ [GF_HNDSK_NULL] = "NULL",
+ [GF_HNDSK_SETVOLUME] = "SETVOLUME",
+ [GF_HNDSK_GETSPEC] = "GETSPEC",
+ [GF_HNDSK_PING] = "PING",
+ [GF_HNDSK_SET_LK_VER] = "SET_LK_VER"
+};
+
+rpc_clnt_prog_t clnt_handshake_prog = {
+ .progname = "GlusterFS Handshake",
+ .prognum = GLUSTER_HNDSK_PROGRAM,
+ .progver = GLUSTER_HNDSK_VERSION,
+ .procnames = clnt_handshake_procs,
+};
+
+char *clnt_dump_proc[GF_DUMP_MAXVALUE] = {
+ [GF_DUMP_NULL] = "NULL",
+ [GF_DUMP_DUMP] = "DUMP",
+};
+
+rpc_clnt_prog_t clnt_dump_prog = {
+ .progname = "GF-DUMP",
+ .prognum = GLUSTER_DUMP_PROGRAM,
+ .progver = GLUSTER_DUMP_VERSION,
+ .procnames = clnt_dump_proc,
+};
+
+char *clnt_pmap_procs[GF_PMAP_MAXVALUE] = {
+ [GF_PMAP_PORTBYBRICK] = "PORTBYBRICK",
+};
+
+rpc_clnt_prog_t clnt_pmap_prog = {
+ .progname = "PORTMAP",
+ .prognum = GLUSTER_PMAP_PROGRAM,
+ .progver = GLUSTER_PMAP_VERSION,
+ .procnames = clnt_pmap_procs,
+};
generated by cgit (git 2.34.1) at 2026-01-05 12:25:19 +0000