aboutsummaryrefslogtreecommitdiff
path: root/src/protocol
diff options
context:
space:
mode:
Diffstat (limited to 'src/protocol')
-rw-r--r--src/protocol/bus/bus.c25
-rw-r--r--src/protocol/pair/pair.c1
-rw-r--r--src/protocol/pipeline/pull.c12
-rw-r--r--src/protocol/pipeline/push.c9
-rw-r--r--src/protocol/pubsub/pub.c9
-rw-r--r--src/protocol/pubsub/sub.c4
-rw-r--r--src/protocol/reqrep/rep.c9
-rw-r--r--src/protocol/reqrep/req.c59
-rw-r--r--src/protocol/survey/respond.c33
-rw-r--r--src/protocol/survey/survey.c10
10 files changed, 103 insertions, 68 deletions
diff --git a/src/protocol/bus/bus.c b/src/protocol/bus/bus.c
index 17ef03bb..3070b90d 100644
--- a/src/protocol/bus/bus.c
+++ b/src/protocol/bus/bus.c
@@ -60,6 +60,7 @@ nni_bus_sock_fini(void *arg)
nni_bus_sock *psock = arg;
if (psock != NULL) {
+ nni_aio_stop(&psock->aio_getq);
nni_aio_fini(&psock->aio_getq);
nni_mtx_fini(&psock->mtx);
NNI_FREE_STRUCT(psock);
@@ -107,15 +108,13 @@ nni_bus_pipe_fini(void *arg)
{
nni_bus_pipe *ppipe = arg;
- if (ppipe != NULL) {
- nni_mtx_fini(&ppipe->mtx);
- nni_aio_fini(&ppipe->aio_getq);
- nni_aio_fini(&ppipe->aio_send);
- nni_aio_fini(&ppipe->aio_recv);
- nni_aio_fini(&ppipe->aio_putq);
- nni_msgq_fini(ppipe->sendq);
- NNI_FREE_STRUCT(ppipe);
- }
+ nni_aio_fini(&ppipe->aio_getq);
+ nni_aio_fini(&ppipe->aio_send);
+ nni_aio_fini(&ppipe->aio_recv);
+ nni_aio_fini(&ppipe->aio_putq);
+ nni_msgq_fini(ppipe->sendq);
+ nni_mtx_fini(&ppipe->mtx);
+ NNI_FREE_STRUCT(ppipe);
}
static int
@@ -183,10 +182,10 @@ nni_bus_pipe_stop(void *arg)
nni_msgq_close(ppipe->sendq);
- nni_aio_cancel(&ppipe->aio_getq, NNG_ECLOSED);
- nni_aio_cancel(&ppipe->aio_send, NNG_ECLOSED);
- nni_aio_cancel(&ppipe->aio_recv, NNG_ECLOSED);
- nni_aio_cancel(&ppipe->aio_putq, NNG_ECLOSED);
+ nni_aio_stop(&ppipe->aio_getq);
+ nni_aio_stop(&ppipe->aio_send);
+ nni_aio_stop(&ppipe->aio_recv);
+ nni_aio_stop(&ppipe->aio_putq);
nni_mtx_lock(&ppipe->psock->mtx);
if (nni_list_active(&psock->pipes, ppipe)) {
diff --git a/src/protocol/pair/pair.c b/src/protocol/pair/pair.c
index f5fec540..55ce5aa9 100644
--- a/src/protocol/pair/pair.c
+++ b/src/protocol/pair/pair.c
@@ -122,7 +122,6 @@ static void
nni_pair_pipe_fini(void *arg)
{
nni_pair_pipe *ppipe = arg;
-
nni_aio_fini(&ppipe->aio_send);
nni_aio_fini(&ppipe->aio_recv);
nni_aio_fini(&ppipe->aio_putq);
diff --git a/src/protocol/pipeline/pull.c b/src/protocol/pipeline/pull.c
index 39e809e6..cde79824 100644
--- a/src/protocol/pipeline/pull.c
+++ b/src/protocol/pipeline/pull.c
@@ -90,11 +90,9 @@ nni_pull_pipe_fini(void *arg)
{
nni_pull_pipe *pp = arg;
- if (pp != NULL) {
- nni_aio_fini(&pp->putq_aio);
- nni_aio_fini(&pp->recv_aio);
- NNI_FREE_STRUCT(pp);
- }
+ nni_aio_fini(&pp->putq_aio);
+ nni_aio_fini(&pp->recv_aio);
+ NNI_FREE_STRUCT(pp);
}
static int
@@ -113,8 +111,8 @@ nni_pull_pipe_stop(void *arg)
{
nni_pull_pipe *pp = arg;
- nni_aio_cancel(&pp->putq_aio, NNG_ECANCELED);
- nni_aio_cancel(&pp->recv_aio, NNG_ECANCELED);
+ nni_aio_stop(&pp->putq_aio);
+ nni_aio_stop(&pp->recv_aio);
}
static void
diff --git a/src/protocol/pipeline/push.c b/src/protocol/pipeline/push.c
index 43c0feaf..b7d4322c 100644
--- a/src/protocol/pipeline/push.c
+++ b/src/protocol/pipeline/push.c
@@ -132,12 +132,11 @@ nni_push_pipe_start(void *arg)
static void
nni_push_pipe_stop(void *arg)
{
- nni_push_pipe *pp = arg;
- nni_push_sock *push = pp->push;
+ nni_push_pipe *pp = arg;
- nni_aio_cancel(&pp->aio_recv, NNG_ECANCELED);
- nni_aio_cancel(&pp->aio_send, NNG_ECANCELED);
- nni_aio_cancel(&pp->aio_getq, NNG_ECANCELED);
+ nni_aio_stop(&pp->aio_recv);
+ nni_aio_stop(&pp->aio_send);
+ nni_aio_stop(&pp->aio_getq);
}
static void
diff --git a/src/protocol/pubsub/pub.c b/src/protocol/pubsub/pub.c
index 316cbf50..e32f179a 100644
--- a/src/protocol/pubsub/pub.c
+++ b/src/protocol/pubsub/pub.c
@@ -83,6 +83,7 @@ nni_pub_sock_fini(void *arg)
{
nni_pub_sock *pub = arg;
+ nni_aio_stop(&pub->aio_getq);
nni_aio_fini(&pub->aio_getq);
nni_mtx_fini(&pub->mtx);
NNI_FREE_STRUCT(pub);
@@ -100,7 +101,6 @@ static void
nni_pub_pipe_fini(void *arg)
{
nni_pub_pipe *pp = arg;
-
nni_aio_fini(&pp->aio_getq);
nni_aio_fini(&pp->aio_send);
nni_aio_fini(&pp->aio_recv);
@@ -172,9 +172,10 @@ nni_pub_pipe_stop(void *arg)
nni_pub_pipe *pp = arg;
nni_pub_sock *pub = pp->pub;
- nni_aio_cancel(&pp->aio_getq, NNG_ECANCELED);
- nni_aio_cancel(&pp->aio_send, NNG_ECANCELED);
- nni_aio_cancel(&pp->aio_recv, NNG_ECANCELED);
+ nni_aio_stop(&pp->aio_getq);
+ nni_aio_stop(&pp->aio_send);
+ nni_aio_stop(&pp->aio_recv);
+
nni_msgq_close(pp->sendq);
nni_mtx_lock(&pub->mtx);
diff --git a/src/protocol/pubsub/sub.c b/src/protocol/pubsub/sub.c
index 36a42c49..03d76e2d 100644
--- a/src/protocol/pubsub/sub.c
+++ b/src/protocol/pubsub/sub.c
@@ -123,8 +123,8 @@ nni_sub_pipe_stop(void *arg)
{
nni_sub_pipe *sp = arg;
- nni_aio_cancel(&sp->aio_putq, NNG_ECANCELED);
- nni_aio_cancel(&sp->aio_recv, NNG_ECANCELED);
+ nni_aio_stop(&sp->aio_putq);
+ nni_aio_stop(&sp->aio_recv);
}
static void
diff --git a/src/protocol/reqrep/rep.c b/src/protocol/reqrep/rep.c
index 013b02fb..049b1422 100644
--- a/src/protocol/reqrep/rep.c
+++ b/src/protocol/reqrep/rep.c
@@ -56,6 +56,7 @@ nni_rep_sock_fini(void *arg)
{
nni_rep_sock *rep = arg;
+ nni_aio_stop(&rep->aio_getq);
nni_aio_fini(&rep->aio_getq);
nni_idhash_fini(rep->pipes);
if (rep->btrace != NULL) {
@@ -192,10 +193,10 @@ nni_rep_pipe_stop(void *arg)
nni_rep_sock *rep = rp->rep;
nni_msgq_close(rp->sendq);
- nni_aio_cancel(&rp->aio_getq, NNG_ECANCELED);
- nni_aio_cancel(&rp->aio_putq, NNG_ECANCELED);
- nni_aio_cancel(&rp->aio_send, NNG_ECANCELED);
- nni_aio_cancel(&rp->aio_recv, NNG_ECANCELED);
+ nni_aio_stop(&rp->aio_getq);
+ nni_aio_stop(&rp->aio_send);
+ nni_aio_stop(&rp->aio_recv);
+ nni_aio_stop(&rp->aio_putq);
nni_idhash_remove(rep->pipes, nni_pipe_id(rp->pipe));
}
diff --git a/src/protocol/reqrep/req.c b/src/protocol/reqrep/req.c
index f13094ff..8e7056f5 100644
--- a/src/protocol/reqrep/req.c
+++ b/src/protocol/reqrep/req.c
@@ -34,6 +34,7 @@ struct nni_req_sock {
nni_time resend;
int raw;
int wantw;
+ int closed;
nni_msg * reqmsg;
nni_req_pipe *pendpipe;
@@ -46,6 +47,7 @@ struct nni_req_sock {
uint32_t nextid; // next id
uint8_t reqid[4]; // outstanding request ID (big endian)
nni_mtx mtx;
+ nni_cv cv;
};
// An nni_req_pipe is our per-pipe protocol private structure.
@@ -81,6 +83,10 @@ nni_req_sock_init(void **reqp, nni_sock *sock)
NNI_FREE_STRUCT(req);
return (rv);
}
+ if ((rv = nni_cv_init(&req->cv, &req->mtx)) != 0) {
+ nni_mtx_fini(&req->mtx);
+ NNI_FREE_STRUCT(req);
+ }
NNI_LIST_INIT(&req->readypipes, nni_req_pipe, node);
NNI_LIST_INIT(&req->busypipes, nni_req_pipe, node);
@@ -108,6 +114,10 @@ nni_req_sock_close(void *arg)
{
nni_req_sock *req = arg;
+ nni_mtx_lock(&req->mtx);
+ req->closed = 1;
+ nni_mtx_unlock(&req->mtx);
+
nni_timer_cancel(&req->timer);
}
@@ -117,10 +127,15 @@ nni_req_sock_fini(void *arg)
nni_req_sock *req = arg;
nni_mtx_lock(&req->mtx);
+ while ((!nni_list_empty(&req->readypipes)) ||
+ (!nni_list_empty(&req->busypipes))) {
+ nni_cv_wait(&req->cv);
+ }
if (req->reqmsg != NULL) {
nni_msg_free(req->reqmsg);
}
nni_mtx_unlock(&req->mtx);
+ nni_cv_fini(&req->cv);
nni_mtx_fini(&req->mtx);
NNI_FREE_STRUCT(req);
}
@@ -171,15 +186,13 @@ nni_req_pipe_fini(void *arg)
{
nni_req_pipe *rp = arg;
- if (rp != NULL) {
- nni_aio_fini(&rp->aio_getq);
- nni_aio_fini(&rp->aio_putq);
- nni_aio_fini(&rp->aio_recv);
- nni_aio_fini(&rp->aio_sendcooked);
- nni_aio_fini(&rp->aio_sendraw);
- nni_mtx_fini(&rp->mtx);
- NNI_FREE_STRUCT(rp);
- }
+ nni_aio_fini(&rp->aio_getq);
+ nni_aio_fini(&rp->aio_putq);
+ nni_aio_fini(&rp->aio_recv);
+ nni_aio_fini(&rp->aio_sendcooked);
+ nni_aio_fini(&rp->aio_sendraw);
+ nni_mtx_fini(&rp->mtx);
+ NNI_FREE_STRUCT(rp);
}
static int
@@ -193,6 +206,10 @@ nni_req_pipe_start(void *arg)
}
nni_mtx_lock(&req->mtx);
+ if (req->closed) {
+ nni_mtx_unlock(&req->mtx);
+ return (NNG_ECLOSED);
+ }
nni_list_append(&req->readypipes, rp);
if (req->wantw) {
nni_req_resend(req);
@@ -210,11 +227,11 @@ nni_req_pipe_stop(void *arg)
nni_req_pipe *rp = arg;
nni_req_sock *req = rp->req;
- nni_aio_cancel(&rp->aio_getq, NNG_ECANCELED);
- nni_aio_cancel(&rp->aio_putq, NNG_ECANCELED);
- nni_aio_cancel(&rp->aio_recv, NNG_ECANCELED);
- nni_aio_cancel(&rp->aio_sendcooked, NNG_ECANCELED);
- nni_aio_cancel(&rp->aio_sendraw, NNG_ECANCELED);
+ nni_aio_stop(&rp->aio_getq);
+ nni_aio_stop(&rp->aio_putq);
+ nni_aio_stop(&rp->aio_recv);
+ nni_aio_stop(&rp->aio_sendcooked);
+ nni_aio_stop(&rp->aio_sendraw);
// At this point there should not be any further AIOs running.
// Further, any completion tasks have completed.
@@ -222,8 +239,11 @@ nni_req_pipe_stop(void *arg)
nni_mtx_lock(&req->mtx);
// This removes the node from either busypipes or readypipes.
// It doesn't much matter which.
- if (nni_list_active(&req->readypipes, rp)) {
- nni_list_remove(&req->readypipes, rp);
+ if (nni_list_node_active(&rp->node)) {
+ nni_list_node_remove(&rp->node);
+ if (req->closed) {
+ nni_cv_wake(&req->cv);
+ }
}
if ((rp == req->pendpipe) && (req->reqmsg != NULL)) {
@@ -443,10 +463,15 @@ nni_req_resend(nni_req_sock *req)
// Note: This routine should be called with the socket lock held.
// Also, this should only be called while handling cooked mode
// requests.
- if (req->reqmsg == NULL) {
+ if ((msg = req->reqmsg) == NULL) {
return;
}
+ if (req->closed) {
+ req->reqmsg = NULL;
+ nni_msg_free(msg);
+ }
+
if (req->wantw) {
req->wantw = 0;
diff --git a/src/protocol/survey/respond.c b/src/protocol/survey/respond.c
index 4a4c8741..089e730e 100644
--- a/src/protocol/survey/respond.c
+++ b/src/protocol/survey/respond.c
@@ -38,6 +38,7 @@ struct nni_resp_sock {
char * btrace;
size_t btrace_len;
nni_aio aio_getq;
+ nni_mtx mtx;
};
// An nni_resp_pipe is our per-pipe protocol private structure.
@@ -57,14 +58,14 @@ nni_resp_sock_fini(void *arg)
{
nni_resp_sock *psock = arg;
- if (psock != NULL) {
- nni_aio_fini(&psock->aio_getq);
- nni_idhash_fini(psock->pipes);
- if (psock->btrace != NULL) {
- nni_free(psock->btrace, psock->btrace_len);
- }
- NNI_FREE_STRUCT(psock);
+ nni_aio_stop(&psock->aio_getq);
+ nni_aio_fini(&psock->aio_getq);
+ nni_idhash_fini(psock->pipes);
+ if (psock->btrace != NULL) {
+ nni_free(psock->btrace, psock->btrace_len);
}
+ nni_mtx_fini(&psock->mtx);
+ NNI_FREE_STRUCT(psock);
}
static int
@@ -83,6 +84,10 @@ nni_resp_sock_init(void **pp, nni_sock *nsock)
psock->btrace_len = 0;
psock->urq = nni_sock_recvq(nsock);
psock->uwq = nni_sock_sendq(nsock);
+
+ if ((rv = nni_mtx_init(&psock->mtx)) != 0) {
+ goto fail;
+ }
if ((rv = nni_idhash_init(&psock->pipes)) != 0) {
goto fail;
}
@@ -177,7 +182,9 @@ nni_resp_pipe_start(void *arg)
ppipe->id = nni_pipe_id(ppipe->npipe);
+ nni_mtx_lock(&psock->mtx);
rv = nni_idhash_insert(psock->pipes, ppipe->id, ppipe);
+ nni_mtx_unlock(&psock->mtx);
if (rv != 0) {
return (rv);
}
@@ -195,13 +202,15 @@ nni_resp_pipe_stop(void *arg)
nni_resp_sock *psock = ppipe->psock;
nni_msgq_close(ppipe->sendq);
- nni_aio_cancel(&ppipe->aio_putq, NNG_ECANCELED);
- nni_aio_cancel(&ppipe->aio_getq, NNG_ECANCELED);
- nni_aio_cancel(&ppipe->aio_send, NNG_ECANCELED);
- nni_aio_cancel(&ppipe->aio_recv, NNG_ECANCELED);
+ nni_aio_stop(&ppipe->aio_putq);
+ nni_aio_stop(&ppipe->aio_getq);
+ nni_aio_stop(&ppipe->aio_send);
+ nni_aio_stop(&ppipe->aio_recv);
if (ppipe->id != 0) {
+ nni_mtx_lock(&psock->mtx);
nni_idhash_remove(psock->pipes, ppipe->id);
+ nni_mtx_unlock(&psock->mtx);
ppipe->id = 0;
}
}
@@ -239,6 +248,7 @@ nni_resp_sock_getq_cb(void *arg)
NNI_GET32(header, id);
nni_msg_trim_header(msg, 4);
+ nni_mtx_lock(&psock->mtx);
rv = nni_idhash_find(psock->pipes, id, (void **) &ppipe);
if (rv != 0) {
@@ -250,6 +260,7 @@ nni_resp_sock_getq_cb(void *arg)
nni_msg_free(msg);
}
}
+ nni_mtx_unlock(&psock->mtx);
}
void
diff --git a/src/protocol/survey/survey.c b/src/protocol/survey/survey.c
index 85657f57..633e1491 100644
--- a/src/protocol/survey/survey.c
+++ b/src/protocol/survey/survey.c
@@ -60,6 +60,7 @@ nni_surv_sock_fini(void *arg)
{
nni_surv_sock *psock = arg;
+ nni_aio_stop(&psock->aio_getq);
nni_aio_fini(&psock->aio_getq);
nni_mtx_fini(&psock->mtx);
NNI_FREE_STRUCT(psock);
@@ -191,10 +192,11 @@ nni_surv_pipe_stop(void *arg)
nni_surv_pipe *ppipe = arg;
nni_surv_sock *psock = ppipe->psock;
- nni_aio_cancel(&ppipe->aio_getq, NNG_ECANCELED);
- nni_aio_cancel(&ppipe->aio_send, NNG_ECANCELED);
- nni_aio_cancel(&ppipe->aio_recv, NNG_ECANCELED);
- nni_aio_cancel(&ppipe->aio_putq, NNG_ECANCELED);
+ nni_aio_stop(&ppipe->aio_getq);
+ nni_aio_stop(&ppipe->aio_send);
+ nni_aio_stop(&ppipe->aio_recv);
+ nni_aio_stop(&ppipe->aio_putq);
+
nni_msgq_close(ppipe->sendq);
nni_mtx_lock(&psock->mtx);
generated by cgit v1.2.3-70-g09d2 (git 2.46.0) at 2026-01-18 17:01:12 +0000