Gdamore/win io lock

src/core/socket.c

@@ -83,7 +83,6 @@ struct nni_socket {
 
 	bool s_closing; // Socket is closing
 	bool s_closed;  // Socket closed, protected by global lock
-	bool s_ctxwait; // Waiting for contexts to close.
 
 	nni_mtx          s_pipe_cbs_mtx;
 	nni_sock_pipe_cb s_pipe_cbs[NNG_PIPE_EV_NUM];
@@ -734,7 +733,6 @@ nni_sock_shutdown(nni_sock *sock)
 	// a chance to do so gracefully.
 
 	while (!nni_list_empty(&sock->s_ctxs)) {
-		sock->s_ctxwait = true;
 		nni_cv_wait(&sock->s_close_cv);
 	}
 	nni_mtx_unlock(&sock_lk);
@@ -798,7 +796,6 @@ nni_sock_close(nni_sock *s)
 
 	// Wait for all other references to drop.  Note that we
 	// have a reference already (from our caller).
-	s->s_ctxwait = true;
 	while ((s->s_ref > 1) || (!nni_list_empty(&s->s_ctxs))) {
 		nni_cv_wait(&s->s_close_cv);
 	}
@@ -1307,9 +1304,7 @@ nni_ctx_rele(nni_ctx *ctx)
 	// tries to avoid ID reuse.
 	nni_id_remove(&ctx_ids, ctx->c_id);
 	nni_list_remove(&sock->s_ctxs, ctx);
-	if (sock->s_closed || sock->s_ctxwait) {
-		nni_cv_wake(&sock->s_close_cv);
-	}
+	nni_cv_wake(&sock->s_close_cv);
 	nni_mtx_unlock(&sock_lk);
 
 	nni_ctx_destroy(ctx);
@@ -1791,9 +1786,7 @@ nni_pipe_remove(nni_pipe *p)
 		d->d_pipe = NULL;
 		dialer_timer_start_locked(d); // Kick the timer to redial.
 	}
-	if (s->s_closing) {
-		nni_cv_wake(&s->s_cv);
-	}
+	nni_cv_wake(&s->s_cv);
 	nni_mtx_unlock(&s->s_mx);
 }
 

src/platform/windows/win_io.c

@@ -23,6 +23,8 @@ static int      win_io_nthr = 0;
 static HANDLE   win_io_h    = NULL;
 static nni_thr *win_io_thrs;
 
+static SRWLock win_io_lock = SRWLOCK_INIT;
+
 static void
 win_io_handler(void *arg)
 {
@@ -45,9 +47,15 @@ win_io_handler(void *arg)
 			break;
 		}
 
+		// The following is done under the win_io_lock to prevent
+		// a use after free race against nni_win_io_fini.
+		AcquireSRWLockShared(&win_io_lock);
 		item = CONTAINING_RECORD(olpd, nni_win_io, olpd);
-		rv   = ok ? 0 : nni_win_error(GetLastError());
-		item->cb(item, rv, (size_t) cnt);
+		if (item->cb != NULL) {
+			rv = ok ? 0 : nni_win_error(GetLastError());
+			item->cb(item, rv, (size_t) cnt);
+		}
+		ReleaseSRWLockShared(&win_io_lock);
 	}
 }
 
@@ -61,26 +69,44 @@ nni_win_io_register(HANDLE h)
 }
 
 int
-nni_win_io_init(nni_win_io *io, nni_win_io_cb cb, void *ptr)
+nni_win_io_init(nni_win_io *io, HANDLE h, nni_win_io_cb cb, void *ptr)
 {
 	ZeroMemory(&io->olpd, sizeof(io->olpd));
 
-	io->cb          = cb;
-	io->ptr         = ptr;
-	io->aio         = NULL;
-	io->olpd.hEvent = CreateEvent(NULL, FALSE, FALSE, NULL);
-	if (io->olpd.hEvent == NULL) {
-		return (nni_win_error(GetLastError()));
-	}
+	io->f   = h;
+	io->cb  = cb;
+	io->ptr = ptr;
+	io->aio = NULL;
 	return (0);
 }
 
 void
 nni_win_io_fini(nni_win_io *io)
 {
-	if (io->olpd.hEvent != NULL) {
-		CloseHandle((HANDLE) io->olpd.hEvent);
+	DWORD num;
+	// Make absolutely sure there is no I/O running.
+	if (io->f != INVALID_HANDLE) {
+		CancelIoEx(io->f, &o->olpd);
+		(void) GetOverlappedResult(io->f, &io->olpd, &num, TRUE);
 	}
+
+	// Now  acquire the win_io_lock to make sure that no threads are
+	// running in this loop.  Note that there is a subtle race here, where
+	// in theory its possible for the callback thread to be just about to
+	// to start the callback (entering this lock).  This is very narrow,
+	// and it's unlikely to be problematic if it does, because we do not
+	// immedately destroy the data behind the overlapped I/O. Closing this
+	// race requires stopping all of the threads in the thread pool (which
+	// means waking them all up).  This may be worth doing in the future.
+	// One problem with this approach is that we have to have a way to wake
+	// *all* of them, not just one.
+	//
+	// One approach to doing this would be to close the handle of the
+	// completion port itself, and generate a new one.  This seems likely
+	// to be rather expensive?
+	AcquireSRWLockExclusive(&win_io_lock);
+	io->cb = NULL;
+	ReleaseSRWLockExclusive(&win_io_lock);
 }
 
 int

src/platform/windows/win_ipcconn.c

@@ -422,8 +422,8 @@ nni_win_ipc_init(
 	c->stream.s_get   = ipc_get;
 	c->stream.s_set   = ipc_set;
 
-	if (((rv = nni_win_io_init(&c->recv_io, ipc_recv_cb, c)) != 0) ||
-	    ((rv = nni_win_io_init(&c->send_io, ipc_send_cb, c)) != 0)) {
+	if (((rv = nni_win_io_init(&c->recv_io, p, ipc_recv_cb, c)) != 0) ||
+	    ((rv = nni_win_io_init(&c->send_io, p, ipc_send_cb, c)) != 0)) {
 		ipc_free(c);
 		return (rv);
 	}

src/platform/windows/win_ipclisten.c

@@ -17,7 +17,7 @@
 
 typedef struct {
 	nng_stream_listener sl;
-	char *              path;
+	char               *path;
 	bool                started;
 	bool                closed;
 	HANDLE              f;
@@ -33,7 +33,7 @@ typedef struct {
 static void
 ipc_accept_done(ipc_listener *l, int rv)
 {
-	nni_aio *   aio;
+	nni_aio    *aio;
 	HANDLE      f;
 	nng_stream *c;
 
@@ -138,7 +138,7 @@ static int
 ipc_listener_set_sec_desc(void *arg, const void *buf, size_t sz, nni_type t)
 {
 	ipc_listener *l = arg;
-	void *        desc;
+	void         *desc;
 	int           rv;
 
 	if ((rv = nni_copyin_ptr(&desc, buf, sz, t)) != 0) {
@@ -200,7 +200,7 @@ ipc_listener_listen(void *arg)
 	ipc_listener *l = arg;
 	int           rv;
 	HANDLE        f;
-	char *        path;
+	char         *path;
 
 	nni_mtx_lock(&l->mtx);
 	if (l->started) {
@@ -331,7 +331,8 @@ nni_ipc_listener_alloc(nng_stream_listener **lp, const nng_url *url)
 	if ((l = NNI_ALLOC_STRUCT(l)) == NULL) {
 		return (NNG_ENOMEM);
 	}
-	if ((rv = nni_win_io_init(&l->io, ipc_accept_cb, l)) != 0) {
+	if ((rv = nni_win_io_init(&l->io, INVALID_HANDLE, ipc_accept_cb, l)) !=
+	    0) {
 		NNI_FREE_STRUCT(l);
 		return (rv);
 	}

src/platform/windows/win_tcpconn.c

@@ -448,8 +448,8 @@ nni_win_tcp_init(nni_tcp_conn **connp, SOCKET s)
 	c->ops.s_get   = tcp_get;
 	c->ops.s_set   = tcp_set;
 
-	if (((rv = nni_win_io_init(&c->recv_io, tcp_recv_cb, c)) != 0) ||
-	    ((rv = nni_win_io_init(&c->send_io, tcp_send_cb, c)) != 0) ||
+	if (((rv = nni_win_io_init(&c->recv_io, s, tcp_recv_cb, c)) != 0) ||
+	    ((rv = nni_win_io_init(&c->send_io, s, tcp_send_cb, c)) != 0) ||
 	    ((rv = nni_win_io_register((HANDLE) s)) != 0)) {
 		tcp_free(c);
 		return (rv);

src/platform/windows/win_tcpdial.c

@@ -208,7 +208,7 @@ nni_tcp_dial(nni_tcp_dialer *d, const nni_sockaddr *sa, nni_aio *aio)
 
 	c->peername = ss;
 
-	if ((rv = nni_win_io_init(&c->conn_io, tcp_dial_cb, c)) != 0) {
+	if ((rv = nni_win_io_init(&c->conn_io, s, tcp_dial_cb, c)) != 0) {
 		nni_aio_finish_error(aio, rv);
 		return;
 	}

src/platform/windows/win_tcplisten.c

@@ -81,13 +81,13 @@ tcp_listener_funcs(nni_tcp_listener *l)
 static void
 tcp_accept_cb(nni_win_io *io, int rv, size_t cnt)
 {
-	nni_tcp_conn *    c = io->ptr;
+	nni_tcp_conn     *c = io->ptr;
 	nni_tcp_listener *l = c->listener;
-	nni_aio *         aio;
+	nni_aio          *aio;
 	int               len1;
 	int               len2;
-	SOCKADDR *        sa1;
-	SOCKADDR *        sa2;
+	SOCKADDR         *sa1;
+	SOCKADDR         *sa2;
 	BOOL              nd;
 	BOOL              ka;
 
@@ -270,7 +270,7 @@ static void
 tcp_accept_cancel(nni_aio *aio, void *arg, int rv)
 {
 	nni_tcp_listener *l = arg;
-	nni_tcp_conn *    c;
+	nni_tcp_conn     *c;
 
 	nni_mtx_lock(&l->mtx);
 	if ((c = nni_aio_get_prov_data(aio)) != NULL) {
@@ -322,7 +322,7 @@ nni_tcp_listener_accept(nni_tcp_listener *l, nni_aio *aio)
 	c->listener = l;
 	c->conn_aio = aio;
 	nni_aio_set_prov_data(aio, c);
-	if (((rv = nni_win_io_init(&c->conn_io, tcp_accept_cb, c)) != 0) ||
+	if (((rv = nni_win_io_init(&c->conn_io, s, tcp_accept_cb, c)) != 0) ||
 	    ((rv = nni_aio_schedule(aio, tcp_accept_cancel, l)) != 0)) {
 		nni_aio_set_prov_data(aio, NULL);
 		nni_mtx_unlock(&l->mtx);

src/platform/windows/win_udp.c

@@ -38,7 +38,7 @@ static void udp_recv_start(nni_plat_udp *);
 int
 nni_plat_udp_open(nni_plat_udp **udpp, nni_sockaddr *sa)
 {
-	nni_plat_udp *   u;
+	nni_plat_udp    *u;
 	SOCKADDR_STORAGE ss;
 	int              sslen;
 	DWORD            no;
@@ -67,7 +67,7 @@ nni_plat_udp_open(nni_plat_udp **udpp, nni_sockaddr *sa)
 	(void) setsockopt(
 	    u->s, IPPROTO_IPV6, IPV6_V6ONLY, (char *) &no, sizeof(no));
 
-	if (((rv = nni_win_io_init(&u->rxio, udp_recv_cb, u)) != 0) ||
+	if (((rv = nni_win_io_init(&u->rxio, u->s, udp_recv_cb, u)) != 0) ||
 	    ((rv = nni_win_io_register((HANDLE) u->s)) != 0)) {
 		nni_plat_udp_close(u);
 		return (rv);
@@ -115,10 +115,10 @@ nni_plat_udp_send(nni_plat_udp *u, nni_aio *aio)
 {
 	SOCKADDR_STORAGE to;
 	int              tolen;
-	nng_sockaddr *   sa;
+	nng_sockaddr    *sa;
 	unsigned         naiov;
-	nni_iov *        aiov;
-	WSABUF *         iov;
+	nni_iov         *aiov;
+	WSABUF          *iov;
 	int              rv;
 	DWORD            nsent;
 
@@ -191,7 +191,7 @@ udp_recv_cb(nni_win_io *io, int rv, size_t num)
 {
 	nni_plat_udp *u = io->ptr;
 	nni_sockaddr *sa;
-	nni_aio *     aio;
+	nni_aio      *aio;
 
 	nni_mtx_lock(&u->lk);
 	if ((aio = nni_list_first(&u->rxq)) == NULL) {
@@ -226,7 +226,7 @@ udp_recv_start(nni_plat_udp *u)
 	DWORD    flags;
 	nni_iov *aiov;
 	unsigned naiov;
-	WSABUF * iov;
+	WSABUF  *iov;
 	nni_aio *aio;
 
 	if ((u->s == INVALID_SOCKET) || (u->closed)) {

src/sp/transport/ipc/ipc.c

@@ -1,5 +1,5 @@
 //
-// Copyright 2021 Staysail Systems, Inc. <info@staysail.tech>
+// Copyright 2024 Staysail Systems, Inc. <info@staysail.tech>
 // Copyright 2018 Capitar IT Group BV <info@capitar.com>
 // Copyright 2019 Devolutions <info@devolutions.net>
 //
@@ -65,7 +65,7 @@ struct ipc_ep {
 	nni_aio             *time_aio;
 	nni_list             busy_pipes; // busy pipes -- ones passed to socket
 	nni_list             wait_pipes; // pipes waiting to match to socket
-	nni_list             neg_pipes;  // pipes busy negotiating
+	nni_list             nego_pipes; // pipes busy negotiating
 	nni_reap_node        reap;
 #ifdef NNG_ENABLE_STATS
 	nni_stat_item st_rcv_max;
@@ -76,7 +76,7 @@ static void ipc_pipe_send_start(ipc_pipe *p);
 static void ipc_pipe_recv_start(ipc_pipe *p);
 static void ipc_pipe_send_cb(void *);
 static void ipc_pipe_recv_cb(void *);
-static void ipc_pipe_neg_cb(void *);
+static void ipc_pipe_nego_cb(void *);
 static void ipc_pipe_fini(void *);
 static void ipc_ep_fini(void *);
 
@@ -165,9 +165,6 @@ static void
 ipc_pipe_reap(ipc_pipe *p)
 {
 	if (!nni_atomic_flag_test_and_set(&p->reaped)) {
-		if (p->conn != NULL) {
-			nng_stream_close(p->conn);
-		}
 		nni_reap(&ipc_pipe_reap_list, p);
 	}
 }
@@ -183,7 +180,7 @@ ipc_pipe_alloc(ipc_pipe **pipe_p)
 	nni_mtx_init(&p->mtx);
 	nni_aio_init(&p->tx_aio, ipc_pipe_send_cb, p);
 	nni_aio_init(&p->rx_aio, ipc_pipe_recv_cb, p);
-	nni_aio_init(&p->neg_aio, ipc_pipe_neg_cb, p);
+	nni_aio_init(&p->neg_aio, ipc_pipe_nego_cb, p);
 	nni_aio_list_init(&p->send_q);
 	nni_aio_list_init(&p->recv_q);
 	nni_atomic_flag_reset(&p->reaped);
@@ -210,7 +207,7 @@ ipc_ep_match(ipc_ep *ep)
 }
 
 static void
-ipc_pipe_neg_cb(void *arg)
+ipc_pipe_nego_cb(void *arg)
 {
 	ipc_pipe *p   = arg;
 	ipc_ep   *ep  = p->ep;
@@ -261,7 +258,7 @@ ipc_pipe_neg_cb(void *arg)
 
 	// We are ready now.  We put this in the wait list, and
 	// then try to run the matcher.
-	nni_list_remove(&ep->neg_pipes, p);
+	nni_list_remove(&ep->nego_pipes, p);
 	nni_list_append(&ep->wait_pipes, p);
 
 	ipc_ep_match(ep);
@@ -276,6 +273,7 @@ ipc_pipe_neg_cb(void *arg)
 	if (rv == NNG_ECLOSED) {
 		rv = NNG_ECONNSHUT;
 	}
+	nni_list_remove(&ep->nego_pipes, p);
 	nng_stream_close(p->conn);
 	// If we are waiting to negotiate on a client side, then a failure
 	// here has to be passed to the user app.
@@ -658,7 +656,7 @@ ipc_pipe_start(ipc_pipe *p, nng_stream *conn, ipc_ep *ep)
 	iov.iov_len     = 8;
 	iov.iov_buf     = &p->tx_head[0];
 	nni_aio_set_iov(&p->neg_aio, 1, &iov);
-	nni_list_append(&ep->neg_pipes, p);
+	nni_list_append(&ep->nego_pipes, p);
 
 	nni_aio_set_timeout(&p->neg_aio, 10000); // 10 sec timeout to negotiate
 	nng_stream_send(p->conn, &p->neg_aio);
@@ -679,7 +677,7 @@ ipc_ep_close(void *arg)
 	if (ep->listener != NULL) {
 		nng_stream_listener_close(ep->listener);
 	}
-	NNI_LIST_FOREACH (&ep->neg_pipes, p) {
+	NNI_LIST_FOREACH (&ep->nego_pipes, p) {
 		ipc_pipe_close(p);
 	}
 	NNI_LIST_FOREACH (&ep->wait_pipes, p) {
@@ -835,7 +833,7 @@ ipc_ep_init(ipc_ep **epp, nni_sock *sock)
 	nni_mtx_init(&ep->mtx);
 	NNI_LIST_INIT(&ep->busy_pipes, ipc_pipe, node);
 	NNI_LIST_INIT(&ep->wait_pipes, ipc_pipe, node);
-	NNI_LIST_INIT(&ep->neg_pipes, ipc_pipe, node);
+	NNI_LIST_INIT(&ep->nego_pipes, ipc_pipe, node);
 
 	ep->proto = nni_sock_proto_id(sock);
 

src/sp/transport/tcp/tcp.c

@@ -293,7 +293,9 @@
 		ep->useraio = NULL;
 		nni_aio_finish_error(uaio, rv);
 	}
+	nni_list_remove(&ep->negopipes, p);
 	nni_mtx_unlock(&ep->mtx);
+
 	tcptran_pipe_reap(p);
 }