diff options
Diffstat (limited to 'rpc/rpc-lib/src')
-rw-r--r-- | rpc/rpc-lib/src/Makefile.am | 2 | ||||
-rw-r--r-- | rpc/rpc-lib/src/libgfrpc.sym | 1 | ||||
-rw-r--r-- | rpc/rpc-lib/src/protocol-common.h | 1 | ||||
-rw-r--r-- | rpc/rpc-lib/src/rpc-clnt-ping.c | 2 | ||||
-rw-r--r-- | rpc/rpc-lib/src/rpc-clnt.c | 27 | ||||
-rw-r--r-- | rpc/rpc-lib/src/rpc-clnt.h | 1 | ||||
-rw-r--r-- | rpc/rpc-lib/src/rpc-drc.c | 13 | ||||
-rw-r--r-- | rpc/rpc-lib/src/rpc-drc.h | 2 | ||||
-rw-r--r-- | rpc/rpc-lib/src/rpcsvc.c | 152 | ||||
-rw-r--r-- | rpc/rpc-lib/src/rpcsvc.h | 7 |
10 files changed, 130 insertions, 78 deletions
diff --git a/rpc/rpc-lib/src/Makefile.am b/rpc/rpc-lib/src/Makefile.am index 81a96476883..35c9db07e7f 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 54d1be1112f..e026d80259b 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 7275d7568b6..0cb5862e9a9 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 2298ef6394f..31f17841bea 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, ¤t); + timespec_now_realtime(¤t); 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 dac707664df..517037c4a5d 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 0d0b115578c..2945265200b 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 d083db24fc5..de8dc630626 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 6aaede0828a..ce66430809b 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 623523c09d2..39910d481bf 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 a88fc841b60..7b3030926c8 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 |