aboutsummaryrefslogtreecommitdiff
path: root/src/protocol/reqrep0
diff options
context:
space:
mode:
Diffstat (limited to 'src/protocol/reqrep0')
-rw-r--r--src/protocol/reqrep0/rep.c80
-rw-r--r--src/protocol/reqrep0/rep_test.c21
-rw-r--r--src/protocol/reqrep0/req.c8
-rw-r--r--src/protocol/reqrep0/xrep.c20
-rw-r--r--src/protocol/reqrep0/xreq.c16
5 files changed, 81 insertions, 64 deletions
diff --git a/src/protocol/reqrep0/rep.c b/src/protocol/reqrep0/rep.c
index a715ab59..a29c3120 100644
--- a/src/protocol/reqrep0/rep.c
+++ b/src/protocol/reqrep0/rep.c
@@ -48,14 +48,14 @@ struct rep0_ctx {
// rep0_sock is our per-socket protocol private structure.
struct rep0_sock {
- nni_mtx lk;
- int ttl;
- nni_idhash * pipes;
- nni_list recvpipes; // list of pipes with data to receive
- nni_list recvq;
- rep0_ctx ctx;
- nni_pollable readable;
- nni_pollable writable;
+ nni_mtx lk;
+ int ttl;
+ nni_idhash * pipes;
+ nni_list recvpipes; // list of pipes with data to receive
+ nni_list recvq;
+ rep0_ctx ctx;
+ nni_pollable readable;
+ nni_pollable writable;
};
// rep0_pipe is our per-pipe protocol private structure.
@@ -63,8 +63,8 @@ struct rep0_pipe {
nni_pipe * pipe;
rep0_sock * rep;
uint32_t id;
- nni_aio * aio_send;
- nni_aio * aio_recv;
+ nni_aio aio_send;
+ nni_aio aio_recv;
nni_list_node rnode; // receivable list linkage
nni_list sendq; // contexts waiting to send
bool busy;
@@ -193,8 +193,8 @@ rep0_ctx_send(void *arg, nni_aio *aio)
if (!p->busy) {
p->busy = true;
len = nni_msg_len(msg);
- nni_aio_set_msg(p->aio_send, msg);
- nni_pipe_send(p->pipe, p->aio_send);
+ nni_aio_set_msg(&p->aio_send, msg);
+ nni_pipe_send(p->pipe, &p->aio_send);
nni_mtx_unlock(&s->lk);
nni_aio_set_msg(aio, NULL);
@@ -273,8 +273,8 @@ rep0_pipe_stop(void *arg)
{
rep0_pipe *p = arg;
- nni_aio_stop(p->aio_send);
- nni_aio_stop(p->aio_recv);
+ nni_aio_stop(&p->aio_send);
+ nni_aio_stop(&p->aio_recv);
}
static void
@@ -283,26 +283,22 @@ rep0_pipe_fini(void *arg)
rep0_pipe *p = arg;
nng_msg * msg;
- if ((msg = nni_aio_get_msg(p->aio_recv)) != NULL) {
- nni_aio_set_msg(p->aio_recv, NULL);
+ if ((msg = nni_aio_get_msg(&p->aio_recv)) != NULL) {
+ nni_aio_set_msg(&p->aio_recv, NULL);
nni_msg_free(msg);
}
- nni_aio_fini(p->aio_send);
- nni_aio_fini(p->aio_recv);
+ nni_aio_fini(&p->aio_send);
+ nni_aio_fini(&p->aio_recv);
}
static int
rep0_pipe_init(void *arg, nni_pipe *pipe, void *s)
{
rep0_pipe *p = arg;
- int rv;
- if (((rv = nni_aio_init(&p->aio_send, rep0_pipe_send_cb, p)) != 0) ||
- ((rv = nni_aio_init(&p->aio_recv, rep0_pipe_recv_cb, p)) != 0)) {
- rep0_pipe_fini(p);
- return (rv);
- }
+ nni_aio_init(&p->aio_send, rep0_pipe_send_cb, p);
+ nni_aio_init(&p->aio_recv, rep0_pipe_recv_cb, p);
NNI_LIST_INIT(&p->sendq, rep0_ctx, sqnode);
@@ -329,7 +325,7 @@ rep0_pipe_start(void *arg)
}
// By definition, we have not received a request yet on this pipe,
// so it cannot cause us to become writable.
- nni_pipe_recv(p->pipe, p->aio_recv);
+ nni_pipe_recv(p->pipe, &p->aio_recv);
return (0);
}
@@ -340,8 +336,8 @@ rep0_pipe_close(void *arg)
rep0_sock *s = p->rep;
rep0_ctx * ctx;
- nni_aio_close(p->aio_send);
- nni_aio_close(p->aio_recv);
+ nni_aio_close(&p->aio_send);
+ nni_aio_close(&p->aio_recv);
nni_mtx_lock(&s->lk);
if (nni_list_active(&s->recvpipes, p)) {
@@ -380,9 +376,9 @@ rep0_pipe_send_cb(void *arg)
nni_msg * msg;
size_t len;
- if (nni_aio_result(p->aio_send) != 0) {
- nni_msg_free(nni_aio_get_msg(p->aio_send));
- nni_aio_set_msg(p->aio_send, NULL);
+ if (nni_aio_result(&p->aio_send) != 0) {
+ nni_msg_free(nni_aio_get_msg(&p->aio_send));
+ nni_aio_set_msg(&p->aio_send, NULL);
nni_pipe_close(p->pipe);
return;
}
@@ -406,8 +402,8 @@ rep0_pipe_send_cb(void *arg)
msg = nni_aio_get_msg(aio);
len = nni_msg_len(msg);
nni_aio_set_msg(aio, NULL);
- nni_aio_set_msg(p->aio_send, msg);
- nni_pipe_send(p->pipe, p->aio_send);
+ nni_aio_set_msg(&p->aio_send, msg);
+ nni_pipe_send(p->pipe, &p->aio_send);
nni_mtx_unlock(&s->lk);
@@ -462,13 +458,13 @@ rep0_ctx_recv(void *arg, nni_aio *aio)
nni_mtx_unlock(&s->lk);
return;
}
- msg = nni_aio_get_msg(p->aio_recv);
- nni_aio_set_msg(p->aio_recv, NULL);
+ msg = nni_aio_get_msg(&p->aio_recv);
+ nni_aio_set_msg(&p->aio_recv, NULL);
nni_list_remove(&s->recvpipes, p);
if (nni_list_empty(&s->recvpipes)) {
nni_pollable_clear(&s->readable);
}
- nni_pipe_recv(p->pipe, p->aio_recv);
+ nni_pipe_recv(p->pipe, &p->aio_recv);
if ((ctx == &s->ctx) && !p->busy) {
nni_pollable_raise(&s->writable);
}
@@ -496,12 +492,12 @@ rep0_pipe_recv_cb(void *arg)
size_t len;
int hops;
- if (nni_aio_result(p->aio_recv) != 0) {
+ if (nni_aio_result(&p->aio_recv) != 0) {
nni_pipe_close(p->pipe);
return;
}
- msg = nni_aio_get_msg(p->aio_recv);
+ msg = nni_aio_get_msg(&p->aio_recv);
nni_msg_set_pipe(msg, p->id);
@@ -521,7 +517,7 @@ rep0_pipe_recv_cb(void *arg)
if (nni_msg_len(msg) < 4) {
// Peer is speaking garbage. Kick it.
nni_msg_free(msg);
- nni_aio_set_msg(p->aio_recv, NULL);
+ nni_aio_set_msg(&p->aio_recv, NULL);
nni_pipe_close(p->pipe);
return;
}
@@ -552,13 +548,13 @@ rep0_pipe_recv_cb(void *arg)
nni_list_remove(&s->recvq, ctx);
aio = ctx->raio;
ctx->raio = NULL;
- nni_aio_set_msg(p->aio_recv, NULL);
+ nni_aio_set_msg(&p->aio_recv, NULL);
if ((ctx == &s->ctx) && !p->busy) {
nni_pollable_raise(&s->writable);
}
// schedule another receive
- nni_pipe_recv(p->pipe, p->aio_recv);
+ nni_pipe_recv(p->pipe, &p->aio_recv);
ctx->btrace_len = len;
memcpy(ctx->btrace, nni_msg_header(msg), len);
@@ -573,8 +569,8 @@ rep0_pipe_recv_cb(void *arg)
drop:
nni_msg_free(msg);
- nni_aio_set_msg(p->aio_recv, NULL);
- nni_pipe_recv(p->pipe, p->aio_recv);
+ nni_aio_set_msg(&p->aio_recv, NULL);
+ nni_pipe_recv(p->pipe, &p->aio_recv);
}
static int
diff --git a/src/protocol/reqrep0/rep_test.c b/src/protocol/reqrep0/rep_test.c
index 879d6ae4..f339e68d 100644
--- a/src/protocol/reqrep0/rep_test.c
+++ b/src/protocol/reqrep0/rep_test.c
@@ -273,6 +273,26 @@ test_rep_close_pipe_during_send(void)
}
void
+test_rep_ctx_recv_aio_stopped(void)
+{
+ nng_socket rep;
+ nng_ctx ctx;
+ nng_aio * aio;
+
+ TEST_NNG_PASS(nng_rep0_open(&rep));
+ TEST_NNG_PASS(nng_aio_alloc(&aio, NULL, NULL));
+ TEST_NNG_PASS(nng_ctx_open(&ctx, rep));
+
+ nng_aio_stop(aio);
+ nng_ctx_recv(ctx, aio);
+ nng_aio_wait(aio);
+ TEST_NNG_FAIL(nng_aio_result(aio), NNG_ECANCELED);
+ TEST_NNG_PASS(nng_ctx_close(ctx));
+ TEST_NNG_PASS(nng_close(rep));
+ nng_aio_free(aio);
+}
+
+void
test_rep_close_pipe_context_send(void)
{
nng_socket rep;
@@ -424,6 +444,7 @@ TEST_LIST = {
{ "rep double recv", test_rep_double_recv },
{ "rep close pipe before send", test_rep_close_pipe_before_send },
{ "rep close pipe during send", test_rep_close_pipe_during_send },
+ { "rep recv aio ctx stopped", test_rep_ctx_recv_aio_stopped },
{ "rep close pipe context send", test_rep_close_pipe_context_send },
{ "rep close context send", test_rep_close_context_send },
{ "rep recv garbage", test_rep_recv_garbage },
diff --git a/src/protocol/reqrep0/req.c b/src/protocol/reqrep0/req.c
index 33629abc..14da7143 100644
--- a/src/protocol/reqrep0/req.c
+++ b/src/protocol/reqrep0/req.c
@@ -184,8 +184,8 @@ req0_pipe_fini(void *arg)
{
req0_pipe *p = arg;
- nni_aio_fini(p->aio_recv);
- nni_aio_fini(p->aio_send);
+ nni_aio_free(p->aio_recv);
+ nni_aio_free(p->aio_send);
}
static int
@@ -194,8 +194,8 @@ req0_pipe_init(void *arg, nni_pipe *pipe, void *s)
req0_pipe *p = arg;
int rv;
- if (((rv = nni_aio_init(&p->aio_recv, req0_recv_cb, p)) != 0) ||
- ((rv = nni_aio_init(&p->aio_send, req0_send_cb, p)) != 0)) {
+ if (((rv = nni_aio_alloc(&p->aio_recv, req0_recv_cb, p)) != 0) ||
+ ((rv = nni_aio_alloc(&p->aio_send, req0_send_cb, p)) != 0)) {
req0_pipe_fini(p);
return (rv);
}
diff --git a/src/protocol/reqrep0/xrep.c b/src/protocol/reqrep0/xrep.c
index 48f74075..308c0f0e 100644
--- a/src/protocol/reqrep0/xrep.c
+++ b/src/protocol/reqrep0/xrep.c
@@ -62,7 +62,7 @@ xrep0_sock_fini(void *arg)
{
xrep0_sock *s = arg;
- nni_aio_fini(s->aio_getq);
+ nni_aio_free(s->aio_getq);
nni_idhash_fini(s->pipes);
nni_mtx_fini(&s->lk);
}
@@ -75,7 +75,7 @@ xrep0_sock_init(void *arg, nni_sock *sock)
nni_mtx_init(&s->lk);
if (((rv = nni_idhash_init(&s->pipes)) != 0) ||
- ((rv = nni_aio_init(&s->aio_getq, xrep0_sock_getq_cb, s)) != 0)) {
+ ((rv = nni_aio_alloc(&s->aio_getq, xrep0_sock_getq_cb, s)) != 0)) {
xrep0_sock_fini(s);
return (rv);
}
@@ -120,10 +120,10 @@ xrep0_pipe_fini(void *arg)
{
xrep0_pipe *p = arg;
- nni_aio_fini(p->aio_getq);
- nni_aio_fini(p->aio_send);
- nni_aio_fini(p->aio_recv);
- nni_aio_fini(p->aio_putq);
+ nni_aio_free(p->aio_getq);
+ nni_aio_free(p->aio_send);
+ nni_aio_free(p->aio_recv);
+ nni_aio_free(p->aio_putq);
nni_msgq_fini(p->sendq);
}
@@ -146,10 +146,10 @@ xrep0_pipe_init(void *arg, nni_pipe *pipe, void *s)
// willing to receive replies. Something to think about for the
// future.)
if (((rv = nni_msgq_init(&p->sendq, 64)) != 0) ||
- ((rv = nni_aio_init(&p->aio_getq, xrep0_pipe_getq_cb, p)) != 0) ||
- ((rv = nni_aio_init(&p->aio_send, xrep0_pipe_send_cb, p)) != 0) ||
- ((rv = nni_aio_init(&p->aio_recv, xrep0_pipe_recv_cb, p)) != 0) ||
- ((rv = nni_aio_init(&p->aio_putq, xrep0_pipe_putq_cb, p)) != 0)) {
+ ((rv = nni_aio_alloc(&p->aio_getq, xrep0_pipe_getq_cb, p)) != 0) ||
+ ((rv = nni_aio_alloc(&p->aio_send, xrep0_pipe_send_cb, p)) != 0) ||
+ ((rv = nni_aio_alloc(&p->aio_recv, xrep0_pipe_recv_cb, p)) != 0) ||
+ ((rv = nni_aio_alloc(&p->aio_putq, xrep0_pipe_putq_cb, p)) != 0)) {
xrep0_pipe_fini(p);
return (rv);
}
diff --git a/src/protocol/reqrep0/xreq.c b/src/protocol/reqrep0/xreq.c
index 7455c986..15652f4f 100644
--- a/src/protocol/reqrep0/xreq.c
+++ b/src/protocol/reqrep0/xreq.c
@@ -96,10 +96,10 @@ xreq0_pipe_fini(void *arg)
{
xreq0_pipe *p = arg;
- nni_aio_fini(p->aio_getq);
- nni_aio_fini(p->aio_putq);
- nni_aio_fini(p->aio_recv);
- nni_aio_fini(p->aio_send);
+ nni_aio_free(p->aio_getq);
+ nni_aio_free(p->aio_putq);
+ nni_aio_free(p->aio_recv);
+ nni_aio_free(p->aio_send);
}
static int
@@ -108,10 +108,10 @@ xreq0_pipe_init(void *arg, nni_pipe *pipe, void *s)
xreq0_pipe *p = arg;
int rv;
- if (((rv = nni_aio_init(&p->aio_getq, xreq0_getq_cb, p)) != 0) ||
- ((rv = nni_aio_init(&p->aio_putq, xreq0_putq_cb, p)) != 0) ||
- ((rv = nni_aio_init(&p->aio_recv, xreq0_recv_cb, p)) != 0) ||
- ((rv = nni_aio_init(&p->aio_send, xreq0_send_cb, p)) != 0)) {
+ if (((rv = nni_aio_alloc(&p->aio_getq, xreq0_getq_cb, p)) != 0) ||
+ ((rv = nni_aio_alloc(&p->aio_putq, xreq0_putq_cb, p)) != 0) ||
+ ((rv = nni_aio_alloc(&p->aio_recv, xreq0_recv_cb, p)) != 0) ||
+ ((rv = nni_aio_alloc(&p->aio_send, xreq0_send_cb, p)) != 0)) {
xreq0_pipe_fini(p);
return (rv);
}
generated by cgit v1.2.3-70-g09d2 (git 2.46.0) at 2026-01-18 17:27:57 +0000