Reduce the number of client tasks and bind them to netmgr queues

Since a client object is bound to a netmgr handle, each client
will always be processed by the same netmgr worker, so we can
simplify the code by binding client->task to the same thread as
the client. Since ns__client_request() now runs in the same event
loop as client->task events, is no longer necessary to pause the
task manager before launching them.

Also removed some functions in isc_task that were not used.

lib/isc/include/isc/task.h

@@ -556,41 +556,6 @@ isc_task_endexclusive(isc_task_t *task);
  *		exclusive access by calling isc_task_spl().
  */
 
-void
-isc_task_pause(isc_task_t *task0);
-void
-isc_task_unpause(isc_task_t *task0);
-/*%<
- * Pause/unpause this task. Pausing a task removes it from the ready
- * queue if it is present there; this ensures that the task will not
- * run again until unpaused. This is necessary when the libuv network
- * thread executes a function which schedules task manager events; this
- * prevents the task manager from executing the next event in a task
- * before the network thread has finished.
- *
- * Requires:
- *\li	'task' is a valid task, and is not already paused or shutting down.
- */
-
-void
-isc_task_getcurrenttime(isc_task_t *task, isc_stdtime_t *t);
-void
-isc_task_getcurrenttimex(isc_task_t *task, isc_time_t *t);
-/*%<
- * Provide the most recent timestamp on the task.  The timestamp is considered
- * as the "current time".
- *
- * isc_task_getcurrentime() returns the time in one-second granularity;
- * isc_task_getcurrentimex() returns it in nanosecond granularity.
- *
- * Requires:
- *\li	'task' is a valid task.
- *\li	't' is a valid non NULL pointer.
- *
- * Ensures:
- *\li	'*t' has the "current time".
- */
-
 bool
 isc_task_exiting(isc_task_t *t);
 /*%<

lib/isc/task.c

@@ -78,15 +78,16 @@
 typedef enum {
 	task_state_idle,    /* not doing anything, events queue empty */
 	task_state_ready,   /* waiting in worker's queue */
-	task_state_paused,  /* not running, paused */
-	task_state_pausing, /* running, waiting to be paused */
 	task_state_running, /* actively processing events */
 	task_state_done	    /* shutting down, no events or references */
 } task_state_t;
 
 #if defined(HAVE_LIBXML2) || defined(HAVE_JSON_C)
 static const char *statenames[] = {
-	"idle", "ready", "paused", "pausing", "running", "done",
+	"idle",
+	"ready",
+	"running",
+	"done",
 };
 #endif /* if defined(HAVE_LIBXML2) || defined(HAVE_JSON_C) */
 
@@ -101,7 +102,6 @@ struct isc_task {
 	int threadid;
 	/* Locked by task lock. */
 	task_state_t state;
-	int pause_cnt;
 	isc_refcount_t references;
 	isc_refcount_t running;
 	isc_eventlist_t events;
@@ -237,7 +237,6 @@ isc_task_create_bound(isc_taskmgr_t *manager, unsigned int quantum,
 
 	isc_mutex_init(&task->lock);
 	task->state = task_state_idle;
-	task->pause_cnt = 0;
 
 	isc_refcount_init(&task->references, 1);
 	isc_refcount_init(&task->running, 0);
@@ -312,8 +311,6 @@ task_shutdown(isc_task_t *task) {
 			was_idle = true;
 		}
 		INSIST(task->state == task_state_ready ||
-		       task->state == task_state_paused ||
-		       task->state == task_state_pausing ||
 		       task->state == task_state_running);
 
 		/*
@@ -437,9 +434,7 @@ task_send(isc_task_t *task, isc_event_t **eventp, int c) {
 		task->state = task_state_ready;
 	}
 	INSIST(task->state == task_state_ready ||
-	       task->state == task_state_running ||
+	       task->state == task_state_running);
-	       task->state == task_state_paused ||
-	       task->state == task_state_pausing);
 	ENQUEUE(task->events, event, ev_link);
 	task->nevents++;
 
@@ -779,26 +774,6 @@ isc_task_gettag(isc_task_t *task) {
 	return (task->tag);
 }
 
-void
-isc_task_getcurrenttime(isc_task_t *task, isc_stdtime_t *t) {
-	REQUIRE(VALID_TASK(task));
-	REQUIRE(t != NULL);
-
-	LOCK(&task->lock);
-	*t = task->now;
-	UNLOCK(&task->lock);
-}
-
-void
-isc_task_getcurrenttimex(isc_task_t *task, isc_time_t *t) {
-	REQUIRE(VALID_TASK(task));
-	REQUIRE(t != NULL);
-
-	LOCK(&task->lock);
-	*t = task->tnow;
-	UNLOCK(&task->lock);
-}
-
 /***
  *** Task Manager.
  ***/
@@ -813,11 +788,7 @@ task_run(isc_task_t *task) {
 	REQUIRE(VALID_TASK(task));
 
 	LOCK(&task->lock);
-	/*
+	/* FIXME */
-	 * It is possible because that we have a paused task in the queue - it
-	 * might have been paused in the meantime and we never hold both queue
-	 * and task lock to avoid deadlocks, just bail then.
-	 */
 	if (task->state != task_state_ready) {
 		goto done;
 	}
@@ -885,24 +856,11 @@ task_run(isc_task_t *task) {
 				 */
 				XTRACE("done");
 				task->state = task_state_done;
-			} else {
+			} else if (task->state == task_state_running) {
-				if (task->state == task_state_running) {
+				XTRACE("idling");
-					XTRACE("idling");
+				task->state = task_state_idle;
-					task->state = task_state_idle;
-				} else if (task->state == task_state_pausing) {
-					XTRACE("pausing");
-					task->state = task_state_paused;
-				}
 			}
 			break;
-		} else if (task->state == task_state_pausing) {
-			/*
-			 * We got a pause request on this task, stop working on
-			 * it and switch the state to paused.
-			 */
-			XTRACE("pausing");
-			task->state = task_state_paused;
-			break;
 		} else if (dispatch_count >= task->quantum) {
 			/*
 			 * Our quantum has expired, but there is more work to be
@@ -1164,65 +1122,6 @@ isc_task_endexclusive(isc_task_t *task) {
 					       &(bool){ true }, false));
 }
 
-void
-isc_task_pause(isc_task_t *task) {
-	REQUIRE(VALID_TASK(task));
-
-	LOCK(&task->lock);
-	task->pause_cnt++;
-	if (task->pause_cnt > 1) {
-		/*
-		 * Someone already paused this task, just increase
-		 * the number of pausing clients.
-		 */
-		UNLOCK(&task->lock);
-		return;
-	}
-
-	INSIST(task->state == task_state_idle ||
-	       task->state == task_state_ready ||
-	       task->state == task_state_running);
-	if (task->state == task_state_running) {
-		task->state = task_state_pausing;
-	} else {
-		task->state = task_state_paused;
-	}
-	UNLOCK(&task->lock);
-}
-
-void
-isc_task_unpause(isc_task_t *task) {
-	bool was_idle = false;
-
-	REQUIRE(VALID_TASK(task));
-
-	LOCK(&task->lock);
-	task->pause_cnt--;
-	INSIST(task->pause_cnt >= 0);
-	if (task->pause_cnt > 0) {
-		UNLOCK(&task->lock);
-		return;
-	}
-
-	INSIST(task->state == task_state_paused ||
-	       task->state == task_state_pausing);
-	/* If the task was pausing we can't reschedule it */
-	if (task->state == task_state_pausing) {
-		task->state = task_state_running;
-	} else {
-		task->state = task_state_idle;
-	}
-	if (task->state == task_state_idle && !EMPTY(task->events)) {
-		task->state = task_state_ready;
-		was_idle = true;
-	}
-	UNLOCK(&task->lock);
-
-	if (was_idle) {
-		task_ready(task);
-	}
-}
-
 void
 isc_taskmgr_setmode(isc_taskmgr_t *manager, isc_taskmgrmode_t mode) {
 	atomic_store(&manager->mode, mode);

lib/isc/tests/task_test.c

@@ -48,7 +48,7 @@ static isc_condition_t cv;
 
 atomic_int_fast32_t counter;
 static int active[10];
-static atomic_bool done, done2;
+static atomic_bool done;
 
 static int
 _setup(void **state) {
@@ -417,83 +417,6 @@ privilege_drop(void **state) {
 	assert_null(task2);
 }
 
-static void
-sleep_cb(isc_task_t *task, isc_event_t *event) {
-	UNUSED(task);
-	int p = *(int *)event->ev_arg;
-	if (p == 1) {
-		/*
-		 * Signal the main thread that we're running, so that
-		 * it can trigger the race.
-		 */
-		LOCK(&lock);
-		atomic_store(&done2, true);
-		SIGNAL(&cv);
-		UNLOCK(&lock);
-		/*
-		 * Wait for the operations in the main thread to be finished.
-		 */
-		LOCK(&lock);
-		while (!atomic_load(&done)) {
-			WAIT(&cv, &lock);
-		}
-		UNLOCK(&lock);
-	} else {
-		/*
-		 * Wait for the operations in the main thread to be finished.
-		 */
-		LOCK(&lock);
-		atomic_store(&done2, true);
-		SIGNAL(&cv);
-		UNLOCK(&lock);
-	}
-	isc_event_free(&event);
-}
-
-static void
-pause_unpause(void **state) {
-	isc_result_t result;
-	isc_task_t *task = NULL;
-	isc_event_t *event1, *event2 = NULL;
-	UNUSED(state);
-	atomic_store(&done, false);
-	atomic_store(&done2, false);
-
-	result = isc_task_create(taskmgr, 0, &task);
-	assert_int_equal(result, ISC_R_SUCCESS);
-
-	event1 = isc_event_allocate(test_mctx, task, ISC_TASKEVENT_TEST,
-				    sleep_cb, &(int){ 1 }, sizeof(isc_event_t));
-	assert_non_null(event1);
-	event2 = isc_event_allocate(test_mctx, task, ISC_TASKEVENT_TEST,
-				    sleep_cb, &(int){ 2 }, sizeof(isc_event_t));
-	assert_non_null(event2);
-	isc_task_send(task, &event1);
-	isc_task_send(task, &event2);
-	/* Wait for event1 to be running */
-	LOCK(&lock);
-	while (!atomic_load(&done2)) {
-		WAIT(&cv, &lock);
-	}
-	UNLOCK(&lock);
-	/* Pause-unpause-detach is what causes the race */
-	isc_task_pause(task);
-	isc_task_unpause(task);
-	isc_task_detach(&task);
-	/* Signal event1 to finish */
-	LOCK(&lock);
-	atomic_store(&done2, false);
-	atomic_store(&done, true);
-	SIGNAL(&cv);
-	UNLOCK(&lock);
-	/* Wait for event2 to finish */
-	LOCK(&lock);
-	while (!atomic_load(&done2)) {
-		WAIT(&cv, &lock);
-	}
-	UNLOCK(&lock);
-}
-
 /*
  * Basic task functions:
  */
@@ -1536,8 +1459,6 @@ main(int argc, char **argv) {
 		cmocka_unit_test_setup_teardown(all_events, _setup, _teardown),
 		cmocka_unit_test_setup_teardown(basic, _setup2, _teardown),
 		cmocka_unit_test_setup_teardown(create_task, _setup, _teardown),
-		cmocka_unit_test_setup_teardown(pause_unpause, _setup,
-						_teardown),
 		cmocka_unit_test_setup_teardown(post_shutdown, _setup2,
 						_teardown),
 		cmocka_unit_test_setup_teardown(privilege_drop, _setup,

lib/isc/win32/libisc.def.in

@@ -634,11 +634,8 @@ isc_task_destroy
 isc_task_detach
 isc_task_endexclusive
 isc_task_exiting
-isc_task_getcurrenttime
-isc_task_getcurrenttimex
 isc_task_getprivilege
 isc_task_onshutdown
-isc_task_pause
 isc_task_privileged
 isc_task_purge
 isc_task_purgeevent
@@ -650,7 +647,6 @@ isc_task_sendtoanddetach
 isc_task_setname
 isc_task_setprivilege
 isc_task_shutdown
-isc_task_unpause
 isc_task_unsend
 isc_taskmgr_excltask
 isc_taskmgr_mode

lib/ns/client.c

@@ -31,6 +31,7 @@
 #include <isc/stdio.h>
 #include <isc/string.h>
 #include <isc/task.h>
+#include <isc/thread.h>
 #include <isc/timer.h>
 #include <isc/util.h>
 
@@ -1660,7 +1661,6 @@ ns__client_request(isc_nmhandle_t *handle, isc_result_t eresult,
 
 	client->state = NS_CLIENTSTATE_READY;
 
-	isc_task_pause(client->task);
 	if (client->handle == NULL) {
 		isc_nmhandle_setdata(handle, client, ns__client_reset_cb,
 				     ns__client_put_cb);
@@ -1701,7 +1701,6 @@ ns__client_request(isc_nmhandle_t *handle, isc_result_t eresult,
 		ns_client_log(client, DNS_LOGCATEGORY_SECURITY,
 			      NS_LOGMODULE_CLIENT, ISC_LOG_DEBUG(10),
 			      "dropped request: suspicious port");
-		isc_task_unpause(client->task);
 		return;
 	}
 #endif /* if NS_CLIENT_DROPPORT */
@@ -1715,7 +1714,6 @@ ns__client_request(isc_nmhandle_t *handle, isc_result_t eresult,
 		ns_client_log(client, DNS_LOGCATEGORY_SECURITY,
 			      NS_LOGMODULE_CLIENT, ISC_LOG_DEBUG(10),
 			      "dropped request: blackholed peer");
-		isc_task_unpause(client->task);
 		return;
 	}
 
@@ -1729,7 +1727,6 @@ ns__client_request(isc_nmhandle_t *handle, isc_result_t eresult,
 		 * There isn't enough header to determine whether
 		 * this was a request or a response.  Drop it.
 		 */
-		isc_task_unpause(client->task);
 		return;
 	}
 
@@ -1746,7 +1743,6 @@ ns__client_request(isc_nmhandle_t *handle, isc_result_t eresult,
 	 */
 	if ((flags & DNS_MESSAGEFLAG_QR) != 0) {
 		CTRACE("unexpected response");
-		isc_task_unpause(client->task);
 		return;
 	}
 
@@ -1814,7 +1810,6 @@ ns__client_request(isc_nmhandle_t *handle, isc_result_t eresult,
 			result = DNS_R_FORMERR;
 		}
 		ns_client_error(client, result);
-		isc_task_unpause(client->task);
 		return;
 	}
 
@@ -1865,7 +1860,6 @@ ns__client_request(isc_nmhandle_t *handle, isc_result_t eresult,
 		 */
 		if ((client->sctx->options & NS_SERVER_EDNSFORMERR) != 0) {
 			ns_client_error(client, DNS_R_FORMERR);
-			isc_task_unpause(client->task);
 			return;
 		}
 
@@ -1874,7 +1868,6 @@ ns__client_request(isc_nmhandle_t *handle, isc_result_t eresult,
 		 */
 		if ((client->sctx->options & NS_SERVER_EDNSNOTIMP) != 0) {
 			ns_client_error(client, DNS_R_NOTIMP);
-			isc_task_unpause(client->task);
 			return;
 		}
 
@@ -1883,7 +1876,6 @@ ns__client_request(isc_nmhandle_t *handle, isc_result_t eresult,
 		 */
 		if ((client->sctx->options & NS_SERVER_EDNSREFUSED) != 0) {
 			ns_client_error(client, DNS_R_REFUSED);
-			isc_task_unpause(client->task);
 			return;
 		}
 
@@ -1892,13 +1884,11 @@ ns__client_request(isc_nmhandle_t *handle, isc_result_t eresult,
 		 */
 		if ((client->sctx->options & NS_SERVER_DROPEDNS) != 0) {
 			ns_client_drop(client, ISC_R_SUCCESS);
-			isc_task_unpause(client->task);
 			return;
 		}
 
 		result = process_opt(client, opt);
 		if (result != ISC_R_SUCCESS) {
-			isc_task_unpause(client->task);
 			return;
 		}
 	}
@@ -1911,7 +1901,6 @@ ns__client_request(isc_nmhandle_t *handle, isc_result_t eresult,
 			result = dns_message_reply(client->message, true);
 			if (result != ISC_R_SUCCESS) {
 				ns_client_error(client, result);
-				isc_task_unpause(client->task);
 				return;
 			}
 
@@ -1920,7 +1909,6 @@ ns__client_request(isc_nmhandle_t *handle, isc_result_t eresult,
 			}
 
 			ns_client_send(client);
-			isc_task_unpause(client->task);
 			return;
 		}
 
@@ -1930,7 +1918,6 @@ ns__client_request(isc_nmhandle_t *handle, isc_result_t eresult,
 		ns_client_dumpmessage(client, "message class could not be "
 					      "determined");
 		ns_client_error(client, notimp ? DNS_R_NOTIMP : DNS_R_FORMERR);
-		isc_task_unpause(client->task);
 		return;
 	}
 
@@ -1985,7 +1972,6 @@ ns__client_request(isc_nmhandle_t *handle, isc_result_t eresult,
 			      "no matching view in class '%s'", classname);
 		ns_client_dumpmessage(client, "no matching view in class");
 		ns_client_error(client, notimp ? DNS_R_NOTIMP : DNS_R_REFUSED);
-		isc_task_unpause(client->task);
 		return;
 	}
 
@@ -2086,7 +2072,6 @@ ns__client_request(isc_nmhandle_t *handle, isc_result_t eresult,
 		      client->message->opcode == dns_opcode_update))
 		{
 			ns_client_error(client, sigresult);
-			isc_task_unpause(client->task);
 			return;
 		}
 	}
@@ -2181,8 +2166,6 @@ ns__client_request(isc_nmhandle_t *handle, isc_result_t eresult,
 		CTRACE("unknown opcode");
 		ns_client_error(client, DNS_R_NOTIMP);
 	}
-
-	isc_task_unpause(client->task);
 }
 
 isc_result_t
@@ -2224,9 +2207,8 @@ get_clienttask(ns_clientmgr_t *manager, isc_task_t **taskp) {
 	MTRACE("clienttask");
 
 	int tid = isc_nm_tid();
-	if (tid < 0) {
+	REQUIRE(tid >= 0);
-		tid = isc_random_uniform(manager->ncpus);
+	REQUIRE(tid < manager->ncpus);
-	}
 
 	isc_task_attach(manager->taskpool[tid], taskp);
 }
@@ -2401,7 +2383,7 @@ clientmgr_destroy(ns_clientmgr_t *manager) {
 		}
 	}
 	isc_mem_put(manager->mctx, manager->taskpool,
-		    manager->ncpus * sizeof(isc_task_t *));
+		    manager->ncpus * sizeof(manager->taskpool[0]));
 	ns_server_detach(&manager->sctx);
 
 	isc_mem_put(manager->mctx, manager, sizeof(*manager));
@@ -2435,8 +2417,8 @@ ns_clientmgr_create(isc_mem_t *mctx, ns_server_t *sctx, isc_taskmgr_t *taskmgr,
 	ns_interface_attach(interface, &manager->interface);
 
 	manager->exiting = false;
-	manager->taskpool = isc_mem_get(mctx,
+	manager->taskpool = isc_mem_get(
-					manager->ncpus * sizeof(isc_task_t *));
+		mctx, manager->ncpus * sizeof(manager->taskpool[0]));
 	for (i = 0; i < manager->ncpus; i++) {
 		manager->taskpool[i] = NULL;
 		result = isc_task_create_bound(manager->taskmgr, 20,