From eae9af1d228cd92387a978e52b9f5032a06dc175 Mon Sep 17 00:00:00 2001 From: Garrett D'Amore Date: Sun, 1 Jan 2017 19:28:46 -0800 Subject: REQ/REP use new style locks. Also, enable compilation for REP. --- src/CMakeLists.txt | 1 + src/protocol/reqrep/rep.c | 66 +++++++++++++++--------------- src/protocol/reqrep/req.c | 102 +++++++++++++++++++++++----------------------- 3 files changed, 86 insertions(+), 83 deletions(-) diff --git a/src/CMakeLists.txt b/src/CMakeLists.txt index 4b09bedc..96ae3fdc 100644 --- a/src/CMakeLists.txt +++ b/src/CMakeLists.txt @@ -67,6 +67,7 @@ set (NNG_SOURCES protocol/pair/pair.c + protocol/reqrep/rep.c protocol/reqrep/req.c transport/inproc/inproc.c diff --git a/src/protocol/reqrep/rep.c b/src/protocol/reqrep/rep.c index ddfcc0c7..7623bd2c 100644 --- a/src/protocol/reqrep/rep.c +++ b/src/protocol/reqrep/rep.c @@ -22,12 +22,12 @@ typedef struct nni_rep_sock nni_rep_sock; // An nni_rep_sock is our per-socket protocol private structure. struct nni_rep_sock { nni_socket * sock; - nni_mutex mx; + nni_mtx mx; nni_msgqueue * uwq; nni_msgqueue * urq; int raw; int ttl; - nni_thread * sender; + nni_thr sender; nni_idhash * pipes; char * btrace; size_t btrace_len; @@ -54,7 +54,7 @@ nni_rep_create(void **repp, nni_socket *sock) if ((rep = nni_alloc(sizeof (*rep))) == NULL) { return (NNG_ENOMEM); } - if ((rv = nni_mutex_init(&rep->mx)) != 0) { + if ((rv = nni_mtx_init(&rep->mx)) != 0) { nni_free(rep, sizeof (*rep)); return (rv); } @@ -64,7 +64,7 @@ nni_rep_create(void **repp, nni_socket *sock) rep->btrace = NULL; rep->btrace_len = 0; if ((rv = nni_idhash_create(&rep->pipes)) != 0) { - nni_mutex_fini(&rep->mx); + nni_mtx_fini(&rep->mx); nni_free(rep, sizeof (*rep)); return (rv); } @@ -72,15 +72,16 @@ nni_rep_create(void **repp, nni_socket *sock) rep->uwq = nni_socket_sendq(sock); rep->urq = nni_socket_recvq(sock); - rv = nni_thread_create(&rep->sender, nni_rep_topsender, rep); + rv = nni_thr_init(&rep->sender, nni_rep_topsender, rep); if (rv != 0) { nni_idhash_destroy(rep->pipes); - nni_mutex_fini(&rep->mx); + nni_mtx_fini(&rep->mx); nni_free(rep, sizeof (*rep)); return (rv); } *repp = rep; nni_socket_senderr(sock, NNG_ESTATE); + nni_thr_run(&rep->sender); return (0); } @@ -90,9 +91,9 @@ nni_rep_destroy(void *arg) { nni_rep_sock *rep = arg; - nni_thread_reap(rep->sender); + nni_thr_fini(&rep->sender); nni_idhash_destroy(rep->pipes); - nni_mutex_fini(&rep->mx); + nni_mtx_fini(&rep->mx); if (rep->btrace != NULL) { nni_free(rep->btrace, rep->btrace_len); } @@ -115,13 +116,13 @@ nni_rep_add_pipe(void *arg, nni_pipe *pipe, void *datap) return (rv); } - nni_mutex_enter(&rep->mx); + nni_mtx_lock(&rep->mx); if ((rv = nni_idhash_insert(rep->pipes, nni_pipe_id(pipe), rp)) != 0) { nni_msgqueue_destroy(rp->sendq); - nni_mutex_exit(&rep->mx); + nni_mtx_unlock(&rep->mx); return (rv); } - nni_mutex_exit(&rep->mx); + nni_mtx_unlock(&rep->mx); return (0); } @@ -132,9 +133,9 @@ nni_rep_rem_pipe(void *arg, void *data) nni_rep_sock *rep = arg; nni_rep_pipe *rp = data; - nni_mutex_enter(&rep->mx); + nni_mtx_lock(&rep->mx); nni_idhash_remove(rep->pipes, nni_pipe_id(rp->pipe)); - nni_mutex_exit(&rep->mx); + nni_mtx_unlock(&rep->mx); nni_msgqueue_destroy(rp->sendq); } @@ -176,9 +177,9 @@ nni_rep_topsender(void *arg) id += header[3]; nni_msg_trim_header(msg, 4); - nni_mutex_enter(&rep->mx); + nni_mtx_lock(&rep->mx); if (nni_idhash_find(rep->pipes, id, (void **) &rp) != 0) { - nni_mutex_exit(&rep->mx); + nni_mtx_unlock(&rep->mx); nni_msg_free(msg); continue; } @@ -190,7 +191,7 @@ nni_rep_topsender(void *arg) // circumstances. nni_msg_free(msg); } - nni_mutex_exit(&rep->mx); + nni_mtx_unlock(&rep->mx); } } @@ -306,14 +307,14 @@ nni_rep_setopt(void *arg, int opt, const void *buf, size_t sz) switch (opt) { case NNG_OPT_MAXTTL: - nni_mutex_enter(&rep->mx); + nni_mtx_lock(&rep->mx); rv = nni_setopt_int(&rep->ttl, buf, sz, 1, 255); - nni_mutex_exit(&rep->mx); + nni_mtx_unlock(&rep->mx); break; case NNG_OPT_RAW: - nni_mutex_enter(&rep->mx); + nni_mtx_lock(&rep->mx); rv = nni_setopt_int(&rep->raw, buf, sz, 0, 1); - nni_mutex_exit(&rep->mx); + nni_mtx_unlock(&rep->mx); break; default: rv = NNG_ENOTSUP; @@ -330,14 +331,14 @@ nni_rep_getopt(void *arg, int opt, void *buf, size_t *szp) switch (opt) { case NNG_OPT_MAXTTL: - nni_mutex_enter(&rep->mx); + nni_mtx_lock(&rep->mx); rv = nni_getopt_int(&rep->ttl, buf, szp); - nni_mutex_exit(&rep->mx); + nni_mtx_unlock(&rep->mx); break; case NNG_OPT_RAW: - nni_mutex_enter(&rep->mx); + nni_mtx_lock(&rep->mx); rv = nni_getopt_int(&rep->raw, buf, szp); - nni_mutex_exit(&rep->mx); + nni_mtx_unlock(&rep->mx); break; default: rv = NNG_ENOTSUP; @@ -352,9 +353,9 @@ nni_rep_sendfilter(void *arg, nni_msg *msg) nni_rep_sock *rep = arg; size_t len; - nni_mutex_enter(&rep->mx); + nni_mtx_lock(&rep->mx); if (rep->raw) { - nni_mutex_exit(&rep->mx); + nni_mtx_unlock(&rep->mx); return (msg); } @@ -364,7 +365,7 @@ nni_rep_sendfilter(void *arg, nni_msg *msg) // If we have a stored backtrace, append it to the header... // if we don't have a backtrace, discard the message. if (rep->btrace == NULL) { - nni_mutex_exit(&rep->mx); + nni_mtx_unlock(&rep->mx); nni_msg_free(msg); return (NULL); } @@ -377,7 +378,7 @@ nni_rep_sendfilter(void *arg, nni_msg *msg) nni_free(rep->btrace, rep->btrace_len); rep->btrace = NULL; rep->btrace_len = 0; - nni_mutex_exit(&rep->mx); + nni_mtx_unlock(&rep->mx); nni_msg_free(msg); return (NULL); } @@ -385,6 +386,7 @@ nni_rep_sendfilter(void *arg, nni_msg *msg) nni_free(rep->btrace, rep->btrace_len); rep->btrace = NULL; rep->btrace_len = 0; + nni_mtx_unlock(&rep->mx); return (msg); } @@ -396,9 +398,9 @@ nni_rep_recvfilter(void *arg, nni_msg *msg) char *header; size_t len; - nni_mutex_enter(&rep->mx); + nni_mtx_lock(&rep->mx); if (rep->raw) { - nni_mutex_exit(&rep->mx); + nni_mtx_unlock(&rep->mx); return (msg); } @@ -410,14 +412,14 @@ nni_rep_recvfilter(void *arg, nni_msg *msg) rep->btrace_len = 0; } if ((rep->btrace = nni_alloc(len)) == NULL) { - nni_mutex_exit(&rep->mx); + nni_mtx_unlock(&rep->mx); nni_msg_free(msg); return (NULL); } rep->btrace_len = len; memcpy(rep->btrace, 0, len); nni_msg_trim_header(msg, len); - nni_mutex_exit(&rep->mx); + nni_mtx_unlock(&rep->mx); return (msg); } diff --git a/src/protocol/reqrep/req.c b/src/protocol/reqrep/req.c index 366e0c57..611baadb 100644 --- a/src/protocol/reqrep/req.c +++ b/src/protocol/reqrep/req.c @@ -22,14 +22,15 @@ typedef struct nni_req_sock nni_req_sock; // An nni_req_sock is our per-socket protocol private structure. struct nni_req_sock { nni_socket * sock; - nni_mutex mx; - nni_cond cv; + nni_mtx mx; + nni_cv cv; nni_msgqueue * uwq; nni_msgqueue * urq; nni_duration retry; nni_time resend; - nni_thread * resender; + nni_thr resender; int raw; + int closing; nni_list pipes; nni_msg * reqmsg; uint32_t nextid; // next id @@ -57,12 +58,12 @@ nni_req_create(void **reqp, nni_socket *sock) if ((req = nni_alloc(sizeof (*req))) == NULL) { return (NNG_ENOMEM); } - if ((rv = nni_mutex_init(&req->mx)) != 0) { + if ((rv = nni_mtx_init(&req->mx)) != 0) { nni_free(req, sizeof (*req)); return (rv); } - if ((rv = nni_cond_init(&req->cv, &req->mx)) != 0) { - nni_mutex_fini(&req->mx); + if ((rv = nni_cv_init(&req->cv, &req->mx)) != 0) { + nni_mtx_fini(&req->mx); nni_free(req, sizeof (*req)); return (rv); } @@ -79,13 +80,14 @@ nni_req_create(void **reqp, nni_socket *sock) req->urq = nni_socket_recvq(sock); *reqp = req; nni_socket_recverr(sock, NNG_ESTATE); - rv = nni_thread_create(&req->resender, nni_req_resender, req); + rv = nni_thr_init(&req->resender, nni_req_resender, req); if (rv != 0) { - nni_cond_fini(&req->cv); - nni_mutex_fini(&req->mx); + nni_cv_fini(&req->cv); + nni_mtx_fini(&req->mx); nni_free(req, sizeof (*req)); return (rv); } + nni_thr_run(&req->resender); return (0); } @@ -94,19 +96,17 @@ static void nni_req_destroy(void *arg) { nni_req_sock *req = arg; - nni_thread *resender; // Shut down the resender. We request it to exit by clearing // its old value, then kick it. - nni_mutex_enter(&req->mx); - resender = req->resender; - req->resender = NULL; - nni_cond_broadcast(&req->cv); - nni_mutex_exit(&req->mx); - - nni_thread_reap(resender); - nni_cond_fini(&req->cv); - nni_mutex_fini(&req->mx); + nni_mtx_lock(&req->mx); + req->closing = 1; + nni_cv_wake(&req->cv); + nni_mtx_unlock(&req->mx); + + nni_thr_fini(&req->resender); + nni_cv_fini(&req->cv); + nni_mtx_fini(&req->mx); nni_free(req, sizeof (*req)); } @@ -122,9 +122,9 @@ nni_req_add_pipe(void *arg, nni_pipe *pipe, void *data) rp->sigclose = 0; rp->req = req; - nni_mutex_enter(&req->mx); + nni_mtx_lock(&req->mx); nni_list_append(&req->pipes, rp); - nni_mutex_exit(&req->mx); + nni_mtx_unlock(&req->mx); return (0); } @@ -135,9 +135,9 @@ nni_req_rem_pipe(void *arg, void *data) nni_req_sock *req = arg; nni_req_pipe *rp = data; - nni_mutex_enter(&req->mx); + nni_mtx_lock(&req->mx); nni_list_remove(&req->pipes, rp); - nni_mutex_exit(&req->mx); + nni_mtx_unlock(&req->mx); } @@ -221,14 +221,14 @@ nni_req_setopt(void *arg, int opt, const void *buf, size_t sz) switch (opt) { case NNG_OPT_RESENDTIME: - nni_mutex_enter(&req->mx); + nni_mtx_lock(&req->mx); rv = nni_setopt_duration(&req->retry, buf, sz); - nni_mutex_exit(&req->mx); + nni_mtx_unlock(&req->mx); break; case NNG_OPT_RAW: - nni_mutex_enter(&req->mx); + nni_mtx_lock(&req->mx); rv = nni_setopt_int(&req->raw, buf, sz, 0, 1); - nni_mutex_exit(&req->mx); + nni_mtx_unlock(&req->mx); break; default: rv = NNG_ENOTSUP; @@ -245,14 +245,14 @@ nni_req_getopt(void *arg, int opt, void *buf, size_t *szp) switch (opt) { case NNG_OPT_RESENDTIME: - nni_mutex_enter(&req->mx); + nni_mtx_lock(&req->mx); rv = nni_getopt_duration(&req->retry, buf, szp); - nni_mutex_exit(&req->mx); + nni_mtx_unlock(&req->mx); break; case NNG_OPT_RAW: - nni_mutex_enter(&req->mx); + nni_mtx_lock(&req->mx); rv = nni_getopt_int(&req->raw, buf, szp); - nni_mutex_exit(&req->mx); + nni_mtx_unlock(&req->mx); break; default: rv = NNG_ENOTSUP; @@ -268,17 +268,17 @@ nni_req_resender(void *arg) int rv; for (;;) { - nni_mutex_enter(&req->mx); - if (req->resender == NULL) { - nni_mutex_exit(&req->mx); + nni_mtx_lock(&req->mx); + if (req->closing) { + nni_mtx_unlock(&req->mx); return; } if (req->reqmsg == NULL) { - nni_cond_wait(&req->cv); - nni_mutex_exit(&req->mx); + nni_cv_wait(&req->cv); + nni_mtx_unlock(&req->mx); continue; } - rv = nni_cond_waituntil(&req->cv, req->resend); + rv = nni_cv_until(&req->cv, req->resend); if ((rv == NNG_ETIMEDOUT) && (req->reqmsg != NULL)) { nni_msg *dup; // XXX: check for final timeout on this? @@ -289,7 +289,7 @@ nni_req_resender(void *arg) } req->resend = nni_clock() + req->retry; } - nni_mutex_exit(&req->mx); + nni_mtx_unlock(&req->mx); } } @@ -301,11 +301,11 @@ nni_req_sendfilter(void *arg, nni_msg *msg) uint32_t id; uint8_t buf[4]; - nni_mutex_enter(&req->mx); + nni_mtx_lock(&req->mx); if (req->raw) { // No automatic retry, and the request ID must // be in the header coming down. - nni_mutex_exit(&req->mx); + nni_mtx_unlock(&req->mx); return (msg); } @@ -322,7 +322,7 @@ nni_req_sendfilter(void *arg, nni_msg *msg) if (nni_msg_append_header(msg, buf, 4) != 0) { // Should be ENOMEM. - nni_mutex_exit(&req->mx); + nni_mtx_unlock(&req->mx); nni_msg_free(msg); return (NULL); } @@ -335,18 +335,18 @@ nni_req_sendfilter(void *arg, nni_msg *msg) // Make a duplicate message... for retries. if (nni_msg_dup(&req->reqmsg, msg) != 0) { - nni_mutex_exit(&req->mx); + nni_mtx_unlock(&req->mx); nni_msg_free(msg); return (NULL); } // Schedule the next retry req->resend = nni_clock() + req->retry; - nni_cond_signal(&req->cv); + nni_cv_wake(&req->cv); // Clear the error condition. nni_socket_recverr(req->sock, 0); - nni_mutex_exit(&req->mx); + nni_mtx_unlock(&req->mx); return (msg); } @@ -359,29 +359,29 @@ nni_req_recvfilter(void *arg, nni_msg *msg) char *header; size_t len; - nni_mutex_enter(&req->mx); + nni_mtx_lock(&req->mx); if (req->raw) { // Pass it unmolested - nni_mutex_exit(&req->mx); + nni_mtx_unlock(&req->mx); return (msg); } header = nni_msg_header(msg, &len); if (len < 4) { - nni_mutex_exit(&req->mx); + nni_mtx_unlock(&req->mx); nni_msg_free(msg); return (NULL); } if (req->reqmsg == NULL) { // We had no outstanding request. - nni_mutex_exit(&req->mx); + nni_mtx_unlock(&req->mx); nni_msg_free(msg); return (NULL); } if (memcmp(header, req->reqid, 4) != 0) { // Wrong request id - nni_mutex_exit(&req->mx); + nni_mtx_unlock(&req->mx); nni_msg_free(msg); return (NULL); } @@ -389,8 +389,8 @@ nni_req_recvfilter(void *arg, nni_msg *msg) nni_socket_recverr(req->sock, NNG_ESTATE); nni_msg_free(req->reqmsg); req->reqmsg = NULL; - nni_cond_signal(&req->cv); - nni_mutex_exit(&req->mx); + nni_cv_wake(&req->cv); + nni_mtx_unlock(&req->mx); return (msg); } -- cgit v1.2.3-70-g09d2