本章导读

TCP 下册讲三块进阶：发送四族（Vec 聚集 / Ref 零拷贝+释放回调 / Take 接管 / File 内核文件发送）与写预算（WriteLimit 限额 / Writable 可写 / Pending 在途——发送背压的三件套）；读取与 Worker 约束（Buffer/Read/Consume 仅 Worker 内——直移缓冲的消费面；跨线程经 Post 投递）；服务端形态（xnetserver——拉取 Accept（队列取连接）、Future Accept 与阻塞 Accept 双面并存、多监听与共享端口）。金标准章（第 67 章）的四态与回显是基础形态——本章是它的完全体。

68.1引入

三个进阶场景。场景一：文件服务器要把 1GB 文件发出去——read 整个文件到内存再 Send 是灾难（内存爆炸）；xrtNetStreamSendFile（内核文件发送+区间有界——第 46 章文件章预告的 sendfile 落点）不经过用户态缓冲。场景二：代理要把上游收到的段链转发下游——复制一份再发是浪费；SendRef 直接引用移交（零拷贝+释放回调）。场景三：服务端要限制单个连接的发送积压（恶意客户端不读——缓冲无限膨胀）；写预算（WriteLimit 设上限、Pending 查在途、Writable 查余量）是背压的发送侧执法。

这三个场景的共同主题是发送面的所有权与预算——第 5 章语言在 IO 出口的全集：Send 复制、SendVec 聚集复制、SendRef 借用+回调、SendTake 接管、SendFile 委托内核。五档按"数据从哪来、谁释放"选——与第 65 章缓冲链四类追加一一对应（那个是入链、这个是出链）。

68.2概念

发送四族与文件发送

入口所有权适用
Send复制小数据安全路径（金标准章）
SendVec聚集复制多缓冲一次发出（第 21 章批量）
SendRef / SendRefs引用+释放回调段链直移（零拷贝——金标准 SendBuffer 的族亲）
SendTake接管xrtMalloc 块（发完引擎释放）
SendFile委托内核文件区间发送（sendfile/mmap 后端）

SendRef 的释放回调：数据发完（或流销毁）回调恰好一次——与第 65 章 AppendRef 同款纪律；SendRefs 是多段变体（段链一次发）。SendFile 的区间：偏移+长度有界——文件的分块发送控制内存（每块内核缓冲而非整文件）。

写预算：发送背压三件套

1WriteLimit(N)单连接发送预算上限（在途+排队 ≤ N）
2Pending()当前在途/排队字节数——水位表
3Writable()余量 = Limit - Pending——可安全提交的量
4超预算 Send失败或阻塞（按调用形态）——背压生效点
5流控配套Pause/Resume——暂停后引擎停发（内存兜底）

背压的完整链条：对端不读 → TCP 窗口收缩 → 引擎缓冲积压 → Pending 逼近 Limit → 新 Send 被拒 → 上游感知（处理者看到发送失败/需等待）——慢消费者无法拖垮服务内存。Pause/Resume 是手动挡（Pausable 语义：Pause 后引擎保留数据不推进——上游可决策丢弃或断开）。第 21 章队列背压的连接版。

读取与 Worker 约束

Worker 内 API（Buffer/Read/Consume 等——必须在 Stream 所属 Worker 线程调用）：直移缓冲的消费面（Buffer 取链、Consume 前进——第 65 章操作在引擎数据上直接用）；跨线程入口（xrtNetPost 投递任务到 Stream 所属 Worker——第 56 章调度器 Post 的引擎版）；等待族（Wait/WaitAvailable 阻塞式——非 Worker 线程的汇合用；WaitAvailableAsync Future 式——编排用）。为什么约束：直移缓冲操作免锁的前提是"单线程"——Worker 约束就是第 56 章调度器绑定纪律的引擎版；跨线程经 Post 传递执行权而非加锁——零锁纪律贯穿。

服务端：双面 Accept 与共享端口

xnetserver（TCP Server）：拉取 Accept（队列式取连接——来一个取一个，上层限流/分发直接控制——tcp_server 范例形态）；Future Accept（事件循环侧——编排友好）；阻塞 Accept（非 Worker 线程——管理线程等连接）。双面并存是迁移期的真实形态（tcp_server_sync 范例：同一 Server 上 Future 与阻塞两种取法同场）。多监听与共享端口：一个 Server 多端点（v4+v6 双栈）；SO_REUSEPORT 式共享端口（server_tour 的 shared-port=1 断言——多进程共享监听的平台支持）。Accept 统计（accepted/queued——服务观测字段）。

停机与风暴：服务端收尾

Listener Close（停止接入——在途 accept 排空）、连接上限（ServerConfig 的队列深度——接入背压：满了拒绝新连接而不是无限排）、停机顺序（Listener → 各 Stream → Engine——第 67 章顺序的服务端版）。连接风暴的处理位：Accept 队列深度限制 + 快速拒绝（错误成本低）——而不是接受后再断（已分配资源）。

68.3示例

完整程序：Stream 全族巡礼

来自仓库范例 examples/network/tcp_stream_tour/main.c——发送四族+预算+流控+半关闭的完整验收：

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)

刚才发生了什么。 ① 配置层：WriteLimit 至少 64k、Writable 起点为 0——写预算三件套的起点。② 发送四族联合：Vec+Ref+Refs+Take+File 各发一段、53 字节全量验证——五档所有权在一条流上全部走到。③ worker-read 行：Worker 内读缓冲三件（Buffer=8 字节可用、Read 取 "buf-"、Consume 前进 4）+ Socket/SetEvents/SetData——Worker 约束 API 的实操。④ waits 行：四种等待形态（阻塞读/Available 快照/Async Future/写等待）各验一次。⑤ flow 行：Pause 后数据 held（引擎保留不推进）、Resume 后 delivered——流控两拍。⑥ shutdown 行：对端读到 EOF、stats 的 sent 计数（65=53+12 等附加字节）、error=(none) 干净终态。这个 tour 是 Stream 数据面的全操作验收——生产代码的检查表。

完整程序：双面 Accept 同场

来自 examples/network/tcp_server_sync/main.c——Future 与阻塞取连接并存：

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

刚才发生了什么。 ① 动态回环端口的 Server 启动（端口 0 + Local 取回——第 67 章姿势）。② Future Accept：事件循环侧取第一条连接——编排友好（后续操作可组合）。③ 阻塞 Accept：管理线程侧取第二条——普通线程的汇合形态。④ 两种姿势同一 Server 同场——迁移期真实形态的实证（从阻塞模型渐进到事件模型不需要一步切换）。server_tour 范例补多端点/共享端口/Accept 统计——服务端全接口。

68.4契约

  • 发送五档：Send 复制 / Vec 聚集 / Ref 引用+回调恰好一次 / Take 接管 / File 委托内核——按来源选（与第 65 章入链四类对应）。
  • 写预算：WriteLimit 上限 + Pending 水位 + Writable 余量——慢消费者背压的发送侧执法。
  • Worker 约束：Buffer/Read/Consume 仅 Worker 内；跨线程经 Post——零锁纪律。
  • 流控：Pause/Resume 手动挡——暂停保留数据，上游决策丢弃或断开。
  • 双面 Accept：拉取/Future/阻塞三形态并存——迁移渐进。
  • 接入背压：Accept 队列深度限制+快速拒绝——风暴挡在门外。

常见错误

坑 1：SendRef 后提前释放数据

症状：接收端偶发收到脏数据——发送回调还没来，数据源已被释放；与第 65 章坑 1 同族。

原因：SendRef 是引用发送——释放回调前数据冻结期（引擎正在读它）。

错误写法
char* Block = xrtMalloc(1024);
Fill(Block);
xrtNetStreamSendRef(Stream, Block, 1024, releaseCb, Ctx);
xrtFree(Block);   /* 回调没来就释放——引擎正在读 */
正确写法
xrtNetStreamSendTake(Stream, TakeBlock, 1024);   /* 接管——引擎发完释放 */
/* 或 SendRef+回调：回调恰好在数据不再被引用后触发——不动它直到回调 */
坑 2：无预算发送打爆内存

症状：服务内存随慢客户端增长——某客户端不读，引擎缓冲无限积压；OOM 崩溃在深夜流量高峰。

原因：没设 WriteLimit——发送无上限，TCP 窗口收缩的积压全部堆在服务内存。

错误写法
xnetstreamconfig Config;
xrtNetStreamConfigInit(&Config);   /* 默认无预算限制 */
xrtNetStreamSend(Stream, HugeData, HugeSize);   /* 无界积压 */
正确写法
xnetstreamconfig Config;
xrtNetStreamConfigInit(&Config);
Config.WriteLimit = 1u << 20;   /* 1MB 预算 */
/* 发送前查余量 */
if ( xrtNetStreamWritable(Stream) < Need ) {
	WaitOrPause(Stream);   /* 背压生效——等或暂停，不上冲 */
}

编程练习

基础：五档对照

同一数据分别用五档发送（File 档用临时文件）——接收端全量验证；各档的释放时机（回调/接管）打印确认。

进阶：背压实验

慢客户端（收 1 字节歇 100ms）+ 服务大量发送：有 WriteLimit vs 无限制对照——观察服务内存曲线（第 6 章统计）与 Pending 水位；超预算的处置路径验证。

挑战：文件分发服务

静态文件服务：多客户端并发下载同一文件——SendFile 区间分块（块预算）+ 每客户端 WriteLimit + 全局接受上限。验收标准：1GB 文件服务内存峰值 <100MB（分块+预算的实证）；慢客户端不影响其他客户端；停机全部收尾（三件套协议）；下载校验逐客户端 Hash64 一致。

小结

知识点速查
发送五档Send 复制/Vec 聚集/Ref 引用+回调/Take 接管/File 内核
写预算WriteLimit 上限 + Pending 水位 + Writable 余量——慢客户端背压
Worker 约束Buffer/Read/Consume 仅 Worker 内——跨线程 Post 投递
流控Pause/Resume——保留数据的手动挡
半关闭ShutdownWrite——对端读 EOF（单向结束）
双面 Accept拉取/Future/阻塞并存——迁移渐进
接入背压Accept 队列深限+快速拒绝——风暴挡门外

📖 本章涉及的 API 参考手册：tcp（81 个条目）tcp_server（22 个条目）net（247 个条目）

点击查看完整签名、参数约束、返回值与错误说明