/* Copyright (c) 2010 Gluster, Inc. This file is part of GlusterFS. GlusterFS is free software; you can redistribute it and/or modify it under the terms of the GNU Affero 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 Affero General Public License for more details. You should have received a copy of the GNU Affero General Public License along with this program. If not, see . */ #ifndef _CONFIG_H #define _CONFIG_H #include "config.h" #endif #include "client.h" #include "xlator.h" #include "defaults.h" #include "glusterfs.h" #include "statedump.h" #include "compat-errno.h" #include "glusterfs3.h" #include "portmap.h" extern rpc_clnt_prog_t clnt3_1_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); /* 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 timeval 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 (¤t, 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_usec = 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 timeval 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_usec = 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, NULL, 0, NULL, 0, NULL); if (ret) goto fail; return; fail: gf_log (THIS->name, GF_LOG_ERROR, "failed to start ping timer"); 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 timeval 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; 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 out; } pthread_mutex_lock (&conn->lock); { timeout.tv_sec = conf->opt.ping_timeout; timeout.tv_usec = 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"); } 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_getspec_rsp (*iov, &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: STACK_UNWIND_STRICT (getspec, frame, rsp.op_ret, rsp.op_errno, rsp.spec); /* Don't use 'GF_FREE', this is allocated by libc */ if (rsp.spec) 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, xdr_from_getspec_req, NULL, 0, NULL, 0, NULL); if (ret) goto unwind; return 0; unwind: gf_log (this->name, GF_LOG_WARNING, "failed to send the request"); STACK_UNWIND_STRICT (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 client3_1_reopen_cbk (struct rpc_req *req, struct iovec *iov, int count, void *myframe) { int32_t ret = -1; gfs3_open_rsp rsp = {0,}; int attempt_lock_recovery = _gf_false; uint64_t fd_count = 0; clnt_local_t *local = NULL; clnt_conf_t *conf = NULL; clnt_fd_ctx_t *fdctx = NULL; call_frame_t *frame = NULL; frame = myframe; if (!frame || !frame->this) goto out; local = frame->local; 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_open_rsp (*iov, &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; } fdctx = local->fdctx; if (!fdctx) { gf_log (frame->this->name, GF_LOG_WARNING, "fdctx not found"); ret = -1; goto out; } pthread_mutex_lock (&conf->lock); { fdctx->remote_fd = rsp.fd; if (!fdctx->released) { list_add_tail (&fdctx->sfd_pos, &conf->saved_fds); if (!list_empty (&fdctx->lock_list)) attempt_lock_recovery = _gf_true; fdctx = NULL; } } pthread_mutex_unlock (&conf->lock); ret = 0; attempt_lock_recovery = _gf_false; /* temporarily */ if (attempt_lock_recovery) { ret = client_attempt_lock_recovery (frame->this, local->fdctx); if (ret < 0) { gf_log (frame->this->name, GF_LOG_DEBUG, "lock recovery not attempted on fd"); } else { gf_log (frame->this->name, GF_LOG_INFO, "need to attempt lock recovery on %"PRIu64 " open fds", fd_count); } } else { fd_count = decrement_reopen_fd_count (frame->this, conf); } out: if (fdctx) client_fdctx_destroy (frame->this, fdctx); if ((ret < 0) && frame && frame->this && conf) decrement_reopen_fd_count (frame->this, conf); frame->local = NULL; STACK_DESTROY (frame->root); client_local_wipe (local); return 0; } int client3_1_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; if (!frame || !frame->this) goto out; local = frame->local; 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_opendir_rsp (*iov, &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; } fdctx = local->fdctx; if (!fdctx) { gf_log (frame->this->name, GF_LOG_WARNING, "fdctx not found"); ret = -1; goto out; } pthread_mutex_lock (&conf->lock); { fdctx->remote_fd = rsp.fd; if (!fdctx->released) { list_add_tail (&fdctx->sfd_pos, &conf->saved_fds); fdctx = NULL; } } pthread_mutex_unlock (&conf->lock); decrement_reopen_fd_count (frame->this, conf); ret = 0; out: if (fdctx) client_fdctx_destroy (frame->this, fdctx); if ((ret < 0) && frame && frame->this && conf) decrement_reopen_fd_count (frame->this, conf); if (frame) { frame->local = NULL; STACK_DESTROY (frame->root); } client_local_wipe (local); return 0; } int protocol_client_reopendir (xlator_t *this, clnt_fd_ctx_t *fdctx) { int ret = -1; gfs3_opendir_req req = {{0,},}; clnt_local_t *local = NULL; inode_t *inode = NULL; char *path = NULL; call_frame_t *frame = NULL; clnt_conf_t *conf = NULL; if (!this || !fdctx) goto out; inode = fdctx->inode; conf = this->private; ret = inode_path (inode, NULL, &path); if (ret < 0) { gf_log (this->name, GF_LOG_WARNING, "couldn't build path from inode %s", uuid_utoa (inode->gfid)); goto out; } local = GF_CALLOC (1, sizeof (*local), gf_client_mt_clnt_local_t); if (!local) { ret = -1; goto out; } local->fdctx = fdctx; local->loc.path = path; path = NULL; frame = create_frame (this, this->ctx->pool); if (!frame) { ret = -1; goto out; } memcpy (req.gfid, inode->gfid, 16); req.path = (char *)local->loc.path; gf_log (frame->this->name, GF_LOG_DEBUG, "attempting reopen on %s", local->loc.path); frame->local = local; local = NULL; ret = client_submit_request (this, &req, frame, conf->fops, GFS3_OP_OPENDIR, client3_1_reopendir_cbk, NULL, xdr_from_opendir_req, NULL, 0, NULL, 0, NULL); if (ret) goto out; return ret; out: gf_log (THIS->name, GF_LOG_ERROR, "failed to send the re-opendir request"); if (frame) { frame->local = NULL; STACK_DESTROY (frame->root); } if (local) client_local_wipe (local); if (path) GF_FREE (path); if ((ret < 0) && this && conf) { decrement_reopen_fd_count (this, conf); } return 0; } int protocol_client_reopen (xlator_t *this, clnt_fd_ctx_t *fdctx) { int ret = -1; gfs3_open_req req = {{0,},}; clnt_local_t *local = NULL; inode_t *inode = NULL; char *path = NULL; call_frame_t *frame = NULL; clnt_conf_t *conf = NULL; if (!this || !fdctx) goto out; inode = fdctx->inode; conf = this->private; ret = inode_path (inode, NULL, &path); if (ret < 0) { gf_log (this->name, GF_LOG_WARNING, "couldn't build path from inode %s", uuid_utoa (inode->gfid)); goto out; } frame = create_frame (this, this->ctx->pool); if (!frame) { ret = -1; goto out; } local = GF_CALLOC (1, sizeof (*local), gf_client_mt_clnt_local_t); if (!local) { ret = -1; goto out; } local->fdctx = fdctx; local->loc.path = path; path = NULL; frame->local = local; memcpy (req.gfid, inode->gfid, 16); req.flags = gf_flags_from_flags (fdctx->flags); req.wbflags = fdctx->wbflags; req.path = (char *)local->loc.path; gf_log (frame->this->name, GF_LOG_DEBUG, "attempting reopen on %s", local->loc.path); local = NULL; ret = client_submit_request (this, &req, frame, conf->fops, GFS3_OP_OPEN, client3_1_reopen_cbk, NULL, xdr_from_open_req, NULL, 0, NULL, 0, NULL); if (ret) goto out; return ret; out: gf_log (THIS->name, GF_LOG_ERROR, "failed to send the re-open request"); if (frame) { frame->local = NULL; STACK_DESTROY (frame->root); } if (local) client_local_wipe (local); if (path) GF_FREE (path); if ((ret < 0) && this && conf) { decrement_reopen_fd_count (this, conf); } return 0; } 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; 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); if (fdctx->is_dir) protocol_client_reopendir (this, fdctx); else protocol_client_reopen (this, fdctx); } } else { gf_log (this->name, GF_LOG_DEBUG, "no open fds - notifying all parents child up"); 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; 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_setvolume_rsp (*iov, &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 (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; } /* 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; /* TODO: more to test */ client_post_handshake (frame, frame->this); out: 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; } if (rsp.dict.dict_val) free (rsp.dict.dict_val); STACK_DESTROY (frame->root); if (reply) dict_unref (reply); return 0; } 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; } } ret = gf_asprintf (&process_uuid_xl, "%s-%s", this->ctx->process_uuid, this->name); 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; } 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'"); } req.dict.dict_len = dict_serialized_length (options); if (req.dict.dict_len < 0) { gf_log (this->name, GF_LOG_ERROR, "failed to get serialized length of dict"); ret = -1; goto fail; } 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, xdr_from_setvolume_req, NULL, 0, NULL, 0, NULL); fail: if (req.dict.dict_val) 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_1_fop_prog.prognum == trav->prognum) && (clnt3_1_fop_prog.progver == trav->progver)) { conf->fops = &clnt3_1_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_pmap_port_by_brick_rsp (*iov, &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"); conf->portmap_err_logged = 1; goto out; } conf->portmap_err_logged = 0; config.remote_port = rsp.port; rpc_clnt_reconfig (conf->rpc, &config); conf->skip_notify = 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); rpc_clnt_reconnect (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; } else { conf->need_different_port = 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, xdr_from_pmap_port_by_brick_req, NULL, 0, NULL, 0, NULL); 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_dump_rsp (*iov, &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, xdr_from_dump_req, NULL, 0, NULL, 0, NULL); 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", }; 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, };