Глава 72 Кадрирование и асинхронные файлы: финал тома 7

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

Сквозная линия тома 7 — «дать данным течь»: адреса (62), порты (63), буферы (64), DNS (65), TCP (66–67), UDP (68), прокси (69), интерфейсы (70). Завершающая глава добавляет два инструмента сборочного уровня — они не вводят новых объектов соединений, а склеивают существующие части в полные решения. Кадровые парсеры отвечают «как лишённый границ байтовый поток TCP снова становится сообщениями» — вход им даёт xnetbuf главы 65, драйвером служат колбэки приёма главы 67; нативный асинхронный файловый IO отвечает «как файловому серверу повесить дисковые чтение/запись в тот же событийный цикл, что и сеть» — правила принадлежности Worker главы 64 здесь исполняются снова. Оба — маленькие модули без зависимостей от движка, но каждый держит на себе текстовые протоколы (в стиле Redis/SMTP), частные бинарные протоколы (RPC/игровые серверы) и прокси-кэши со статической раздачей — несколько больших классов реальных нагрузок.

72.1引入

Два сценария, одна общность. Сценарий один: вы реализуете текстовый протокол над TCP; клиент шлёт PING\r\n, сервер отвечает PONG\r\n. Звучит просто, но TCP не обещает «один send — один recv»: три отправки могут прийти склеенными (липкие пакеты), одна отправка может прийти разорванной пополам (полупакеты). Что если PING\r пришло первым? Вручную — лихорадочная арифметика индексов. Сценарий два: вы пишете прокси-кэш или файловый сервис; сетевые чтение/запись идут асинхронно на Engine, но чтение файла с диска — блокирующий вызов: один Worker читает диск — и сотни соединений на том же Worker застревают.

Общее у двух сценариев — проблема границы. Первому не хватает границ сообщений — их добавляют кадровые парсеры; второму не хватает «единой событийной границы диска и сети» — её добавляют нативные асинхронные файлы: файловые операции подаются прямо в порт завершения Worker, Windows идёт через ReadFile/WriteFile IOCP, Linux — через READV/WRITEV io_uring, с диспетчеризацией из того же пула, что и события сокетов; Worker никогда не блокируется на диске.

72.2概念

Трёхстатный разбор: общий ритм инкрементных кадрировщиков

Два кадрировщика (по строкам, по префиксу длины) делят одну трёхстатную модель возврата:

XNET_FRAME_MORE — нормальный инкрементный контрольный результат, ошибки не ставит — это и есть «липкие/полупакеты, поглощённые машиной состояний». При XNET_FRAME_READY выходной xnetframe описывает один полный кадр в голове текущего ввода: PayloadOffset/PayloadSize локализуют нагрузку, FrameSize — байты целого кадра, Declared хранит исходное значение поля длины протокола. XNET_FRAME_ERROR несёт структурированную ошибку (четыре вида: FRAME_CONFIG/FRAME_STATE/FRAME_LIMIT/FRAME_LENGTH) и обычно означает, что поток протокола больше не заслуживает доверия — закройте соединение или отбросьте ввод и сбросьте кадрировщик.

Строчное кадрирование: инкрементный поиск с произвольным разделителем

У xnetlineconfig три элемента: Delimiter (разделитель — заимствованное представление, обязанное жить, пока жив кадрировщик; без ограничения длины — годятся \n, \r\n и даже многобайтовые сигналы), MaxPayload (потолок нагрузки, по умолчанию 8192; SIZE_MAX — явно без границ), IncludeDelimiter (оставлять ли разделитель; по умолчанию убирается). Конфигурация по умолчанию — разделение по LF.

Кадрировщик внутренне хранит указатель блока и смещение внутри блока (xnetlineframer), поэтому даже когда ввод рассыпан по множеству однобайтовых ссылочных блоков, позиционирование старого смещения не требует повторного обхода головы цепочки — это водораздел от «сканировать с начала каждый раз» и правильный способ потреблять ссылочные добавления главы 65. Разделитель на границе буферных блоков, самопересечение (разделитель \r\n при вводе \r\r\n), точное попадание на потолок нагрузки — всё поддерживается.

Один ключевой контракт: после возврата MORE от Next вызывающий может только сохранить исходный префикс ввода и добавлять в хвост того же xnetbuf; перед потреблением, заменой, сменой буфера или перестановкой ввода обязателен xrtNetLineReset (сохраняет конфигурацию, отбрасывает инкрементный прогресс). Другое применение Reset — переиспользование кадрировщика при «начале диалога заново» на том же соединении, экономя одну инициализацию.

Кадрирование с префиксом длины: одна формула на все случаи

Самое ходовое кадрирование частных бинарных протоколов: поле длины в 1–8 байт заголовка + нагрузка. Общая длина кадра определяется одной формулой:

FrameSize = LengthOffset + LengthSize + Declared + Adjustment

• LengthOffset/LengthSize: позиция и ширина поля длины внутри кадра (1..8 байт); Order явно фиксирует порядок байт, независимо от хостового.
• Adjustment: знаковая поправка. Если поле длины протокола уже включает заголовок, отнимите его отрицательным значением.
• Strip: начало нагрузки — сохранить весь заголовок, убрать только поле длины или срезать весь прикладной заголовок; одно поле, три выбора.
• MaxFrame: жёсткий потолок целого кадра (по умолчанию 1 МиБ) — злобные или битые заявленные длины останавливаются здесь.

Конфигурация по умолчанию — «4-байтовая длина big-endian, поле длины убирается, потолок 1 МиБ» — формат большинства RPC. Все границы — конец поля, беззнаковое заявленное значение, знаковая поправка, преобразование size_t, strip и потолок — проверяются до доступа к нагрузке; пути «сначала поверить длине, потом читать за границей» не существует.

Два способа взять кадр: копия и ноль копий

После того как Next выдал описание кадра, нагрузку берут двумя способами. Удобный путь xrtNetFrameCopy: копирует байты нагрузки (не сверх ёмкости выхода) в буфер вызывающего — маленькое сообщение в одну строку. Путь без копий: прямо через xrtNetBufSpans/xrtNetBufPeek главы 65 взять представления нагрузки — большие кадры копировать не нужно. После взятия xrtNetFrameConsume точно убирает целый кадр из головы ввода (он сначала проверяет, что диапазон кадра всё ещё целиком в голове; если ввод меняли — отказ с FRAME_STATE). Заметьте: описание кадра заимствует текущий ввод — после потребления, замены или перестановки префикса ввода старым описанием кадра пользоваться нельзя.

Нативные асинхронные файлы: диск на порту завершения

Модуль net_file подаёт позиционные чтение/запись обычных файлов в порт завершения Network Engine — для файловых сервисов, прокси-кэшей и собственных конвейеров хранения: не создаёт Future, не занимает TaskPool, не копирует нагрузку вызывающего. Пять пунктов:

• Открытие: xrtNetFileOpen автоматически добавляет флаг XFILE_ASYNC; опции — тот же набор, что у xrtFileOpen главы 45 (по умолчанию только чтение; новый файл требует явного XFILE_CREATE|XFILE_WRITE).
• Принадлежность Worker: Read/Write подаются только в потоке владеющего Worker — после первой подачи файл навсегда закрепляется за одним Worker; межпоточная подача получает XERR_STATE. Главному потоку нужно инициировать — xrtNetEnginePost доставляет действие подачи на Worker (стандартная поза примеров этой главы).
• Абсолютные смещения: чтение/запись несут iOffset и не зависят от файлового курсора — множественные конкурентные чтения одного файла не мешают друг другу.
• Единственный финал: каждая операция возвращает ненулевой идентификатор, а колбэк xnetcompletion получает ровно один финальныйevento на Worker (Result/Bytes/Id события xnetportevent описывают результат). Файл, буферы и completion обязаны жить до прибытия финала. xrtNetFileCancel только запрашивает отмену — исходная операция всё равно завершается через completion: финалом может быть XNET_RESULT_CANCELLED, а может операция уже успела — тогда OK.
• Бит возможностей: если у порта Worker нет нативного файлового IO (например, подстраховочный бэкенд SELECT), подача получает XERR_UNSUPPORTED — явный отказ, без деградации в блокировку Worker.

Когда все операции завершены, файл закрывается через xrtClose — тип дескриптора тот же xfile главы 45; всё файловое семейство делит этот единственный вход закрытия, отдельной функции закрытия для асинхронных файлов нет.

72.3示例

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

Следующая программа из examples/network/frame_line/main.c: ввод нарочно приходит кусками first\r и \nsecond\r\n (полупакет переходит разделитель), кадрировщик восстанавливает кадры по одному и демонстрирует переиспользование Reset:

#include <stdio.h>

#include <xrt.h>



















 */


/* 从分块输入中逐条取出 CRLF 文本帧。 */
int main(void)
{
	xnetlineconfig Config;
	xnetlineframer Framer;
	xnetframe Frame;
	xnetbuf Input;
	char sLine[32];

	xrtNetLineConfigInit(&Config);
	Config.Delimiter.Data = (cbytes)"\r\n";
	Config.Delimiter.Size = 2;
	if ( !xrtNetLineInit(&Framer, &Config) ||
		 !xrtNetBufInit(&Input, NULL) ||
		 !xrtNetBufAppend(&Input, "first\r", 6) ) {
		return 1;
	}
	if ( xrtNetLineNext(&Framer, &Input, &Frame) != XNET_FRAME_MORE ||
		 !xrtNetBufAppend(&Input, "\nsecond\r\n", 10) ) {
		xrtNetBufClear(&Input);
		return 1;
	}
	while ( xrtNetLineNext(&Framer, &Input, &Frame) ==
		XNET_FRAME_READY ) {
		size_t iSize = xrtNetFrameCopy(
			&Input, &Frame, sLine, sizeof(sLine) - 1u
		);

		sLine[iSize] = 0;
		printf("%s\n", sLine);
		if ( !xrtNetFrameConsume(&Input, &Frame) ) {
			xrtNetBufClear(&Input);
			return 1;
		}
	}
	xrtNetBufClear(&Input);
	
	 *   时复用帧器（省一次初始化）；Reset 后可继续喂新流。 */
	(void)xrtNetLineReset(&Framer);
	printf("reset=ok\n");
	return 0;
}

Что только что произошло. (1) Разделитель установлен в \r\n (заимствованное представление прямо на литерал); первый кусок first\r даёт MORE — разделитель пришёл лишь наполовину, кадрировщик записал неразрешённый префикс в инкрементный курсор. (2) Второй кусок \nsecond\r\n дописывается в хвост того же xnetbuf (корректное применение контракта MORE), цикл Next последовательно разбирает first и second как два полных кадра. (3) Каждый кадр проходит три шага «Copy в стековый буфер → печать → Consume точное потребление», sizeof(sLine) - 1u оставляет место под хвостовой ноль. (4) В конце xrtNetLineReset сохраняет конфигурацию и отбрасывает прогресс — новая сессия переиспользует кадрировщик. Это фундамент кадрирования текстовых протоколов класса Redis/SMTP: липкие и полупакеты поглощаются машиной состояний парсера, бизнес-код видит только полные сообщения.

Вторая полная программа: разбор кадров с четырёхбайтовым префиксом длины

Вторая программа из examples/network/frame_length/main.c, стандартный образец частного бинарного протокола — 00 00 00 05 + hello:

#include <stdio.h>

#include <xrt.h>

















 */


/* 解析四字节网络序长度前缀，并直接读取 payload。 */
int main(void)
{
	static const uint8 Packet[] = {
		0, 0, 0, 5, 'h', 'e', 'l', 'l', 'o'
	};
	xnetlengthconfig Config;
	xnetlengthframer Framer;
	xnetframe Frame;
	xnetbuf Input;
	char sPayload[6] = { 0 };

	xrtNetLengthConfigInit(&Config);
	if ( !xrtNetLengthInit(&Framer, &Config) ||
		 !xrtNetBufInit(&Input, NULL) ||
		 !xrtNetBufAppend(&Input, Packet, sizeof(Packet)) ||
		 (xrtNetLengthNext(&Framer, &Input, &Frame) !=
			XNET_FRAME_READY) ||
		 (xrtNetFrameCopy(
			&Input, &Frame, sPayload, sizeof(sPayload) - 1u
		 ) != 5) ) {
		xrtNetBufClear(&Input);
		return 1;
	}
	printf("%s\n", sPayload);
	xrtNetBufClear(&Input);
	return 0;
}

Что только что произошло. Конфигурация по умолчанию (xrtNetLengthConfigInit) — ровно четырёхбайтовая длина big-endian, поле длины убирается, потолок 1 МиБ — прямое попадание в формат этого протокола. LengthNext за один шаг выдаёт описание кадра, FrameCopy копирует нагрузку в 5-байтовый буфер (возврат 5 — ровно длина нагрузки, плюс один байт под хвостовой ноль). Сравните с моделью возврата строчного примера: LengthNext при «поле длины неполно» или «заявленная длина ещё не собралась» так же возвращает MORE, и цикл «добавил — повторил» у вызывающего полностью изоморфен строчному — смени формат кадра, не меняя ритм драйвера.

Третья полная программа: файловые чтение/запись и отмена на порту завершения

Третья программа из examples/network/file_tour/main.c проходит «подача — финал — отмена» целиком: запись с абсолютным смещением, чтение с поверкой, единственный финал после отмены чтения в полёте:

 */

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

#include <math.h>
static inline double exampleTimerLimit(int64 Timeout)
{
    return Timeout == XRT_WAIT_FOREVER ? INFINITY : xrtTimer() + (double)Timeout / 1000.0;
}
static inline bool exampleTimerExpired(double Limit)
{
    return xrtTimer() >= Limit;
}
static inline int64 exampleTimerRemaining(double Limit)
{
    double Ms;
    if (Limit == INFINITY) return XRT_WAIT_FOREVER;
    Ms = ceil((Limit - xrtTimer()) * 1000.0);
    return Ms <= 0 ? 0 : Ms >= 0x1p63 ? INT64_MAX : (int64)Ms;
}

static const char* sFile = "net_file_tour_tmp.bin";

/* Completion 终态记录。 */
typedef struct exampleio {
	xnetcompletion Completion;
	volatile bool bDone;
	xnetresult Result;
	size_t iBytes;
	uint64 iId;
} exampleio;

static void exampleDone(xnetworker* pWorker,
	const xnetportevent* pEvent, ptr pData)
{
	exampleio* pIo = (exampleio*)pData;

	(void)pWorker;
	pIo->Result = pEvent->Result;
	pIo->iBytes = pEvent->Bytes;
	pIo->iId = pEvent->Id;
	pIo->bDone = true;
}

static bool exampleSpin(exampleio* pIo, double iEnd)
{
	while ( !pIo->bDone ) {
		if ( exampleTimerExpired(iEnd) ) {
			return false;
		}
		xrtThreadYield();
	}
	return true;
}

/* Worker 任务：提交写并登记 Completion。 */
typedef struct exampletask {
	xnetworker* pWorker;
	xfile File;
	exampleio* pIo;
	const void* pData;
	size_t iSize;
	uint64 iOffset;
	volatile bool bDone;
	bool bOk;
	uint64 iId;
} exampletask;

static bool exampleSpinUntilTask(exampletask* pTask);

static void exampleWriteTask(xnetworker* pWorker, ptr pUserData)
{
	exampletask* pTask = (exampletask*)pUserData;

	(void)pWorker;
	pTask->iId = xrtNetFileWrite(pTask->pWorker, pTask->File,
		pTask->iOffset, pTask->pData, pTask->iSize,
		&pTask->pIo->Completion);
	pTask->bOk = (pTask->iId != 0u);
	pTask->bDone = true;
}

static void exampleReadTask(xnetworker* pWorker, ptr pUserData)
{
	exampletask* pTask = (exampletask*)pUserData;

	(void)pWorker;
	pTask->iId = xrtNetFileRead(pTask->pWorker, pTask->File,
		pTask->iOffset, (void*)pTask->pData, pTask->iSize,
		&pTask->pIo->Completion);
	pTask->bOk = (pTask->iId != 0u);
	pTask->bDone = true;
}

static bool examplePost(xnetengine* pEngine, xnetpost* pPost,
	void (*pProc)(xnetworker*, ptr), exampletask* pTask)
{
	(void)pPost;
	return xrtNetEnginePost(pEngine, 0u, pProc, (ptr)pTask) &&
		exampleSpin((exampleio*)pTask->pIo,
			exampleTimerLimit(3000)) &&
		((exampleio*)pTask->pIo)->bDone;
}

/* 等待任务提交本身完成（不含 IO 终态）。 */
static bool exampleSpinUntilTask(exampletask* pTask)
{
	double iEnd = exampleTimerLimit(3000);

	while ( !pTask->bDone ) {
		if ( exampleTimerExpired(iEnd) ) {
			return false;
		}
		xrtThreadYield();
	}
	return true;
}

int main(void)
{
	xnetengine* pEngine = NULL;
	xnetpost Post;
	xfile File = 0;
	exampleio Io;
	exampletask Task;
	uint8 arrRead[16];
	int iResult = 1;

	(void)remove(sFile);
	pEngine = xrtNetEngineCreate(NULL);
	if ( (pEngine == NULL) || !xrtNetEngineStart(pEngine) ||
		!xrtNetPostInit(&Post) ) {
		goto Cleanup;
	}

    /* Native file completion requires IOCP or an available io_uring port. */
    {
        xnetport* pPort = xrtNetWorkerPort(xrtNetEngineWorker(pEngine, 0u));
        if (pPort == NULL) goto Cleanup;
        if ((xrtNetPortCapabilities(pPort) & XNET_PORT_CAP_FILE_IO) == 0u) {
            printf("file: native file completion unavailable on this backend\n");
            iResult = 0;
            goto Cleanup;
        }
    }
	/* ---- 打开 + 写：绝对偏移 0 写 8 字节。 ---- */
	{
		xfileoptions Options;

		/* 默认选项是只读——新文件须显式 CREATE|WRITE。 */
		xrtFileOptionsInit(&Options);
		Options.Flags = XFILE_READ | XFILE_WRITE |
			XFILE_CREATE | XFILE_TRUNCATE;
		File = xrtNetFileOpen(sFile, &Options);
	}
	memset(&Io, 0, sizeof(Io));
	xrtNetCompletionInit(&Io.Completion, exampleDone, (ptr)&Io);
	memset(&Task, 0, sizeof(Task));
	Task.pWorker = xrtNetEngineWorker(pEngine, 0u);
	Task.File = File;
	Task.pIo = &Io;
	Task.pData = "filedata";
	Task.iSize = 8u;
	Task.iOffset = 0u;
	if ( (File == 0) ||
		!examplePost(pEngine, &Post, exampleWriteTask, &Task) ||
		(Io.Result != XNET_RESULT_OK) ||
		(Io.iBytes != 8u) ||
		(Io.iId != Task.iId) ) {
		goto Cleanup;
	}
	printf("file: open(async) + write completed at offset 0 ok\n");

	/* ---- 读回验证 + 追加偏移写。 ---- */
	memset(&Io, 0, sizeof(Io));
	xrtNetCompletionInit(&Io.Completion, exampleDone, (ptr)&Io);
	memset(arrRead, 0, sizeof(arrRead));
	memset(&Task, 0, sizeof(Task));
	Task.pWorker = xrtNetEngineWorker(pEngine, 0u);
	Task.File = File;
	Task.pIo = &Io;
	Task.pData = arrRead;
	Task.iSize = 8u;
	Task.iOffset = 0u;
	if ( !examplePost(pEngine, &Post, exampleReadTask, &Task) ||
		(Io.Result != XNET_RESULT_OK) ||
		(Io.iBytes != 8u) ||
		(memcmp(arrRead, "filedata", 8u) != 0) ) {
		goto Cleanup;
	}
	printf("file: read-back verified + reopen append ok\n");

	/* ---- 取消：提交一个大读后立即取消，仍收唯一终态。 ---- */
	{
		static uint8 arrBig[4096];
		exampleio CancelIo;
		uint64 iCancelId;

		memset(&CancelIo, 0, sizeof(CancelIo));
		xrtNetCompletionInit(&CancelIo.Completion, exampleDone,
			(ptr)&CancelIo);
		memset(&Task, 0, sizeof(Task));
		Task.pWorker = xrtNetEngineWorker(pEngine, 0u);
		Task.File = File;
		Task.pIo = &CancelIo;
		Task.pData = arrBig;
		Task.iSize = sizeof(arrBig);
		Task.iOffset = 0u;
		if ( !xrtNetEnginePost(pEngine, 0u, exampleReadTask,
				(ptr)&Task) ||
			!exampleSpinUntilTask(&Task) ||
			(Task.iId == 0u) ) {
			goto Cleanup;
		}
		iCancelId = Task.iId;
		
		 * 到 EOF）——先等一小段，未完成才取消。 */
		{
			int iSpin;

			for ( iSpin = 0; (iSpin < 50) &&
				!CancelIo.bDone; iSpin++ ) {
				xrtSleep(1u);
			}
			if ( !CancelIo.bDone ) {
				if ( !xrtNetFileCancel(Task.pWorker,
						iCancelId) ||
					!exampleSpin(&CancelIo,
						exampleTimerLimit(
							3000)) ) {
					goto Cleanup;
				}
			}
		}
		/* 终态必达：结果是 OK（读已完成）或 CANCELLED。 */
		if ( (CancelIo.Result != XNET_RESULT_OK) &&
			(CancelIo.Result != XNET_RESULT_CANCELLED) ) {
			goto Cleanup;
		}
	}
	printf("file: cancel in-flight read -> terminal event ok\n");
	iResult = 0;

Cleanup:
	if ( File != 0 ) {
		xrtClose(File);
	}
	(void)xrtNetEngineStop(pEngine);
	(void)xrtNetEngineDestroy(pEngine);
	(void)remove(sFile);
	return iResult;
}

Что только что произошло. (1) Файловые опции явно XFILE_READ|XFILE_WRITE|XFILE_CREATE|XFILE_TRUNCATE — по умолчанию только чтение, создание объявляется явно, а xrtNetFileOpen поверх автоматически добавляет асинхронный флаг. (2) Подача идёт стандартными двумя скачками: главный поток xrtNetEnginePost доставляет exampleWriteTask на Worker 0, а функция задачи на Worker вызывает xrtNetFileWrite и регистрирует completion — приземлённая форма контракта «подача только во владеющем Worker». Финальный колбэк exampleDone записывает Result/Bytes/Id, главный поток крутится в ожидании bDone и сверяет три вещи: результат OK, байты сходятся, Id события совпадает с возвратом подачи. (3) Чтение с поверкой — та же поза, memcmp проверяет содержимое. (4) Секция отмены сначала подаёт большое чтение на 4096 байт, крутится немного, убеждаясь, что оно не завершилось, затем xrtNetFileCancel — финал приходит всегда, и результат может быть только OK (чтение как раз успело) или CANCELLED; заметьте, что чтение 4096 байт из 8-байтового файла сразу завершится по EOF, поэтому пример сначала ждёт и лишь потом решает, отменять ли — эта обработка «отмена может проиграть гонку завершению» и есть вид настоящего кода. (5) Уборка: xrtClose(File) → остановка Engine → удаление временного файла, порядок менять нельзя.

72.4契约

• Жизненный цикл кадрировщика: кадрировщик — объект-значение на стеке вызывающего, без разделяемого состояния; разные экземпляры конкурентно в любых потоках, один экземпляр — последовательные вызовы: всё безопасно; конфигурация (представление разделителя) обязана жить до ухода кадрировщика.
• Контракт MORE: после MORE разрешено только сохранить исходный префикс ввода и добавлять в тот же xnetbuf; перед потреблением/заменой/сменой буфера/перестановкой — обязателен Reset; MORE ошибки не ставит.
• Кадр заимствует ввод: смещения xnetframe относительны головы текущего ввода; после изменения префикса старый кадр недействителен; Consume проверяет диапазон и потребляет точно.
• Потолки кадров: нагрузка строки сверх MaxPayload — FRAME_LIMIT; переполнение/превышение/нелегальная поправка заявленной длины — FRAME_LENGTH; заведомо нелегальная конфигурация — FRAME_CONFIG.
• Принадлежность файла: Read/Write/Cancel действительны только в потоке владеющего Worker (XERR_STATE); после первой подачи файл закреплён за одним Worker; главный поток передаёт через EnginePost.
• Файловые финалы: операция возвращает ненулевой Id; completion получает ровно один финал (Result/Bytes/Id); файл, буферы и completion живут до финала; Cancel только запрашивает — финал OK или CANCELLED.
• Возможности файла: бэкенд без нативного файлового IO (подстраховка SELECT) — XERR_UNSUPPORTED, без деградации в блокировку; Windows IOCP, Linux io_uring.
• Закрытие: xrtClose после завершения всех операций; смещение + длина за пределами выразимого — XNET_ERROR_PORT_SUBMIT.

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

if ( xrtNetLineNext(&Framer, &Input, &Frame) ==
	XNET_FRAME_MORE ) {
	xrtNetBufConsume(&Input, 8);   /* тронули префикс: инкрементный курсор повис */
	xrtNetLineNext(&Framer, &Input, &Frame); /* FRAME_STATE */
}

if ( xrtNetLineNext(&Framer, &Input, &Frame) ==
	XNET_FRAME_MORE ) {
	/* единственное законное действие: исходный префикс сохранён, новые данные — в хвост */
	xrtNetBufAppend(&Input, pChunk, iSize);
}
/* если текущий ввод действительно надо отбросить (например, сброс протокола): сначала Reset, затем новый поток */
(void)xrtNetLineReset(&Framer);

/* главный поток: не во владеющем Worker, XERR_STATE, возврат 0 */
iId = xrtNetFileRead(pWorker, File, 0, Buffer,
	sizeof(Buffer), &Completion);

/* стандартные два скачка: Post доставляет задачу на Worker, подача — в функции задачи */
static void submitRead(xnetworker* pWorker, ptr pUserData) {
	exampletask* pTask = (exampletask*)pUserData;
	pTask->iId = xrtNetFileRead(pWorker, pTask->File,
		pTask->iOffset, (void*)pTask->pData, pTask->iSize,
		&pTask->pIo->Completion);
}
xrtNetEnginePost(pEngine, 0u, submitRead, (ptr)&Task);

iId = xrtNetFileRead(pWorker, File, 0, Buffer,
	sizeof(Buffer), &Completion);
if ( iId == 0u ) {
	/* предположили временную ошибку и зациклились навсегда: вечно UNSUPPORTED */
}

iId = xrtNetFileRead(pWorker, File, 0, Buffer,
	sizeof(Buffer), &Completion);
if ( iId == 0u && xrtErrorIs(xrtGetError(),
	XERR_UNSUPPORTED) ) {
	
	   модуль главы 48 или выделенный дисковый поток, а не на повторы */
}

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

Итоги