Глава 68 TCP (часть 2): противодавление, ссылки и сервер

 */

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

 */


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