Глава 112 xws (часть 3): группы соединений и широковещание

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

Базовая операция push-системы — «отправить одно сообщение всем подключённым» — xwsgroup превращает «множество соединений» в корректный конкурентный контейнер: единственное членство (одно соединение, добавленное дважды, — всё ещё один член), жизненный цикл ссылок (Add берёт ссылку, Remove/Clear/Destroy возвращает — соединения не закрываются и не прерываются), жёсткая ёмкость (когда Limit полон, новый член получает повторяемую ошибку ёмкости), запечатывание (Seal навсегда останавливает приём — стандартное действие сервера, прекращающего сессии), стабильные снимки (Snapshot копирует членов в порядке вступления и наращивает ссылки — выделение вне лок-а, большие снимки не блокируют Add/Remove) и пакетное широковещание (синхронная и асинхронная формы; асинхронная возвращает ожидаемый объект операции — отмена ожидания не разрушает владение данными, уже принятыми каждым соединением). Широковещание не обходит потолки одиночного соединения: медленные соединения пропускаются, ожидаются или закрываются по вашей политике — никогда не затягиваются в неограниченную общую очередь.

112.1引入

«Поддерживать список онлайн-соединений» выглядит как 数组+锁 (массив + лок) — упражнение для начинающих, пока не выстраиваются детали конкурентности: Open- и Close-обратные вызовы одного соединения исполняются конкурентно на разных Worker, Add и Remove соперничают; вступление и выход членов во время обхода широковещания (инвалидация итератора — старая проблема главы 18); медленное соединение держит весь пакет при рассылке (один тащит N); при остановке сервера «сначала не принимать новых, старых закрыть грациозно» (полузакрытое состояние). Каждая строка — будни реальной push-системы.

xwsgroup отвечает построчно: семантика единственного членства поглощает повторные Add; «линеаризовать под лок-ом, выделять вне его» у Snapshot даёт обходу стабильное множество (снимок держит собственные ссылки — даже уничтожение исходной группы ничего не меняет; инвалидация итератора становится невозможной); широковещание — независимый приём каждым соединением (медленное не тащит остальных) + агрегированное ожидание (объект асинхронной операции); Seal даёт полузакрытое состояние. Ценность главы не в числе API (около десятка входов), а в том, что каждая конкурентная семантика продумана — читайте её как «урок проектирования конкурентных контейнеров» на примере WebSocket.

112.2概念

Жизненный цикл и единственное членство

Смысл единственного членства: повторные Open-обратные вызовы (случайность сетевого слоя) не могут протечь двойными ссылками; Remove возвращает ровно ту долю, что взял Add. Семантика ёмкости: когда Limit полон, Add падает с XERR_AGAIN / XWS_GROUP_ERROR_CAPACITY — повторяемо (после удаления члена можно добавить снова), не протокольная ошибка; этот класс ошибок разделяет «перегружено» и «нелегально» (модель категорий главы 4 на слое контейнеров).

Запечатывание: полузакрытое состояние

xrtWsGroupSeal постоянно и идемпотентно: после запечатывания новые члены не входят, но повторное добавление существующего члена всё ещё успешно (идемпотентная семантика продолжается), а remove/очистка/снимок работают как обычно. Стандартный порядок остановки: прекратить приём новых соединений (слой прослушивания) → Seal группы (идущие Open больше не входят) → закрыть или прервать каждую существующую сессию грациозно — «сперва закрыть вход, потом убирать зал».

Снимок: корректный способ стабильного обхода

xrtWsGroupSnapshotCreate: копирует текущих членов в порядке вступления и наращивает ссылку каждому; ключевая инженерная деталь — хранилище непрерывного снимка выделяется вне лок-а группы: сперва без лок-а пробуем выделение; взяв лок снова, если члены переросли ёмкость — освободить и повторить; лишь при достаточной ёмкости наращиваются ссылки по порядку в точке линеаризации. Эффект: работа по выделению большого снимка (способная задеть лок аллокатора) не блокирует горячий путь Add/Remove/Clear. Снимок не держит группу — после уничтожения исходной группы члены снимка остаются действительными (их ссылки живут на снимке); отпустите SnapshotDestroy по завершении. Это ровно решённая форма «правил инвалидации контейнеров» главы 18: нужен стабильный обход — копируйте снимок с наращенными ссылками, а не gamble по исходному контейнеру.

Пакетное широковещание: независимый приём и агрегированное ожидание

Синхронная форма: обход снимка и отправка по соединениям (AGAIN/отказ каждого обрабатываются независимо — пропустить, пометить как медленное, закрыть по политике). Асинхронная форма: входы вроде xrtWsGroupTextAsync возвращают объект операции xwsgroupop — GroupOpWait ждёт завершения всей партии, destroy освобождает; отмена агрегированного ожидания не разрушает владение данными, уже принятыми каждым соединением (пакетная версия «отмена ≠ уничтожение» главы 110). Широковещание не обходит потолки отправки одиночного соединения — очередь медленного соединения ограничена как обычно, и его судьбу решает ваша политика.

Сочетание с тремя состояниями отправки

Нагрузка широковещания естественно ложится на форму ref (глава 111): один статичный блок, ссылки передаются в очереди N соединений — обратный вызов release ведёт счёт (настоящее освобождение только после последней записи). Асинхронные входы группового широковещания внутри — именно пакетное применение этой формы; самописный обход собирает то же самое через xrtWsConnBinaryRef (упражнение главы 111 это делало).

112.3示例

Первая полная программа: каркас жизненного цикла группы

Программа ниже из examples/websocket/group — стандартная форма подключения Add/Remove:

#include <stdio.h>

#include <xws.h>



/* Upgrade 成功后把借用连接加入应用连接组。 */
static void connectionOpen(xwsgroup* pGroup, xwsconn* pConnection)
{
	(void)xrtWsGroupAdd(pGroup, pConnection);
}



/* 唯一 Close 回调中移除连接，归还连接组持有的引用。 */
static void connectionClose(xwsgroup* pGroup, xwsconn* pConnection)
{
	(void)xrtWsGroupRemove(pGroup, pConnection);
}



/* 示例保留回调形状，实际服务器可直接把组放入路由上下文。 */
int main(void)
{
	xwsgroup* pGroup = xrtWsGroupCreate(0);

	if ( pGroup == NULL ) {
		return 1;
	}
	(void)connectionOpen;
	(void)connectionClose;
	printf("WebSocket connection group is ready\n");
	xrtWsGroupDestroy(pGroup);
	return 0;
}

Что только что произошло. (1) xrtWsGroupCreate(0) строит группу без потолка (продакшн передаёт Limit — ёмкость есть линия перегрузки). (2) Две функции — шаблон подключения: connectionOpen (в обратном вызове успешного обновления GroupAdd — соединение входит), connectionClose (в единственном Close-обратном вызове GroupRemove — возврат ссылки). Реальный сервер кладёт группу в контекст маршрутизации (глава 113 — здесь Data-поле на своём месте) — однострочное подключение Open/Close, и вся конкурентная семантика группы наследуется. (3) main образца лишь проверяет готовность (шаблонная форма); ссылки (void) в connectionOpen/Close — доказательство на этапе компиляции «формы обратных вызовов сохранены».

Вторая полная программа: ожидаемое широковещание пустой группы

Вторая программа из examples/websocket/group_future — полный жизненный цикл объекта асинхронной операции:

#include <stdio.h>

#include <xws.h>



/* 展示空组广播也会返回可等待、可检查的完整操作对象。 */
int main(void)
{
	xwsgroup* pGroup = xrtWsGroupCreate(0);
	xwsgroupop* pOperation;

	if ( pGroup == NULL ) {
		return 1;
	}
	pOperation = xrtWsGroupTextAsync(
		pGroup,
		XRT_STR_LITERAL("service-ready")
	);
	if ( (pOperation == NULL) ||
		(xrtWsGroupOpWait(pOperation) != XWAIT_OK) ) {
		xrtWsGroupOpDestroy(pOperation);
		xrtWsGroupDestroy(pGroup);
		return 1;
	}
	printf(
		"members=%zu accepted=%zu rejected=%zu\n",
		xrtWsGroupOpCount(pOperation),
		xrtWsGroupOpAccepted(pOperation),
		xrtWsGroupOpRejected(pOperation)
	);
	xrtWsGroupOpDestroy(pOperation);
	xrtWsGroupDestroy(pGroup);
	return 0;
}

Что только что произошло. (1) xrtWsGroupTextAsync(组, 消息) (группа, сообщение) возвращает полный объект операции даже для пустой группы — заголовочный комментарий прямо формулирует замысел: «широковещание в пустую группу тоже возвращает полный ожидаемый, проверяемый объект операции» — пустое множество не ошибка; агрегатная семантика самосогласована (ноль соединений — все успешны). (2) GroupOpWait ждёт всю партию (XWAIT_OK), GroupOpDestroy освобождает объект операции — трёхшаговый жизненный цикл асинхронного широковещания (запуск/ожидание/освобождение). (3) Прочтите оба образца вместе: после входа Open в группу TextAsync рассылает — каждый член принимает независимо; когда ожидание возвращается, вся партия имеет вердикты. Политика медленных соединений (пропустить/закрыть) живёт в конфигурации и вашем коде обхода — группа не решает за вас.

112.4契约

• Единственное членство: повторный Add того же указателя успешен без удвоения; Remove/Clear/Destroy возвращают ровно те ссылки, что взял Add; операции группы не закрывают и не прерывают соединения.
• Ёмкость: Limit ноль = без потолка; заполненный Add возвращает XERR_AGAIN/CAPACITY (повторяемо, не протокольная ошибка).
• Запечатывание: Seal постоянно и идемпотентно; после — новых нет, повторное добавление старых успешно; remove/очистка/снимок как обычно — при остановке сперва Seal, потом уборка.
• Снимок: выделение вне лок-а + линеаризованное наращивание внутри; порядок вступления сохранён; группу не держит (после уничтожения группы члены действительны); Destroy по завершении.
• Запросы: Has/Count/Limit/Sealed — запертые мгновенные снимки — не замена стабильному обходу (берите Snapshot).
• Независимость широковещания: приём по соединениям; потолки одиночного соединения не обходятся; политика медленных (пропустить/ждать/закрыть) — вызывающему.
• Объекты асинхронных операций: семейство TextAsync возвращает ожидаемые проверяемые объекты; отмена ожидания не разрушает принятого владения.
• Дисциплина лок-а: Clear меняет множество на пустое под лок-ом и возвращает старые ссылки вне его — разрушение и работа аллокатора не занимают лок группы.
• Opaque-проверки: группа/снимок/операция — opaque; перед разыменованием проверяются фиксированный заголовок и непрерывное хранилище; заворот отвергается как ошибка параметра; Destroy(NULL) — пустая операция.
• Обрезка: WEBSOCKET_GROUP/_GROUP_FUTURE независимы; группа зависит от соединений и set/mutex — кадровый слой ничего из этого не тянет.

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

size_t n = xrtWsGroupCount(pGroup);
for ( size_t i = 0; i < n; i++ ) {
	/* взять i-го члена - Remove мог уже изменить множество */
}

xwsgroupsnapshot* pSnap = xrtWsGroupSnapshotCreate(pGroup);
for ( size_t i = 0; i < xrtWsGroupSnapshotCount(pSnap); i++ ) {
	xwsconn* pConn = xrtWsGroupSnapshotGet(pSnap, i);
	send_or_skip(pConn);   /* стабильное множество: входы/выходы во время обхода этот снимок не трогают */
}
xrtWsGroupSnapshotDestroy(pSnap);

xrtWsGroupRemove(pGroup, pConn);
/* показалось, что соединение закрыто - противоположная сторона онлайн */

xrtWsGroupRemove(pGroup, pConn);       /* сперва выйти из группы */
if ( !xrtWsConnClose(pConn, XWS_CLOSE_NORMAL,
		XRT_STR_LITERAL("kicked")) ) {
	xrtWsConnAbort(pConn);              /* потом закрыть по политике */
}

stop_listener();
xrtWsGroupDestroy(pGroup);   /* Add идущих Open ещё не закончились */

stop_listener();             /* 1. не принимать новые соединения */
xrtWsGroupSeal(pGroup);      /* 2. запечатать: Add идущих Open отвергаются */
drain_workers();             /* 3. дождаться молчания обратных вызовов (Open/Close) */
close_all_members(pGroup);   /* 4. грациозно закрыть существующие сессии */
xrtWsGroupDestroy(pGroup);   /* 5. уничтожить группу последней */

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

Итоги