diff options
Diffstat (limited to 'src/protocol/pipeline')
| -rw-r--r-- | src/protocol/pipeline/pull.c | 16 | ||||
| -rw-r--r-- | src/protocol/pipeline/push.c | 71 |
2 files changed, 48 insertions, 39 deletions
diff --git a/src/protocol/pipeline/pull.c b/src/protocol/pipeline/pull.c index eb14be81..eb66bc21 100644 --- a/src/protocol/pipeline/pull.c +++ b/src/protocol/pipeline/pull.c @@ -107,7 +107,6 @@ nni_pull_pipe_start(void *arg) nni_pull_pipe *pp = arg; // Start the pending pull... - nni_pipe_hold(pp->pipe); nni_pull_recv(pp); return (0); @@ -115,12 +114,11 @@ nni_pull_pipe_start(void *arg) static void -nni_pull_pipe_stop(void *arg) +nni_pull_pipe_stop(nni_pull_pipe *pp) { - nni_pull_pipe *pp = arg; - // Cancel any pending sendup. nni_msgq_aio_cancel(pp->pull->urq, &pp->putq_aio); + nni_pipe_remove(pp->pipe); } @@ -133,8 +131,7 @@ nni_pull_recv_cb(void *arg) if (nni_aio_result(aio) != 0) { // Failed to get a message, probably the pipe is closed. - nni_pipe_close(pp->pipe); - nni_pipe_rele(pp->pipe); + nni_pull_pipe_stop(pp); return; } @@ -157,8 +154,7 @@ nni_pull_putq_cb(void *arg) // we can do. Just close the pipe. nni_msg_free(aio->a_msg); aio->a_msg = NULL; - nni_pipe_close(pp->pipe); - nni_pipe_rele(pp->pipe); + nni_pull_pipe_stop(pp); return; } @@ -172,8 +168,7 @@ nni_pull_recv(nni_pull_pipe *pp) { // Schedule the aio with callback. if (nni_pipe_aio_recv(pp->pipe, &pp->recv_aio) != 0) { - nni_pipe_close(pp->pipe); - nni_pipe_rele(pp->pipe); + nni_pipe_remove(pp->pipe); } } @@ -230,7 +225,6 @@ static nni_proto_pipe_ops nni_pull_pipe_ops = { .pipe_init = nni_pull_pipe_init, .pipe_fini = nni_pull_pipe_fini, .pipe_start = nni_pull_pipe_start, - .pipe_stop = nni_pull_pipe_stop, }; static nni_proto_sock_ops nni_pull_sock_ops = { diff --git a/src/protocol/pipeline/push.c b/src/protocol/pipeline/push.c index 9554b2be..e69ebdbf 100644 --- a/src/protocol/pipeline/push.c +++ b/src/protocol/pipeline/push.c @@ -39,6 +39,8 @@ struct nni_push_pipe { nni_aio aio_recv; nni_aio aio_send; nni_aio aio_getq; + int refcnt; + nni_mtx mtx; }; static int @@ -70,6 +72,19 @@ nni_push_sock_fini(void *arg) } +static void +nni_push_pipe_fini(void *arg) +{ + nni_push_pipe *pp = arg; + + nni_aio_fini(&pp->aio_recv); + nni_aio_fini(&pp->aio_send); + nni_aio_fini(&pp->aio_getq); + nni_mtx_fini(&pp->mtx); + NNI_FREE_STRUCT(pp); +} + + static int nni_push_pipe_init(void **ppp, nni_pipe *pipe, void *psock) { @@ -80,19 +95,17 @@ nni_push_pipe_init(void **ppp, nni_pipe *pipe, void *psock) return (NNG_ENOMEM); } if ((rv = nni_aio_init(&pp->aio_recv, nni_push_recv_cb, pp)) != 0) { - NNI_FREE_STRUCT(pp); - return (rv); + goto fail; } if ((rv = nni_aio_init(&pp->aio_send, nni_push_send_cb, pp)) != 0) { - nni_aio_fini(&pp->aio_recv); - NNI_FREE_STRUCT(pp); + goto fail; return (rv); } if ((rv = nni_aio_init(&pp->aio_getq, nni_push_getq_cb, pp)) != 0) { - nni_aio_fini(&pp->aio_send); - nni_aio_fini(&pp->aio_recv); - NNI_FREE_STRUCT(pp); - return (rv); + goto fail; + } + if ((rv = nni_mtx_init(&pp->mtx)) != 0) { + goto fail; } NNI_LIST_NODE_INIT(&pp->node); @@ -100,18 +113,10 @@ nni_push_pipe_init(void **ppp, nni_pipe *pipe, void *psock) pp->push = psock; *ppp = pp; return (0); -} - -static void -nni_push_pipe_fini(void *arg) -{ - nni_push_pipe *pp = arg; - - nni_aio_fini(&pp->aio_recv); - nni_aio_fini(&pp->aio_send); - nni_aio_fini(&pp->aio_getq); - NNI_FREE_STRUCT(pp); +fail: + nni_push_pipe_fini(pp); + return (rv); } @@ -125,6 +130,10 @@ nni_push_pipe_start(void *arg) return (NNG_EPROTO); } + nni_mtx_lock(&pp->mtx); + pp->refcnt = 2; + nni_mtx_unlock(&pp->mtx); + // Schedule a receiver. This is mostly so that we can detect // a closed transport pipe. nni_pipe_hold(pp->pipe); @@ -139,12 +148,22 @@ nni_push_pipe_start(void *arg) static void -nni_push_pipe_stop(void *arg) +nni_push_pipe_stop(nni_push_pipe *pp) { - nni_push_pipe *pp = arg; nni_push_sock *push = pp->push; + int refcnt; nni_msgq_aio_cancel(push->uwq, &pp->aio_getq); + + nni_mtx_lock(&pp->mtx); + NNI_ASSERT(pp->refcnt > 0); + pp->refcnt--; + refcnt = pp->refcnt; + nni_mtx_unlock(&pp->mtx); + + if (refcnt == 0) { + nni_pipe_remove(pp->pipe); + } } @@ -156,8 +175,7 @@ nni_push_recv_cb(void *arg) // We normally expect to receive an error. If a pipe actually // sends us data, we just discard it. if (nni_aio_result(&pp->aio_recv) != 0) { - nni_pipe_close(pp->pipe); - nni_pipe_rele(pp->pipe); + nni_push_pipe_stop(pp); return; } nni_msg_free(pp->aio_recv.a_msg); @@ -175,8 +193,7 @@ nni_push_send_cb(void *arg) if (nni_aio_result(&pp->aio_send) != 0) { nni_msg_free(pp->aio_send.a_msg); pp->aio_send.a_msg = NULL; - nni_pipe_close(pp->pipe); - nni_pipe_rele(pp->pipe); + nni_push_pipe_stop(pp); return; } @@ -192,8 +209,7 @@ nni_push_getq_cb(void *arg) if (nni_aio_result(aio) != 0) { // If the socket is closing, nothing else we can do. - nni_pipe_close(pp->pipe); - nni_pipe_rele(pp->pipe); + nni_push_pipe_stop(pp); return; } @@ -244,7 +260,6 @@ static nni_proto_pipe_ops nni_push_pipe_ops = { .pipe_init = nni_push_pipe_init, .pipe_fini = nni_push_pipe_fini, .pipe_start = nni_push_pipe_start, - .pipe_stop = nni_push_pipe_stop, }; static nni_proto_sock_ops nni_push_sock_ops = { |