summaryrefslogtreecommitdiffstats
path: root/rpc/rpc-lib/src
diff options
context:
space:
mode:
Diffstat (limited to 'rpc/rpc-lib/src')
-rw-r--r--rpc/rpc-lib/src/Makefile.am2
-rw-r--r--rpc/rpc-lib/src/libgfrpc.sym1
-rw-r--r--rpc/rpc-lib/src/protocol-common.h1
-rw-r--r--rpc/rpc-lib/src/rpc-clnt-ping.c2
-rw-r--r--rpc/rpc-lib/src/rpc-clnt.c27
-rw-r--r--rpc/rpc-lib/src/rpc-clnt.h1
-rw-r--r--rpc/rpc-lib/src/rpc-drc.c13
-rw-r--r--rpc/rpc-lib/src/rpc-drc.h2
-rw-r--r--rpc/rpc-lib/src/rpcsvc.c152
-rw-r--r--rpc/rpc-lib/src/rpcsvc.h7
10 files changed, 130 insertions, 78 deletions
diff --git a/rpc/rpc-lib/src/Makefile.am b/rpc/rpc-lib/src/Makefile.am
index 81a9647..35c9db0 100644
--- a/rpc/rpc-lib/src/Makefile.am
+++ b/rpc/rpc-lib/src/Makefile.am
@@ -2,7 +2,7 @@ lib_LTLIBRARIES = libgfrpc.la
libgfrpc_la_SOURCES = auth-unix.c rpcsvc-auth.c rpcsvc.c auth-null.c \
rpc-transport.c xdr-rpc.c xdr-rpcclnt.c rpc-clnt.c auth-glusterfs.c \
- rpc-drc.c $(CONTRIBDIR)/sunrpc/xdr_sizeof.c rpc-clnt-ping.c \
+ rpc-drc.c rpc-clnt-ping.c \
autoscale-threads.c mgmt-pmap.c
EXTRA_DIST = libgfrpc.sym
diff --git a/rpc/rpc-lib/src/libgfrpc.sym b/rpc/rpc-lib/src/libgfrpc.sym
index 54d1be1..e026d80 100644
--- a/rpc/rpc-lib/src/libgfrpc.sym
+++ b/rpc/rpc-lib/src/libgfrpc.sym
@@ -65,3 +65,4 @@ rpc_transport_unix_options_build
rpc_transport_unref
rpc_clnt_mgmt_pmap_signout
rpcsvc_autoscale_threads
+rpcsvc_statedump
diff --git a/rpc/rpc-lib/src/protocol-common.h b/rpc/rpc-lib/src/protocol-common.h
index 7275d75..0cb5862 100644
--- a/rpc/rpc-lib/src/protocol-common.h
+++ b/rpc/rpc-lib/src/protocol-common.h
@@ -309,6 +309,7 @@ enum glusterd_mgmt_v3_procnum {
GLUSTERD_MGMT_V3_PRE_VALIDATE,
GLUSTERD_MGMT_V3_BRICK_OP,
GLUSTERD_MGMT_V3_COMMIT,
+ GLUSTERD_MGMT_V3_POST_COMMIT,
GLUSTERD_MGMT_V3_POST_VALIDATE,
GLUSTERD_MGMT_V3_UNLOCK,
GLUSTERD_MGMT_V3_MAXVALUE,
diff --git a/rpc/rpc-lib/src/rpc-clnt-ping.c b/rpc/rpc-lib/src/rpc-clnt-ping.c
index 2298ef6..31f1784 100644
--- a/rpc/rpc-lib/src/rpc-clnt-ping.c
+++ b/rpc/rpc-lib/src/rpc-clnt-ping.c
@@ -122,7 +122,7 @@ rpc_clnt_ping_timer_expired(void *rpc_ptr)
goto out;
}
- clock_gettime(CLOCK_REALTIME, &current);
+ timespec_now_realtime(&current);
pthread_mutex_lock(&conn->lock);
{
unref = rpc_clnt_remove_ping_timer_locked(rpc);
diff --git a/rpc/rpc-lib/src/rpc-clnt.c b/rpc/rpc-lib/src/rpc-clnt.c
index dac7076..517037c 100644
--- a/rpc/rpc-lib/src/rpc-clnt.c
+++ b/rpc/rpc-lib/src/rpc-clnt.c
@@ -97,7 +97,7 @@ call_bail(void *data)
struct saved_frame *saved_frame = NULL;
struct saved_frame *trav = NULL;
struct saved_frame *tmp = NULL;
- char frame_sent[256] = {
+ char frame_sent[GF_TIMESTR_SIZE] = {
0,
};
struct timespec timeout = {
@@ -105,7 +105,6 @@ call_bail(void *data)
};
char peerid[UNIX_PATH_MAX] = {0};
gf_boolean_t need_unref = _gf_false;
- int len;
GF_VALIDATE_OR_GOTO("client", data, out);
@@ -165,11 +164,8 @@ call_bail(void *data)
list_for_each_entry_safe(trav, tmp, &list, list)
{
- gf_time_fmt(frame_sent, sizeof frame_sent, trav->saved_at.tv_sec,
- gf_timefmt_FT);
- len = strlen(frame_sent);
- snprintf(frame_sent + len, sizeof(frame_sent) - len,
- ".%" GF_PRI_SUSECONDS, trav->saved_at.tv_usec);
+ gf_time_fmt_tv(frame_sent, sizeof frame_sent, &trav->saved_at,
+ gf_timefmt_FT);
gf_log(conn->name, GF_LOG_ERROR,
"bailing out frame type(%s), op(%s(%d)), xid = 0x%x, "
@@ -317,20 +313,15 @@ saved_frames_unwind(struct saved_frames *saved_frames)
{
struct saved_frame *trav = NULL;
struct saved_frame *tmp = NULL;
- char timestr[1024] = {
+ char timestr[GF_TIMESTR_SIZE] = {
0,
};
- int len;
list_splice_init(&saved_frames->lk_sf.list, &saved_frames->sf.list);
list_for_each_entry_safe(trav, tmp, &saved_frames->sf.list, list)
{
- gf_time_fmt(timestr, sizeof timestr, trav->saved_at.tv_sec,
- gf_timefmt_FT);
- len = strlen(timestr);
- snprintf(timestr + len, sizeof(timestr) - len, ".%" GF_PRI_SUSECONDS,
- trav->saved_at.tv_usec);
+ gf_time_fmt_tv(timestr, sizeof timestr, &trav->saved_at, gf_timefmt_FT);
if (!trav->rpcreq || !trav->rpcreq->prog)
continue;
@@ -928,7 +919,7 @@ rpc_clnt_notify(rpc_transport_t *trans, void *mydata,
}
case RPC_TRANSPORT_MSG_RECEIVED: {
- clock_gettime(CLOCK_REALTIME, &conn->last_received);
+ timespec_now_realtime(&conn->last_received);
pollin = data;
if (pollin->is_reply)
@@ -942,8 +933,7 @@ rpc_clnt_notify(rpc_transport_t *trans, void *mydata,
}
case RPC_TRANSPORT_MSG_SENT: {
- clock_gettime(CLOCK_REALTIME, &conn->last_sent);
-
+ timespec_now_realtime(&conn->last_sent);
ret = 0;
break;
}
@@ -960,6 +950,7 @@ rpc_clnt_notify(rpc_transport_t *trans, void *mydata,
conn->config.remote_port = 0;
conn->connected = 1;
conn->disconnected = 0;
+ pthread_cond_broadcast(&conn->cond);
}
pthread_mutex_unlock(&conn->lock);
@@ -1005,6 +996,7 @@ rpc_clnt_connection_init(struct rpc_clnt *clnt, glusterfs_ctx_t *ctx,
conn = &clnt->conn;
pthread_mutex_init(&clnt->conn.lock, NULL);
+ pthread_cond_init(&clnt->conn.cond, NULL);
conn->name = gf_strdup(name);
if (!conn->name) {
@@ -1830,6 +1822,7 @@ rpc_clnt_destroy(struct rpc_clnt *rpc)
saved_frames_destroy(saved_frames);
pthread_mutex_destroy(&rpc->lock);
pthread_mutex_destroy(&rpc->conn.lock);
+ pthread_cond_destroy(&rpc->conn.cond);
/* mem-pool should be destroyed, otherwise,
it will cause huge memory leaks */
diff --git a/rpc/rpc-lib/src/rpc-clnt.h b/rpc/rpc-lib/src/rpc-clnt.h
index 0d0b115..2945265 100644
--- a/rpc/rpc-lib/src/rpc-clnt.h
+++ b/rpc/rpc-lib/src/rpc-clnt.h
@@ -127,6 +127,7 @@ struct rpc_clnt_config {
struct rpc_clnt_connection {
pthread_mutex_t lock;
+ pthread_cond_t cond;
rpc_transport_t *trans;
struct rpc_clnt_config config;
gf_timer_t *reconnect;
diff --git a/rpc/rpc-lib/src/rpc-drc.c b/rpc/rpc-lib/src/rpc-drc.c
index d083db2..de8dc63 100644
--- a/rpc/rpc-lib/src/rpc-drc.c
+++ b/rpc/rpc-lib/src/rpc-drc.c
@@ -190,7 +190,7 @@ rpcsvc_get_drc_client(rpcsvc_drc_globals_t *drc,
if (!client)
goto out;
- client->ref = 0;
+ GF_ATOMIC_INIT(client->ref, 0);
client->sock_union = (union gf_sock_union) * sockaddr;
client->op_count = 0;
INIT_LIST_HEAD(&client->client_list);
@@ -246,7 +246,7 @@ static drc_client_t *
rpcsvc_drc_client_ref(drc_client_t *client)
{
GF_ASSERT(client);
- client->ref++;
+ GF_ATOMIC_INC(client->ref);
return client;
}
@@ -261,11 +261,12 @@ rpcsvc_drc_client_ref(drc_client_t *client)
static drc_client_t *
rpcsvc_drc_client_unref(rpcsvc_drc_globals_t *drc, drc_client_t *client)
{
+ uint32_t refcount;
+
GF_ASSERT(drc);
- GF_ASSERT(client->ref);
- client->ref--;
- if (!client->ref) {
+ refcount = GF_ATOMIC_DEC(client->ref);
+ if (!refcount) {
drc->client_count--;
rpcsvc_remove_drc_client(client);
client = NULL;
@@ -589,7 +590,7 @@ rpcsvc_drc_priv(rpcsvc_drc_globals_t *drc)
}
gf_proc_dump_build_key(key, "client", "%d.ref_count", i);
- gf_proc_dump_write(key, "%d", client->ref);
+ gf_proc_dump_write(key, "%" PRIu32, GF_ATOMIC_GET(client->ref));
gf_proc_dump_build_key(key, "client", "%d.op_count", i);
gf_proc_dump_write(key, "%d", client->op_count);
i++;
diff --git a/rpc/rpc-lib/src/rpc-drc.h b/rpc/rpc-lib/src/rpc-drc.h
index 6aaede0..ce66430 100644
--- a/rpc/rpc-lib/src/rpc-drc.h
+++ b/rpc/rpc-lib/src/rpc-drc.h
@@ -24,7 +24,7 @@ struct drc_client {
struct rb_table *rbtree;
/* no. of ops currently cached */
uint32_t op_count;
- uint32_t ref;
+ gf_atomic_uint32_t ref;
struct list_head client_list;
};
diff --git a/rpc/rpc-lib/src/rpcsvc.c b/rpc/rpc-lib/src/rpcsvc.c
index 623523c..39910d4 100644
--- a/rpc/rpc-lib/src/rpcsvc.c
+++ b/rpc/rpc-lib/src/rpcsvc.c
@@ -13,6 +13,7 @@
#include <glusterfs/dict.h>
#include <glusterfs/byte-order.h>
#include <glusterfs/compat-errno.h>
+#include <glusterfs/statedump.h>
#include "xdr-rpc.h"
#include <glusterfs/iobuf.h>
#include "xdr-common.h"
@@ -41,6 +42,10 @@
#include "xdr-rpcclnt.h"
#include <glusterfs/glusterfs-acl.h>
+#ifndef PTHREAD_MUTEX_ADAPTIVE_NP
+#define PTHREAD_MUTEX_ADAPTIVE_NP PTHREAD_MUTEX_DEFAULT
+#endif
+
static struct rpcsvc_program gluster_dump_prog;
#define rpcsvc_alloc_request(svc, request) \
@@ -66,59 +71,33 @@ rpcsvc_request_handler(void *arg);
static int
rpcsvc_match_subnet_v4(const char *addrtok, const char *ipaddr);
-void
+static void
rpcsvc_toggle_queue_status(rpcsvc_program_t *prog,
- rpcsvc_request_queue_t *queue, char status[])
+ rpcsvc_request_queue_t *queue,
+ unsigned long status[])
{
- int queue_index = 0, status_index = 0, set_bit = 0;
-
- if (queue != &prog->request_queue[0]) {
- queue_index = (queue - &prog->request_queue[0]);
- }
-
- status_index = queue_index / 8;
- set_bit = queue_index % 8;
-
- status[status_index] ^= (1 << set_bit);
+ unsigned queue_index = queue - prog->request_queue;
- return;
+ status[queue_index / __BITS_PER_LONG] ^= (1UL << (queue_index %
+ __BITS_PER_LONG));
}
int
rpcsvc_get_free_queue_index(rpcsvc_program_t *prog)
{
- int queue_index = 0, max_index = 0, i = 0;
- unsigned int right_most_unset_bit = 0;
-
- right_most_unset_bit = 8;
+ unsigned i, j = 0;
- max_index = gf_roof(EVENT_MAX_THREADS, 8) / 8;
- for (i = 0; i < max_index; i++) {
- if (prog->request_queue_status[i] == 0) {
- right_most_unset_bit = 0;
+ for (i = 0; i < EVENT_MAX_THREADS / __BITS_PER_LONG; i++)
+ if (prog->request_queue_status[i] != ULONG_MAX) {
+ j = __builtin_ctzl(~prog->request_queue_status[i]);
break;
- } else {
- /* get_rightmost_set_bit (sic)*/
- right_most_unset_bit = __builtin_ctz(
- ~prog->request_queue_status[i]);
- if (right_most_unset_bit < 8) {
- break;
- }
}
- }
-
- if (right_most_unset_bit > 7) {
- queue_index = -1;
- } else {
- queue_index = i * 8;
- queue_index += right_most_unset_bit;
- }
- if (queue_index != -1) {
- prog->request_queue_status[i] |= (0x1 << right_most_unset_bit);
- }
+ if (i == EVENT_MAX_THREADS / __BITS_PER_LONG)
+ return -1;
- return queue_index;
+ prog->request_queue_status[i] |= (1UL << j);
+ return i * __BITS_PER_LONG + j;
}
rpcsvc_notify_wrapper_t *
@@ -363,6 +342,10 @@ rpcsvc_program_actor(rpcsvc_request_t *req)
goto err;
}
+ if (svc->xl->ctx->measure_latency) {
+ timespec_now(&req->begin);
+ }
+
req->ownthread = program->ownthread;
req->synctask = program->synctask;
@@ -1512,10 +1495,18 @@ rpcsvc_submit_generic(rpcsvc_request_t *req, struct iovec *proghdr,
size_t hdrlen = 0;
char new_iobref = 0;
rpcsvc_drc_globals_t *drc = NULL;
+ gf_latency_t *lat = NULL;
if ((!req) || (!req->trans))
return -1;
+ if (req->prog && req->begin.tv_sec) {
+ if ((req->procnum >= 0) && (req->procnum < req->prog->numactors)) {
+ timespec_now(&req->end);
+ lat = &req->prog->latencies[req->procnum];
+ gf_latency_update(lat, &req->begin, &req->end);
+ }
+ }
trans = req->trans;
for (i = 0; i < hdrcount; i++) {
@@ -1846,6 +1837,15 @@ rpcsvc_submit_message(rpcsvc_request_t *req, struct iovec *proghdr,
iobref);
}
+void
+rpcsvc_program_destroy(rpcsvc_program_t *program)
+{
+ if (program) {
+ GF_FREE(program->latencies);
+ GF_FREE(program);
+ }
+}
+
int
rpcsvc_program_unregister(rpcsvc_t *svc, rpcsvc_program_t *program)
{
@@ -1855,6 +1855,18 @@ rpcsvc_program_unregister(rpcsvc_t *svc, rpcsvc_program_t *program)
goto out;
}
+ pthread_rwlock_rdlock(&svc->rpclock);
+ {
+ list_for_each_entry(prog, &svc->programs, program)
+ {
+ if ((prog->prognum == program->prognum) &&
+ (prog->progver == program->progver)) {
+ break;
+ }
+ }
+ }
+ pthread_rwlock_unlock(&svc->rpclock);
+
ret = rpcsvc_program_unregister_portmap(program);
if (ret == -1) {
gf_log(GF_RPCSVC, GF_LOG_ERROR,
@@ -1871,17 +1883,6 @@ rpcsvc_program_unregister(rpcsvc_t *svc, rpcsvc_program_t *program)
goto out;
}
#endif
- pthread_rwlock_rdlock(&svc->rpclock);
- {
- list_for_each_entry(prog, &svc->programs, program)
- {
- if ((prog->prognum == program->prognum) &&
- (prog->progver == program->progver)) {
- break;
- }
- }
- }
- pthread_rwlock_unlock(&svc->rpclock);
gf_log(GF_RPCSVC, GF_LOG_DEBUG,
"Program unregistered: %s, Num: %d,"
@@ -1902,6 +1903,8 @@ rpcsvc_program_unregister(rpcsvc_t *svc, rpcsvc_program_t *program)
ret = 0;
out:
+ rpcsvc_program_destroy(prog);
+
if (ret == -1) {
if (program) {
gf_log(GF_RPCSVC, GF_LOG_ERROR,
@@ -2285,6 +2288,11 @@ rpcsvc_program_register(rpcsvc_t *svc, rpcsvc_program_t *program,
}
memcpy(newprog, program, sizeof(*program));
+ newprog->latencies = gf_latency_new(program->numactors);
+ if (!newprog->latencies) {
+ rpcsvc_program_destroy(newprog);
+ goto out;
+ }
INIT_LIST_HEAD(&newprog->program);
pthread_mutexattr_init(&thr_attr);
@@ -3230,6 +3238,48 @@ out:
return ret;
}
+void
+rpcsvc_program_dump(rpcsvc_program_t *prog)
+{
+ char key_prefix[GF_DUMP_MAX_BUF_LEN];
+ char key[GF_DUMP_MAX_BUF_LEN];
+ int i;
+
+ snprintf(key_prefix, GF_DUMP_MAX_BUF_LEN, "%s", prog->progname);
+ gf_proc_dump_add_section("%s", key_prefix);
+
+ gf_proc_dump_build_key(key, key_prefix, "program-number");
+ gf_proc_dump_write(key, "%d", prog->prognum);
+
+ gf_proc_dump_build_key(key, key_prefix, "program-version");
+ gf_proc_dump_write(key, "%d", prog->progver);
+
+ strncat(key_prefix, ".latency",
+ sizeof(key_prefix) - strlen(key_prefix) - 1);
+
+ for (i = 0; i < prog->numactors; i++) {
+ gf_proc_dump_build_key(key, key_prefix, "%s", prog->actors[i].procname);
+ gf_latency_statedump_and_reset(key, &prog->latencies[i]);
+ }
+}
+
+void
+rpcsvc_statedump(rpcsvc_t *svc)
+{
+ rpcsvc_program_t *prog = NULL;
+ int ret = 0;
+ ret = pthread_rwlock_tryrdlock(&svc->rpclock);
+ if (ret)
+ return;
+ {
+ list_for_each_entry(prog, &svc->programs, program)
+ {
+ rpcsvc_program_dump(prog);
+ }
+ }
+ pthread_rwlock_unlock(&svc->rpclock);
+}
+
static rpcsvc_actor_t gluster_dump_actors[GF_DUMP_MAXVALUE] = {
[GF_DUMP_NULL] = {"NULL", NULL, NULL, GF_DUMP_NULL, DRC_NA, 0},
[GF_DUMP_DUMP] = {"DUMP", rpcsvc_dump, NULL, GF_DUMP_DUMP, DRC_NA, 0},
diff --git a/rpc/rpc-lib/src/rpcsvc.h b/rpc/rpc-lib/src/rpcsvc.h
index a88fc84..7b30309 100644
--- a/rpc/rpc-lib/src/rpcsvc.h
+++ b/rpc/rpc-lib/src/rpcsvc.h
@@ -264,6 +264,8 @@ struct rpcsvc_request {
gf_boolean_t ownthread;
gf_boolean_t synctask;
+ struct timespec begin; /*req handling start time*/
+ struct timespec end; /*req handling end time*/
};
#define rpcsvc_request_program(req) ((rpcsvc_program_t *)((req)->prog))
@@ -419,6 +421,7 @@ struct rpcsvc_program {
supported by the system. */
/* Program specific state handed to actors */
void *private;
+ gf_latency_t *latencies; /*Tracks latency statistics for the rpc call*/
/* This upcall is provided by the program during registration.
* It is used to notify the program about events like connection being
@@ -457,7 +460,7 @@ struct rpcsvc_program {
gf_boolean_t alive;
gf_boolean_t synctask;
- char request_queue_status[EVENT_MAX_THREADS / 8 + 1];
+ unsigned long request_queue_status[EVENT_MAX_THREADS / __BITS_PER_LONG];
};
typedef struct rpcsvc_cbk_program {
@@ -686,4 +689,6 @@ rpcsvc_autoscale_threads(glusterfs_ctx_t *ctx, rpcsvc_t *rpc, int incr);
extern int
rpcsvc_destroy(rpcsvc_t *svc);
+void
+rpcsvc_statedump(rpcsvc_t *svc);
#endif