aboutsummaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--src/CMakeLists.txt1
-rw-r--r--src/protocol/reqrep/rep.c66
-rw-r--r--src/protocol/reqrep/req.c102
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);
}
generated by cgit v1.2.3-70-g09d2 (git 2.46.0) at 2026-01-18 11:46:28 +0000