Глава 56 Корутины (часть 2): планировщик и таймеры

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

Средняя глава о корутинах повышает ручные три такта первой (Create/Resume/чтение финала) до опеки планировщика: xrtCoSchedCreate строит планировщик, xrtCoGo (или xrtCoSchedPost подачи функции) запускает корутину, xrtCoSleep — таймированная уступка внутри корутины, xrtCoSchedRun ведёт насос (до завершения всех корутин) с PollFor/PollUntil — шаговое ожидание событий; таймеры едины с планировщиком (уступку Sleep будит таймерное колесо); один экземпляр планировщика на поток — стандартная модель: привязка корутин к потоку планировщиком выполняется сама собой. После средней главы бизнес-код почти не касается примитивов первой.

56.1引入

Зародыш планировщика из первой главы (три задачи вперемешку) оставил вопрос: кто зовёт Resume? Если ответ «главный цикл вручную», то «какая корутина первой, кто будит спящую секунды, как новые корутины вклиниваются» — всё придётся писать самому, — и этот «код политики доводки» и есть планировщик. Три его ключевых решения: очередь готовности (Post ставит, Run забирает и доводит — FIFO или по политике); таймерное колесо (спящая корутина висит на таймерной структуре, по истечении возвращается в очередь — ответ на «кто будит»); насосный цикл (RunUntil доводит с лимитом — ответ на «когда остановиться»: возврат при пустой очереди без таймеров или по истечении срока).

Стандартная форма, собранная из этих механизмов, — планировщик на поток + насосный цикл: главный цикл потока это while ( running ) SchedRunUntil(sched, deadline) — поданные корутины доводятся, Sleep-ы просыпают по времени, внешние события (обратные вызовы завершения IO) подают Post-ом новые корутины. Цикл событий сетевого движка тома 7 — индустриализированная форма этого же: событие IO → пробуждение корутины соединения → насосный цикл продолжается.

56.2概念

Ядро планировщика: тройка

xrtCoGo — стандартный вход: создание и постановка в очередь за один шаг (в большинстве сцен не нужно сперва Create, потом подавать); xrtCoSchedPost подаёт обычную функцию (сигнатура xcoschedpostproc — лёгкое событие «подали — исполнилось»); Post несёт потолок противодавления (подача при заполненном CreateLimit отказывает — планировочная редакция клапана из главы 21). Семантические пункты Sleep: вызов внутри корутины (в обычном потоке не определён); уступка в таймерное колесо, по истечении будит насос — поток продолжает доводить других.Пара Park/Wake: Park подвешивает текущую корутину до Wake (пробуждение из внешнего потока безопасно), отмены или срока — стандартный канал, каким внешние события (завершение IO) будят корутины.

Насосный цикл: Run и RunUntil

xrtCoSchedRun доводит до завершения всех корутин (форма дежурства пакетных задач); xrtCoSchedStep — одиночный шаг (доводит максимум одну готовую корутину — форма встраивания в существующий цикл событий); PollFor/PollUntil в пределах срока ждут события и шагают (внешний Post будит немедленно — стандартный такт резидентного насоса: while(running) PollFor(sched, 10ms)). xrtCoSchedAlive сообщает число живых корутин (метрика мониторинга), xrtCoSchedClose запрашивает отмену всем и прекращает приём (вход остановки планировщика — верхняя кнопка дерева отмены главы 54). Две хозяйственные формы насосного цикла: резидентный поток — цикл while(running) PollFor(sched, 10ms), внешние события будят через Post (форма сетевого движка); встроенная — внешний цикл событий сам зовёт Step/PollFor (простаивающие срезы времени GUI-каркаса — планировщик встроен в готовый цикл).

Семейство таймеров

Таймерные способности представлены тремя семействами: xrtCoSleep (сон внутри корутины), xrtCoParkFor/ParkUntil (подвеска до срока или Wake — внешнее событие и таймер, кто первый), отложенное исполнение подачи (функция Post исполняется на следующем такте насоса — естественная семантика «следующего кадра»). Все идут через один насос — таймерные пробуждения и восстановления корутин сериализуются в одном потоке (дивиденд безблокировочности однопоточного планировщика).

По одному на поток: модель и резоны

Стандартная модель — один экземпляр планировщика на поток. Резоны трёхслойны: привязка корутины к потоку (железная дисциплина первой главы) выполняется автоматически правилом «корутины бегут только в планировщике своего потока»; планировщик внутри одного потока без блокировок (очередь и таймерное колесо трогает только свой поток — крупнейшая зона исключений из главы 53 «когда блокировки не нужны»); много-поточное расширение = несколько независимых планировщиков + Post через потоки (подача потокобезопасна — внешние события из любого потока Post-ят в планировщик правильного потока). Это и «work-stealing»-многопоточность исполнителя/пула задач главы 59 — две модели, каждая со своей ареной (сопоставление в следующем разделе).

Планировщик против пула задач: выбор из двух моделей

Мнемоника выбора: ждём много — планировщик (подвеска дёшева), считаем много — пул задач (пропускная способность прежде); смешанная нагрузка употребляет обе (CPU-задача в пуле по завершении Post-ит обратно в корутину планировщика — межмодельная форма продолжения главы 58).

56.3示例

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

Из образца репозитория examples/concurrency/coroutine_scheduler/main.c:

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



/* 示例任务异步等待后打印编号。 */
static ptr exampleTask(ptr pData)
{
	int iTask = (int)(intptr_t)pData;

	(void)xrtCoSleep((uint64)iTask * 1000u);
	printf("task %d\n", iTask);
	return pData;
}



/* 调度器 post 在所属线程中创建真正的协程任务。 */
static void examplePost(xcosched* pSched, ptr pData)
{
	(void)xrtCoGo(pSched, exampleTask, pData, NULL);
}





















 */


/* 先投递多个轻量任务，再运行到投递和协程全部完成。 */
int main(void)
{
	xcosched* pSched = xrtCoSchedCreate();

	if ( pSched == NULL ) {
		return 1;
	}
	for ( int i = 1; i <= 3; i++ ) {
		if ( !xrtCoSchedPost(
			pSched,
			examplePost,
			(ptr)(intptr_t)i
		) ) {
			return 2;
		}
	}
	if ( !xrtCoSchedRun(pSched) ) {
		return 3;
	}
	(void)xrtCoSchedDestroy(pSched);
	(void)xrtCoThreadDetach();
	return 0;
}

Что только что произошло. (1) Три корутины xrtCoGo в планировщик — созданы и поставлены. (2) Насос (Run или шаги PollFor) доводит: каждую до точки приостановки, ротация, доводка до конца — порядок вывода task 1/2/3 решают вместе очередь и точки уступок (полный разбор политики — в образце tour). (3) После завершения всех xrtCoSchedRun возвращается; обнуление Alive утверждаемо; финалы трёх корутин проверяемы по одной. (4) Контраст с «ручными тремя тактами» первой главы: здесь нет явных Create/Resume — отделение запуска от доводки и есть вся абстракция планировщика.

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

Из examples/concurrency/coroutine_event/main.c — события корутин и таймерные пробуждения:

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



/* 生产者完成异步工作后置位事件。 */
static ptr producer(ptr pData)
{
	xcoevent* pEvent = (xcoevent*)pData;

	if ( xrtCoSleep(1) != XWAIT_OK ) {
		return NULL;
	}
	return xrtCoEventSet(pEvent) ? pEvent : NULL;
}



/* 消费者不阻塞调度线程地等待生产者。 */
static ptr consumer(ptr pData)
{
	xcoevent* pEvent = (xcoevent*)pData;

	if ( xrtCoEventAwait(pEvent) != XWAIT_OK ) {
		return NULL;
	}
	printf("event received\n");
	return pEvent;
}



















 */


/* 展示同一调度器中生产者和消费者的事件同步。 */
int main(void)
{
	xcoevent tEvent;
	xcosched* pSched;
	xcoro* pConsumer;
	bool bOkay;

	if ( !xrtCoEventInit(&tEvent, false, false) ) {
		return 1;
	}
	pSched = xrtCoSchedCreate();
	if ( pSched == NULL ) {
		(void)xrtCoEventUnit(&tEvent);
		return 1;
	}
	pConsumer = xrtCoSpawn(pSched, consumer, &tEvent, NULL);
	bOkay =
		(pConsumer != NULL) &&
		xrtCoGo(pSched, producer, &tEvent, NULL) &&
		xrtCoSchedRun(pSched) &&
		(xrtCoResult(pConsumer) == &tEvent) &&
		xrtCoDestroy(pConsumer);
	(void)xrtCoSchedDestroy(pSched);
	(void)xrtCoEventUnit(&tEvent);
	(void)xrtCoThreadDetach();
	return bOkay ? 0 : 1;
}

Что только что произошло. (1) Корутина подвешена xrtCoPark-ом (ждёт внешнего пробуждения); по истечении таймера или по вызову Wake она возвращается в очередь готовности — путь таймерного/событийного пробуждения пройден целиком (возврат Park различает естественное пробуждение/отмену/срок). (2) Пока шло ожидание, насосный цикл доводил других корутин (или возвращался пустым) — «конкурентное ожидание» в одном потоке и есть комбинация уступки и пробуждения. (3) Таймерные пробуждения и восстановления корутин сериализуются в одном потоке — передача данных между корутинами без блокировок (эмпирическое доказательство дивиденда однопоточного планирования; через потоки безопасны только два входа — Post и Wake). Образец coroutine_tour оставлен чтением полного интерфейса: конфигурация планировщика, варианты политик, статистика — по группе утверждений на каждую.

56.4契约

• Отделение запуска от доводки: Go/Post только ставят в очередь, доводит только насос — подача и Wake потокобезопасны (межпоточно легальны), доводка принадлежит исключительно потоку планировщика.
• Семантика Sleep/Park: вызов внутри корутин (в обычном потоке не определён); Sleep на таймерном колесе, Park ждёт Wake/отмены/срока — поток не спит ни в том, ни в другом случае.
• Формы насоса: Run до полного завершения; PollFor/Until — ожидание событий со сроком и шаг (такт резидентного цикла); Step — одиночный шаг встраивания.
• Безблокировочный дивиденд: внутри однопоточного планировщика очереди/таймерное колесо/обратные вызовы без блокировок — синхронизация нужна только межэкземплярному сотрудничеству (через Post или очереди главы 21).
• Выбор модели: ждём много (IO-плотно) — планировщик, считаем много (CPU-плотно) — пул задач; смешанная нагрузка — обе модели в ансамбле.

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

Сетевой движок (арена тома 7): резидентный поток с насосом PollFor(10ms) + обратные вызовы завершения IO Wake-ят корутины соединений — по корутине на соединение, прямолинейная обработка чтения-записи; событийный слой движка (с главы 63) делит поток с насосом планировщика — событийная доводка и восстановление корутин в одном потоке, ноль блокировок. Оркестровка асинхронного бизнеса: многошаговые процессы (кеш→база→отложенная запись) на каждом шаге Park ждут внешнего события (обратный вызов DB будит Wake-ом) — прямой код, асинхронное исполнение; против цепочки Future: у корутин стек и локальные переменные, состояние не переезжает в замыкания. Модель Actor: по корутине планировщика на актора, сообщения подаются Post-ом (подача — «отправка письма») — одиночная корутина гарантирует отсутствие блокировок внутри актора. Общая форма трёх хозяев: насос — сердце, Post/Wake — артерии, корутины — клетки — понимая это кровообращение, понимаешь все употребления планировщика.

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

Остановка планировщика — место встречи главы 49 о сигналах и главы 54 об отмене в мире корутин. Полный путь: внешний триггер (сигнал или управленческая команда) → xrtCoSchedClose (запрос отмены всем корутинам и прекращение приёма — большая кнопка дерева отмены главы 54) → насос продолжает доводить до обнуления Alive (каждая корутина ощущает отмену в своей точке Park/Sleep/уступки и завершает через очистный стек) → xrtCoSchedDestroy разрушает. Три дисциплины: после Close насос продолжает работать (отмена кооперативная — нет насоса, некому завершать); Alive в ноль прежде Destroy (тот же корень, что «Destroy успешен только завершённой» первой главы); дежурство со сроком (насос ждёт завершения с deadline — по истечении отказ от грации, запись числа незавершённых и принудительный выход — редакция пятисекундной конвенции главы 49).

Напоминание о наблюдении: метрики здоровья планировщика

Планировщику нужно собственное наблюдение (пост конвейера главы 50 в параллельном мире): Alive (число живых корутин — мера стационарного потолка и дрейфа утечек), глубина очереди Post (уровень противодавления — число переполнений CreateLimit есть сигнал перегрузки), длительность цикла насоса (фактическое время одного такта PollFor — растягивание циклов значит, что какая-то корутина делает тяжёлую работу до уступки — рантаймный отпечаток ловушки 1), задержка пробуждения Sleep/Park (дрейф таймерной точности — свидетельство, что ритм насоса тащат). Четыре метрики в структурный журнал (поля главы 38), и поведение планировщика превращается из чёрного ящика в кривые — при параллельном «кажется не так» сперва кривые, потом догадки о коде.

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

static ptr worker(ptr pData)
{
	BlockingRead(Fd, Buf, Size);   /* блокирующее IO - не Yield! */
	return NULL;                    /* насос заклинен: все остальные ждут */
}

static ptr worker(ptr pData)
{
	/* блокирующую операцию вынести из планировщика: подать в пул задач (глава 59), по завершении Post назад */
	xrtTaskSubmit(gPool, blockingTask, pData, NULL);   /* CPU/IO-задача в пул */
	xrtCoPark();   /* подвеска - обратный вызов пула разбудит Wake-ом */
	/* обратный вызов пула подаёт Post-ом новую корутину обратно в этот планировщик (форма продолжения главы 58) */
	return NULL;
}

/* поток B */
/* прямая манипуляция внутренностями планировщика - гонка с насосным циклом, не определено */

/* поток B */
xrtCoSchedPost(gSched, shutdownProc, NULL);   /* Post «функции остановки» */
/* остановочная корутина исполняет очистку в потоке планировщика - правильный поток делает правильное дело */

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

Итоги