Chapter 59 Executors and Tasks

Chapter Orientation

The executor is the abstraction of execution resources: tasks are submitted to it rather than opening threads directly — pooled reuse and work stealing are implemented at this layer; the business side sees only submission and completion. Twin task shapes: submit-and-forget (xrtExecutorSubmit — runs and done, no return value; the executor sample's 1000-task counting is this shape) and Future tasks (xrtTaskSubmit returning a Future — Chapter 58's currency); a batch interface (SubmitBatch — amortizing synchronization cost); and a shutdown protocol (xrtExecutorClose stops intake + Wait drains — isomorphic to Chapter 21's queue close-and-shutdown). Chapter 48's file-async task pool (xtaskpool) is the executor family's specialized form — this chapter teaches the general one.

59.1引入

The "open a thread per task" model goes bankrupt in three places.Volume: ten thousand concurrent tasks = ten thousand threads = memory explosion (1MB stack each) — pooled reuse runs ten thousand tasks on a few dozen threads.Overhead: thread creation is a millisecond-scale system call — for frequent short tasks the overhead exceeds the task itself; pool submission is milliseconds (a lock-free queue enqueue).Management: a thousand bare threads each minding its own stopping, waiting, and error collection; the executor gives this batch of threads one uniform Close+Wait shutdown protocol and one statistical language.

So the executor's value formula: thread reuse (volume) + lightweight submission (overhead) + uniform shutdown (management). The business-side change is just replacing "open a thread" with "submit a task" — you already did so in Chapter 48 (the file-async task pool); this chapter makes the executor's full interface surface and shutdown protocol clear.

59.2概念

Twin task shapes

Forget versus Future is not a rank distinction — it is a fork of needs: pure side-effect tasks (log flushing, cache refresh, statistics reporting) waste a Future (nobody waits); result-bearing ones (computation, queries) must have a Future (Chapter 58's delivery model). The cancellation-token parameter rides naturally in the Future shape (Chapter 54's tree threading the task pool).

The minimal submit-complete loop

xrtExecutorCreate(配置) (config) builds the instance (thread count and queue depth configurable — depth is the backpressure valve; a full pool fails submission rather than piling infinitely); the task function's signature is kin to the coroutine proc (ptr(ptr) or a void shape per the entrance); xrtExecutorWait waits for all to complete (WaitFor/Until timed variants). The executor sample's loop: submit 1000 forget tasks (each atomically incrementing a counter) → Wait → completed: 1000 — the simplest form of submit, execute, join.

The shutdown protocol: Close + Wait

Fully isomorphic to Chapter 21's queue Close section (Chapter 48's shutdown protocol) — closing is "stop intake", not "void": backlog tasks run to completion, Future tasks' results remain collectable. The executor_tour sample verifies: close+wait closed=1 queued=0 — Close then Wait to drain, the queue at zero: machine testimony.

Task pool vs scheduler: the endgame comparison

Chapter 56's table upgrades at the volume's end into its endgame edition (both chapters now learned): the task pool runs functions (stackless — run to completion, no mid-flight yield); the scheduler runs coroutines (stackful — Park/Sleep yield and resume).The combined shape is engineering's standard answer: blocking/CPU work met inside a coroutine is Posted to the task pool; the completion callback delivers a Future/continuation the result back to the scheduler-side coroutine (Chapter 58's continuation across models) — coroutines are the waiting skeleton, the task pool the computing muscle. Volume 7's network engine is exactly this combination in full.

59.3示例

Complete program: the thousand-task loop

From the repository sample examples/concurrency/executor/main.c:

#include <stdio.h>
#include <xrt.h>



















 */


/* 示例工作不需要返回值或 Future。 */
static void executorExampleRun(ptr pData)
{
	xatomic32* pCount = (xatomic32*)pData;

	(void)xrtAtomic32FetchAdd(pCount, 1, XMEMORY_RELAXED);
}



int main(void)
{
	xexecutor* pExecutor = xrtExecutorCreate(NULL);
	xatomic32 Count;

	xrtAtomic32Init(&Count, 0);
	if ( pExecutor == NULL ) {
		return 1;
	}
	for ( size_t i = 0; i < 1000; i++ ) {
		if ( !xrtExecutorSubmit(
			pExecutor,
			executorExampleRun,
			&Count,
			NULL,
			NULL
		) ) {
			(void)xrtExecutorCancel(pExecutor);
			(void)xrtExecutorDestroy(pExecutor);
			return 2;
		}
	}
	if ( !xrtExecutorDestroy(pExecutor) ) {
		return 3;
	}
	printf("completed: %u\n", xrtAtomic32Load(&Count, XMEMORY_ACQUIRE));
	return 0;
}

What just happened. (1) xrtExecutorCreate(NULL) — NULL takes the default configuration (thread count and queue depth per platform; pass a config to customize — depth is backpressure). (2) 1000 forget tasks submitted — each atomically increments a counter (Chapter 9's atomics — the counter is the tasks' only sharing; the join statistics complete lock-free). (3) Wait for all — completed: 1000 proves all 1000 tasks executed on pool threads, none lost. (4) Note no thread was opened for any task — the pool reused fixed threads; submission cost is a millisecond-scale enqueue. This 62-line loop is the executor's minimal complete usage.

Complete program: batch and shutdown

From examples/concurrency/executor_tour/main.c — batch interface and the Close+Wait protocol:

 */

#include <stdio.h>
#include <string.h>
#include <xrt.h>

#include <math.h>
static inline double exampleTimerLimit(int64 Timeout)
{
    return Timeout == XRT_WAIT_FOREVER ? INFINITY : xrtTimer() + (double)Timeout / 1000.0;
}
static inline bool exampleTimerExpired(double Limit)
{
    return xrtTimer() >= Limit;
}
static inline int64 exampleTimerRemaining(double Limit)
{
    double Ms;
    if (Limit == INFINITY) return XRT_WAIT_FOREVER;
    Ms = ceil((Limit - xrtTimer()) * 1000.0);
    return Ms <= 0 ? 0 : Ms >= 0x1p63 ? INT64_MAX : (int64)Ms;
}

static xatomic32 g_Ran;

static void exampleJob(ptr pData)
{
	(void)pData;
	(void)xrtAtomic32FetchAdd(&g_Ran, 1u, XMEMORY_RELAXED);
}

int main(void)
{
	xexecutorconfig Config = { 2, 8, 0 };
	xexecutoritem Items[3];
	xexecutorstats Stats;
	xexecutor* pExecutor = NULL;
	int iResult = 1;

	memset(&Stats, 0, sizeof(Stats));
	pExecutor = xrtExecutorCreate(&Config);
	if ( pExecutor == NULL ) {
		goto Cleanup;
	}
	/* SubmitBatch：三项原子受理。 */
	Items[0].Proc = exampleJob;
	Items[0].Data = NULL;
	Items[0].Destroy = NULL;
	Items[0].DestroyContext = NULL;
	Items[1] = Items[0];
	Items[2] = Items[0];
	xrtAtomic32Init(&g_Ran, 0u);
	if ( !xrtExecutorSubmitBatch(pExecutor, Items, 3u) ) {
		goto Cleanup;
	}
	
	 * 未关闭时调用返回 ERROR，不是等新工作的手段）。 */
	{
		double iDeadline = exampleTimerLimit(INT64_C(3000));

		while ( true ) {
			if ( !xrtExecutorGet(pExecutor, &Stats) ) {
				goto Cleanup;
			}
			if ( Stats.Completed == 3u ) {
				break;
			}
			if ( exampleTimerExpired(iDeadline) ) {
				goto Cleanup;
			}
			xrtThreadYield();
		}
	}
	/* 统计：受理 >= 3、完成 = 3。 */
	if ( !xrtExecutorGet(pExecutor, &Stats) ||
		(Stats.Submitted < 3u) ||
		(Stats.Completed != 3u) ||
		(xrtAtomic32Load(&g_Ran, XMEMORY_RELAXED) != 3u) ||
		(Stats.Threads != 2u) ) {
		goto Cleanup;
	}
	printf("executor: batch=3 stats(submitted>=%llu completed=%llu)"
		" ok\n",
		(unsigned long long)Stats.Submitted,
		(unsigned long long)Stats.Completed);
	/* Close → WaitFor（有界等待）与 Wait 收口。 */
	if ( !xrtExecutorClose(pExecutor) ||
		(xrtExecutorWaitFor(pExecutor,
			INT64_C(3000)) != XWAIT_OK) ||
		(xrtExecutorWait(pExecutor) != XWAIT_OK) ||
		!xrtExecutorGet(pExecutor, &Stats) ||
		!Stats.Closed ||
		(Stats.Queued != 0u) ) {
		goto Cleanup;
	}
	printf("executor: close+wait closed=1 queued=0 ok\n");
	iResult = 0;

Cleanup:
	xrtExecutorDestroy(pExecutor);
	return iResult;
}

What just happened. (1) SubmitBatch enqueues three tasks at once — batching amortizes the synchronous cost (the executor edition of Chapter 21's queue batch interface); the statistics (submitted/completed) reconcile per batch. (2) After Close, further submissions fail (intake-stop verified) while the enqueued keep running; Wait drains — queued=0 testifies the queue reached zero. (3) The protocol's two steps are non-interchangeable: Close first (else Wait may never finish between old tasks and new submissions), Wait after (Destroy only once the backlog executes). (4) Future-shaped tasks (xrtTaskSubmit with a cancellation token) unfold in the future/executor sample family — Chapter 58 taught them; this chapter composes.

59.4契约

• Twin shapes: forget (side effects) / Future (result + cancellation) — a fork of needs, not a rank.
• Backpressure built in: the queue-depth config is the cap — a full pool fails submission rather than piling (Chapter 21's discipline).
• Shutdown protocol: Close stops intake (backlog continues) → Wait drains → Destroy — the order is iron; isomorphic to queue closing.
• Batch interface: a group of tasks enqueued at once — amortized synchronization; statistics reconcile per batch.
• Task discipline: task functions don't loop forever (unless honoring a cancellation token); long tasks accept the token and set checkpoints (Chapter 54).
• Model combination: the scheduler manages the waiting skeleton, the pool the computing muscle — cross-model via Future/Post.

From examples to engineering: three hosts of the executor

Parallel computation (the Future shape's home turf): batch processing, image processing, data analysis — data fragment shards submitted as Future tasks, joined via future_combine or Wait one by one; pool depth = CPU core count as the starting point (the compute boundary), excess submissions queue (backpressure).The async-IO foundation (Chapter 48's shape): file async, DNS resolution, external-process waiting — asynchronous-blocking IO wrapped as tasks into the pool, the caller takes a Future without waiting; depth by IO concurrency needs (IO-dense may exceed core count — waiting costs no CPU).The scheduler's compute annex (the combined shape): a coroutine meeting a CPU or blocking stretch → Post to the pool → the pool's completed Future/continuation → back to the scheduler-side coroutine — Volume 7's network engine's standard division. The three hosts' depth configs differ: CPU boundary by core count, IO boundary by concurrency budget, mixed by measurement — depth is a performance parameter (an object of Chapter 137's benchmarking methods).

The executor family panorama

XRT's "execution resources" come in more than one implementation; one panorama prevents confusion.xexecutor (this chapter): the general executor — forget + batch + Close/Wait protocol; the task model is "a function that runs to completion".xtaskpool (Chapter 48): the task pool — xrtTaskSubmit returns a Future, cancellation token a first-class parameter; the dedicated foundation of file async; its relation to the executor is "same family, different emphasis" — taskpool's Future/cancellation integration is deeper, executor's batch and statistics fuller.xcosched (Chapter 56): the coroutine scheduler — not a task pool (runs coroutines, not functions), but its pump loop fills the same-position "execution resource" role. The three share one base vocabulary (queues, backpressure, shutdown); selection follows the task shape: functions → executor/taskpool, coroutines → sched, mixed → combination.

An operations view: the pool's capacity economics

Pool depth is not a guess — it has economics.Too small: insufficient parallelism (CPU cores idle) or insufficient IO concurrency (throughput capped by the pool) — symptoms: queue depth chronically full, long task waits.Too large: thread memory (stacks), context switching (CPU-type excess does no good — there are only so many cores), downstream pressure (IO-type excess knocks over database connections) — symptoms: memory climbing, switching high, downstream erroring.Measure and tune: three datasets read together — queue-depth curves (chronically full → add or expand downstream), task-duration distributions (Chapter 6's statistics or Chapter 42's timing), system metrics (CPU/memory/switching). Starting points: CPU-type = core count, IO-type = concurrency budget (80% of downstream capacity), mixed starts CPU-type and grows by measurement.Depth is not bigger-is-better — it is "the gatekeeper of downstream capacity".

Common Mistakes

static void taskForever(ptr pData)
{
	while ( true ) {
		Poll();   /* infinite loop - never checks the cancellation token */
	}
}

static void taskCancellable(ptr pData, xcancel* pCancel)   /* the Future shape carries a token */
{
	while ( !xrtCancelRequested(pCancel) ) {   /* checkpoint (Chapter 54) */
		Poll();
	}
	/* cancellation hit - the task returns, draining advances */
}

static void taskLeak(ptr pData)
{
	xvalue* pTmp = xrtValueInt(42);   /* created and dropped - nobody frees */
	Report(pTmp);
	/* Release missing - the forget task's leak window */
}

static void taskClean(ptr pData)
{
	xvalue* pTmp = xrtValueInt(42);
	Report(pTmp);
	xrtValueRelease(pTmp);   /* created in the task, finished in the task */
}

Programming Exercises

Summary