第 72 章 分帧与异步文件：卷七收官

FrameSize = LengthOffset + LengthSize + Declared + Adjustment

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

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

 */

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