Глава 57 Channel и Select

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

Channel — самый высокочастотный инструмент бизнес-слоя параллельной системы: типобезопасная передача сообщений (семантика ptr), точная ёмкость с противодавлением (заполнен — жди; ёмкость 0 — рандевуная синхронная передача), семантика закрытия в стиле Go (после Close отправка ошибается, приём осушает остаток и только потом завершается — гарантия остановки «ни данные не потеряны, ни отправки не повисли»), мультиплексное ожидание Select (одновременное ожидание нескольких каналов и таймеров — кто раньше, тот и выиграл) и отменяемый приём (RecvCancel — двух-в-одном ожидание с токеном главы 54). Безблокировочная очередь главы 21 — её транспортная основа; Channel добавляет поверх протокол закрытия и семантику ожидания.

57.1引入

Производитель-потребитель, собранный вручную из Mutex+Condition главы 53 (условное ожидание + взаимоисключающая защита + согласование пробуждений), — тридцать строк и легко ошибиться (забытый Signal, ложные пробуждения, границы буфера); безблокировочная очередь главы 21 решает транспорт (без блокировок, высокая пропускная способность), но не имеет закрытия — откуда потребитель знает, что «больше не будет»? А если при остановке производитель ещё шлёт?

Channel упаковывает две вещи: транспорт (очередь с ёмкостью — заполнена, Send ждёт; пуста, Recv ждёт) и протокол (правила после Close: отправитель получает явную ошибку; приёмник, осушив остаток, получает явный сигнал конца). Семантика закрытия — краеугольный камень остановки системы сообщений: «отправок больше не будет» это событие, а не угадывание по тайм-ауту. Select добавляет мультиплексное ожидание: одна корутина/поток ждёт одновременно два канала (любой из них) или с таймером (тайм-аут первым) — базовая поза цикла событий.

57.2概念

Тройка канала: ёмкость, отправка, приём

Семантика ёмкости: 0 — рандеву, отправка ждёт прибытия приёмника, переправа один на один (стандартная форма синхронной передачи); N — буферная, не более N в запасе, при заполнении Send ждёт (противодавление встроено). Назначение TrySend: щуп на неблокирующих путях (цикл событий блокироваться не смеет) — при заполнении ложь, и вызывающий решает, выбросить или повторить позже (это не ожидание).

Семантика закрытия: трёхстороннее правило в стиле Go

Семантика осушения (форма цикла образца channel) — ключ остановки: Close не «очистка» — уже поставленные в очередь данные потребляются как обычно; «закрытие» лишь объявляет будущее закрытым. Это гарантирует «ни данные не потеряны, ни отправки не повисли» — редакция сообщения у раздела Close очереди главы 21 (протокол остановки главы 48).

Select: мультиплексное ожидание

xrtChannelSelect (демонстрация в образце select): одновременно регистрируется намерение приёма на нескольких каналах плюс опциональный таймер — кто раньше прибыл, тот и вернулся (с отметкой, какой канал сработал). Форма: массив ожиданий + тайм-аут → возврат готового пункта. Справедливость: при одновременной готовности нескольких — порядок регистрации или политика реализации — случайной справедливости не обещается (взвешенная справедливость достигается ротацией порядка регистрации на бизнес-слое). Select — стандартная поза «один потребитель обслуживает много источников» и «ожидание с тайм-аутом» — хребет циклов событий и контроля тайм-аутов.

Слияние с токеном отмены

xrtChannelRecvCancel (образец channel_cancel): приём и отмена в одном — экземпляр отменяемого ожидания главы 54 на Channel. Складывается в трёхстороннее ожидание: данные пришли / канал закрыт (сигнал конца) / отмена поразила — возврат xwaitresult сообщает единообразно. У отправки тоже отменяемый вариант (при остановке застрявший на заполненном канале отправитель тоже разбуживается отменой). Все формы ожидания Channel (Send/Recv/Select) конфигурируются токеном — дерево отмены проходит через каналы насквозь.

Channel против очереди: что когда

Разделение безблокировочной очереди главы 21 и Channel этой главы: очередь — примитив транспорта, максимум пропускной способности (без блокировок), пакетные интерфейсы, специализация по топологиям SPSC/MPSC/MPMC; Channel — примитив протокола: семантика закрытия, мультиплексный Select, интеграция отмены, близость планировщику (ожидание корутины не занимает поток — дивиденд главы 56). Выбор: горячий путь данных (сетевой приём-передача, раздача задач) — очередь; протокольная плоскость сотрудничества (запрос-ответ, диспетчеризация событий, осушение при остановке) — Channel. Комбинируются: Channel внутри использует очередь как основу (слой реализации), внешние интерфейсы каждый при своём деле.

57.3示例

Полная программа: ёмкость, TrySend и осушение после закрытия

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

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





















 */


/* 演示精确容量、非阻塞发送和关闭后排空。 */
int main(void)
{
	xchannel* pChannel = xrtChannelCreate(2u);
	ptr pItem = NULL;

	if ( pChannel == NULL ) {
		return 1;
	}
	if (
		(xrtChannelTrySend(pChannel, (ptr)(uintptr_t)10u) != XCHANNEL_OK) ||
		(xrtChannelTrySend(pChannel, (ptr)(uintptr_t)20u) != XCHANNEL_OK)
	) {
		(void)xrtChannelDestroy(pChannel);
		return 2;
	}
	xrtChannelClose(pChannel);
	while ( xrtChannelRecv(pChannel, &pItem) == XWAIT_OK ) {
		printf("%llu\n", (unsigned long long)(uintptr_t)pItem);
	}
	return xrtChannelDestroy(pChannel) ? 0 : 3;
}

Что только что произошло. (1) xrtChannelCreate(2) — точная ёмкость 2: оба xrtChannelTrySend (10 и 20) успешно встают в очередь (при незаполненности истина); третья дала бы ложь (заполнено — образец третью не шлёт). (2) xrtChannelClose закрывает — остаток на месте: два Recv исправно забирают 10 и 20 (осушение); (3) осушив остаток, Recv возвращает сигнал конца (не OK) — цикл приёма завершается естественно. (4) Это полная эмпирика «трёхстороннего правила закрытия»: отправки остановились (после Close), остаток ушёл (10/20 потреблены), конец явен (сигнал после осушения). Образец channel_select демонстрирует Select: два канала (быстрый и медленный) ожидаются одновременно — побеждает быстрый (20 приходит с той дороги, что раньше).

Полная программа: мультиплексное ожидание Select

Из examples/concurrency/channel_select/main.c:

#include <stdio.h>

#include <xrt.h>



















 */


/* 从两个 Channel 中接收先到达的消息。 */
int main(void)
{
	xchannel tFirst;
	xchannel tSecond;
	xchannelcase arrCase[2];
	xchannelselectresult tResult;
	ptr pFirst = NULL;
	ptr pSecond = NULL;
	int iExit = 0;

	if (
		!xrtChannelInit(&tFirst, 1u) ||
		!xrtChannelInit(&tSecond, 1u)
	) {
		return 1;
	}
	if (
		xrtChannelTrySend(
			&tSecond,
			(ptr)(uintptr_t)20u
		) != XCHANNEL_OK
	) {
		iExit = 2;
	} else {
		arrCase[0] = xrtChannelCaseRecv(&tFirst, &pFirst);
		arrCase[1] = xrtChannelCaseRecv(&tSecond, &pSecond);
		tResult = xrtChannelSelect(arrCase, 2u);
		if (
			(tResult.Wait != XWAIT_OK) ||
			(tResult.Index != 1u)
		) {
			iExit = 3;
		} else {
			printf("%llu\n", (unsigned long long)(uintptr_t)pSecond);
		}
	}
	(void)xrtChannelUnit(&tFirst);
	(void)xrtChannelUnit(&tSecond);
	return iExit;
}

Что только что произошло. (1) На двух каналах по сообщению, но время прихода разное — Select регистрирует намерение приёма на обеих дорогах; прибывшая первой (та, где 20) побеждает и принимается. (2) Вывод 20 доказывает вердикт Select: из многих — первый пришедший обслуживается — это не опрос двух дорог (это занятое ожидание), а одно ожидание нескольких событий. (3) Данные второй дороги на месте (не потреблены — последующий Select или Recv возьмут); Select не потребляет неготовые дороги. (4) Вставьте эту форму в цикл событий: много источников (таймеры, уведомления IO, командный канал) на один Select — стандартный хребет цикла событий. Образцы channel_cancel и deadline закрывают третье измерение: слияние отмены и тайм-аута в одном ожидании.

57.4契约

• Ёмкость — противодавление: заполнено — Send ждёт; ёмкость 0 — рандеву (точка синхронной передачи); TrySend — неблокирующий щуп, не ожидание.
• Трёхстороннее закрытие: отправитель ошибается / приёмник после осушения получает сигнал конца / Close идемпотентен — стандартный протокол остановочного осушения.
• Select — первый пришедший: при одновременной готовности — политика реализации; неготовые дороги не потребляются — справедливость ротацией на бизнес-слое.
• Интеграция отмены: все формы Send/Recv/Select конфигурируются токеном — трёхстороннее ожидание (данные/закрытие/отмена) единообразно возвращает xwaitresult.
• Принадлежность сообщений: передача ptr — отправитель строит, приёмник обрабатывает; объекты со счётом ссылок передаются напрямую (дисциплина ссылок значений главы 31).
• Разделение с очередями: горячий путь данных — очереди главы 21; протокольная плоскость сотрудничества — Channel; комбинировать, не путать.

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

Раздача задач (форма обработки запросов): корутина-приёмник RecvCancel-ит запросы → пул корутин-обработчиков потребляет → канал результатов возвращает — Channel сшивает конвейер, ёмкость и есть потолок конкурентности (противодавление), закрытие и есть остановочное осушение (сообщений редакция пятисекундной конвенции главы 49). Шина событий: несколько источников (сигналы, таймеры, уведомления IO) каждое своей корутиной производят события в один Channel — один потребительский Select обслуживает все источники (веерный вход); центральный узел, диспетчеризирующий по типам событий. Почтовый ящик Actor: по командному каналу на актора (ёмкость маленькая и определённая), актор в цикле Recv обрабатывает — естественная сериализация (внутри актора без блокировок); закрытие ящика — выход актора на пенсию. Три хозяина делят одну ментальную модель: Channel — это «вызов метода» параллельного мира — сообщения вместо общей памяти, ожидание вместо блокировок, закрытие вместо флагов жизненного цикла (максима школы CSP, приземлённая в XRT).

CSP и общая память: два стиля параллелизма

Школа за Channel (CSP — Communicating Sequential Processes) заслуживает формального сопоставления со стилем общей памяти главы 53. Школа общей памяти: данные в разделяемых структурах, доступ под блокировками — прямо и производительно, но правильность блокировок держится целиком на дисциплине (порядок, гранулярность, deadlock-и). Школа сообщений: данные не разделяются, принадлежность передаётся через каналы — мировоззрение без блокировок (данные в единоличном владении), ясная компонуемость (канал как интерфейс), но копирование/передача стоят задержки. Инженерная реальность — смесь: внутри структур данных — блокировки или безблокировочные примитивы (очереди главы 21), между модулями — сообщения (Channel) — «внутри компактно, снаружи ясно». XRT даёт оба в полном объёме (примитивы главы 53 + Channel этой главы); выбор определяется слоем, а не верой.

Взгляд на дизайн: три числа ёмкости

Ёмкость канала — не одно число, а проектное решение из трёх. 0: рандеву — синхронная передача, требующая «другая сторона подтвердила получение» (семантика RPC); малое определённое значение (1~16): противодавление передаётся быстро — верхний поток почти сразу чувствует медленный нижний (малая ёмкость и есть сигнальная линия, перегрузка видна рано); большое значение или выделенная очередь: сглаживание пиков — буферизация всплесков (но с наблюдением: кривая глубины очереди видна до выхода из-под контроля). Три числа — три обещания: синхронность, раннее противодавление, сглаживание. Отправные точки: каналы сотрудничества — малые (8); конвейеры запросов — по задержке нижнего потока и бюджету памяти; «безлимит» — никогда (дисциплина клапана главы 21 действует и здесь — потолок подачи CreateLimit есть сестринский механизм на стороне планировщика).

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

xrtChannelClose(pChannel);
/* приёмник решил, что закрыто значит пусто - вышел напрямую, остаток потерян */

xrtChannelClose(pChannel);
/* цикл осушения: остаток продолжает приходить до сигнала конца */
while ( xrtChannelRecv(pChannel, &Item) == XWAIT_OK ) {
	Consume(Item);   /* во время остатка OK; после осушения - сигнал конца */
}
/* выход по сигналу конца - потерь ноль (форма образца) */

xchannel* Ch = xrtChannelCreate(0);   /* рандеву */
xrtChannelSend(Ch, Item);              /* ждёт приёмника - ожидание "отправил и пошёл" рушится */
DoNextThing();                          /* тайминговая зависимость не случилась */

xchannel* Ch = xrtChannelCreate(8);  /* буферная: запас в 8 */
xrtChannelSend(Ch, Item);             /* встал в очередь и вернулся (не заполнен) - отправил и пошёл */
DoNextThing();
/* синхронная передача (продолжить только после подтверждения получения) - только для ёмкости 0: ёмкость выбирается по семантике */

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

Итоги