Глава 69 UDP-датаграммы

 */


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



/* 在截止时间内等待 UDP 状态转换。 */
static bool exampleUdpWaitState(xnetudp* pUdp, xnetudpstate State)
{
	double iDeadline = exampleTimerLimit(3000);

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



/* 在截止时间内拉取一个 UDP 包。 */
static xnetudppacket* exampleUdpReceive(xnetudp* pUdp)
{
	double iDeadline = exampleTimerLimit(3000);
	xnetudppacket* pPacket;

	for ( ;; ) {
		pPacket = xrtNetUdpReceive(pUdp);
		if ( pPacket != NULL ) {
			return pPacket;
		}
		if ( exampleTimerExpired(iDeadline) ) {
			return NULL;
		}
		xrtThreadYield();
	}
}



/* 建立绑定端和连接端，完成一次 UDP 回环收发。 */
int main(void)
{
	xnetengineconfig EngineConfig;
	xnetudpconfig UdpConfig;
	xnetengine* pEngine = NULL;
	xnetudp* pServer = NULL;
	xnetudp* pClient = NULL;
	xnetudppacket* pPacket = NULL;
	xnetaddr Address;
	xnetdgramcontrol Control;
	str sEndpoint = NULL;
	int iResult = 1;

	xrtNetEngineConfigInit(&EngineConfig);
	EngineConfig.Workers = 2;
	xrtNetUdpConfigInit(&UdpConfig);
	UdpConfig.ReceiveMeta = XNET_DGRAM_META_DESTINATION;
	pEngine = xrtNetEngineCreate(&EngineConfig);
	if ( (pEngine == NULL) || !xrtNetEngineStart(pEngine) ) {
		goto Cleanup;
	}
	(void)xrtNetAddrLoopback(&Address, XNET_FAMILY_IPV4, 0);
	pServer = xrtNetUdpBind(pEngine, &Address, 0,
		&UdpConfig, NULL, NULL);
	if ( (pServer == NULL) || !xrtNetUdpLocal(pServer, &Address) ) {
		iResult = 2;
		goto Cleanup;
	}
	sEndpoint = xrtNetAddrEndpointString(&Address);
	if ( sEndpoint == NULL ) {
		iResult = 3;
		goto Cleanup;
	}
	printf("UDP server: %s\n", sEndpoint);
	xrtFree(sEndpoint);
	sEndpoint = NULL;
	pClient = xrtNetUdpConnect(pEngine, &Address, 1,
		NULL, NULL, NULL);
	if ( (pClient == NULL) ||
		 !exampleUdpWaitState(pServer, XNET_UDP_OPEN) ||
		 !exampleUdpWaitState(pClient, XNET_UDP_OPEN) ) {
		iResult = 4;
		goto Cleanup;
	}
	if ( xrtNetUdpSend(pClient, "hello UDP", 9) !=
		 XNET_RESULT_OK ) {
		iResult = 5;
		goto Cleanup;
	}
	pPacket = exampleUdpReceive(pServer);
	if ( pPacket == NULL ) {
		iResult = 6;
		goto Cleanup;
	}
	printf("received: %.*s\n",
		(int)xrtNetUdpPacketSize(pPacket),
		(cstr)xrtNetUdpPacketData(pPacket));
	if ( ((xrtNetUdpPacketMeta(pPacket)->Flags &
		  XNET_DGRAM_META_DESTINATION) == 0) ||
		 ((xrtNetUdpSendControlAvailable(pServer) &
		  XNET_DGRAM_CONTROL_SOURCE) == 0) ) {
		iResult = 7;
		goto Cleanup;
	}
	memset(&Control, 0, sizeof(Control));
	Control.Flags = XNET_DGRAM_CONTROL_SOURCE;
	Control.Source = xrtNetUdpPacketMeta(pPacket)->Destination;
	Control.Source.Port = 0;
	if ( xrtNetUdpSendMsg(
		pServer,
		xrtNetUdpPacketRemote(pPacket),
		&Control,
		"reply",
		5
	) != XNET_RESULT_OK ) {
		iResult = 8;
		goto Cleanup;
	}
	xrtNetUdpPacketDestroy(pPacket);
	pPacket = exampleUdpReceive(pClient);
	if ( (pPacket == NULL) || (xrtNetUdpPacketSize(pPacket) != 5) ||
		 (memcmp(xrtNetUdpPacketData(pPacket), "reply", 5) != 0) ) {
		iResult = 9;
		goto Cleanup;
	}
	printf("response: %.*s\n",
		(int)xrtNetUdpPacketSize(pPacket),
		(cstr)xrtNetUdpPacketData(pPacket));
	iResult = 0;

Cleanup:
	xrtNetUdpPacketDestroy(pPacket);
	xrtFree(sEndpoint);
	if ( pClient != NULL ) {
		(void)xrtNetUdpClose(pClient);
		(void)exampleUdpWaitState(pClient, XNET_UDP_CLOSED);
	}
	if ( pServer != NULL ) {
		(void)xrtNetUdpClose(pServer);
		(void)exampleUdpWaitState(pServer, XNET_UDP_CLOSED);
	}
	xrtNetUdpDestroy(pClient);
	xrtNetUdpDestroy(pServer);
	if ( (pEngine != NULL) && !xrtNetEngineDestroy(pEngine) &&
		 (iResult == 0) ) {
		iResult = 10;
	}
	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;
}



/* 等 Future 成功解析。 */
static bool exampleFutureWait(xfuture* pFuture)
{
	return (pFuture != NULL) &&
		(xrtFutureWaitFor(pFuture, INT64_C(3000)) == XWAIT_OK) &&
		(xrtFutureState(pFuture) == XFUTURE_RESOLVED);
}



/* 在截止时间内等 UDP 进入指定状态。 */
static bool exampleWaitState(xnetudp* pUdp, xnetudpstate State)
{
	double iDeadline = exampleTimerLimit(3000);

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



/* 收满 count 条（普通单包拉取，等待全部到齐）。 */
static size_t exampleDrainCount(xnetudp* pUdp, size_t iWant)
{
	size_t iGot = 0;
	double iDeadline = exampleTimerLimit(3000);

	while ( iGot < iWant ) {
		xnetudppacket* pPacket = xrtNetUdpReceive(pUdp);

		if ( pPacket != NULL ) {
			xrtNetUdpPacketDestroy(pPacket);
			++iGot;
			iDeadline = exampleTimerLimit(3000);
			continue;
		}
		if ( exampleTimerExpired(iDeadline) ) {
			break;
		}
		xrtThreadYield();
	}
	return iGot;
}



/* 服务端绑小接收槽演示截断；客户端连发演示三种批量接收。 */
int main(void)
{
	xnetengineconfig EngineConfig;
	xnetudpconfig UdpConfig;
	xnetengine* pEngine = NULL;
	xnetudp* pServer = NULL;
	xnetudp* pClient = NULL;
	xnetaddr Address;
	xnetudppacket* pPackets[8];
	xnetudppacket* pKept = NULL;
	xnetudpbatch* pBatch = NULL;
	xnetudpbatch* pBatchRef = NULL;
	xfuture* pFuture = NULL;
	xfuture* pWritable = NULL;
	xfuture* pErrorFuture = NULL;
	char Big[64];
	size_t iCount;
	size_t i;
	int iResult = 1;

	xrtNetEngineConfigInit(&EngineConfig);
	EngineConfig.Workers = 1;
	xrtNetUdpConfigInit(&UdpConfig);
	UdpConfig.ReceiveSize = 16u;  /* 对端发 64 字节 → 截断前缀 */
	UdpConfig.Truncation = XNET_UDP_TRUNCATE_DELIVER;
	pEngine = xrtNetEngineCreate(&EngineConfig);
	if ( (pEngine == NULL) || !xrtNetEngineStart(pEngine) ) {
		goto Cleanup;
	}
	(void)xrtNetAddrLoopback(&Address, XNET_FAMILY_IPV4, 0);
	pServer = xrtNetUdpBind(pEngine, &Address, 0, &UdpConfig,
		NULL, NULL);
	if ( (pServer == NULL) || !xrtNetUdpLocal(pServer, &Address) ) {
		iResult = 2;
		goto Cleanup;
	}
	pClient = xrtNetUdpConnect(pEngine, &Address, 0, NULL,
		NULL, NULL);
	if ( (pClient == NULL) ||
		 !exampleWaitState(pServer, XNET_UDP_OPEN) ||
		 !exampleWaitState(pClient, XNET_UDP_OPEN) ) {
		iResult = 3;
		goto Cleanup;
	}

	/* 截断演示：64 字节载荷到达 16 字节接收槽。 */
	memset(Big, 'A', sizeof(Big));
	if ( xrtNetUdpSend(pClient, Big, sizeof(Big)) != XNET_RESULT_OK ) {
		goto Cleanup;
	}
	iCount = exampleDrainCount(pServer, 1);
	if ( iCount != 1u ) {
		iResult = 4;
		goto Cleanup;
	}
	{
		/* 重发一条并直接取包核对截断标志与前缀长度。 */
		xnetudppacket* pPacket = NULL;
		double iDeadline = exampleTimerLimit(3000);

		if ( xrtNetUdpSend(pClient, Big, sizeof(Big)) !=
			 XNET_RESULT_OK ) {
			goto Cleanup;
		}
		while ( (pPacket = xrtNetUdpReceive(pServer)) == NULL ) {
			if ( exampleTimerExpired(iDeadline) ) {
				break;
			}
			xrtThreadYield();
		}
		if ( (pPacket == NULL) || !xrtNetUdpPacketTruncated(pPacket) ||
			 (xrtNetUdpPacketSize(pPacket) != 16u) ) {
			xrtNetUdpPacketDestroy(pPacket);
			iResult = 5;
			goto Cleanup;
		}
		printf("batch: truncated prefix=%zu of %zu\n",
			xrtNetUdpPacketSize(pPacket), sizeof(Big));
		xrtNetUdpPacketDestroy(pPacket);
	}

	/* 拉取批量：三条齐发，等队列到齐后一次 ReceiveBatch 全取。 */
	for ( i = 0; i < 3u; ++i ) {
		if ( xrtNetUdpSend(pClient, "bb", 2) != XNET_RESULT_OK ) {
			goto Cleanup;
		}
	}
	memset(pPackets, 0, sizeof(pPackets));
	{
		double iDeadline = exampleTimerLimit(3000);

		while ( xrtNetUdpQueued(pServer) < 3u ) {
			if ( exampleTimerExpired(iDeadline) ) {
				iResult = 6;
				goto Cleanup;
			}
			xrtThreadYield();
		}
		iCount = xrtNetUdpReceiveBatch(pServer, pPackets, 8);
		if ( iCount != 3u ) {
			iResult = 6;
			goto Cleanup;
		}
	}
	printf("pull batches: %zu packets in 1 batch\n", iCount);
	for ( i = 0; i < iCount; ++i ) {
		xrtNetUdpPacketDestroy(pPackets[i]);
	}

	/* 阻塞批量：两条齐发，BatchPacket 借用 + BatchTake 转移。 */
	for ( i = 0; i < 2u; ++i ) {
		if ( xrtNetUdpSend(pClient, "ww", 2) != XNET_RESULT_OK ) {
			goto Cleanup;
		}
	}
	{
		double iDeadline = exampleTimerLimit(3000);

		while ( xrtNetUdpQueued(pServer) < 2u ) {
			if ( exampleTimerExpired(iDeadline) ) {
				iResult = 7;
				goto Cleanup;
			}
			xrtThreadYield();
		}
	}
	pBatch = xrtNetUdpReceiveBatchWait(pServer, 4,3000, NULL);
	if ( (pBatch == NULL) || (xrtNetUdpBatchCount(pBatch) < 1u) ) {
		iResult = 7;
		goto Cleanup;
	}
	{
		xnetudppacket* pBorrowed = xrtNetUdpBatchPacket(pBatch, 0);
		xnetudppacket* pTaken = xrtNetUdpBatchTake(pBatch,
			xrtNetUdpBatchCount(pBatch) - 1u);

		if ( (pBorrowed == NULL) || (pTaken == NULL) ||
			 (xrtNetUdpPacketSize(pBorrowed) != 2u) ||
			 (xrtNetUdpBatchPacket(pBatch,
				xrtNetUdpBatchCount(pBatch) - 1u) != NULL) ) {
			xrtNetUdpPacketDestroy(pTaken);
			iResult = 8;
			goto Cleanup;
		}
		printf("wait batch: %zu packets (borrow=take=ok)\n",
			xrtNetUdpBatchCount(pBatch));
		xrtNetUdpPacketDestroy(pTaken);
	}
	xrtNetUdpBatchDestroy(pBatch);
	pBatch = NULL;

	/* Future 批量：两条齐发 + 双引用把包与批量带出生命周期。 */
	for ( i = 0; i < 2u; ++i ) {
		if ( xrtNetUdpSend(pClient, "ff", 2) != XNET_RESULT_OK ) {
			goto Cleanup;
		}
	}
	{
		double iDeadline = exampleTimerLimit(3000);

		while ( xrtNetUdpQueued(pServer) < 2u ) {
			if ( exampleTimerExpired(iDeadline) ) {
				iResult = 9;
				goto Cleanup;
			}
			xrtThreadYield();
		}
	}
	pFuture = xrtNetUdpReceiveBatchAsync(pServer, 4);
	if ( !exampleFutureWait(pFuture) ) {
		iResult = 9;
		goto Cleanup;
	}
	pBatch = (xnetudpbatch*)xrtFutureValue(pFuture);
	if ( (pBatch == NULL) || (xrtNetUdpBatchCount(pBatch) < 1u) ) {
		iResult = 10;
		goto Cleanup;
	}
	
	 * 必须自己 Ref（包与整批各一份），不能替 Future 释放。 */
	pKept = xrtNetUdpPacketRef(xrtNetUdpBatchPacket(pBatch, 0));
	pBatchRef = xrtNetUdpBatchRef(pBatch);
	if ( (pKept == NULL) || (pBatchRef == NULL) ) {
		iResult = 11;
		goto Cleanup;
	}
	printf("future batch: %zu packets, ref kept size=%zu\n",
		xrtNetUdpBatchCount(pBatch),
		xrtNetUdpPacketSize(pKept));
	/* 先销毁 Future（释放其引用）：两份 Ref 让包与批量继续存活。 */
	xrtFutureDestroy(pFuture);
	pFuture = NULL;
	if ( (xrtNetUdpBatchCount(pBatchRef) < 1u) ||
		 (xrtNetUdpPacketSize(pKept) != 2u) ) {
		iResult = 12;
		goto Cleanup;
	}
	xrtNetUdpPacketDestroy(pKept);
	pKept = NULL;
	xrtNetUdpBatchDestroy(pBatchRef);
	pBatchRef = NULL;
	pBatch = NULL;

	/* 错误队列空路径：拉取 NULL/0 + Future 由关闭终结。 */
	{
		xnetudperrorpacket* pErrors[4];

		printf("error queue: empty pull=%zu/%zu",
			(size_t)(xrtNetUdpReceiveError(pServer) != NULL ?
				1u : 0u),
			xrtNetUdpReceiveErrorBatch(pServer, pErrors, 4));
	}
	pErrorFuture = xrtNetUdpReceiveErrorAsync(pServer);
	(void)xrtNetUdpAbort(pServer);
	if ( !exampleWaitState(pServer, XNET_UDP_CLOSED) ) {
		iResult = 13;
		goto Cleanup;
	}
	/* 关闭使未决错误 Future 进入终态（RESOLVED=1..CLOSED=4）。 */
	(void)xrtFutureWaitFor(pErrorFuture, INT64_C(3000));
	printf(", async terminal=%d\n",
		(int)xrtFutureState(pErrorFuture));

	/* 发送条件：空闲队列立即可写（同步 + Future 双形态）。 */
	printf("writable: sync=%d",
		xrtNetUdpWritable(pClient, 64,3000,
			NULL) ? 1 : 0);
	pWritable = xrtNetUdpWritableAsync(pClient, 64);
	printf(" async=%d\n", exampleFutureWait(pWritable) ? 1 : 0);
	iResult = 0;

Cleanup:
	xrtFutureDestroy(pWritable);
	xrtFutureDestroy(pErrorFuture);
	xrtFutureDestroy(pFuture);
	xrtNetUdpBatchDestroy(pBatch);
	xrtNetUdpBatchDestroy(pBatchRef);
	xrtNetUdpPacketDestroy(pKept);
	if ( pClient != NULL ) {
		(void)xrtNetUdpAbort(pClient);
		(void)exampleWaitState(pClient, XNET_UDP_CLOSED);
	}
	if ( pServer != NULL ) {
		(void)xrtNetUdpAbort(pServer);
		(void)exampleWaitState(pServer, XNET_UDP_CLOSED);
	}
	xrtNetUdpDestroy(pClient);
	xrtNetUdpDestroy(pServer);
	if ( (pEngine != NULL) && !xrtNetEngineDestroy(pEngine) &&
		 (iResult == 0) ) {
		iResult = 14;
	}
	return iResult;
}