Глава 59 Исполнители и задачи

Ориентация главы

Исполнитель — абстракция ресурса исполнения: задачи подаются ему, а не открывают потоки напрямую — пул-переиспользование и перенос работы (work stealing) реализованы на этом слое; бизнес видит только подачу и завершение. Двойная форма задач: подал и забыл (xrtExecutorSubmit — исполнилось и всё, без возвращаемого значения; тысяча задач-счётчиков образца executor — эта форма) и Future-задачи (xrtTaskSubmit возвращает Future — валюта главы 58); пакетный интерфейс (SubmitBatch — амортизация синхронных расходов); протокол остановки (xrtExecutorClose прекращает приём + Wait осушает — изоморф закрытию очереди главы 21). Пул задач (xtaskpool) файловой асинхронности главы 48 — специализированная форма семейства исполнителей; эта глава разбирает общую версию.

59.1引入

Модель «каждой задаче по потоку» банкротится в трёх местах. Масштаб: десять тысяч конкурентных задач = десять тысяч потоков = взрыв памяти (по 1 МБ стека) — пул-переиспользование гоняет десять тысяч задач на нескольких десятках потоков. Накладные расходы: создание потока — системный вызов миллисекундного порядка; у частых коротких задач расходы превышают саму задачу; подача в пул — миллисекунды (постановка в безблокировочную очередь). Управление: у тысячи голых потоков остановка, ожидание и сбор ошибок каждый сам по себе; исполнитель даёт этой пачке потоков единый протокол остановки Close+Wait и единую статистику.

Формула ценности исполнителя: переиспользование потоков (масштаб) + лёгкая подача (расходы) + единая остановка (управление). Со стороны бизнеса меняется лишь замена «открыть поток» на «подать задачу» — в главе 47 вы уже так делали (пул задач файловой асинхронности); эта глава проясняет полный интерфейс исполнителя и его протокол остановки.

59.2概念

Двойная форма задач

«Забыть» и Future — не иерархия рангов, а развилка потребностей: задачи с чистым побочным эффектом (сброс журналов на диск, обновление кеша, статистическая отчётность) с Future — расточительство (никто не ждёт); задачи с результатом (вычисление, запрос) — обязательно Future (модель доставки главы 58). Параметр токена отмены естественно живёт в Future-форме (дерево главы 54 прошивает пул задач).

Минимальный контур подачи-завершения

xrtExecutorCreate(配置) (конфигурация) строит экземпляр (число потоков и глубина очереди настраиваемы — глубина это клапан противодавления: заполнено — подача отказывает, а не копится бесконечно); сигнатура задачи семейственна процедуре корутины (ptr(ptr) или void-форма по входу); xrtExecutorWait ждёт всех (WaitFor/Until — варианты со сроком). Контур образца executor: подать 1000 «забытых» задач (каждая атомарно наращивает счётчик) → Wait → completed: 1000 — простейшая форма подачи, исполнения, сведение.

Протокол остановки: Close + Wait

Полностью изоморф разделу Close очереди главы 21 (протокол остановки главы 48) — закрытие это «прекратить приём», а не «аннулировать»: остаток задач исполняется, результаты Future-задач забираются как обычно. Образец executor_tour проверяет: close+wait closed=1 queued=0 — Close, затем Wait до осушения и нулевая очередь: машинное свидетельство.

Пул задач против планировщика: финальное сопоставление

Таблица главы 56 в конце тома повышается до финальной редакции (обе главы изучены): пул задач исполняет функции (без стека — до конца, без уступки на середине); планировщик исполняет корутины (со стеком — Park/Sleep уступают и восстанавливают). Комбинированная форма — инженерный стандартный ответ: блокирующую/CPU-работу, встреченную внутри корутины, Post-ят в пул задач; по завершении Future/продолжение доставляет результат обратно в корутину на стороне планировщика (межмодельная форма продолжения главы 58) — корутины — скелет ожидания, пул задач — мышцы вычислений. Сетевой движок тома 7 — именно эта комбинация в полном виде.

59.3示例

Полная программа: контур тысячи задач

Из образца репозитория 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;
}

Что только что произошло. (1) xrtExecutorCreate(NULL) — NULL берёт конфигурацию по умолчанию (потоки и глубина очереди по платформе; для настройки передайте конфигурацию — глубина и есть противодавление). (2) 1000 «забытых» задач поданы — каждая атомарно наращивает счётчик (атомики главы 9 — единственное разделение между задачами; сводная статистика завершена без блокировок). (3) Wait ждёт всех — completed: 1000 доказывает: все 1000 задач исполнены на потоках пула, ни одна не потеряна. (4) Обратите внимание: ни для одной задачи не открыт поток — пул переиспользовал фиксированные потоки; расходы подачи — миллисекундная постановка в очередь. Этот 62-строчный контур — минимальное полное употребление исполнителя.

Полная программа: пакет и остановка

Из examples/concurrency/executor_tour/main.c — пакетный интерфейс и протокол Close+Wait:

 */

#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;
}

Что только что произошло. (1) SubmitBatch одной операцией ставит три задачи — пакет амортизирует синхронные расходы (редакция пакетного интерфейса очереди главы 21 для исполнителя); статистика (submitted/completed) сверяется по пакетам. (2) После Close повторная подача отказывает (проверка прекращения приёма), поставленные продолжают исполняться; Wait до осушения — queued=0 свидетельствует обнуление очереди. (3) Порядок двух шагов протокола нерушим: сперва Close (иначе Wait мог бы вечно ждать между старыми задачами и новыми подачами), потом Wait (Destroy лишь после исполнения остатка). (4) Future-формы задач (xrtTaskSubmit с токеном отмены) развёрнуты в семействе образцов future/executor — глава 58 их уже научила, эта глава компонует.

59.4契约

• Двойная форма: «забыть» (побочный эффект) / Future (результат + отмена) — развилка потребностей, не ранг.
• Противодавление встроено: настроенная глубина очереди и есть потолок — заполнено, подача отказывает, а не копится (дисциплина главы 21).
• Протокол остановки: Close прекращает приём (остаток идёт) → Wait осушает → Destroy — порядок железный; изоморф закрытию очереди.
• Пакетный интерфейс: группа задач в очередь за раз — амортизированная синхронизация; статистика сверяется по пакетам.
• Дисциплина задач: функции задач без вечных циклов (кроме как с уважением к токену отмены); длинные задачи принимают токен и ставят контрольные точки (глава 54).
• Комбинация моделей: планировщик ведает скелетом ожидания, пул — мышцами вычислений; между моделями — Future/Post.

От примеров к инженерии: три хозяина исполнителя

Параллельные вычисления (арена Future-формы): пакетная обработка, обработка изображений, анализ данных — фрагменты данных подаются Future-задачами, сведение через future_combine или Wait по одной; глубина пула = числу ядер как отправная точка (граница вычислительности), излишек подач стоит в очереди (противодавление). Фундамент асинхронного IO (форма главы 48): файловая асинхронность, DNS-разрешение, ожидание внешних процессов — блокирующий IO оборачивается в задачу пула, вызывающий берёт Future и не ждёт; глубина по потребностям IO-конкурентности (IO-плотно можно сверх числа ядер — ожидание CPU не занимает). Вычислительный пристрой планировщика (комбинированная форма): корутина встретила CPU-или-блокирующий участок → Post в пул → завершённый Future/продолжение → обратно в корутину планировщика — стандартное разделение труда сетевого движка тома 7. Глубина пула у трёх хозяев разная: граница CPU — по ядрам, граница IO — по бюджету конкурентности, смешанно — по измерениям: глубина это параметр производительности (объект методов бенчмаркинга главы 137).

Панорама семейства исполнителей

«Ресурс исполнения» в XRT не один; панорама бережёт от путаницы. xexecutor (эта глава): общий исполнитель — «забыть» + пакет + протокол Close/Wait; модель задачи — «функция до конца». xtaskpool (глава 48): пул задач — xrtTaskSubmit возвращает Future, токен отмены — параметр первого класса; специализированный фундамент файловой асинхронности; отношение с исполнителем — «одно семейство, разные акценты»: у taskpool глубже интеграция Future/отмены, у исполнителя полнее пакет и статистика. xcosched (глава 56): планировщик корутин — не пул задач (гоняет корутины, не функции), но его насосный цикл занимает ту же позицию «ресурса исполнения». Трое делят общий базовый словарь (очередь, противодавление, остановка); выбор по форме задачи: функции → executor/taskpool, корутины → sched, смешанно → комбинация.

Взгляд эксплуатации: экономика ёмкости пула

Глубина пула не выбирается на глаз — у неё есть экономика. Мало: параллелизма недостаточно (ядра простаивают) или IO-конкурентности недостаточно (пропускная способность упирается в пул) — симптомы: глубина очереди хронически полна, время ожидания задач велико. Много: память потоков (стеки), переключение контекста (CPU-перебор бесполезен — ядер столько), давление на нижний поток (IO-перебор валит число соединений базы) — симптомы: память растёт, переключения высоки, нижний поток ошибается. Настройка по измерениям: три набора данных вместе — кривые глубины очереди (хронически полна → добавить или расширить нижний поток), распределение длительностей задач (статистика главы 6 или таймирование главы 42), системные метрики (CPU/память/переключения). Отправные значения: CPU-тип = ядра, IO-тип = бюджет конкурентности (80% ёмкости нижнего потока), смешанный стартует с CPU-типа и растёт по измерениям. Глубина не «чем больше, тем лучше» — она «привратник ёмкости нижнего потока».

Типичные ошибки

static void taskForever(ptr pData)
{
	while ( true ) {
		Poll();   /* вечный цикл - токен отмены не смотрится */
	}
}

static void taskCancellable(ptr pData, xcancel* pCancel)   /* Future-форма несёт токен */
{
	while ( !xrtCancelRequested(pCancel) ) {   /* контрольная точка (глава 54) */
		Poll();
	}
	/* отмена попала - задача возвращается, осушение движется */
}

static void taskLeak(ptr pData)
{
	xvalue* pTmp = xrtValueInt(42);   /* создал и бросил - освобождать некому */
	Report(pTmp);
	/* нет Release - окно утечки забытой задачи */
}

static void taskClean(ptr pData)
{
	xvalue* pTmp = xrtValueInt(42);
	Report(pTmp);
	xrtValueRelease(pTmp);   /* создано в задаче - в задаче и завершено */
}

Практические задания

Итоги