Synchronous send on xfrin dispatch

lib/dns/dispatch.c

@@ -113,10 +113,11 @@ struct dns_dispatch {
 	isc_socktype_t socktype;
 	isc_refcount_t references;
 	isc_mem_t *mctx;
-	dns_dispatchmgr_t *mgr; /*%< dispatch manager */
-	isc_nmhandle_t *handle; /*%< netmgr handle for TCP connection */
-	isc_sockaddr_t local;	/*%< local address */
-	isc_sockaddr_t peer;	/*%< peer address (TCP) */
+	dns_dispatchmgr_t *mgr;	     /*%< dispatch manager */
+	isc_nmhandle_t *handle;	     /*%< netmgr handle for TCP connection */
+	isc_sockaddr_t local;	     /*%< local address */
+	isc_sockaddr_t peer;	     /*%< peer address (TCP) */
+	dns_dispatch_sharetag_t tag; /*%< tag for sharing partitioning */
 
 	dns_dispatchopt_t options;
 	dns_dispatchstate_t state;
@@ -1134,6 +1135,7 @@ dispatch_allocate(dns_dispatchmgr_t *mgr, isc_socktype_t type, uint32_t tid,
 struct dispatch_key {
 	const isc_sockaddr_t *local;
 	const isc_sockaddr_t *peer;
+	const dns_dispatch_sharetag_t tag;
 };
 
 static uint32_t
@@ -1142,6 +1144,7 @@ dispatch_hash(struct dispatch_key *key) {
 	if (key->local) {
 		hashval ^= isc_sockaddr_hash(key->local, true);
 	}
+	hashval ^= isc_hash32(&key->tag, sizeof(key->tag), true);
 
 	return (hashval);
 }
@@ -1161,14 +1164,16 @@ dispatch_match(struct cds_lfht_node *node, const void *key0) {
 		peer = disp->peer;
 	}
 
-	return (isc_sockaddr_equal(&peer, key->peer) &&
+	return (disp->tag == key->tag && isc_sockaddr_equal(&peer, key->peer) &&
 		(key->local == NULL || isc_sockaddr_equal(&local, key->local)));
 }
 
 isc_result_t
 dns_dispatch_createtcp(dns_dispatchmgr_t *mgr, const isc_sockaddr_t *localaddr,
 		       const isc_sockaddr_t *destaddr,
-		       dns_dispatchopt_t options, dns_dispatch_t **dispp) {
+		       dns_dispatchopt_t options,
+		       const dns_dispatch_sharetag_t tag,
+		       dns_dispatch_t **dispp) {
 	dns_dispatch_t *disp = NULL;
 	uint32_t tid = isc_tid();
 
@@ -1179,6 +1184,7 @@ dns_dispatch_createtcp(dns_dispatchmgr_t *mgr, const isc_sockaddr_t *localaddr,
 
 	disp->options = options;
 	disp->peer = *destaddr;
+	disp->tag = tag;
 
 	if (localaddr != NULL) {
 		disp->local = *localaddr;
@@ -1195,6 +1201,7 @@ dns_dispatch_createtcp(dns_dispatchmgr_t *mgr, const isc_sockaddr_t *localaddr,
 	struct dispatch_key key = {
 		.local = &disp->local,
 		.peer = &disp->peer,
+		.tag = tag,
 	};
 
 	if ((disp->options & DNS_DISPATCHOPT_UNSHARED) == 0) {
@@ -1222,7 +1229,8 @@ dns_dispatch_createtcp(dns_dispatchmgr_t *mgr, const isc_sockaddr_t *localaddr,
 
 isc_result_t
 dns_dispatch_gettcp(dns_dispatchmgr_t *mgr, const isc_sockaddr_t *destaddr,
-		    const isc_sockaddr_t *localaddr, dns_dispatch_t **dispp) {
+		    const isc_sockaddr_t *localaddr,
+		    const dns_dispatch_sharetag_t tag, dns_dispatch_t **dispp) {
 	dns_dispatch_t *disp_connected = NULL;
 	dns_dispatch_t *disp_fallback = NULL;
 	isc_result_t result = ISC_R_NOTFOUND;
@@ -1235,6 +1243,7 @@ dns_dispatch_gettcp(dns_dispatchmgr_t *mgr, const isc_sockaddr_t *destaddr,
 	struct dispatch_key key = {
 		.local = localaddr,
 		.peer = destaddr,
+		.tag = tag,
 	};
 
 	rcu_read_lock();
@@ -2020,9 +2029,12 @@ tcp_dispatch_connect(dns_dispatch_t *disp, dns_dispentry_t *resp) {
 			tcp_startrecv(disp, resp);
 		}
 
-		/* Already connected; call the connected cb asynchronously */
+		/* We must call resp_connected synchronously to avoid a race
+		 * condition with the socket being marked as ready to read
+		 * on TCP disconnect
+		 */
 		dns_dispentry_ref(resp); /* DISPENTRY005 */
-		isc_async_run(resp->loop, resp_connected, resp);
+		resp_connected(resp);
 		break;
 
 	default:

lib/dns/include/dns/dispatch.h

@@ -77,6 +77,12 @@ typedef enum dns_dispatchopt {
 	DNS_DISPATCHOPT_UNSHARED = 1 << 1, /* Don't share this connection */
 } dns_dispatchopt_t;
 
+typedef enum dns_dispatch_sharetag {
+	DNS_DISPATCH_SHARETAG_UNSET = 0,
+	DNS_DISPATCH_SHARETAG_REQUEST = 1,
+	DNS_DISPATCH_SHARETAG_XFRIN = 2
+} dns_dispatch_sharetag_t;
+
 isc_result_t
 dns_dispatchmgr_create(isc_mem_t *mctx, isc_loopmgr_t *loopmgr, isc_nm_t *nm,
 		       dns_dispatchmgr_t **mgrp);
@@ -184,11 +190,16 @@ dns_dispatch_createudp(dns_dispatchmgr_t *mgr, const isc_sockaddr_t *localaddr,
 
 isc_result_t
 dns_dispatch_createtcp(dns_dispatchmgr_t *mgr, const isc_sockaddr_t *localaddr,
-		       const isc_sockaddr_t *destaddr,
-		       dns_dispatchopt_t options, dns_dispatch_t **dispp);
+		       const isc_sockaddr_t	    *destaddr,
+		       dns_dispatchopt_t	     options,
+		       const dns_dispatch_sharetag_t tag,
+		       dns_dispatch_t		   **dispp);
 /*%<
  * Create a new TCP dns_dispatch.
  *
+ * If the 'DNS_DISPATCHOPT_UNSHARED' option bit is not set, 'tag' can be used
+ * to partition the sharing of the dispatch.
+ *
  * Requires:
  *
  *\li	mgr is a valid dispatch manager.
@@ -255,7 +266,8 @@ dns_dispatch_resume(dns_dispentry_t *resp, uint16_t timeout);
 
 isc_result_t
 dns_dispatch_gettcp(dns_dispatchmgr_t *mgr, const isc_sockaddr_t *destaddr,
-		    const isc_sockaddr_t *localaddr, dns_dispatch_t **dispp);
+		    const isc_sockaddr_t	 *localaddr,
+		    const dns_dispatch_sharetag_t tag, dns_dispatch_t **dispp);
 /*
  * Attempt to connect to a existing TCP connection.
  */

lib/dns/request.c

@@ -342,8 +342,9 @@ tcp_dispatch(bool newtcp, dns_requestmgr_t *requestmgr,
 	isc_result_t result;
 
 	if (!newtcp) {
-		result = dns_dispatch_gettcp(requestmgr->dispatchmgr, destaddr,
-					     srcaddr, dispatchp);
+		result = dns_dispatch_gettcp(
+			requestmgr->dispatchmgr, destaddr, srcaddr,
+			DNS_DISPATCH_SHARETAG_REQUEST, dispatchp);
 		if (result == ISC_R_SUCCESS) {
 			char peer[ISC_SOCKADDR_FORMATSIZE];
 
@@ -354,8 +355,9 @@ tcp_dispatch(bool newtcp, dns_requestmgr_t *requestmgr,
 		}
 	}
 
-	result = dns_dispatch_createtcp(requestmgr->dispatchmgr, srcaddr,
-					destaddr, 0, dispatchp);
+	result = dns_dispatch_createtcp(
+		requestmgr->dispatchmgr, srcaddr, destaddr, 0,
+		DNS_DISPATCH_SHARETAG_REQUEST, dispatchp);
 	return (result);
 }
 

lib/dns/resolver.c

@@ -2045,7 +2045,8 @@ fctx_query(fetchctx_t *fctx, dns_adbaddrinfo_t *addrinfo,
 
 		result = dns_dispatch_createtcp(
 			res->view->dispatchmgr, &addr, &sockaddr,
-			DNS_DISPATCHOPT_UNSHARED, &query->dispatch);
+			DNS_DISPATCHOPT_UNSHARED, DNS_DISPATCH_SHARETAG_UNSET,
+			&query->dispatch);
 		if (result != ISC_R_SUCCESS) {
 			goto cleanup_query;
 		}

lib/dns/xfrin.c

@@ -1224,6 +1224,29 @@ xfrin_create(isc_mem_t *mctx, dns_zone_t *zone, dns_db_t *db, isc_loop_t *loop,
 	*xfrp = xfr;
 }
 
+static isc_result_t
+tcp_dispatch(dns_xfrin_t *xfr, dns_dispatchmgr_t *dispmgr,
+	     const isc_sockaddr_t *srcaddr, const isc_sockaddr_t *destaddr,
+	     unsigned int options, const dns_dispatch_sharetag_t tag,
+	     dns_dispatch_t **dispatchp) {
+	isc_result_t result;
+
+	result = dns_dispatch_gettcp(dispmgr, destaddr, srcaddr, tag,
+				     dispatchp);
+	if (result == ISC_R_SUCCESS) {
+		char peer[ISC_SOCKADDR_FORMATSIZE];
+
+		isc_sockaddr_format(destaddr, peer, sizeof(peer));
+		xfrin_log(xfr, ISC_LOG_DEBUG(1),
+			  "attached to TCP connection to %s", peer);
+		return (result);
+	}
+
+	result = dns_dispatch_createtcp(dispmgr, srcaddr, destaddr, options,
+					tag, dispatchp);
+	return (result);
+}
+
 static isc_result_t
 xfrin_start(dns_xfrin_t *xfr) {
 	isc_result_t result = ISC_R_FAILURE;
@@ -1239,9 +1262,10 @@ xfrin_start(dns_xfrin_t *xfr) {
 		result = ISC_R_SHUTTINGDOWN;
 		goto failure;
 	} else {
-		result = dns_dispatch_createtcp(
-			dispmgr, &xfr->sourceaddr, &xfr->primaryaddr,
-			DNS_DISPATCHOPT_UNSHARED, &xfr->disp);
+		result = tcp_dispatch(xfr, dispmgr, &xfr->sourceaddr,
+				      &xfr->primaryaddr,
+				      0, DNS_DISPATCH_SHARETAG_XFRIN,
+				      &xfr->disp);
 		dns_dispatchmgr_detach(&dispmgr);
 		if (result != ISC_R_SUCCESS) {
 			goto failure;

tests/dns/dispatch_test.c

@@ -507,8 +507,9 @@ connected_gettcp(isc_result_t eresult ISC_ATTR_UNUSED,
 		.dispatchmgr = dns_dispatchmgr_ref(test1->dispatchmgr),
 	};
 
-	result = dns_dispatch_gettcp(test2->dispatchmgr, &tcp_server_addr,
-				     &tcp_connect_addr, &test2->dispatch);
+	result = dns_dispatch_gettcp(
+		test2->dispatchmgr, &tcp_server_addr, &tcp_connect_addr,
+		DNS_DISPATCH_SHARETAG_UNSET, &test2->dispatch);
 	assert_int_equal(result, ISC_R_SUCCESS);
 
 	assert_ptr_equal(test1->dispatch, test2->dispatch);
@@ -536,13 +537,15 @@ connected_newtcp(isc_result_t eresult ISC_ATTR_UNUSED,
 	*test4 = (test_dispatch_t){
 		.dispatchmgr = dns_dispatchmgr_ref(test3->dispatchmgr),
 	};
-	result = dns_dispatch_gettcp(test4->dispatchmgr, &tcp_server_addr,
-				     &tcp_connect_addr, &test4->dispatch);
+	result = dns_dispatch_gettcp(
+		test4->dispatchmgr, &tcp_server_addr, &tcp_connect_addr,
+		DNS_DISPATCH_SHARETAG_UNSET, &test4->dispatch);
 	assert_int_equal(result, ISC_R_NOTFOUND);
 
 	result = dns_dispatch_createtcp(
 		test4->dispatchmgr, &tcp_connect_addr, &tcp_server_addr,
-		DNS_DISPATCHOPT_UNSHARED, &test4->dispatch);
+		DNS_DISPATCHOPT_UNSHARED, DNS_DISPATCH_SHARETAG_UNSET,
+		&test4->dispatch);
 	assert_int_equal(result, ISC_R_SUCCESS);
 
 	assert_ptr_not_equal(test3->dispatch, test4->dispatch);
@@ -592,8 +595,9 @@ ISC_LOOP_TEST_IMPL(dispatch_timeout_tcp_connect) {
 					&test->dispatchmgr);
 	assert_int_equal(result, ISC_R_SUCCESS);
 
-	result = dns_dispatch_createtcp(test->dispatchmgr, &tcp_connect_addr,
-					&tcp_server_addr, 0, &test->dispatch);
+	result = dns_dispatch_createtcp(
+		test->dispatchmgr, &tcp_connect_addr, &tcp_server_addr, 0,
+		DNS_DISPATCH_SHARETAG_UNSET, &test->dispatch);
 	assert_int_equal(result, ISC_R_SUCCESS);
 
 	result = dns_dispatch_add(test->dispatch, isc_loop_main(loopmgr), 0,
@@ -637,8 +641,9 @@ ISC_LOOP_TEST_IMPL(dispatch_timeout_tcp_response) {
 					&test->dispatchmgr);
 	assert_int_equal(result, ISC_R_SUCCESS);
 
-	result = dns_dispatch_createtcp(test->dispatchmgr, &tcp_connect_addr,
-					&tcp_server_addr, 0, &test->dispatch);
+	result = dns_dispatch_createtcp(
+		test->dispatchmgr, &tcp_connect_addr, &tcp_server_addr, 0,
+		DNS_DISPATCH_SHARETAG_UNSET, &test->dispatch);
 	assert_int_equal(result, ISC_R_SUCCESS);
 
 	result = dns_dispatch_add(test->dispatch, isc_loop_main(loopmgr), 0,
@@ -672,8 +677,9 @@ ISC_LOOP_TEST_IMPL(dispatch_tcp_response) {
 					&test->dispatchmgr);
 	assert_int_equal(result, ISC_R_SUCCESS);
 
-	result = dns_dispatch_createtcp(test->dispatchmgr, &tcp_connect_addr,
-					&tcp_server_addr, 0, &test->dispatch);
+	result = dns_dispatch_createtcp(
+		test->dispatchmgr, &tcp_connect_addr, &tcp_server_addr, 0,
+		DNS_DISPATCH_SHARETAG_UNSET, &test->dispatch);
 	assert_int_equal(result, ISC_R_SUCCESS);
 
 	result = dns_dispatch_add(
@@ -710,8 +716,9 @@ ISC_LOOP_TEST_IMPL(dispatch_tls_response) {
 					&test->dispatchmgr);
 	assert_int_equal(result, ISC_R_SUCCESS);
 
-	result = dns_dispatch_createtcp(test->dispatchmgr, &tls_connect_addr,
-					&tls_server_addr, 0, &test->dispatch);
+	result = dns_dispatch_createtcp(
+		test->dispatchmgr, &tls_connect_addr, &tls_server_addr, 0,
+		DNS_DISPATCH_SHARETAG_UNSET, &test->dispatch);
 	assert_int_equal(result, ISC_R_SUCCESS);
 
 	result = dns_dispatch_add(test->dispatch, isc_loop_main(loopmgr), 0,
@@ -815,8 +822,9 @@ ISC_LOOP_TEST_IMPL(dispatch_gettcp) {
 	assert_int_equal(result, ISC_R_SUCCESS);
 
 	/* Client */
-	result = dns_dispatch_createtcp(test->dispatchmgr, &tcp_connect_addr,
-					&tcp_server_addr, 0, &test->dispatch);
+	result = dns_dispatch_createtcp(
+		test->dispatchmgr, &tcp_connect_addr, &tcp_server_addr, 0,
+		DNS_DISPATCH_SHARETAG_UNSET, &test->dispatch);
 	assert_int_equal(result, ISC_R_SUCCESS);
 
 	result = dns_dispatch_add(
@@ -849,7 +857,8 @@ ISC_LOOP_TEST_IMPL(dispatch_newtcp) {
 
 	result = dns_dispatch_createtcp(
 		test->dispatchmgr, &tcp_connect_addr, &tcp_server_addr,
-		DNS_DISPATCHOPT_UNSHARED, &test->dispatch);
+		DNS_DISPATCHOPT_UNSHARED, DNS_DISPATCH_SHARETAG_UNSET,
+		&test->dispatch);
 	assert_int_equal(result, ISC_R_SUCCESS);
 
 	result = dns_dispatch_add(