From 77311bfcc94bba96cdee73ddcd1ac9a6d0ed17d2 Mon Sep 17 00:00:00 2001 From: Garrett D'Amore Date: Tue, 14 Aug 2018 15:57:57 +0500 Subject: fixes #208 pipe start should occur before connect / accept fixes #599 nng_dial sync should not return until added to socket This reintroduces the changes for the above fixes, building upon the transport modifications that we have made to eliminate the separate transport pipe start entry point. It also includes slightly reworked code during start to put a hold on the pipe when it is created, which we we drop at the end, hopefully fixing a use-after-free. --- src/core/pipe.c | 69 ++++++--------------------------------------------------- 1 file changed, 7 insertions(+), 62 deletions(-) (limited to 'src/core/pipe.c') diff --git a/src/core/pipe.c b/src/core/pipe.c index 4f50ac7c..66cd87cc 100644 --- a/src/core/pipe.c +++ b/src/core/pipe.c @@ -75,9 +75,6 @@ pipe_destroy(nni_pipe *p) } nni_mtx_unlock(&nni_pipe_lk); - // Wait for neg callbacks to finish. (Already closed). - nni_aio_stop(p->p_start_aio); - if (p->p_proto_data != NULL) { p->p_proto_ops.pipe_stop(p->p_proto_data); } @@ -93,7 +90,6 @@ pipe_destroy(nni_pipe *p) if (p->p_tran_data != NULL) { p->p_tran_ops.p_fini(p->p_tran_data); } - nni_aio_fini(p->p_start_aio); nni_cv_fini(&p->p_cv); nni_mtx_fini(&p->p_mtx); NNI_FREE_STRUCT(p); @@ -154,9 +150,6 @@ nni_pipe_send(nni_pipe *p, nni_aio *aio) void nni_pipe_close(nni_pipe *p) { - // abort any pending negotiation/start process. - nni_aio_close(p->p_start_aio); - nni_mtx_lock(&p->p_mtx); if (p->p_closed) { // We already did a close. @@ -178,49 +171,12 @@ nni_pipe_close(nni_pipe *p) nni_reap(&p->p_reap, (nni_cb) pipe_destroy, p); } -bool -nni_pipe_closed(nni_pipe *p) -{ - bool rv; - nni_mtx_lock(&p->p_mtx); - rv = p->p_closed; - nni_mtx_unlock(&p->p_mtx); - return (rv); -} - uint16_t nni_pipe_peer(nni_pipe *p) { return (p->p_tran_ops.p_peer(p->p_tran_data)); } -static void -nni_pipe_start_cb(void *arg) -{ - nni_pipe *p = arg; - nni_sock *s = p->p_sock; - nni_aio * aio = p->p_start_aio; - - if (nni_aio_result(aio) != 0) { - nni_pipe_close(p); - return; - } - - nni_pipe_run_cb(p, NNG_PIPE_EV_ADD_PRE); - if (nni_pipe_closed(p)) { - nni_pipe_close(p); - return; - } - - if ((p->p_proto_ops.pipe_start(p->p_proto_data) != 0) || - nni_sock_closing(s)) { - nni_pipe_close(p); - return; - } - - nni_pipe_run_cb(p, NNG_PIPE_EV_ADD_POST); -} - int nni_pipe_create(nni_pipe **pp, nni_sock *sock, nni_tran *tran, void *tdata) { @@ -228,6 +184,7 @@ nni_pipe_create(nni_pipe **pp, nni_sock *sock, nni_tran *tran, void *tdata) int rv; void * sdata = nni_sock_proto_data(sock); nni_proto_pipe_ops *pops = nni_sock_proto_pipe_ops(sock); + uint64_t id; if ((p = NNI_ALLOC_STRUCT(p)) == NULL) { // In this case we just toss the pipe... @@ -236,7 +193,6 @@ nni_pipe_create(nni_pipe **pp, nni_sock *sock, nni_tran *tran, void *tdata) } // Make a private copy of the transport ops. - p->p_start_aio = NULL; p->p_tran_ops = *tran->tran_pipe; p->p_tran_data = tdata; p->p_proto_ops = *pops; @@ -253,18 +209,17 @@ nni_pipe_create(nni_pipe **pp, nni_sock *sock, nni_tran *tran, void *tdata) nni_mtx_init(&p->p_mtx); nni_cv_init(&p->p_cv, &nni_pipe_lk); - if ((rv = nni_aio_init(&p->p_start_aio, nni_pipe_start_cb, p)) == 0) { - uint64_t id; - nni_mtx_lock(&nni_pipe_lk); - if ((rv = nni_idhash_alloc(nni_pipes, &id, p)) == 0) { - p->p_id = (uint32_t) id; - } - nni_mtx_unlock(&nni_pipe_lk); + nni_mtx_lock(&nni_pipe_lk); + if ((rv = nni_idhash_alloc(nni_pipes, &id, p)) == 0) { + p->p_id = id; + p->p_refcnt = 1; } + nni_mtx_unlock(&nni_pipe_lk); if ((rv != 0) || ((rv = pops->pipe_init(&p->p_proto_data, p, sdata)) != 0)) { nni_pipe_close(p); + nni_pipe_rele(p); return (rv); } @@ -296,16 +251,6 @@ nni_pipe_getopt( return (NNG_ENOTSUP); } -void -nni_pipe_start(nni_pipe *p) -{ - if (p->p_tran_ops.p_start == NULL) { - nni_aio_finish(p->p_start_aio, 0, 0); - } else { - p->p_tran_ops.p_start(p->p_tran_data, p->p_start_aio); - } -} - void * nni_pipe_get_proto_data(nni_pipe *p) { -- cgit v1.2.3-70-g09d2