Chapter 72 Framing and Async Files: The Volume 7 Finale

Chapter Orientation

Volume 7's through-line is "get data flowing": addresses (62), ports (63), buffers (64), DNS (65), TCP (66–67), UDP (68), proxies (69), interfaces (70). The closing chapter adds two assembly-grade tools: they introduce no new connection objects but glue existing parts into complete solutions. The frame parsers answer "how do I turn TCP's boundary-less byte stream back into messages" — Chapter 65's xnetbuf is their input, Chapter 67's receive callbacks are their drive; native async file I/O answers "how does a file server hang disk reads and writes into the same event loop as the network" — Chapter 64's Worker-ownership rules pay off again here. Both are small zero-engine-dependency modules, yet they respectively support text protocols (Redis/SMTP style), private binary protocols (RPC/game servers), and proxy caches and static file services — several large classes of real workloads.

72.1引入

Two scenarios, one commonality. Scenario one: you implement a text protocol over TCP; the client sends PING\r\n, the server replies PONG\r\n. Sounds simple, but TCP does not promise "one send matches one recv" — three sends may arrive glued as one (sticky packets), and one send may arrive split in two (partial packets). What if PING\r arrives first? Hand-rolling it is frantic index arithmetic. Scenario two: you write a proxy cache or file service; the network's reads and writes advance asynchronously on the Engine, but reading a file from disk is a blocking call — one Worker reads disk and hundreds of connections on the same Worker all stall.

The two scenarios share one thing: the boundary problem. The former lacks message boundaries — the frame parsers supply them; the latter lacks "a unified event boundary for disk and network" — native async files supply it — file operations submit directly to the Worker's completion port, Windows going through IOCP's ReadFile/WriteFile, Linux through io_uring's READV/WRITEV, dispatched from the same pool as socket events, so a Worker never blocks on disk.

72.2概念

Three-state parsing: the incremental framers' shared rhythm

The two framers (line-delimited, length-prefixed) share one three-state return model:

XNET_FRAME_MORE is a normal incremental control result and sets no error — it is precisely "sticky/partial packets absorbed by the state machine". On XNET_FRAME_READY, the output xnetframe describes one complete frame at the head of the current input: PayloadOffset/PayloadSize locate the payload, FrameSize is the whole frame's byte count, and Declared preserves the protocol length field's raw value. XNET_FRAME_ERROR carries a structured error (four kinds: FRAME_CONFIG/FRAME_STATE/FRAME_LIMIT/FRAME_LENGTH), usually meaning the protocol stream is no longer trustworthy — close the connection or drop the input and reset the framer.

Line framing: incremental search with arbitrary delimiters

xnetlineconfig has three elements: Delimiter (a borrowing view whose source must live as long as the framer is used; no length limit — \n, \r\n, even multi-byte sentinels work), MaxPayload (default 8192; SIZE_MAX is explicitly unbounded), and IncludeDelimiter (whether to keep the delimiter; removed by default). The default configuration is LF-delimited.

The framer internally keeps the block pointer and in-block offset (xnetlineframer), so even when input is scattered across many single-byte reference blocks, locating the old offset never re-walks the chain head — this is the watershed from "rescanning from the start every time" and the correct way to consume Chapter 65's reference-block appends. Delimiters crossing buffer blocks, self-overlapping (delimiter \r\n meeting input \r\r\n), and landing exactly on the payload limit are all supported.

One key contract: after Next returns MORE, the caller may only keep the original input prefix and append at the tail of the same xnetbuf; before consuming, replacing, switching buffers, or reordering input, you must call xrtNetLineReset (keeps the config, discards incremental progress). Reset's other use is reusing the framer when the same connection "restarts the conversation", saving one initialization.

Length-prefixed framing: one formula fits all

The most common framing for private binary protocols: a 1–8 byte length field in the header + payload. The total frame size follows one formula:

FrameSize = LengthOffset + LengthSize + Declared + Adjustment

• LengthOffset/LengthSize: the length field's position and width within the frame (1..8 bytes); Order fixes endianness explicitly, independent of host byte order.
• Adjustment: a signed adjustment. If the protocol's length field already includes the header, subtract it back with a negative value.
• Strip: the payload start — keep the whole header, remove only the length field, or remove the entire application header; one field, three choices.
• MaxFrame: the hard total-frame cap (default 1 MiB) — malicious or corrupt declared lengths are stopped here.

The default configuration is "4-byte big-endian payload length, length field removed, 1 MiB cap" — exactly most RPCs' format. Every boundary — field end, unsigned declared value, signed adjustment, size_t conversion, strip and cap — is validated before the payload is accessed; there is no "trust the length first, read out of bounds later" path.

Two ways to take a frame: copy and zero-copy

After Next produces a frame description, the payload can be taken two ways. The convenience path xrtNetFrameCopy: copies the payload bytes (up to output capacity) into the caller's buffer — small messages in one line. The zero-copy path: use Chapter 65's xrtNetBufSpans/xrtNetBufPeek directly for payload views — big frames need no copy. After taking, xrtNetFrameConsume removes exactly the whole frame from the input head (it first verifies the frame range is still wholly at the head; if the input was altered, it refuses with FRAME_STATE). Note the frame description borrows the current input: after the input prefix is consumed, replaced, or reordered, old frame descriptions must not be used.

Native async files: the disk on the completion port

The net_file module submits positioned reads and writes of ordinary files to the Network Engine's completion port, for file services, proxy caches, and custom storage pipelines — no Futures created, no TaskPool occupied, no caller-payload copies. Five key points:

• Open: xrtNetFileOpen automatically adds the XFILE_ASYNC flag; options are the same set as Chapter 45's xrtFileOpen (read-only by default; creating a file needs explicit XFILE_CREATE|XFILE_WRITE).
• Worker ownership: Read/Write may be submitted only on the owning Worker thread — after a file's first submission, it is permanently pinned to the same Worker; a cross-thread submission gets XERR_STATE. For the main thread to initiate, use xrtNetEnginePost to deliver the submission action onto a Worker (the standard posture of this chapter's examples).
• Absolute offsets: reads and writes carry iOffset and don't depend on the file cursor — multiple concurrent reads of the same file never interfere.
• The sole terminal state: every operation returns a non-zero identifier, and the xnetcompletion callback receives exactly one final-state event on the Worker (described by the xnetportevent's Result/Bytes/Id). File, buffers, and the completion must be kept alive until the terminal state arrives. xrtNetFileCancel only requests cancellation; the original operation still concludes through the completion — the terminal state may be XNET_RESULT_CANCELLED, or the operation may already have finished and be OK.
• Capability bit: when the Worker's port backend has no native file I/O capability (e.g. the SELECT fallback backend), submission gets XERR_UNSUPPORTED — reported explicitly, never degraded to blocking the Worker.

After all operations conclude, close the file with xrtClose — the handle type is Chapter 45's xfile; the whole file family shares this one close entry, with no separate close function for async files.

72.3示例

First complete program: CRLF line framing and partial-packet absorption

The following program comes from examples/network/frame_line/main.c: the input deliberately arrives split as first\r and \nsecond\r\n (a partial packet straddling the delimiter); the framer restores each frame and demonstrates Reset reuse:

#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;
}

What just happened. (1) The delimiter is set to \r\n (a borrowed view pointing straight at the literal); the first block first\r fed in yields MORE — only half the delimiter has arrived, and the framer records the pending prefix in its incremental cursor. (2) The second block \nsecond\r\n is appended to the tail of the same xnetbuf (the MORE contract used correctly), and the Next loop resolves first and second in turn as two complete frames. (3) Each frame walks the three steps "Copy into a stack buffer → print → Consume precisely", with sizeof(sLine) - 1u leaving room for the trailing zero. (4) At the end, xrtNetLineReset keeps the config and discards progress — a new session reuses the framer. This is the framing foundation of Redis/SMTP-class text protocols: sticky and partial packets are absorbed by the parser's state machine, and business code sees only complete messages one by one.

Second complete program: four-byte length-prefixed deframing

The second program comes from examples/network/frame_length/main.c, the standard sample of a private binary protocol — 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;
}

What just happened. The default configuration (xrtNetLengthConfigInit) is exactly four-byte big-endian length, length field removed, 1 MiB cap — hitting this protocol's format directly. LengthNext resolves the frame description in one step, and FrameCopy copies the payload into a 5-byte buffer (returning 5, exactly the payload length, one byte over for the trailing zero). Compare with the line-framing example's return model: LengthNext returns MORE equally for "length field incomplete" or "declared length not fully arrived", and the caller's append-retry loop is fully isomorphic with line framing — switch frame formats without changing the driving rhythm.

Third complete program: file read/write and cancellation on the completion port

The third program comes from examples/network/file_tour/main.c, walking "submit — terminal state — cancel" completely: absolute-offset write, read-back verification, and a unique terminal state still arriving after cancelling an in-flight read:

 */

#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;
}

What just happened. (1) The file options explicitly use XFILE_READ|XFILE_WRITE|XFILE_CREATE|XFILE_TRUNCATE — read-only by default; creating a file must be declared, and xrtNetFileOpen automatically adds the async flag on top. (2) Submission takes the standard two hops: the main thread xrtNetEnginePosts exampleWriteTask onto Worker 0, and the task function calls xrtNetFileWrite on the Worker and registers the completion — this is the landed form of the "submit only on the owning Worker" contract. The terminal callback exampleDone records Result/Bytes/Id, and the main thread spins waiting for bDone and cross-checks three things: result OK, byte count matching, and the event Id equal to the submission's return. (3) The read-back uses the same posture, with memcmp verifying contents. (4) The cancel section first submits a big 4096-byte read, spins briefly to confirm it hasn't finished, then xrtNetFileCancel — the terminal state always arrives and the result can only be OK (the read happened to finish) or CANCELLED; note that reading 4096 bytes from an 8-byte file completes immediately at EOF, so the example waits first before deciding whether to cancel — this "cancellation may lose the race condition to completion" handling is exactly what real code looks like. (5) The wind-down xrtClose(File) → stop the Engine → delete the temp file, in an order that must not be swapped.

72.4契约

• Framer lifecycle: framers are value objects on the caller's stack, no shared state; different instances run concurrently on any threads, and one instance is safe under sequential calls; the config (delimiter view) must live until the framer is discarded.
• The MORE contract: after MORE, only keeping the original input prefix and appending to the same xnetbuf is allowed; before consuming/replacing/switching buffers/reordering, Reset is mandatory; MORE sets no error.
• Frames borrow input: an xnetframe's offsets are relative to the current input head; old frames die once the prefix changes; Consume verifies the range and then consumes precisely.
• Frame limits: a line payload over MaxPayload reports FRAME_LIMIT; a length declaration overflowing/over-limit/an illegal adjustment reports FRAME_LENGTH; a config that cannot possibly be legal reports FRAME_CONFIG.
• File ownership: Read/Write/Cancel are valid only on the owning Worker thread (XERR_STATE); after a file's first submission it stays pinned to one Worker; the main thread hands over via EnginePost.
• File terminal states: operations return a non-zero Id; the completion receives exactly one terminal state (Result/Bytes/Id); file, buffers, and completion are kept until the terminal state; Cancel only requests — the terminal is OK or CANCELLED.
• File capability: a backend without native file I/O (SELECT fallback) reports XERR_UNSUPPORTED, never degrading to blocking; Windows IOCP, Linux io_uring.
• Close: xrtClose after all operations conclude; offset + length beyond the expressible range reports XNET_ERROR_PORT_SUBMIT.

Common Mistakes

if ( xrtNetLineNext(&Framer, &Input, &Frame) ==
	XNET_FRAME_MORE ) {
	xrtNetBufConsume(&Input, 8);   /* touched the prefix: the incremental cursor dangles */
	xrtNetLineNext(&Framer, &Input, &Frame); /* FRAME_STATE */
}

if ( xrtNetLineNext(&Framer, &Input, &Frame) ==
	XNET_FRAME_MORE ) {
	/* the only legal action: keep the prefix, append new data at the tail */
	xrtNetBufAppend(&Input, pChunk, iSize);
}
/* to truly discard the current input (e.g. protocol reset): Reset first, then feed the new stream */
(void)xrtNetLineReset(&Framer);

/* main thread: not on the owning Worker, XERR_STATE, returns 0 */
iId = xrtNetFileRead(pWorker, File, 0, Buffer,
	sizeof(Buffer), &Completion);

/* the standard two hops: Post delivers the task to the Worker, and the task function submits */
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 ) {
	/* assuming a transient error, retry in a dead loop: forever UNSUPPORTED */
}

iId = xrtNetFileRead(pWorker, File, 0, Buffer,
	sizeof(Buffer), &Completion);
if ( iId == 0u && xrtErrorIs(xrtGetError(),
	XERR_UNSUPPORTED) ) {
	
	   async file module or a dedicated disk thread, not retries */
}

Programming Exercises

Summary