Глава 111 xws (часть 2): путь отправки — writer, ссылки и сжатие

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

Глава 110 утвердила три состояния владения отправкой (copy/ref/take); эта глава проходит путь отправки до конца. writer (xwswriter): потоковая фрагментация длинных сообщений — xrtWsConnBeginText открывает, WriterWrite идёт кусками, WriterFinish завершает; заголовок и состояние сжатия writer имеют стабильные размеры (содержимое не меняет раскладку сетевого объекта); сжатая отправка: writer сжатия BeginTextCompressed (форма permessage-deflate, согласованного в главе 96, на отправляющей стороне — при включённом сжатии действует та же семантика противодавления и владения); матрица владения: copy/ref/take × text/binary × синхронно/Future — полная комбинация; «девять видов отправки» строки 3 обхода и есть эта матрица. Отказы отправки соблюдают атомарное владение: непринятое остаётся у вызывающего, после приёма освобождает соединение.

111.1引入

Отправить снимок журнала на 10 МБ: путь copy сперва копирует 10 МБ (память ×2, потом очередь); путь ref требует «один непрерывный блок памяти» (снимок и так один — подходит); но что если снимок генерируется потоково (ещё не накоплен)? Путь writer: генерируешь и тут же WriterWrite — каждый кусок идёт прямо в поток кадров, память постоянна, фрагментация автоматична (опкод первого кадра, continuation, FIN — автоматизация правил фрагментации главы 95 на отправке). Шаг дальше: снимок — сильно повторяющийся текст, сжатие экономит 90% — BeginTextCompressed открывает writer сжатия, интерфейс производителя не меняется (всё те же Write/Finish), сжатие происходит внутри writer — «потоковость» и «сжатие» комбинируются ортогонально.

Инженерные детали writer на высоте: «заголовок и состояние сжатия стабильны по размеру; hover, события или динамическое содержимое не меняют раскладку сетевого объекта» — форма памяти сетевого объекта предсказуема (экономика Worker главы 64, продлённая на слой раскладки); атомарность пути отказа (запись упала на середине — принятые куски освобождает соединение, непринятые ваши) позволяет восстанавливаться после ошибек без угадывания состояния.

111.2概念

Жизненный цикл writer

Три входа соответствуют типу сообщения и сжатию: BeginText/BeginBinary (обычные), BeginTextCompressed/BeginBinaryCompressed (сжатые — требуют, чтобы соединение согласовало permessage-deflate). Границы кусков не связаны с границами кадров: вы пишете Write в ритме бизнеса, writer кадрует по протокольной политике — большие куски режутся автоматически, маленькие могут склеиваться; FIN появляется только на Finish (ровно один финал на сообщение).

Writer сжатия: та же семантика

При включённом сжатии writer несёт внутри deflater (интеграция xwsdeflater главы 97 в соединение): каждый кусок сжимается в поток кадров, RSV1 помечает только первый кадр, no-context-takeover сбрасывает словарь на границах сообщений по согласованию — интерфейс Write/Finish для вызывающего полностью идентичен обычному writer. «При включённом сжатии держится та же семантика противодавления и владения» — AGAIN всё тот же AGAIN, освобождение всё то же ровно-один-раз — поведение объектов сжатия главы 97, без потерь перенесённое на путь отправки соединения.

Матрица владения и атомарное правило

Девять синхронных отправок (обход) = три состояния владения × {text, binary, ping/управление}; формы Future добавляют измерение способа ожидания. Правило выбора: короткие сообщения copy (цена копии пренебрежима); статические/разделяемые большие блоки ref (одно выделение, много отправок — примитив широковещательной рассылки главы 112); уже выделенная передача take (экономия последней копии).

Writer и цепочки буферов

Формы take/buffer могут передать соединению цепочку сетевых буферов XRT (глава 65) напрямую — сгенерированная цепочка кусков попадает в очередь отправки без «сплющивающего копирования»; заголовок кадрирования writer собирается с головой цепочки одной Vec-отправкой (глава 68). Нулевое копирование пути отправки распространяется с «ссылок» на «цепочки».

111.3示例

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

Программа ниже из examples/websocket/writer — шаблон writer для многочастных сообщений:

#include <stdio.h>

#include <xws.h>



/* 在 Connection 所属 Worker 上把多个业务分块写成一条 Text 消息。 */
static xnetresult sendChunks(
	xwsconn* pConnection,
	const xstrview* pChunks,
	size_t iCount
)
{
	xwswriter* pWriter;
	xnetresult Result;

	if ( (pChunks == NULL) || (iCount == 0) ) {
		return XNET_RESULT_ERROR;
	}
	pWriter = xrtWsConnBeginText(pConnection);
	if ( pWriter == NULL ) {
		return XNET_RESULT_ERROR;
	}
	for ( size_t i = 0; i < iCount; i++ ) {
		xbytesview Chunk = {
			(cbytes)pChunks[i].Data,
			pChunks[i].Size
		};

		Result = (i + 1u) == iCount ?
			xrtWsWriterFinish(pWriter, Chunk) :
			xrtWsWriterWrite(pWriter, Chunk);
		if ( Result != XNET_RESULT_OK ) {
			xrtWsWriterDestroy(pWriter);
			return Result;
		}
	}
	xrtWsWriterDestroy(pWriter);
	return XNET_RESULT_OK;
}



/* 示例函数由已经建立连接的 Worker 回调调用。 */
int main(void)
{
	(void)sendChunks;
	printf("WebSocket Writer example is ready\n");
	return 0;
}

Что только что произошло. (1) xrtWsConnBeginText(连接) (соединение) открывает текстовый writer; цикл зовёт WriterWrite на каждый кусок, последний кусок идёт через WriterFinish (запись + FIN за один вызов) — стандартная форма цикла writer. (2) Отказ на середине — WriterDestroy: путь отказа для незавершённого сообщения (соединение остаётся рабочим; не отправлено только это сообщение). (3) Сигнатура функции берёт массив кусков — образец имеет шаблонную форму (main проверяет готовность); реальные вызовы происходят в обратных вызовах Worker (дисциплина главы 110). Почему writer, а не накопить-и-отправить: куски приходят из потоковой генерации (курсоры/конвейеры/дампы) — накопление целого сообщения и есть пик памяти; writer развязывает «ритм генерации» и «ритм отправки».

Вторая полная программа: путь сжатой отправки

Вторая программа из examples/websocket/writer_deflate — проверка изоморфизма writer сжатия:

#include <stdio.h>

#include <xws.h>



/* 在 Connection 所属 Worker 上把业务分块写成一条压缩 Text 消息。 */
static xnetresult sendCompressedChunks(
	xwsconn* pConnection,
	const xstrview* pChunks,
	size_t iCount
)
{
	xwswriter* pWriter;
	xnetresult Result;

	if ( (pChunks == NULL) || (iCount == 0) ) {
		return XNET_RESULT_ERROR;
	}
	pWriter = xrtWsConnBeginTextCompressed(pConnection);
	if ( pWriter == NULL ) {
		return XNET_RESULT_ERROR;
	}
	for ( size_t i = 0; i < iCount; i++ ) {
		xbytesview Chunk = {
			(cbytes)pChunks[i].Data,
			pChunks[i].Size
		};

		Result = (i + 1u) == iCount ?
			xrtWsWriterFinish(pWriter, Chunk) :
			xrtWsWriterWrite(pWriter, Chunk);
		if ( Result != XNET_RESULT_OK ) {
			xrtWsWriterDestroy(pWriter);
			return Result;
		}
	}
	xrtWsWriterDestroy(pWriter);
	return XNET_RESULT_OK;
}



/* 示例函数由已经协商 permessage-deflate 的 Connection Worker 回调调用。 */
int main(void)
{
	(void)sendCompressedChunks;
	printf("WebSocket compressed Writer example is ready\n");
	return 0;
}

Что только что произошло. (1) Единственное отличие — вход: xrtWsConnBeginTextCompressed; дальше цикл Write/Finish/Destroy дословно совпадает с обычным writer. «Сжатие — свойство writer, а не переписывание процесса отправки» — код вызывающего получает сжатие без единой правки. (2) Предусловие: соединение согласовало permessage-deflate (слой согласования главы 97, конфигурация маршрутизации главы 113); открытие writer сжатия на несогласованном соединении падает на входе. (3) Семантику сжатия гарантирует writer: RSV1 на первом кадре, сброс словаря по согласованию, потолок распаковки у противоположной стороны — отправляющей стороне не нужно вникать в детали правил сжатия главы 97; согласование зафиксировало поведение — поведение определено. Вместе с writer_ref (вариант writer для ссылочных кусков — потоковый финал статичных больших блоков) три образца пути отправки образуют полную матрицу.

111.4契约

• Жизненный цикл writer: Begin (три входа: text/binary/compressed) → Write×N → Finish (FIN) или Destroy (отказ) — одно из двух обязательно; вызов на Worker-владельце соединения.
• Развязка кусок-кадр: ритм кусков вызывающего не связан с протокольным кадрированием; FIN ровно один раз; управляющие кадры нельзя вставлять внутрь сообщения своего writer (отправная сторона правила обхода главы 95 — чтобы послать Ping, сперва Finish или через вход уровня соединения).
• Writer сжатия: отличие — только вход; интерфейс Write/Finish как у обычного writer; предусловие — согласование; RSV1/словарь/потолок гарантирует writer.
• Стабильные размеры: заголовок и состояние сжатия writer фиксированы — раскладка сетевого объекта не зависит от содержимого.
• Атомарность владения: непринятое остаётся у вызывающего, после приёма соединение освобождает ровно один раз; take может передавать цепочки буферов (голова цепочки достигает нулевого копирования).
• Выбор трёх состояний: короткое = copy; статичное/разделяемое = ref; передача выделенного = take — девять синхронных комбинаций плюс семейство Future.
• Единое противодавление: записи writer также ограничены отметками — при AGAIN кусок не принят (можно повторить в точности).
• Обрезка: websocket_writer/websocket_deflater независимы — соединения без writer не платят ничего.

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

pWriter = xrtWsConnBeginText(pConn);
write_some(pWriter);
/* Finish/Destroy забыты: отправочная сторона соединения заперта */

pWriter = xrtWsConnBeginText(pConn);
if ( pWriter == NULL ) { return XNET_RESULT_ERROR; }
for ( ... ) {
	if ( WriterWrite(...) != XNET_RESULT_OK ) {
		xrtWsWriterDestroy(pWriter);   /* путь отказа обязателен */
		return Result;
	}
}
return xrtWsWriterFinish(pWriter, LastChunk);

/* deflate не предлагался при согласовании */
pWriter = xrtWsConnBeginTextCompressed(pConn);  /* NULL: способности нет */

/* конфигурация сервера/клиента объявляет deflate (глава 113: маршрутизация/конфигурация клиента) */
/* после успешного согласования выбирать вход по способности соединения */
if ( conn_deflate_negotiated(pConn) ) {
	pWriter = xrtWsConnBeginTextCompressed(pConn);
} else {
	pWriter = xrtWsConnBeginText(pConn);   /* откат на обычный */
}

xrtWsConnBinaryRef(pConn, &Ref);
/* ... где-то на асинхронном пути ... */
use(Buf);   /* release мог уже произойти: висячий */

   (глава 22) и брать снова только через пул */
static void releaseToPool(ptr pCtx, cbytes pData, size_t iSize) {
	pool_return((pool*)pCtx, (ptr)pData);   /* вернуть, а не освободить */
}

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

Итоги