Chapter Orientation

TCP's second half covers three advanced blocks: the four send families (Vec gather / Ref zero-copy + release callback / Take takeover / File kernel file sending) with the write budget (WriteLimit cap / Writable headroom / Pending in-flight — the send-backpressure trio); reads and Worker constraints (Buffer/Read/Consume inside a Worker only — the consumption face of direct-move buffers; crossing threads goes through Post); and the server shape (xnetserver — pull Accept (dequeuing connections), Future Accept and blocking Accept coexisting, multi-listen and shared ports). The gold standard's four states and echo (Chapter 67) are the basic form — this chapter is its full bloom.

68.1引入

Three advanced scenarios. Scenario one: a file server must send a 1GB file — reading the whole file into memory then Sending is a disaster (memory explosion); xrtNetStreamSendFile (kernel file sending + bounded ranges — the sendfile landing point previewed in Chapter 46) passes through no user-space buffer. Scenario two: a proxy must forward an upstream span chain downstream — copying before sending is waste; SendRef hands it over by reference directly (zero-copy + release callback). Scenario three: a server must cap a single connection's send backlog (a malicious client that never reads — buffers inflate without bound); the write budget (WriteLimit cap, Pending in-flight, Writable headroom) is backpressure's send-side enforcement.

The common theme is ownership and budget on the send face — Chapter 5's language in full on the IO exit: Send copies, SendVec gathers-and-copies, SendRef borrows + callback, SendTake takes over, SendFile delegates to the kernel. Five tiers chosen by "where the data comes from and who frees it" — corresponding one-to-one with Chapter 65's four buffer append kinds (those chain in, these chain out).

68.2概念

The four send families and file sending

EntryOwnershipFits
SendCopySmall-data safe path (gold-standard chapter)
SendVecGather-copyMultiple buffers sent at once (Chapter 21's batching)
SendRef / SendRefsReference + release callbackDirect span-chain handoff (zero-copy — kin of the gold standard's SendBuffer)
SendTakeTakeoverAn xrtMalloc block (engine frees when sent)
SendFileKernel delegationFile-range sending (sendfile/mmap backend)

SendRef's release callback: fired exactly once when the data is sent (or the stream destroyed) — the same discipline as Chapter 65's AppendRef; SendRefs is the multi-span variant (a whole span chain in one send). SendFile's range: offset + length bounded — chunked file sending controls memory (a kernel buffer per chunk, not the whole file).

The write budget: the send-backpressure trio

1WriteLimit(N)per-connection send budget cap (in-flight + queued <= N)
2Pending()current in-flight/queued bytes — the watermark gauge
3Writable()headroom = Limit - Pending — the safe amount to submit
4Send over budgetfails or blocks (per call form) — where backpressure bites
5Flow control to matchPause/Resume — after Pause the engine holds data (memory backstop)

The complete backpressure chain: the peer stops reading → the TCP window shrinks → engine buffers back up → Pending approaches Limit → new Sends are refused → the upstream notices (the handler sees send failure / must wait) — a slow consumer cannot drag down service memory. Pause/Resume is the manual gear (Pausable semantics: after Pause the engine retains data without advancing — the upstream may decide to drop or disconnect). Chapter 21's queue backpressure, connection edition.

Reads and Worker constraints

In-Worker APIs (Buffer/Read/Consume etc. — must be called on the Stream's owning Worker thread): the consumption face of direct-move buffers (Buffer fetches the chain, Consume advances — Chapter 65's operations used directly on engine data); the cross-thread entry (xrtNetPost delivers a task to the Stream's Worker — the engine edition of Chapter 56's scheduler Post); the wait family (Wait/WaitAvailable blocking — rendezvous from non-Worker threads; WaitAvailableAsync Future-style — for orchestration). Why the constraint: the precondition of lock-free direct-move buffer operations is "one thread" — the Worker constraint is the engine edition of Chapter 56's scheduler-binding discipline; crossing threads transfers execution via Post rather than adding locks — the zero-lock discipline throughout.

The server: dual-face Accept and shared ports

xnetserver (TCP Server): pull Accept (dequeue connections — take one as one arrives; upper-layer rate limiting/dispatch controls directly — the tcp_server example's shape); Future Accept (event-loop side — orchestration-friendly); blocking Accept (non-Worker threads — a management thread waiting for connections). Coexistence of both faces is the real migration-era shape (the tcp_server_sync example: Future and blocking takes on the same Server, side by side). Multi-listen and shared ports: one Server, multiple endpoints (v4 + v6 dual-stack); SO_REUSEPORT-style shared ports (server_tour's shared-port=1 assertion — platform support for multi-process shared listening). Accept statistics (accepted/queued — service observation fields).

Shutdown and storms: the server's wind-down

Listener Close (stop admissions — drain in-flight accepts), the connection cap (ServerConfig's queue depth — admission backpressure: when full, refuse new connections rather than queueing forever), shutdown order (Listener → Streams → Engine — Chapter 67's order, server edition). Connection storms' handling point: Accept queue-depth limits + fast refusal (low error cost) — rather than accepting then disconnecting (resources already allocated).

68.3示例

Complete program: the Stream full-family tour

From the repository example examples/network/tcp_stream_tour/main.c — the complete acceptance of the four send families + budget + flow control + half-close:

examples/network/tcp_stream_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;
}

#define EXAMPLE_DEADLINE_MS	INT64_C(3000)

/* 释放回调计数：SendRef/SendRefs 的零复制载荷离队时各执行一次。 */
static uint32 g_Releases = 0;

static void exampleCountRelease(ptr pContext, cbytes pData,
	size_t iSize)
{
	(void)pContext;
	(void)pData;
	(void)iSize;
	g_Releases = g_Releases + 1u;
}

/* Worker 任务上下文。 */
typedef struct examplestreamtask {
	xnetstream* pStream;
	volatile bool bDone;
	size_t iBufferSize;
	size_t iRead;
	size_t iConsume;
	char ReadOut[8];
	bool bSocket;
	bool bSetEvents;
	bool bSetData;
} examplestreamtask;

/* 最小事件集（仅 Close），不改变拉取模式。 */
static void exampleStreamClose(xnetstream* pStream,
	xnetresult Result, const xerror* pError, ptr pData)
{
	(void)pStream;
	(void)Result;
	(void)pError;
	(void)pData;
}


 * SetEvents + SetData——六件 Worker 专用操作一批完成。 */
static void exampleStreamTask(xnetworker* pWorker, ptr pData)
{
	examplestreamtask* pTask = (examplestreamtask*)pData;
	const xnetbuf* pBuffer;
	static xnetstreamevents s_Events = { 0 };
	static uint64 s_Tag = 0;

	(void)pWorker;
	s_Events.Close = exampleStreamClose;
	pBuffer = xrtNetStreamBuffer(pTask->pStream);
	pTask->iBufferSize = pBuffer != NULL ? xrtNetBufSize(pBuffer) : 0;
	pTask->iRead = xrtNetStreamRead(pTask->pStream,
		pTask->ReadOut, 4u);
	pTask->iConsume = xrtNetStreamConsume(pTask->pStream, 4u);
	pTask->bSocket = xrtNetStreamSocket(pTask->pStream) != NULL;
	pTask->bSetEvents = xrtNetStreamSetEvents(pTask->pStream,
		&s_Events, NULL);
	pTask->bSetData = xrtNetStreamSetData(pTask->pStream, &s_Tag);
	pTask->bDone = true;
}

static bool exampleWaitDone(volatile bool* pFlag)
{
	double iDeadline = exampleTimerLimit(EXAMPLE_DEADLINE_MS);

	while ( !*pFlag ) {
		if ( exampleTimerExpired(iDeadline) ) {
			return false;
		}
		xrtThreadYield();
	}
	return true;
}

static bool exampleWaitStreamState(xnetstream* pStream,
	xnetstreamstate State)
{
	double iDeadline = exampleTimerLimit(EXAMPLE_DEADLINE_MS);

	while ( xrtNetStreamState(pStream) != State ) {
		if ( exampleTimerExpired(iDeadline) ) {
			return false;
		}
		xrtThreadYield();
	}
	return true;
}

int main(void)
{
	static const char sExpected[] =
		"vec-joinref-zero-copyrefs-arefs-btake-ownedfile-block";
	/* take-owned 与 file-block 各 10 字节，不带结尾零。 */
	xnetengineconfig EngineConfig;
	xnetstreamconfig StreamConfig;
	xnetlistenconfig ListenConfig;
	xnetengine* pEngine = NULL;
	xnetlistener* pListener = NULL;
	xnetstream* pClient = NULL;
	xnetstream* pServer = NULL;
	xnetaddr Address;
	xnetref Refs[2];
	xnetspan Vec[2];
	xnetpost Post;
	examplestreamtask Task;
	xfuture* pAvailable = NULL;
	xnetstreamstats Stats;
	ptr pTake = NULL;
	xfile File;
	uint64 iFileSize = 0;
	int iResult = 1;

	/* 配置：默认值即自适应读取 + 背压水位。 */
	xrtNetStreamConfigInit(&StreamConfig);
	xrtNetEngineConfigInit(&EngineConfig);
	EngineConfig.Workers = 1;
	xrtNetListenConfigInit(&ListenConfig);
	(void)xrtNetAddrLoopback(&Address, XNET_FAMILY_IPV4, 0);
	ListenConfig.Address = Address;
	pEngine = xrtNetEngineCreate(&EngineConfig);
	if ( (pEngine == NULL) || !xrtNetEngineStart(pEngine) ) {
		goto Cleanup;
	}
	pListener = xrtNetListen(pEngine, &ListenConfig, NULL, NULL, NULL);
	if ( (pListener == NULL) ||
		 !xrtNetListenerLocal(pListener, &Address) ) {
		iResult = 2;
		goto Cleanup;
	}
	pClient = xrtNetStreamConnect(pEngine, &Address, 0,
		&StreamConfig, NULL, NULL);
	if ( (pClient == NULL) ||
		 !exampleWaitStreamState(pClient, XNET_STREAM_OPEN) ) {
		iResult = 3;
		goto Cleanup;
	}
	{
		double iDeadline = exampleTimerLimit(EXAMPLE_DEADLINE_MS);

		while ( (pServer = xrtNetListenerAccept(pListener)) ==
			NULL ) {
			if ( exampleTimerExpired(iDeadline) ) {
				iResult = 4;
				goto Cleanup;
			}
			xrtThreadYield();
		}
	}
	if ( !exampleWaitStreamState(pServer, XNET_STREAM_OPEN) ) {
		iResult = 4;
		goto Cleanup;
	}
	printf("stream-tour: config write-limit>=%s writable>=%s\n",
		xrtNetStreamWriteLimit(pClient) >= 65536u ? "64k" : "?",
		xrtNetStreamWritable(pClient) > 0u ? "0" : "?");

	/* 五种发送形态拼出 55 字节已知流。 */
	Vec[0].Data = (cbytes)"vec-";
	Vec[0].Size = 4;
	Vec[1].Data = (cbytes)"join";
	Vec[1].Size = 4;
	if ( xrtNetStreamSendVec(pClient, Vec, 2) != XNET_RESULT_OK ) {
		goto Cleanup;
	}
	if ( xrtNetStreamSendRef(pClient, "ref-zero-copy", 13,
		exampleCountRelease, &g_Releases) != XNET_RESULT_OK ) {
		goto Cleanup;
	}
	Refs[0].Data = (cbytes)"refs-a";
	Refs[0].Size = 6;
	Refs[0].Release = exampleCountRelease;
	Refs[0].Context = &g_Releases;
	Refs[1].Data = (cbytes)"refs-b";
	Refs[1].Size = 6;
	Refs[1].Release = exampleCountRelease;
	Refs[1].Context = &g_Releases;
	if ( xrtNetStreamSendRefs(pClient, Refs, 2) != XNET_RESULT_OK ) {
		goto Cleanup;
	}
	pTake = xrtMalloc(10);
	if ( pTake == NULL ) {
		goto Cleanup;
	}
	memcpy(pTake, "take-owned", 10);
	if ( xrtNetStreamSendTake(pClient, pTake, 10) ==
		 XNET_RESULT_OK ) {
		pTake = NULL;
	}
	else {
		goto Cleanup;
	}
	/* SendFile：临时文件一段已知区间。 */
	File = xrtOpen("xrt-stream-tour.tmp",
		XFILE_WRITE | XFILE_CREATE | XFILE_TRUNCATE);
	if ( (File == NULL) ||
		 !xrtWriteFull(File, "file-block", 10, NULL) ) {
		iResult = 5;
		goto Cleanup;
	}
	(void)xrtClose(File);
	File = xrtOpen("xrt-stream-tour.tmp", XFILE_READ);
	if ( (File == NULL) ||
		 (xrtNetStreamSendFile(pClient, File, 0, 10) !=
		  XNET_RESULT_OK) ) {
		iResult = 6;
		goto Cleanup;
	}
	(void)xrtClose(File);
	File = NULL;

	/* 对端整体核对 55 字节。 */
	if ( !xrtNetStreamWaitAvailable(pServer, sizeof(sExpected) - 1u,EXAMPLE_DEADLINE_MS, NULL) ) {
		iResult = 7;
		goto Cleanup;
	}
	{
		xnetbytes* pBytes = xrtNetStreamRecv(pServer,
			sizeof(sExpected) - 1u,EXAMPLE_DEADLINE_MS, NULL);
		xbytesview View = xrtNetBytesView(pBytes);
		bool bMatch = (View.Size == sizeof(sExpected) - 1u) &&
			(memcmp(View.Data, sExpected, View.Size) == 0);

		xrtNetBytesDestroy(pBytes);
		if ( !bMatch ) {
			iResult = 8;
			goto Cleanup;
		}
	}
	printf("sends: vec+ref+refs+take+file = %zu bytes verified\n",
		sizeof(sExpected) - 1u);
	/* 发送队列排空：Pending 归零。 */
	{
		double iDeadline = exampleTimerLimit(EXAMPLE_DEADLINE_MS);

		while ( xrtNetStreamPending(pClient) != 0u ) {
			if ( exampleTimerExpired(iDeadline) ) {
				iResult = 9;
				goto Cleanup;
			}
			xrtThreadYield();
		}
	}

	/* Worker 读取族：先备好 8 字节再投递任务。 */
	if ( xrtNetStreamSend(pClient, "buf-demo", 8) != XNET_RESULT_OK ) {
		goto Cleanup;
	}
	if ( !xrtNetStreamWaitAvailable(pServer, 8u,EXAMPLE_DEADLINE_MS, NULL) ) {
		iResult = 10;
		goto Cleanup;
	}
	memset(&Task, 0, sizeof(Task));
	Task.pStream = pServer;
	if ( !xrtNetPostInit(&Post) || !xrtNetPost(
		xrtNetStreamWorker(pServer), &Post, exampleStreamTask,
		&Task) || !exampleWaitDone(&Task.bDone) ) {
		iResult = 11;
		goto Cleanup;
	}
	printf("worker-read: buffer=%zu read=\"%.*s\" consume=%zu",
		Task.iBufferSize, (int)Task.iRead, Task.ReadOut,
		Task.iConsume);
	printf(" socket+setevents+setdata=%s\n",
		(Task.bSocket && Task.bSetEvents && Task.bSetData &&
		 (xrtNetStreamData(pServer) != NULL)) ? "ok" : "fail");
	if ( (Task.iBufferSize != 8u) || (Task.iRead != 4u) ||
		 (memcmp(Task.ReadOut, "buf-", 4) != 0) ||
		 (Task.iConsume != 4u) || !Task.bSocket ||
		 !Task.bSetEvents || !Task.bSetData ) {
		iResult = 12;
		goto Cleanup;
	}

	/* 等待族：READ 条件 / 字节数门槛 / Future 门槛 / WRITE 条件。 */
	if ( xrtNetStreamSend(pServer, "wake", 4) != XNET_RESULT_OK ) {
		goto Cleanup;
	}
	printf("waits: read=%d",
		xrtNetStreamWait(pClient, XNET_STREAM_WAIT_READ,EXAMPLE_DEADLINE_MS, NULL) ? 1 : 0);
	printf(" available=%d",
		xrtNetStreamWaitAvailable(pClient, 4u,EXAMPLE_DEADLINE_MS, NULL) ? 1 : 0);
	{
		/* BytesRef：接收结果可共享持有——引用与原件各销毁一次。 */
		xnetbytes* pBytes = xrtNetStreamRecv(pClient, 4,EXAMPLE_DEADLINE_MS, NULL);
		xnetbytes* pShared = xrtNetBytesRef(pBytes);

		xrtNetBytesDestroy(pShared);
		xrtNetBytesDestroy(pBytes);
	}
	if ( xrtNetStreamSend(pServer, "more", 4) != XNET_RESULT_OK ) {
		goto Cleanup;
	}
	pAvailable = xrtNetStreamWaitAvailableAsync(pClient, 4u);
	printf(" available-async=%d",
		(pAvailable != NULL) &&
			(xrtFutureWaitFor(pAvailable, EXAMPLE_DEADLINE_MS) ==
			 XWAIT_OK) ? 1 : 0);
	{
		xnetbytes* pBytes = xrtNetStreamRecv(pClient, 4,EXAMPLE_DEADLINE_MS, NULL);

		xrtNetBytesDestroy(pBytes);
	}
	printf(" write=%d\n",
		xrtNetStreamWait(pClient, XNET_STREAM_WAIT_WRITE,EXAMPLE_DEADLINE_MS, NULL) ? 1 : 0);

	/* 流控：Pause 后数据滞留内核，Resume 后到达。 */
	xrtNetStreamPause(pServer);
	if ( xrtNetStreamSend(pClient, "held", 4) != XNET_RESULT_OK ) {
		goto Cleanup;
	}
	{
		double iHold = exampleTimerLimit(200);
		bool bHeld = true;

		while ( exampleTimerExpired(iHold) == false ) {
			if ( xrtNetStreamAvailable(pServer) != 0u ) {
				bHeld = false;
				break;
			}
			xrtThreadYield();
		}
		printf("flow: pause=%s", bHeld ? "held" : "leaked");
	}
	if ( !xrtNetStreamResume(pServer) ||
		 !xrtNetStreamWaitAvailable(pServer, 4u,EXAMPLE_DEADLINE_MS, NULL) ) {
		iResult = 13;
		goto Cleanup;
	}
	{
		xnetbytes* pBytes = xrtNetStreamRecv(pServer, 4,EXAMPLE_DEADLINE_MS, NULL);

		xrtNetBytesDestroy(pBytes);
		printf(" resume=delivered\n");
	}

	/* 半关闭：客户端 ShutdownWrite → 服务端读到 EOF（Recv 空 + 终态）。 */
	if ( !xrtNetStreamShutdownWrite(pClient) ) {
		iResult = 14;
		goto Cleanup;
	}
	{
		xnetbytes* pBytes = xrtNetStreamRecv(pServer, 0,EXAMPLE_DEADLINE_MS, NULL);
		xbytesview View = xrtNetBytesView(pBytes);
		int iEof = (View.Size == 0u);

		xrtNetBytesDestroy(pBytes);
		if ( !iEof ) {
			iResult = 15;
			goto Cleanup;
		}
		printf("shutdown: peer-eof=1");
	}

	/* 统计与终态错误。 */
	if ( !xrtNetStreamStats(pClient, &Stats) ||
		 (Stats.SentBytes < 63u) ) {
		iResult = 16;
		goto Cleanup;
	}
	printf(" stats(sent=%llu)",
		(unsigned long long)Stats.SentBytes);
	printf(" error=%s\n",
		xrtNetStreamError(pClient) == NULL ? "(none)" : "err");
	(void)xrtNetStreamLocal(pClient, &Address);
	(void)iFileSize;
	iResult = 0;

Cleanup:
	xrtFree(pTake);
	xrtFutureDestroy(pAvailable);
	(void)xrtFileDelete("xrt-stream-tour.tmp");
	if ( pClient != NULL ) {
		(void)xrtNetStreamAbort(pClient);
		(void)exampleWaitStreamState(pClient, XNET_STREAM_CLOSED);
	}
	if ( pServer != NULL ) {
		(void)xrtNetStreamAbort(pServer);
		(void)exampleWaitStreamState(pServer, XNET_STREAM_CLOSED);
	}
	if ( pListener != NULL ) {
		double iEnd = exampleTimerLimit(EXAMPLE_DEADLINE_MS);

		(void)xrtNetListenerClose(pListener);
		while ( xrtNetListenerState(pListener) !=
			XNET_LISTENER_CLOSED ) {
			if ( exampleTimerExpired(iEnd) ) {
				break;
			}
			xrtThreadYield();
		}
	}
	xrtNetStreamDestroy(pClient);
	xrtNetStreamDestroy(pServer);
	xrtNetListenerDestroy(pListener);
	if ( (pEngine != NULL) && !xrtNetEngineDestroy(pEngine) &&
		 (iResult == 0) ) {
		iResult = 17;
	}
	return iResult;
}
$ gcc -O1 -DXRT_MODULE_ALL -I single -include xrt.h impl.c examples/network/tcp_stream_tour/main.c -lws2_32 -liphlpapistream-tour: config write-limit>=64k writable>=0sends: vec+ref+refs+take+file = 53 bytes verifiedworker-read: buffer=8 read="buf-" consume=4 socket+setevents+setdata=okwaits: read=1 available=1 available-async=1 write=1flow: pause=held resume=deliveredshutdown: peer-eof=1 stats(sent=65) error=(none)

What just happened. (1) Configuration tier: WriteLimit at least 64k, Writable starting at 0 — the write-budget trio's starting point. (2) The four send families in joint performance: Vec+Ref+Refs+Take+File each send one stretch, 53 bytes verified in full — all five ownership tiers walked on one stream. (3) The worker-read line: the in-Worker read trio (Buffer = 8 bytes available, Read fetches "buf-", Consume advances 4) + Socket/SetEvents/SetData — hands-on with the Worker-constraint APIs. (4) The waits line: four wait forms (blocking read / Available snapshot / Async Future / write wait), one each. (5) The flow line: after Pause, data held (engine retains, doesn't advance); after Resume, delivered — the two beats of flow control. (6) The shutdown line: the peer reads EOF, stats' sent count (65 = 53 + 12 assorted extra bytes), error=(none) — a clean final state. This tour is the Stream data plane's full-operation acceptance — a checklist for production code.

Complete program: dual-face Accept, side by side

From examples/network/tcp_server_sync/main.c — Future and blocking takes coexisting:

examples/network/tcp_server_sync/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;
}



typedef struct example_tcp_server_wait {
	xnetengine* Engine;
	xnetserver* Server;
	xnetstream* ClientFuture;
	xnetstream* AcceptedFuture;
	xnetstream* ClientSync;
	xnetstream* AcceptedSync;
	xnetaddr Local;
} example_tcp_server_wait;



/* 启动使用动态回环端口的拉取式 TCP Server。 */
static bool exampleTcpServerWaitStart(example_tcp_server_wait* pState)
{
	xnetengineconfig EngineConfig;
	xnetserverconfig ServerConfig;

	memset(pState, 0, sizeof(*pState));
	xrtNetEngineConfigInit(&EngineConfig);
	EngineConfig.Workers = 2u;
	pState->Engine = xrtNetEngineCreate(&EngineConfig);
	if ( (pState->Engine == NULL) ||
		!xrtNetEngineStart(pState->Engine) ) {
		return false;
	}
	xrtNetServerConfigInit(&ServerConfig);
	if ( !xrtNetAddrLoopback(
		&ServerConfig.Listen.Address,
		XNET_FAMILY_IPV4,
		0
	) ) {
		return false;
	}
	pState->Server = xrtNetServerStart(
		pState->Engine,
		&ServerConfig,
		NULL,
		NULL,
		NULL
	);
	return (pState->Server != NULL) && xrtNetServerLocal(
		pState->Server,
		0,
		&pState->Local
	);
}



/* 连接刚启动的本地 Server。 */
static xnetstream* exampleTcpServerWaitConnect(
	example_tcp_server_wait* pState
)
{
	return xrtNetStreamConnect(
		pState->Engine,
		&pState->Local,
		1u,
		NULL,
		NULL,
		NULL
	);
}



/* 终止连接、关闭 Server，并等待 Engine 中的对象全部退出。 */
static bool exampleTcpServerWaitStop(example_tcp_server_wait* pState)
{
	xnetstream* Streams[] = {
		pState->ClientFuture,
		pState->AcceptedFuture,
		pState->ClientSync,
		pState->AcceptedSync
	};
	double Deadline = exampleTimerLimit(INT64_C(5000));
	bool bClosed = true;
	size_t i;

	for ( i = 0; i < (sizeof(Streams) / sizeof(Streams[0])); i++ ) {
		if ( Streams[i] != NULL ) {
			(void)xrtNetStreamAbort(Streams[i]);
		}
	}
	if ( pState->Server != NULL ) {
		(void)xrtNetServerClose(pState->Server);
	}
	for ( ;; ) {
		bool bDone = (pState->Server == NULL) ||
			(xrtNetServerState(pState->Server) == XNET_SERVER_CLOSED);

		for ( i = 0; i < (sizeof(Streams) / sizeof(Streams[0])); i++ ) {
			bDone = bDone && ((Streams[i] == NULL) ||
				(xrtNetStreamState(Streams[i]) == XNET_STREAM_CLOSED));
		}
		if ( bDone ) {
			break;
		}
		if ( exampleTimerExpired(Deadline) ) {
			bClosed = false;
			break;
		}
		xrtThreadYield();
	}
	for ( i = 0; i < (sizeof(Streams) / sizeof(Streams[0])); i++ ) {
		xrtNetStreamDestroy(Streams[i]);
	}
	xrtNetServerDestroy(pState->Server);
	if ( (pState->Engine != NULL) &&
		!xrtNetEngineDestroy(pState->Engine) ) {
		bClosed = false;
	}
	return bClosed;
}



/* 依次演示 Future Accept 和非 Worker 线程上的阻塞 Accept。 */
int main(void)
{
	example_tcp_server_wait State;
	xfutureresult Result;
	xfuture* pAccept = NULL;
	int iResult = 1;

	if ( !exampleTcpServerWaitStart(&State) ) {
		goto Cleanup;
	}
	pAccept = xrtNetServerAcceptAsync(State.Server);
	State.ClientFuture = exampleTcpServerWaitConnect(&State);
	if ( (pAccept == NULL) || (State.ClientFuture == NULL) ||
		(xrtFutureWaitFor(
			pAccept,
			INT64_C(5000)
		 ) != XWAIT_OK) ||
		!xrtFutureResult(pAccept, &Result) ||
		(Result.State != XFUTURE_RESOLVED) ) {
		goto Cleanup;
	}
	State.AcceptedFuture = xrtNetStreamRef(
		(xnetstream*)Result.Value
	);
	xrtFutureDestroy(pAccept);
	pAccept = NULL;
	if ( State.AcceptedFuture == NULL ) {
		goto Cleanup;
	}

	State.ClientSync = exampleTcpServerWaitConnect(&State);
	if ( State.ClientSync == NULL ) {
		goto Cleanup;
	}
	State.AcceptedSync = xrtNetServerAcceptWait(
		State.Server,INT64_C(5000),
		NULL
	);
	if ( State.AcceptedSync == NULL ) {
		goto Cleanup;
	}
	printf(
		"accepted Future and synchronous connections on port %u\n",
		(unsigned)State.Local.Port
	);
	iResult = 0;

Cleanup:
	if ( (pAccept != NULL) &&
		(xrtFutureState(pAccept) == XFUTURE_PENDING) ) {
		(void)xrtFutureCancel(pAccept);
		(void)xrtFutureWaitFor(pAccept, INT64_C(5000));
	}
	xrtFutureDestroy(pAccept);
	if ( !exampleTcpServerWaitStop(&State) && (iResult == 0) ) {
		iResult = 2;
	}
	return iResult;
}
$ gcc -O1 -DXRT_MODULE_ALL -I single -include xrt.h impl.c examples/network/tcp_server_sync/main.c -lws2_32 -liphlpapiaccepted Future and synchronous connections on port NNNNN

What just happened. (1) A Server started on a dynamic loopback port (port 0 + Local retrieval — Chapter 67's posture). (2) Future Accept: the event-loop side takes the first connection — orchestration-friendly (later operations compose). (3) Blocking Accept: the management-thread side takes the second — the rendezvous shape for ordinary threads. (4) Both postures on the same Server at once — proof of the real migration-era shape (moving from the blocking to the event model needs no one-step switchover). The server_tour example adds multi-endpoint/shared-port/Accept statistics — the server's full interface.

68.4契约

  • Five send tiers: Send copy / Vec gather / Ref reference + callback exactly once / Take takeover / File kernel delegation — chosen by source (corresponding to Chapter 65's four chain-in kinds).
  • Write budget: WriteLimit cap + Pending watermark + Writable headroom — send-side enforcement against slow consumers.
  • Worker constraint: Buffer/Read/Consume inside a Worker only; cross-thread via Post — the zero-lock discipline.
  • Flow control: Pause/Resume as the manual gear — pause retains data; the upstream decides to drop or disconnect.
  • Dual-face Accept: pull/Future/blocking coexisting — gradual migration.
  • Admission backpressure: Accept queue-depth limit + fast refusal — storms stopped at the door.

Common Mistakes

Pitfall 1: freeing data early after SendRef：Pitfall 1: freeing data early after SendRef

Symptoms: the receiver occasionally gets dirty data — the send callback hasn't arrived and the data source is already freed; same family as Chapter 65's Pitfall 1.

Cause: SendRef is reference sending — a data freeze window before the release callback (the engine is reading it).

错误写法
char* Block = xrtMalloc(1024);
Fill(Block);
xrtNetStreamSendRef(Stream, Block, 1024, releaseCb, Ctx);
xrtFree(Block);   /* freed before the callback — the engine is still reading */
正确写法
xrtNetStreamSendTake(Stream, TakeBlock, 1024);   /* takeover — the engine frees when sent */
/* or SendRef + callback: the callback fires exactly when the data is no longer referenced — don't touch it until then */
Pitfall 2: budget-less sending blowing up memory：Pitfall 2: budget-less sending blowing up memory

Symptoms: service memory grows with slow clients — one client never reads, engine buffers back up without bound; an OOM crash at the late-night traffic peak.

Cause: no WriteLimit set — sending is uncapped, and the backlog from the shrinking TCP window all piles into service memory.

错误写法
xnetstreamconfig Config;
xrtNetStreamConfigInit(&Config);   /* no budget limit by default */
xrtNetStreamSend(Stream, HugeData, HugeSize);   /* unbounded backlog */
正确写法
xnetstreamconfig Config;
xrtNetStreamConfigInit(&Config);
Config.WriteLimit = 1u << 20;   /* 1MB budget */
/* check headroom before sending */
if ( xrtNetStreamWritable(Stream) < Need ) {
	WaitOrPause(Stream);   /* backpressure bites — wait or pause, don't pile on */
}

Programming Exercises

Basic: five-tier comparison

Send the same data through each of the five tiers (File tier via a temp file) — verify in full at the receiver; print and confirm each tier's release timing (callback/takeover).

Advanced: a backpressure experiment

A slow client (read 1 byte, sleep 100ms) + the server sending heavily: with WriteLimit versus unlimited — watch the service memory curve (Chapter 6's stats) and the Pending watermark; verify the over-budget handling path.

Challenge: a file-distribution service

A static file service: multiple clients concurrently downloading the same file — SendFile chunked by range (chunk budget) + per-client WriteLimit + a global admission cap. Acceptance: serving a 1GB file keeps peak service memory <100MB (proof of chunking + budget); slow clients don't affect the others; shutdown winds down fully (the trio protocol); per-client downloads verify identical Hash64.

Summary

TopicQuick reference
Five send tiersSend copy / Vec gather / Ref reference + callback / Take takeover / File kernel
Write budgetWriteLimit cap + Pending watermark + Writable headroom — backpressure against slow clients
Worker constraintBuffer/Read/Consume inside a Worker only — cross-thread via Post delivery
Flow controlPause/Resume — the manual gear that retains data
Half-closeShutdownWrite — the peer reads EOF (one-way end)
Dual-face Acceptpull/Future/blocking coexisting — gradual migration
Admission backpressureAccept queue-depth limit + fast refusal — storms stopped outside the door

📖 API references in this chapter:tcp (81 entries)tcp_server (22 entries)net (247 entries)

Click for full signatures, parameter constraints, return values and error notes