Fix more races between connect and shutdown

There were more races that could happen while connecting to a
socket while closing or shutting down the same socket.  This
commit introduces a .closing flag to guard the socket from
being closed twice.

lib/isc/netmgr/netmgr-int.h

@@ -201,6 +201,7 @@ typedef struct isc__nm_uvreq {
 	uv_pipe_t ipc;	      /* used for sending socket
 			       * uv_handles to other threads */
 	union {
+		uv_handle_t handle;
 		uv_req_t req;
 		uv_getaddrinfo_t getaddrinfo;
 		uv_getnameinfo_t getnameinfo;
@@ -465,6 +466,7 @@ struct isc_nmsocket {
 	 * If active==false but closed==false, that means the socket
 	 * is closing.
 	 */
+	atomic_bool closing;
 	atomic_bool closed;
 	atomic_bool listening;
 	atomic_bool listen_error;

lib/isc/netmgr/netmgr.c

@@ -1081,6 +1081,7 @@ isc__nmsocket_init(isc_nmsocket_t *sock, isc_nm_t *mgr, isc_nmsocket_type type,
 	atomic_init(&sock->overlimit, false);
 	atomic_init(&sock->processing, false);
 	atomic_init(&sock->readpaused, false);
+	atomic_init(&sock->closing, false);
 
 	sock->magic = NMSOCK_MAGIC;
 }
@@ -1094,6 +1095,8 @@ isc__nmsocket_clearcb(isc_nmsocket_t *sock) {
 	sock->recv_cbarg = NULL;
 	sock->accept_cb = NULL;
 	sock->accept_cbarg = NULL;
+	sock->connect_cb = NULL;
+	sock->connect_cbarg = NULL;
 }
 
 void

lib/isc/netmgr/tcp.c

@@ -114,8 +114,10 @@ failed_accept_cb(isc_nmsocket_t *sock, isc_result_t eresult) {
 }
 
 static void
-failed_connect_cb(isc_nmsocket_t *sock, isc_result_t eresult) {
-	isc__nm_uvreq_t *req;
+failed_connect_cb(isc_nmsocket_t *sock, isc__nm_uvreq_t *req,
+		  isc_result_t eresult) {
+	REQUIRE(VALID_NMSOCK(sock));
+	REQUIRE(VALID_UVREQ(req));
 	REQUIRE(sock->tid == isc_nm_tid());
 
 	if (sock->timer_running) {
@@ -129,8 +131,6 @@ failed_connect_cb(isc_nmsocket_t *sock, isc_result_t eresult) {
 
 	atomic_store(&sock->connecting, false);
 
-	req = uv_handle_get_data((uv_handle_t *)&sock->timer);
-
 	isc__nmsocket_clearcb(sock);
 	if (req->cb.connect != NULL) {
 		req->cb.connect(NULL, eresult, req->cbarg);
@@ -149,7 +149,7 @@ connecttimeout_cb(uv_timer_t *handle) {
 
 	REQUIRE(sock->tid == isc_nm_tid());
 
-	failed_connect_cb(sock, ISC_R_TIMEDOUT);
+	failed_connect_cb(sock, req, ISC_R_TIMEDOUT);
 }
 
 static int
@@ -162,20 +162,17 @@ tcp_connect_direct(isc_nmsocket_t *sock, isc__nm_uvreq_t *req) {
 
 	worker = &sock->mgr->workers[sock->tid];
 
-	if (!sock->timer_initialized) {
-		uv_timer_init(&worker->loop, &sock->timer);
-		uv_handle_set_data((uv_handle_t *)&sock->timer, req);
-		sock->timer_initialized = true;
-	}
-
 	atomic_store(&sock->connecting, true);
 
 	r = uv_tcp_init(&worker->loop, &sock->uv_handle.tcp);
 	if (r != 0) {
 		isc__nm_incstats(sock->mgr, sock->statsindex[STATID_OPENFAIL]);
+		atomic_store(&sock->closing, true);
 		atomic_store(&sock->closed, true);
 		atomic_store(&sock->result, isc__nm_uverr2result(r));
 		atomic_store(&sock->connect_error, true);
+		failed_connect_cb(sock, req, isc__nm_uverr2result(r));
+		atomic_store(&sock->active, false);
 		return (r);
 	}
 
@@ -186,14 +183,21 @@ tcp_connect_direct(isc_nmsocket_t *sock, isc__nm_uvreq_t *req) {
 					 sock->statsindex[STATID_BINDFAIL]);
 			atomic_store(&sock->result, isc__nm_uverr2result(r));
 			atomic_store(&sock->connect_error, true);
-			failed_connect_cb(sock, isc__nm_uverr2result(-22));
-			tcp_close_direct(sock);
+			failed_connect_cb(sock, req, isc__nm_uverr2result(r));
+			atomic_store(&sock->active, false);
+			isc__nm_tcp_close(sock);
 			return (r);
 		}
 	}
 
-	uv_handle_set_data(&sock->uv_handle.handle, sock);
+	if (!sock->timer_initialized) {
+		uv_timer_init(&worker->loop, &sock->timer);
+		uv_handle_set_data((uv_handle_t *)&sock->timer, req);
+		sock->timer_initialized = true;
+	}
 
+	uv_handle_set_data(&sock->uv_handle.handle, sock);
+	uv_handle_set_data(&req->uv_req.handle, req);
 	r = uv_tcp_connect(&req->uv_req.connect, &sock->uv_handle.tcp,
 			   &req->peer.type.sa, tcp_connect_cb);
 	if (r != 0) {
@@ -201,8 +205,9 @@ tcp_connect_direct(isc_nmsocket_t *sock, isc__nm_uvreq_t *req) {
 				 sock->statsindex[STATID_CONNECTFAIL]);
 		atomic_store(&sock->result, isc__nm_uverr2result(r));
 		atomic_store(&sock->connect_error, true);
-		failed_connect_cb(sock, isc__nm_uverr2result(-22));
-		tcp_close_direct(sock);
+		failed_connect_cb(sock, req, isc__nm_uverr2result(r));
+		atomic_store(&sock->active, false);
+		isc__nm_tcp_close(sock);
 		return (r);
 	}
 	isc__nm_incstats(sock->mgr, sock->statsindex[STATID_CONNECT]);
@@ -245,18 +250,14 @@ isc__nm_async_tcpconnect(isc__networker_t *worker, isc__netievent_t *ev0) {
 static void
 tcp_connect_cb(uv_connect_t *uvreq, int status) {
 	isc_result_t result;
-	isc__nm_uvreq_t *req = (isc__nm_uvreq_t *)uvreq->data;
+	isc__nm_uvreq_t *req = uv_handle_get_data((uv_handle_t *)uvreq);
 	isc_nmsocket_t *sock = uv_handle_get_data((uv_handle_t *)uvreq->handle);
 	struct sockaddr_storage ss;
 	isc_nmhandle_t *handle = NULL;
 
+	REQUIRE(VALID_NMSOCK(sock));
 	REQUIRE(sock->tid == isc_nm_tid());
 
-	if (status != 0) {
-		failed_connect_cb(sock, isc__nm_uverr2result(status));
-		return;
-	}
-
 	if (sock->timer_running) {
 		uv_timer_stop(&sock->timer);
 		sock->timer_running = false;
@@ -266,11 +267,14 @@ tcp_connect_cb(uv_connect_t *uvreq, int status) {
 		return;
 	}
 
-	atomic_store(&sock->connecting, false);
-
 	REQUIRE(VALID_UVREQ(req));
 
-	sock = uv_handle_get_data((uv_handle_t *)uvreq->handle);
+	if (status != 0) {
+		failed_connect_cb(sock, req, isc__nm_uverr2result(status));
+		return;
+	}
+
+	atomic_store(&sock->connecting, false);
 
 	isc__nm_incstats(sock->mgr, sock->statsindex[STATID_CONNECT]);
 	uv_tcp_getpeername(&sock->uv_handle.tcp, (struct sockaddr *)&ss,
@@ -1282,11 +1286,9 @@ tcp_close_cb(uv_handle_t *uvhandle) {
 
 static void
 timer_close_cb(uv_handle_t *uvhandle) {
-	isc_nmsocket_t *sock = uv_handle_get_data(uvhandle);
+	uv_handle_t *handle = uv_handle_get_data(uvhandle);
 
-	REQUIRE(VALID_NMSOCK(sock));
-
-	uv_close(&sock->uv_handle.handle, tcp_close_cb);
+	uv_close(handle, tcp_close_cb);
 }
 
 static void
@@ -1308,7 +1310,13 @@ tcp_close_direct(isc_nmsocket_t *sock) {
 
 	if (sock->timer_initialized) {
 		sock->timer_initialized = false;
-		uv_handle_set_data((uv_handle_t *)&sock->timer, sock);
+		/*
+		 * The read and timer is stopped and the socket will be
+		 * scheduled to be closed, so we can override the data that the
+		 * timer handle holds.
+		 */
+		uv_handle_set_data((uv_handle_t *)&sock->timer,
+				   &sock->uv_handle.handle);
 		uv_close((uv_handle_t *)&sock->timer, timer_close_cb);
 	} else {
 		uv_close(&sock->uv_handle.handle, tcp_close_cb);
@@ -1321,6 +1329,11 @@ isc__nm_tcp_close(isc_nmsocket_t *sock) {
 	REQUIRE(sock->type == isc_nm_tcpsocket);
 	REQUIRE(!isc__nmsocket_active(sock));
 
+	if (!atomic_compare_exchange_strong(&sock->closing, &(bool){ false },
+					    true)) {
+		return;
+	}
+
 	if (sock->tid == isc_nm_tid()) {
 		tcp_close_direct(sock);
 	} else {
@@ -1339,10 +1352,15 @@ isc__nm_tcp_close(isc_nmsocket_t *sock) {
 void
 isc__nm_async_tcpclose(isc__networker_t *worker, isc__netievent_t *ev0) {
 	isc__netievent_tcpclose_t *ievent = (isc__netievent_tcpclose_t *)ev0;
+	isc_nmsocket_t *sock = ievent->sock;
 
-	REQUIRE(worker->id == ievent->sock->tid);
+	REQUIRE(VALID_NMSOCK(sock));
 
-	tcp_close_direct(ievent->sock);
+	UNUSED(worker);
+
+	REQUIRE(sock->tid == isc_nm_tid());
+
+	tcp_close_direct(sock);
 }
 
 void
@@ -1355,7 +1373,11 @@ isc__nm_tcp_shutdown(isc_nmsocket_t *sock) {
 	}
 
 	if (atomic_load(&sock->connecting)) {
-		failed_connect_cb(sock, ISC_R_CANCELED);
+		if (sock->timer_initialized) {
+			isc__nm_uvreq_t *req =
+				uv_handle_get_data((uv_handle_t *)&sock->timer);
+			failed_connect_cb(sock, req, ISC_R_CANCELED);
+		}
 		return;
 	}
 

lib/isc/netmgr/udp.c

@@ -217,8 +217,9 @@ isc__nm_async_udplisten(isc__networker_t *worker, isc__netievent_t *ev0) {
 static void
 udp_stop_cb(uv_handle_t *handle) {
 	isc_nmsocket_t *sock = uv_handle_get_data(handle);
-	atomic_store(&sock->closed, true);
 
+	isc__nm_incstats(sock->mgr, sock->statsindex[STATID_CLOSE]);
+	atomic_store(&sock->closed, true);
 	isc__nmsocket_detach((isc_nmsocket_t **)&sock->uv_handle.udp.data);
 }
 
@@ -229,9 +230,12 @@ stop_udp_child(isc_nmsocket_t *sock) {
 
 	uv_udp_recv_stop(&sock->uv_handle.udp);
 
-	uv_close((uv_handle_t *)&sock->uv_handle.udp, udp_stop_cb);
+	if (!atomic_compare_exchange_strong(&sock->closing, &(bool){ false },
+					    true)) {
+		return;
+	}
 
-	isc__nm_incstats(sock->mgr, sock->statsindex[STATID_CLOSE]);
+	uv_close(&sock->uv_handle.handle, udp_stop_cb);
 
 	LOCK(&sock->parent->lock);
 	atomic_fetch_sub(&sock->parent->rchildren, 1);
@@ -630,19 +634,24 @@ udp_connect_direct(isc_nmsocket_t *sock, isc__nm_uvreq_t *req) {
 	r = uv_udp_init(&worker->loop, &sock->uv_handle.udp);
 	if (r != 0) {
 		isc__nm_incstats(sock->mgr, sock->statsindex[STATID_OPENFAIL]);
-		/* Socket was never opened; no need for udp_close_direct() */
+		/* Socket was never opened; no need for isc__nm_udp_close() */
+		atomic_store(&sock->closing, true);
 		atomic_store(&sock->closed, true);
 		atomic_store(&sock->result, isc__nm_uverr2result(r));
 		atomic_store(&sock->connect_error, true);
+		failed_connect_cb(sock, req, isc__nm_uverr2result(r));
+		atomic_store(&sock->active, false);
 		return (r);
 	}
 
 	r = uv_udp_open(&sock->uv_handle.udp, sock->fd);
 	if (r != 0) {
 		isc__nm_incstats(sock->mgr, sock->statsindex[STATID_OPENFAIL]);
-		atomic_store(&sock->closed, true);
 		atomic_store(&sock->connect_error, true);
 		atomic_store(&sock->result, isc__nm_uverr2result(r));
+		failed_connect_cb(sock, req, isc__nm_uverr2result(r));
+		atomic_store(&sock->active, false);
+		isc__nm_udp_close(sock);
 		return (r);
 	}
 	isc__nm_incstats(sock->mgr, sock->statsindex[STATID_OPEN]);
@@ -657,7 +666,9 @@ udp_connect_direct(isc_nmsocket_t *sock, isc__nm_uvreq_t *req) {
 		isc__nm_incstats(sock->mgr, sock->statsindex[STATID_BINDFAIL]);
 		atomic_store(&sock->connect_error, true);
 		atomic_store(&sock->result, isc__nm_uverr2result(r));
-		udp_close_direct(sock);
+		failed_connect_cb(sock, req, isc__nm_uverr2result(r));
+		atomic_store(&sock->active, false);
+		isc__nm_udp_close(sock);
 		return (r);
 	}
 
@@ -669,7 +680,9 @@ udp_connect_direct(isc_nmsocket_t *sock, isc__nm_uvreq_t *req) {
 				 sock->statsindex[STATID_CONNECTFAIL]);
 		atomic_store(&sock->connect_error, true);
 		atomic_store(&sock->result, isc__nm_uverr2result(r));
-		udp_close_direct(sock);
+		failed_connect_cb(sock, req, isc__nm_uverr2result(r));
+		atomic_store(&sock->active, false);
+		isc__nm_udp_close(sock);
 		return (r);
 	}
 	isc__nm_incstats(sock->mgr, sock->statsindex[STATID_CONNECT]);
@@ -714,7 +727,6 @@ isc__nm_async_udpconnect(isc__networker_t *worker, isc__netievent_t *ev0) {
 
 	r = udp_connect_direct(sock, req);
 	if (r != 0) {
-		failed_connect_cb(sock, req, isc__nm_uverr2result(r));
 		LOCK(&sock->lock);
 		SIGNAL(&sock->cond);
 		UNLOCK(&sock->lock);
@@ -977,11 +989,9 @@ udp_close_cb(uv_handle_t *uvhandle) {
 
 static void
 timer_close_cb(uv_handle_t *uvhandle) {
-	isc_nmsocket_t *sock = uv_handle_get_data(uvhandle);
+	uv_handle_t *handle = uv_handle_get_data(uvhandle);
 
-	REQUIRE(VALID_NMSOCK(sock));
-
-	uv_close(&sock->uv_handle.handle, udp_close_cb);
+	uv_close(handle, udp_close_cb);
 }
 
 static void
@@ -995,7 +1005,13 @@ udp_close_direct(isc_nmsocket_t *sock) {
 
 	if (sock->timer_initialized) {
 		sock->timer_initialized = false;
-		uv_handle_set_data((uv_handle_t *)&sock->timer, sock);
+		/*
+		 * The read and timer is stopped and the socket will be
+		 * scheduled to be closed, so we can override the data that the
+		 * timer handle holds.
+		 */
+		uv_handle_set_data((uv_handle_t *)&sock->timer,
+				   &sock->uv_handle.handle);
 		uv_close((uv_handle_t *)&sock->timer, timer_close_cb);
 	} else {
 		uv_close(&sock->uv_handle.handle, udp_close_cb);
@@ -1016,9 +1032,13 @@ void
 isc__nm_udp_close(isc_nmsocket_t *sock) {
 	REQUIRE(VALID_NMSOCK(sock));
 	REQUIRE(sock->type == isc_nm_udpsocket);
-
 	REQUIRE(!isc__nmsocket_active(sock));
 
+	if (!atomic_compare_exchange_strong(&sock->closing, &(bool){ false },
+					    true)) {
+		return;
+	}
+
 	if (sock->tid == isc_nm_tid()) {
 		udp_close_direct(sock);
 	} else {
@@ -1048,8 +1068,6 @@ failed_connect_cb(isc_nmsocket_t *sock, isc__nm_uvreq_t *req,
 
 	INSIST(req != NULL);
 
-	req = uv_handle_get_data((uv_handle_t *)&sock->timer);
-
 	isc__nmsocket_clearcb(sock);
 
 	if (req->cb.connect != NULL) {
@@ -1071,7 +1089,11 @@ isc__nm_udp_shutdown(isc_nmsocket_t *sock) {
 	}
 
 	if (atomic_load(&sock->connecting)) {
-		failed_connect_cb(sock, NULL, ISC_R_CANCELED);
+		if (sock->timer_initialized) {
+			isc__nm_uvreq_t *req =
+				uv_handle_get_data((uv_handle_t *)&sock->timer);
+			failed_connect_cb(sock, req, ISC_R_CANCELED);
+		}
 		return;
 	}