Глава 61 Планировщик на практике: корутины, каналы и задачи в совместном бою

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

Заключительная глава тома 6 (композиционного типа) собирает инструменты всего тома в запускаемый скелет службы: насос планировщика как цикл событий (глава 56), по корутине на соединение для прямого обслуживания (главы 55/56), Channel как внутренняя шина сообщений (глава 57), вычислительные/CPU-участки в пристрое пула задач (глава 59), Future/продолжения мостят оба мира (глава 58), группы задач — структурный жизненный цикл и грациозная остановка (глава 60 + дерево отмены главы 54 + сигнальный триггер главы 49). Тезис композиционной главы: ценность параллельных инструментов не в одиночном употреблении, а в сборке в целую программу, которая «запускается, наблюдаема и останавливается» — скелет этой главы и есть репетиция сетевого движка тома 7.

61.1引入

Полный список требований mini-HTTP-службы: принимать соединения (IO-событийно); прямая логика на каждое соединение (прочитать запрос → обработать → записать ответ); части запросов требуют тяжёлых вычислений (цикл событий блокировать нельзя); сотрудничество внутренних компонентов (ограничитель частоты, агрегатор статистики); грациозная остановка (сигнал → прекратить приём → осушить в полёте → принудительный выход по сроку). Каждое требование соответствует инструменту тома 6 — но инструмент порознь полного требования не закрывает: планировщик без пула — вычисления клинят насос; пул без группы — остановке не в чем сводиться; группа без дерева отмены — остановка не распространяется.

Сборка — единственный путь. Глава проходит полный скелет вдоль потока данных — каждая станция помечена, какая глава что дала и почему так выбрано. Снимите со скелета детали протокола — и получите сетевой движок тома 7 (с главы 64); там вы увидите ту же основу в индустриальной форме, нагруженную TCP/TLS/HTTP.

61.2概念

Панорама скелета

Три ключевых решения сборки

Решение первое: граница раздела между потоком-насосом и пулом. Поток цикла событий (планировщик) ведёт всю ожидающую логику (соединения, протоколы, координацию); пул задач ведёт все вычислительные участки (тяжёлый счёт, блокирующий IO). Критерий границы: этот код уступает или держит — уступающий (ждёт данных, ждёт событий) принадлежит планировщику, держащий (считает, блокирующе зовёт) — пулу. Что при смешении: вычисления клинят насос (ловушка 1 главы 56), или ожидание занимает пул (растрата конкурентности пула). Решение второе: сообщения против общей памяти. Корутины соединений не делят ничего (у каждой свой стек); потоки данных между корутинами (статистическая отчётность) идут через Channel — приземление «внутри компактно, снаружи ясно» (разделение CSP главы 57). Агрегатор статистики — единственная корутина с состоянием (владеет счётчиками), все обновления сообщениями. Решение третье: этажность остановки. Сигнал → корневой токен → Close планировщика (глава 56) + Cancel служебной группы (глава 60) — двойной канал: планировщик прекращает насос, группа завершает; они сотрудничают, а не дублируют (Close ведает «новые работы не доводятся», Cancel — «всё в полёте завершается»).

Прогулка по потоку данных

Запрос с тяжёлым вычислением от начала до конца: (1) корутина-приёмник получила новое соединение (событие слушателя разбудило её Wake-ом) → xrtCoGo корутина соединения; (2) корутина соединения читает запрос (Park-уступка, пока событие IO в ожидании); (3) обработка нашла тяжёлый участок → xrtTaskSubmit в пул (с Future подгруппы) → корутина соединения Park-ом ждёт результата; (4) поток пула вычислил → Promise Resolve → xrtCoWake корутины соединения; (5) корутина соединения восстановилась, собрала ответ, записала (событие записи); (6) статистическое сообщение TrySend-ом в Channel (заполнен — выброшено: деградационная семантика «статистику можно терять») → агрегатор потребляет. Весь путь: в потоке-насосе ноль блокировок, в потоках пула ноль ожидания, в сообщениях ноль блокировок — три «ноля» суть проверяемые метрики правильности сборки.

Прогулка по остановке

Последовательность остановки (полная вторая половина главы 49 о сигналах): (1) обратный вызов сигнала ставит флаг остановки (лёгкая работа — дисциплина главы 49); (2) главный цикл видит флаг: слушатель закрыт (прекратить приём) → Request корневого токена (глава 54) → Cancel служебной группы + Close планировщика; (3) каждая корутина соединения ощущает отмену в точке Park/Yield → очистный стек в обратном порядке (глава 55) → финал Cancelled; (4) задачи пула в полёте выходят на контрольных точках (хорошая форма ловушки 1 главы 59); (5) WaitFor группы (льготные 5 секунд — конвенция главы 49) → все финалы — грациозный выход; по истечении — запись числа незавершённых и Destroy на выход. (6) Итоговый отчёт статистики (stats группы — слияние наблюдения главы 60) в журнал.

Контрольный список сборки

Пишете свой параллельный скелет — проходите пять пунктов: в насосе ноль блокировок (на потоке планировщика ни одного блокирующего вызова — пройдитесь по коду всех корутин); всякое ожидание отменяемо (у каждого Park/Wait/Recv есть токен — на остановке нет мёртвых ожиданий); финалы троичны (успех/отказ/отмена считаются и рапортуются раздельно — ловушка 2 главы 60); жизненные циклы совпадают (границы областей каждого класса ресурсов ясны: подгруппа соединения = соединение, группа службы = служба, пул = движок — порядок разрушения обратен зависимостям); наблюдение насквозь (глубина очереди, Active группы, счётчики отмен, цикл насоса — посты конвейера главы 50 в параллелизме открыты все). Пять зелёных — и скелет перерастает «запускается» в «готов к выкладке».

61.3示例

Полная программа: ядро-кольцо скелета (цикл событий + корутины + пул)

Минимальное запускаемое ядро сборки — насос, корутины соединений, вычислительный пристрой, шина сообщений, каждый на своём месте:

/* service_core.c - ядро-кольцо скелета тома 6: планировщик + корутины + пул + канал */
#define XRT_MODULE_ALL
#define XRT_IMPLEMENTATION
#include <xrt.h>
#include <stdio.h>

static xcosched* gSched;
static xtaskpool* gPool;
static xchannel* gStats;
static xatomic32 gDone;

static void aggregate(ptr pData)          /* тело корутины-агрегатора статистики */
{
	ptr Item;
	while ( xrtChannelRecv(gStats, &Item) == XWAIT_OK ) {
		(void)xrtAtomic32FetchAdd((xatomic32*)pData, 1, XMEMORY_RELAXED);
	}
}

static int computeHeavy(int Input)        /* имитация тяжёлого вычисления (в пуле) */
{
	int Sum = 0;
	for ( int i = 0; i < Input * 100; ++i ) {
		Sum += i % 7;
	}
	return Sum;
}

static ptr workerTask(ptr pData)          /* задача пула: вычислить и опубликовать */
{
	int Input = (int)(uintptr_t)pData;
	int Result = computeHeavy(Input);
	return (ptr)(uintptr_t)Result;
}

static ptr connection(ptr pData)          /* корутина соединения: прямая логика */
{
	int Input = (int)(uintptr_t)pData;
	xfuture* Job = xrtTaskSubmit(gPool, workerTask,
		(ptr)(uintptr_t)Input, NULL);      /* пристрой-пул (глава 59) */
	if ( Job == NULL ) {
		return NULL;
	}
	if ( xrtFutureWait(Job) != XWAIT_OK ) {  /* в насосе без блокировок - прямой ожидание здесь ради простоты демонстрации */
		xrtFutureDestroy(Job);
		return NULL;
	}
	int Value = (int)(uintptr_t)xrtFutureValue(Job);  /* заимствованное чтение (глава 58) */
	xrtFutureDestroy(Job);
	(void)xrtChannelTrySend(gStats, (ptr)(uintptr_t)Value);  /* отчёт по шине (глава 57) */
	return (ptr)(uintptr_t)Value;
}

int main(void)
{
	xatomic32 Count;
	xcoro* Agg;
	xtaskpoolconfig PoolCfg = { 2, 32, 0 };

	xrtAtomic32Init(&Count, 0);
	xrtAtomic32Init(&gDone, 0);
	gSched = xrtCoSchedCreate();               /* цикл событий (глава 56) */
	gPool = xrtTaskPoolCreate(&PoolCfg);       /* вычислительный пристрой (глава 59) */
	gStats = xrtChannelCreate(64);             /* шина сообщений (глава 57) */
	if ( (gSched == NULL) || (gPool == NULL) || (gStats == NULL) ) {
		return 1;
	}
	Agg = xrtCoGo(gSched, aggregate, &Count);  /* корутина-агрегатор */
	for ( int i = 1; i <= 3; ++i ) {
		(void)xrtCoGo(gSched, connection, (ptr)(uintptr_t)i);
	}
	(void)xrtCoSchedRun(gSched);              /* насос до завершения всех корутин */
	xrtChannelClose(gStats);                   /* осушение (трёхстороннее правило главы 57) */
	(void)xrtCoSchedRun(gSched);               /* агрегатор потребляет остаток */
	xrtCoSchedDestroy(gSched);
	(void)xrtTaskPoolClose(gPool);             /* протокол остановки пула (глава 59) */
	xrtTaskPoolWait(gPool);
	xrtTaskPoolDestroy(gPool);
	printf("aggregated=%d\n", (int)xrtAtomic32Load(&Count, XMEMORY_ACQUIRE));
	return 0;
}

Где стоит этот код. Инструменты тома в одной раме: насос xrtCoSchedCreate как главная линия, три корутины соединений xrtCoGo с конкурентной прямой логикой, тяжёлое вычисление xrtTaskSubmit-ом в пристрой пула, канал xrtChannelCreate как статистическая шина, корутина-агрегатор потребляет. Трёхфазный финал в обратном порядке зависимостей: насос Run до пустоты → Close канала и осушение → Run снова (агрегатор потребляет остаток) → Close+Wait пула — протоколы остановки каждого слоя (главы 56/57/59) сцеплены в правильном порядке. aggregated=3 свидетельствует: все три сообщения соединений дошли до шины и все потреблены. Обратите внимание: xrtFutureWait в корутине соединения в настоящем скелете должен быть Park-ожиданием Wake (в насосе блокировок нет — здесь прямой Wait ради простоты демонстрации; аннотация ловушки 1 главы 56 применима).

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

Из examples/concurrency/task_group_scope/main.c (глава 60) + образец cancel главы 54 в совместном употреблении — остановочное кольцо скелета:

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





















 */





 */
int main(void)
{
	int iResult = 1;
	xtaskgroup* pParent = NULL;
	xtaskgroup* pChild = NULL;
	xfuture* pLeaf = NULL;
	xfuture* pDone = NULL;
	xpromise* pLeafPromise = NULL;
	xcancel* pLeafCancel = NULL;
	xtaskgroupstats tStats;

	/* 子组作为父组中的一项，叶 Future 再由子组跟踪。 */
	pParent = xrtTaskGroupCreate(NULL);
	if ( pParent == NULL ) {
		goto cleanup;
	}
	pChild = xrtTaskGroupChild(pParent, NULL);
	pLeafPromise = xrtPromiseCreate(&pLeaf, NULL);
	if ( (pChild == NULL) || (pLeafPromise == NULL) ||
		!xrtTaskGroupAdd(pChild, pLeaf) ) {
		goto cleanup;
	}
	pDone = xrtTaskGroupFuture(pParent);
	pLeafCancel = xrtFutureCancelToken(pLeaf);
	if ( (pDone == NULL) || (pLeafCancel == NULL) ) {
		goto cleanup;
	}

	/* 父级只发出协作取消请求，叶生产端负责确认自己的终态。 */
	if ( !xrtTaskGroupCancel(pParent) ||
		!xrtCancelRequested(pLeafCancel) ||
		!xrtPromiseCancel(pLeafPromise) ||
		(xrtFutureWait(pDone) != XWAIT_OK) ) {
		goto cleanup;
	}
	if ( (xrtFutureState(pLeaf) != XFUTURE_CANCELLED) ||
		!xrtTaskGroupGet(pParent, &tStats) ) {
		goto cleanup;
	}
	printf(
		"parent completed = %llu, cancelled = %llu\n",
		(unsigned long long)tStats.Completed,
		(unsigned long long)tStats.Cancelled
	);
	iResult = 0;

cleanup:
	/* 子组和父组允许在仍有活动监听时延迟完成内部回收。 */
	xrtCancelDestroy(pLeafCancel);
	xrtFutureDestroy(pDone);
	xrtPromiseDestroy(pLeafPromise);
	xrtFutureDestroy(pLeaf);
	xrtTaskGroupDestroy(pChild);
	xrtTaskGroupDestroy(pParent);
	return iResult;
}

Где стоит этот код. Эмпирический скелет остановочного кольца: Cancel родительской группы (большая кнопка служебной группы) → подгруппы и листья получают запрос по всему дереву → листья подтверждают финалы каждый сам (один завершён, один отменён — самостоятельность производящей стороны) → Wait родителя берёт смешанную статистику. Подставьте его в шаги (3)~(5) последовательности остановки скелета, добавьте сигнальный триггер (глава 49) и дежурство со сроком (WaitFor) — полный контур грациозной остановки. Два образца вместе закрывают все станции скелета: образец один собирает исполнение (насос/корутины/пул/шина), образец два — остановку (группы/дерево отмены/статистика финалов): один бежит, другой останавливает, скелет цел.

Три новых пункта проверки станций (четыре тома 4 — параллельная редакция)

Глава 36 дала четыре пункта для конвейера данных, глава 50 — редакцию наблюдения; параллельный скелет добавляет три. Пункт первый: единоличное владение данными. Каждая параллельная сущность (корутина/задача) владеет своими данными; между сущностями — только сообщения или Future; агрегатор статистики — единственная корутина, владеющая счётчиками (все обновления через канал); если две корутины могут писать одну переменную, сборка уже ошибочна. Пункт второй: пути ошибок тоже структурны. Написано ли поведение скелета после отказа задачи — подгруппа соединения вешает в группу и провалившийся Future (виден в stats), отказные ответы идут тем же каналом (потребитель разбирает по классу финала); «успех прогнали, про отказ не думали» в параллелизме — семя остановочной аварии. Пункт третий: вложенность разложима. Каждый слой скелета (насос/шина/пул/группа) тестируется отдельно — применение конвейерного мышления главы 50 на слое сборки: входы-выходы каждого слоя ясны (сообщения/задачи/финалы), слои организованы формой станций-функций главы 36 — только тестируемые слои смеют компоноваться.

От скелета к сетевому движку: анонс тома 7

Отображение этого скелета на сетевой движок тома 7 стоит прописать в заключительной главе. Насос планировщика → порт событий (пять бэкендов главы 64: IOCP/epoll/kqueue/io_uring/select — PollFor насоса соответствует интерфейсу ожидания порта); корутина соединения → событийные обратные вызовы Stream движка (таблицы колбэков Accept/Read/Close золотого стандарта главы 66 — том 7 выражает их обратными вызовами, а не корутинами напрямую, но планировщик может обернуть колбэчи корутинами); пристрой пула задач → Worker-пул сетевого движка (изоморф — вычисления не клинят IO-потоки); шина сообщений → внутренние статистические/управленческие каналы движка; служебная группа → последовательность остановки движка (глава 66: «колбэчи Close могут ещё завершаться на Worker-ах; разрушение Engine ждёт» — редакция Wait группы задач для движка). Отображение не тождество — том 7 добавляет над скелетом слои протоколов (автомат состояний TCP, TLS, HTTP) и продакшен-детали (уровни противодавления, буферы с подсчётом ссылок) — но пятипунктовый контрольный список скелета дословно применим к сетевому движку. Читайте том 7 с диаграммой скелета этой главы, спрашивая у каждой главы «на каком месте скелета крепится этот слой».

Сетевые группы задач (task_net): мост из тома 6 в том 7

Между скелетом и сетевым движком стоит готовый мост: необязательный мост от системы задач к сетевому Engine (XRT_FEATURE_TASK_NET — однонаправленный; сеть не проникает обратно в ядро Future/Task). xrtTaskNet подаёт задачу на назначенного аффинного Worker-а: сигнатура xtasknetproc заимствует xnetworker впереди обычных аргументов задачи — прямой доступ к пулу буферов Worker-а и контексту Engine, а контракты отмены, значений задач, структурных ошибок, временной арены и владения данными полностью совпадают с обычными задачами; варианты After/Until отдают «отложенное / по дедлайну» таймеру Engine (математика дедлайнов главы 9 в издании движка). Групповая форма xrtTaskGroupNetFor резервирует слот активности группы задач до подачи — когда группа закрыта или достигнут потолок, задача вообще не стартует; групповая отмена распространяется на Future и ждёт настоящей отмены таймера. Два жёстких правила: сетевые задачи исполняются в потоке цикла событий — никаких блокировок и длинных вычислений (тяжесть уходит в xtaskpool — дисциплина «насос никогда не блокирует» в модульном изводе); при остановке Engine немедленные задачи всё же исполняются в фазе стекинга, а отложенные терпят структурную неудачу XERR_CLOSED — до публикации терминального состояния таймер, слушатель отмены и разрушение данных завершаются целиком (потребитель никогда не видит контекст, ещё используемый Worker-ом).

Из репозиторного образца examples/network/task/main.c — немедленная/отложенная/по дедлайну/групповая, четыре вида подачи одним туром:

 */

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



/* 在 Engine Worker 上生成一个轻量结果。 */
static xtaskoutcome buildValue(
	xnetworker* pWorker,
	xcancel* pCancel,
	ptr pData,
	xtaskvalue* pResult
)
{
	(void)pCancel;
	printf("worker=%u\n", xrtNetWorkerIndex(pWorker));
	pResult->Value = pData;
	return XTASK_SUCCESS;
}





/* 演示立即网络任务与统一 Future 等待。 */
int main(void)
{
	xnetengine* pEngine = xrtNetEngineCreate(NULL);
	xfuture* pFuture;
	int iValue = 42;

	if ( (pEngine == NULL) || !xrtNetEngineStart(pEngine) ) {
		return 1;
	}
	pFuture = xrtTaskNet(
		pEngine,
		0,
		buildValue,
		&iValue,
		NULL
	);
	if ( (pFuture == NULL) ||
		(xrtFutureWaitFor(pFuture, 3000) != XWAIT_OK) ||
		(xrtFutureState(pFuture) != XFUTURE_RESOLVED) ) {
		xrtFutureDestroy(pFuture);
		(void)xrtNetEngineDestroy(pEngine);
		return 1;
	}
	printf("value=%d\n", *(int*)xrtFutureValue(pFuture));
	xrtFutureDestroy(pFuture);

	/* ---- After / Until：延迟与截止时间两种提交形态。 ---- */
	pFuture = xrtTaskNetAfter(pEngine, 0, buildValue, &iValue,
		NULL, 0);
	if ( (pFuture == NULL) ||
		(xrtFutureWaitFor(pFuture, 3000) != XWAIT_OK) ) {
		xrtFutureDestroy(pFuture);
		(void)xrtNetEngineDestroy(pEngine);
		return 2;
	}
	printf("after: value=%d\n", *(int*)xrtFutureValue(pFuture));
	xrtFutureDestroy(pFuture);
	pFuture = xrtTaskNetFor(pEngine, 0, buildValue, &iValue,
		NULL,0);
	if ( (pFuture == NULL) ||
		(xrtFutureWaitFor(pFuture, 3000) != XWAIT_OK) ) {
		xrtFutureDestroy(pFuture);
		(void)xrtNetEngineDestroy(pEngine);
		return 3;
	}
	printf("until: value=%d\n", *(int*)xrtFutureValue(pFuture));
	xrtFutureDestroy(pFuture);

	/* ---- GroupNetUntil：延迟任务原子纳入任务组。 ---- */
	{
		xtaskgroup* pGroup;

		pGroup = xrtTaskGroupCreate(NULL);
		if ( (pGroup == NULL) ||
			(xrtTaskGroupNetFor(pGroup, pEngine, 0,
				buildValue, &iValue, NULL,0) == NULL) ||
			!xrtTaskGroupClose(pGroup) ||
			(xrtTaskGroupWaitFor(pGroup, 3000) !=
				XWAIT_OK) ) {
			xrtTaskGroupDestroy(pGroup);
			(void)xrtNetEngineDestroy(pEngine);
			return 4;
		}
		printf("group-until: done\n");
		xrtTaskGroupDestroy(pGroup);
	}
	return xrtNetEngineDestroy(pEngine) ? 0 : 1;
}

Что только что произошло. (1) xrtTaskNet подаёт немедленную задачу — xrtFutureWaitFor ждёт с семантикой дедлайна главы 9, значение читается через заимствование xrtFutureValue (дисциплина главы 58). (2) After/Until демонстрируют подачу по таймеру — разница только в «относительных миллисекундах» против «абсолютного дедлайна», а вывод доказывает: задача действительно исполнилась после истечения. (3) GroupNetUntil атомарно принимает отложенную задачу в группу — групповая отмена и защита потолком гарантированы зарезервированным слотом. (4) Все четыре сегмента печатают worker=0 — аффинная подача попадает на одного Worker-а, ровно форма «мелкой работы в потоке IO».

Итоги тома: опись активов тома 6

Том 6 закрывается; переучёт. Слой примитивов: четырёхшаговый жизненный цикл потока, четвёрка (мьютекс/условие/квота/rw-блокировка), четыре условия deadlock-а и профилактика порядком блокировок (глава 53). Слой сотрудничества: тройка токена отмены и дерево распространения (53), четыре состояния-три финала и очистный стек корутины (54), тройка планировщика и три формы семейства насоса (55), ёмкость/трёхстороннее закрытие/Select/интеграция отмены Channel (56), четыре состояния/продолжения/комбинаторы/семейство ожидания Future (57), двойная форма/протокол остановки/пакет исполнителя (58). Структурный слой: пятёрка группы задач/перечень финалов/области родитель-потомок/Done Future (59). Ментальные модели: линия отмены, трёхслойный конвейер сборки, общее-против-сообщений, лестница стоимости ожидания, три гранулярности, генеалогия структурного параллелизма. Плюс навык сборки этой главы — этот список и есть фундамент «читать любую современную параллельную библиотеку, проектировать собственную параллельную архитектуру». С тома 7 все эти инструменты входят в бой.

61.4契约

• Три ноля: поток-насос без блокировок, потоки пула без ожидания, пути сообщений без взаимной блокировки — проверяемые метрики правильности сборки.
• Критерий разделения: уступающее — планировщику, держащее — пулу; граница по признаку «уступает ли».
• Двойной канал остановки: Close планировщика (прекратить доводку) + Cancel группы (завершить в полёте) — сотрудничество, не дублирование; порядок и дежурство со сроком — в паре.
• Финал в обратном порядке: насос → шина → пул (обратный зависимостям); у каждого слоя свой протокол остановки (главы 56/57/59).
• Совпадение жизненных циклов: подгруппа соединения = соединение, группа службы = служба — граница области и есть граница ресурсов.
• Пять пунктов наблюдения: глубина очереди, Active группы, счётчики отмен, цикл насоса, троица конечных состояний — порог выкладки.

Путь эволюции: три уровня итераций скелета

Скелет не проектируется одним заходом; у трёх уровней своя отдача. Уровень первый (запускается): насос + корутины + пул, минимальное кольцо тройки — образец один этой главы; проверка трёх нолей, проход потока данных. Уровень второй (останавливается): добавить сигналы, дерево отмены, служебную группу, дежурство со сроком — остановочное кольцо образца два главы 60; проверка последних трёх пунктов списка (ожидания отменяемы/финалы троичны/циклы совпадают). Уровень третий (наблюдаемо): подключить конвейер главы 50 — глубина очереди, Active группы, цикл насоса в структурный журнал; добавить внедрение отказов (глава 6) для проверки путей отказа. Большинство проектов останавливается на уровне «полтора» — «запускается, но не умеет грациозно останавливаться» — самый частый технический долг параллельных систем; второй уровень — порог выкладки, третий — порог эксплуатации. Примерно по неделе на уровень — вложения в скелет окупаются полностью с первой ночной тревогой.

Взгляд на производительность: три точки настройки скелета

Узкие места производительности скелета стоят в трёх предсказуемых местах. Цикл насоса (срок PollFor): длинный — задержка отклика на события, короткий — холостое вращение жжёт CPU — старт с 10 мс, настройка по требованиям задержки (интерактив ~1 мс, пакетный ~50 мс); доля вращения — обратная связь настройки (пункт приёмки упражнения-вызова). Глубина пула (экономика главы 59): CPU-тип = ядра, IO-тип = бюджет конкурентности — решает доля тяжёлых вычислений; после измерения распределения длительностей задач — по формулам главы 59. Ёмкость канала (три числа главы 57): статистическая шина рапортует TrySend-ом с выбросом при заполнении (ёмкость 64 для сглаживания) — деградационная семантика выбрасываемых сообщений записана в проекте; канал запросов — малые значения по потребностям передачи противодавления. Общий принцип трёх точек: сперва на умолчаниях, подключить наблюдение, настроить по данным — параллельный выпуск главы 137 об анализе производительности вернётся с инструментами.

Последнее напоминание: скелет — живой документ

Код скелета (с контрольным списком из пяти пунктов и комментариями решений сборки) должен поддерживаться как живой документ команды: новичок входит через скелет (быстрее, чем десять глав книги — теория вернётся позже); новая служба стартует копией скелета (решения сборки уже проверены); изменение архитектуры сперва в скелете (три ноля зелёные — потом бизнес-код). Симптомы гниения скелета: какой-то пункт списка хронически красен, комментарии решений разошлись с кодом, никто не может изложить протокол остановки какого-то слоя — любой симптом открывает окно рефакторинга. Сопровождаемость параллельной системы не в объёме кода, а в ясности структуры — скелет и есть физический носитель этой структуры. Несите его в том 7.

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

(void)xrtCoGo(gSched, worker, NULL);    /* сперва подали корутину */
gSched = xrtCoSchedCreate();             /* потом построили насос - цель подачи недействительна */

gSched = xrtCoSchedCreate();             /* сперва насос (фундамент зависимостей) */
gPool = xrtTaskPoolCreate(&PoolCfg);     /* затем пристрой */
gStats = xrtChannelCreate(64);           /* затем шина */
(void)xrtCoGo(gSched, worker, NULL);     /* работу подаем последней - зависимости готовы */

xrtCoSchedClose(gSched);
(void)xrtCoSchedRun(gSched);   /* насос осушен */
xrtTaskPoolDestroy(gPool);      /* пул разрушен сразу - статистика в шине застряла навсегда */

xrtCoSchedClose(gSched);
(void)xrtCoSchedRun(gSched);   /* насос осушен (производители остановлены) */
xrtChannelClose(gStats);
(void)xrtCoSchedRun(gSched);   /* шина осушена (потребители завершились) */
(void)xrtTaskPoolClose(gPool);
xrtTaskPoolWait(gPool);        /* пул осушен - все три слоя собраны */
xrtTaskPoolDestroy(gPool);

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

Итоги