summaryrefslogtreecommitdiffstats
path: root/rpc/rpc-lib
diff options
context:
space:
mode:
Diffstat (limited to 'rpc/rpc-lib')
-rw-r--r--rpc/rpc-lib/src/rpc-transport.c7
-rw-r--r--rpc/rpc-lib/src/rpcsvc.c262
-rw-r--r--rpc/rpc-lib/src/rpcsvc.h7
3 files changed, 87 insertions, 189 deletions
diff --git a/rpc/rpc-lib/src/rpc-transport.c b/rpc/rpc-lib/src/rpc-transport.c
index e3bc519fc97..adbc0fe1e2b 100644
--- a/rpc/rpc-lib/src/rpc-transport.c
+++ b/rpc/rpc-lib/src/rpc-transport.c
@@ -1119,8 +1119,11 @@ rpc_transport_notify (rpc_transport_t *this, rpc_transport_event_t event,
goto out;
}
- //ret = this->notify (this, this->notify_data, event, data);
- ret = this->notify (this, this->mydata, event, data);
+ if (this->notify != NULL) {
+ ret = this->notify (this, this->mydata, event, data);
+ } else {
+ ret = 0;
+ }
out:
return ret;
}
diff --git a/rpc/rpc-lib/src/rpcsvc.c b/rpc/rpc-lib/src/rpcsvc.c
index 2b6ed58f07b..88675205c90 100644
--- a/rpc/rpc-lib/src/rpcsvc.c
+++ b/rpc/rpc-lib/src/rpcsvc.c
@@ -1360,7 +1360,7 @@ out:
}
-int
+inline int
rpcsvc_program_unregister_portmap (rpcsvc_program_t *prog)
{
if (!prog)
@@ -1376,6 +1376,38 @@ rpcsvc_program_unregister_portmap (rpcsvc_program_t *prog)
}
+int32_t
+rpcsvc_get_listener_port (rpcsvc_listener_t *listener)
+{
+ int32_t listener_port = -1;
+
+ if ((listener == NULL) || (listener->trans == NULL)) {
+ goto out;
+ }
+
+ switch (listener->trans->myinfo.sockaddr.ss_family) {
+ case AF_INET:
+ listener_port = ((struct sockaddr_in6 *)&listener->trans->myinfo.sockaddr)->sin6_port;
+ break;
+
+ case AF_INET6:
+ listener_port = ((struct sockaddr_in *)&listener->trans->myinfo.sockaddr)->sin_port;
+ break;
+
+ default:
+ gf_log (GF_RPCSVC, GF_LOG_DEBUG,
+ "invalid address family (%d)",
+ listener->trans->myinfo.sockaddr.ss_family);
+ goto out;
+ }
+
+ listener_port = ntohs (listener_port);
+
+out:
+ return listener_port;
+}
+
+
rpcsvc_listener_t *
rpcsvc_get_listener (rpcsvc_t *svc, uint16_t port, rpc_transport_t *trans)
{
@@ -1398,34 +1430,21 @@ rpcsvc_get_listener (rpcsvc_t *svc, uint16_t port, rpc_transport_t *trans)
continue;
}
-
- switch (listener->trans->myinfo.sockaddr.ss_family) {
- case AF_INET:
- listener_port
- = ((struct sockaddr_in6 *)&listener->trans->myinfo.sockaddr)->sin6_port;
- break;
-
- case AF_INET6:
- listener_port
- = ((struct sockaddr_in *)&listener->trans->myinfo.sockaddr)->sin_port;
- break;
- default:
+ listener_port = rpcsvc_get_listener_port (listener);
+ if (listener_port == -1) {
gf_log (GF_RPCSVC, GF_LOG_DEBUG,
- "invalid address family (%d)",
- listener->trans->myinfo.sockaddr.ss_family);
- goto unlock;
+ "invalid port for listener %s",
+ listener->trans->name);
+ continue;
}
- listener_port = ntohs (listener_port);
-
if (listener_port == port) {
found = 1;
break;
}
}
}
-unlock:
pthread_mutex_unlock (&svc->rpclock);
if (!found) {
@@ -1457,31 +1476,38 @@ rpcsvc_submit_message (rpcsvc_request_t *req, struct iovec *proghdr,
int
-rpcsvc_program_unregister (rpcsvc_t *svc, rpcsvc_program_t prog)
+rpcsvc_program_unregister (rpcsvc_t *svc, rpcsvc_program_t *prog)
{
int ret = -1;
- if (!svc)
- return -1;
+ if (!svc || !prog) {
+ goto out;
+ }
- /* TODO: De-init the listening connection for this program. */
- ret = rpcsvc_program_unregister_portmap (&prog);
+ ret = rpcsvc_program_unregister_portmap (prog);
if (ret == -1) {
gf_log (GF_RPCSVC, GF_LOG_ERROR, "portmap unregistration of"
" program failed");
- goto err;
+ goto out;
}
- ret = 0;
gf_log (GF_RPCSVC, GF_LOG_DEBUG, "Program unregistered: %s, Num: %d,"
- " Ver: %d, Port: %d", prog.progname, prog.prognum,
- prog.progver, prog.progport);
+ " Ver: %d, Port: %d", prog->progname, prog->prognum,
+ prog->progver, prog->progport);
-err:
- if (ret == -1)
+ pthread_mutex_lock (&svc->rpclock);
+ {
+ list_del (&prog->program);
+ }
+ pthread_mutex_unlock (&svc->rpclock);
+
+ ret = 0;
+out:
+ if (ret == -1) {
gf_log (GF_RPCSVC, GF_LOG_ERROR, "Program unregistration failed"
- ": %s, Num: %d, Ver: %d, Port: %d", prog.progname,
- prog.prognum, prog.progver, prog.progport);
+ ": %s, Num: %d, Ver: %d, Port: %d", prog->progname,
+ prog->prognum, prog->progver, prog->progport);
+ }
return ret;
}
@@ -1524,16 +1550,16 @@ rpcsvc_transport_create (rpcsvc_t *svc, dict_t *options, char *name)
goto out;
}
- ret = rpc_transport_register_notify (trans, rpcsvc_notify, svc);
+ ret = rpc_transport_listen (trans);
if (ret == -1) {
- gf_log (GF_RPCSVC, GF_LOG_DEBUG, "registering notify failed");
+ gf_log (GF_RPCSVC, GF_LOG_DEBUG,
+ "listening on transport failed");
goto out;
}
- ret = rpc_transport_listen (trans);
+ ret = rpc_transport_register_notify (trans, rpcsvc_notify, svc);
if (ret == -1) {
- gf_log (GF_RPCSVC, GF_LOG_DEBUG,
- "listening on transport failed");
+ gf_log (GF_RPCSVC, GF_LOG_DEBUG, "registering notify failed");
goto out;
}
@@ -1657,127 +1683,37 @@ out:
}
-int
-rpcsvc_program_register (rpcsvc_t *svc, rpcsvc_program_t program)
+inline int
+rpcsvc_program_register (rpcsvc_t *svc, rpcsvc_program_t *program)
{
- rpcsvc_program_t *newprog = NULL;
int ret = -1;
- rpcsvc_listener_t *listener = NULL;
- data_t *listen_port_data = NULL;
- uint16_t listen_port = 0;
-
- if (!svc)
- return -1;
-
- newprog = GF_CALLOC (1, sizeof(*newprog),
- gf_common_mt_rpcsvc_program_t);
- if (!newprog)
- return -1;
-
- if (!program.actors)
- goto free_prog;
-
- memcpy (newprog, &program, sizeof (program));
-
- listen_port = RPCSVC_DEFAULT_LISTEN_PORT;
- if (program.options != NULL) {
- /*
- * FIXME: use a method which does not hard-code each transport's
- * option keys.
- */
- listen_port_data = dict_get (program.options, "listen-port");
- if (listen_port_data != NULL) {
- listen_port = data_to_uint16 (listen_port_data);
- } else if ((listen_port_data
- = dict_get (program.options,
- "transport.socket.listen-port"))
- != NULL) {
- listen_port = data_to_uint16 (listen_port_data);
- } else if ((listen_port_data
- = dict_get (program.options,
- "transport.rdma.listen-port"))
- != NULL) {
- listen_port = data_to_uint16 (listen_port_data);
- }
- } else if (program.progport != 0) {
- listen_port = program.progport;
- }
- listener = rpcsvc_get_listener (svc, listen_port, NULL);
- if ((listener == NULL) && (listen_port == RPCSVC_DEFAULT_LISTEN_PORT)) {
- listener = rpcsvc_get_listener (svc, 6969, NULL);
+ if (!svc) {
+ goto out;
}
- if ((listener == NULL) || (listener->trans == NULL)) {
- if ((listener != NULL) && (listener->trans == NULL)) {
- gf_log (GF_RPCSVC, GF_LOG_DEBUG,
- "empty listener without transport found, "
- "destroying it");
- rpcsvc_listener_destroy (listener);
- }
-
- if ((program.progport != 0)
- && (listen_port == program.progport)) {
- ret = dict_set (program.options,
- "transport.socket.listen-port",
- data_from_uint16 (listen_port));
- if (ret == -1) {
- gf_log (GF_RPCSVC, GF_LOG_DEBUG,
- "setting listening port (%d) specified "
- "by program (%s) in options dictionary "
- "failed", listen_port,
- program.progname);
- goto free_prog;
- }
-
- ret = dict_set (program.options,
- "transport.rdma.listen-port",
- data_from_uint16 (listen_port));
- if (ret == -1) {
- gf_log (GF_RPCSVC, GF_LOG_DEBUG,
- "setting listening port (%d) specified "
- "by program (%s) in options dictionary "
- "failed", listen_port,
- program.progname);
- goto free_prog;
- }
- }
-
- listener = rpcsvc_create_listener (svc, program.options,
- program.progname);
- if (listener == NULL) {
- ret = -1;
- gf_log (GF_RPCSVC, GF_LOG_DEBUG,
- "creation of listener for program (%s) failed",
- program.progname);
- goto free_prog;
- }
+ if (program->actors == NULL) {
+ goto out;
}
- ret = rpcsvc_program_register_portmap (newprog, program.progport);
- if (ret == -1) {
- gf_log (GF_RPCSVC, GF_LOG_ERROR, "portmap registration of"
- " program failed");
- goto free_prog;
- }
+ INIT_LIST_HEAD (&program->program);
pthread_mutex_lock (&svc->rpclock);
{
- list_add_tail (&newprog->program, &svc->programs);
+ list_add_tail (&program->program, &svc->programs);
}
pthread_mutex_unlock (&svc->rpclock);
ret = 0;
gf_log (GF_RPCSVC, GF_LOG_DEBUG, "New program registered: %s, Num: %d,"
- " Ver: %d, Port: %d", newprog->progname, newprog->prognum,
- newprog->progver, newprog->progport);
+ " Ver: %d, Port: %d", program->progname, program->prognum,
+ program->progver, program->progport);
-free_prog:
+out:
if (ret == -1) {
gf_log (GF_RPCSVC, GF_LOG_ERROR, "Program registration failed:"
- " %s, Num: %d, Ver: %d, Port: %d", newprog->progname,
- newprog->prognum, newprog->progver, newprog->progport);
- GF_FREE (newprog);
+ " %s, Num: %d, Ver: %d, Port: %d", program->progname,
+ program->prognum, program->progver, program->progport);
}
return ret;
@@ -1884,9 +1820,6 @@ rpcsvc_init (glusterfs_ctx_t *ctx, dict_t *options)
{
rpcsvc_t *svc = NULL;
int ret = -1, poolcount = 0;
- rpcsvc_listener_t *listener = NULL;
- uint32_t listen_port = 0;
- data_t *listen_port_data = NULL;
if ((!ctx) || (!options))
return NULL;
@@ -1929,50 +1862,9 @@ rpcsvc_init (glusterfs_ctx_t *ctx, dict_t *options)
svc->ctx = ctx;
gf_log (GF_RPCSVC, GF_LOG_DEBUG, "RPC service inited.");
- listen_port = RPCSVC_DEFAULT_LISTEN_PORT;
-
- /*
- * FIXME: use a method which does not hard-code each transport's
- * option keys.
- */
- listen_port_data = dict_get (options, "listen-port");
- if (listen_port_data != NULL) {
- listen_port = data_to_uint16 (listen_port_data);
- } else if ((listen_port_data
- = dict_get (options, "transport.socket.listen-port"))
- != NULL) {
- listen_port = data_to_uint16 (listen_port_data);
- } else if ((listen_port_data
- = dict_get (options, "transport.rdma.listen-port"))
- != NULL) {
- listen_port = data_to_uint16 (listen_port_data);
- }
-
- /* One listen port per RPC */
- listener = rpcsvc_get_listener (svc, 0, NULL);
- if (!listener) {
- /* FIXME: listener is given the name of first program that
- * creates it. This is not always correct. For eg., multiple
- * programs can be listening on the same listener
- * (glusterfs 3.1.0, 3.1.2, 3.1.3 etc).
- */
- listener = rpcsvc_create_listener (svc, options, "RPC");
- if (!listener) {
- gf_log (GF_RPCSVC, GF_LOG_ERROR, "creation of listener"
- " for program failed");
- goto free_svc;
- }
- }
-
- if (!listener->trans) {
- gf_log (GF_RPCSVC, GF_LOG_ERROR, "listener with no transport "
- "found");
- goto free_svc;
- }
-
gluster_dump_prog.options = options;
- ret = rpcsvc_program_register (svc, gluster_dump_prog);
+ ret = rpcsvc_program_register (svc, &gluster_dump_prog);
if (ret) {
gf_log (GF_RPCSVC, GF_LOG_ERROR,
"failed to register DUMP program");
diff --git a/rpc/rpc-lib/src/rpcsvc.h b/rpc/rpc-lib/src/rpcsvc.h
index 25381af7798..0a3e91da7b2 100644
--- a/rpc/rpc-lib/src/rpcsvc.h
+++ b/rpc/rpc-lib/src/rpcsvc.h
@@ -379,10 +379,10 @@ struct rpcsvc_program {
* procedure handlers.
*/
extern int
-rpcsvc_program_register (rpcsvc_t *svc, rpcsvc_program_t program);
+rpcsvc_program_register (rpcsvc_t *svc, rpcsvc_program_t *program);
extern int
-rpcsvc_program_unregister (rpcsvc_t *svc, rpcsvc_program_t program);
+rpcsvc_program_unregister (rpcsvc_t *svc, rpcsvc_program_t *program);
/* This will create and add a listener to listener pool. Programs can
* use any of the listener in this pool. A single listener can be used by
@@ -395,6 +395,9 @@ rpcsvc_program_unregister (rpcsvc_t *svc, rpcsvc_program_t program);
extern rpcsvc_listener_t *
rpcsvc_create_listener (rpcsvc_t *svc, dict_t *options, char *name);
+void
+rpcsvc_listener_destroy (rpcsvc_listener_t *listener);
+
extern int
rpcsvc_program_register_portmap (rpcsvc_program_t *newprog, uint32_t port);
generated by cgit (git 2.34.1) at 2024-11-12 19:24:11 +0000