Chapter 67 TCP Streams (Part 1): Connections and Reading/Writing

Chapter Orientation

Chapters 63 through 65 settled "where addresses come from" — parsing, ports, buffers, and DNS. This chapter starts moving real data: building an Engine, listening on a port, dialing out, sending and receiving bytes, and finally shutting everything down cleanly. The TCP module is the foundation for everything that follows — Chapter 68's backpressure and server-side shapes, Volume 8's TLS, and Volume 9's HTTP all build on the single object xnetstream. This chapter first masters two faces: the blocking synchronous face (the shortest code path for utility programs) and the event-callback face (the skeleton of production services), and thoroughly explains the two most error-prone things — the "state machine" and the "closing sequence".

67.1引入

Imagine writing a small data-migration tool: read a batch of data from one local service and forward it to another. Such a tool's networking needs are plain yet complete — connect, send, receive fully, close cleanly. Anyone who has written it on raw sockets knows the real effort is never the five syscalls, but the wind-down: what if the connection is stuck half-open? Can you close while unsent data still sits in the buffer? When is it safe to release the object?

XRT's TCP layer builds these traps into object contracts: streams have an explicit state machine, CLOSED is the only final state, and no reference is released before it is seen; closing comes in two semantics — "drain" and "abort now"; a single Engine drives the event loops of all connections. This chapter walks the model through two complete programs.

67.2概念

Three objects, one engine

• xnetengine: the network engine. It manages the platform event backend (IOCP / epoll / kqueue / io_uring) and a pool of Worker threads; every Listener and Stream is attached to some Engine to work. In the config, Workers sets the thread count. Rules of thumb for Worker counts: utilities and unit tests need just 1–2 — an Engine's threads only do event dispatch and callbacks, never business computation (to schedule small tasks directly onto the Worker loop, use Chapter 61's network task groups task_net); server-side capacity planning (Worker counts, queue depth, backpressure watermarks) is the subject of Chapter 68. Multiple Engines can coexist, but the convention in this and later chapters is one Engine per process.
• xnetlistener: the listener. Binds an address (port 0 lets the system assign one) and accepts incoming connections.
• xnetstream: the stream. The read/write face of one TCP connection; the client and the server each hold one.

The states of both Stream and Listener only move forward:

CLOSED is the only final state, and it is guaranteed before publication: the socket, the buffers, in-flight operations, and the active usage of the Engine have all concluded. In other words, only when you see CLOSED may you release the caller-side reference.

One face, two calling conventions

xnetstream offers two calling conventions at once:

• Synchronous face: xrtNetListenerAcceptWait blocks for a connection, xrtNetStreamRecv blocks to receive data, and returns an owning xnetbytes byte result. Utility-style programs (liveness probes, one-shot migrations, simple clients) use the synchronous face for the shortest code.
• Event face: register a callback table (Accept/Read/Close function pointers), and the Engine's Worker threads invoke them as events arrive. Production services use the event face for high concurrency and hard backpressure.

Both faces share the same objects and state machine; this chapter demonstrates one complete program for each.

Two closing semantics and reference counting

• xrtNetStreamClose: requests a draining close — finish sending in-flight data, complete the protocol wind-down, then enter CLOSED.
• xrtNetStreamAbort: aborts immediately — discard in-flight data and reach CLOSED as fast as possible. Error paths and Cleanup sections use it to avoid hanging.

Objects returned by creation functions carry one caller-side reference; the runtime holds another internal reference until the sole Close callback finishes. So the correct exit order is: request the close (Close or Abort) → wait for the state to reach CLOSED → xrtNetStreamDestroy to release the caller reference. Ref/Destroy may be used across threads, but Ref cannot resurrect an already-finalized object from zero.

67.3示例

First complete program: synchronous-face echo

The following program comes from the repository example examples/network/tcp_sync/main.c: start a loopback listener, connect to it and send a hello, and have the server receive and print it. This is the canonical sample of "the least code that walks the full lifecycle":

#include <stdio.h>
#include <string.h>
#include <xrt.h>





















 */


/* 使用现有 Engine 完成一次本地阻塞 Accept 和接收。 */
int main(void)
{
	xnetengineconfig EngineConfig;
	xnetlistenconfig ListenConfig;
	xnetengine* pEngine;
	xnetlistener* pListener;
	xnetstream* pClient;
	xnetstream* pServer;
	xnetbytes* pBytes;
	xbytesview View;
	xnetaddr Address;
	bool bReceived;

	xrtNetEngineConfigInit(&EngineConfig);
	EngineConfig.Workers = 2;
	pEngine = xrtNetEngineCreate(&EngineConfig);
	if ( (pEngine == NULL) || !xrtNetEngineStart(pEngine) ) {
		return 1;
	}
	xrtNetListenConfigInit(&ListenConfig);
	(void)xrtNetAddrLoopback(
		&ListenConfig.Address,
		XNET_FAMILY_IPV4,
		0
	);
	pListener = xrtNetListen(pEngine, &ListenConfig, NULL, NULL, NULL);
	if ( (pListener == NULL) || !xrtNetListenerLocal(pListener, &Address) ) {
		return 2;
	}
	pClient = xrtNetStreamConnect(
		pEngine,
		&Address,
		1,
		NULL,
		NULL,
		NULL
	);
	pServer = xrtNetListenerAcceptWait(
		pListener,3000,
		NULL
	);
	if ( (pClient == NULL) || (pServer == NULL) ||
		 (xrtNetStreamSend(pClient, "hello", 5) != XNET_RESULT_OK) ) {
		return 3;
	}
	pBytes = xrtNetStreamRecv(
		pServer,
		0,3000,
		NULL
	);
	View = xrtNetBytesView(pBytes);
	bReceived = pBytes != NULL;
	if ( pBytes != NULL ) {
		printf("received: %.*s\n", (int)View.Size, (cstr)View.Data);
	}
	xrtNetBytesDestroy(pBytes);
	(void)xrtNetStreamAbort(pClient);
	(void)xrtNetStreamAbort(pServer);
	(void)xrtNetListenerClose(pListener);
	while ( (xrtNetStreamState(pClient) != XNET_STREAM_CLOSED) ||
		 (xrtNetStreamState(pServer) != XNET_STREAM_CLOSED) ||
		 (xrtNetListenerState(pListener) != XNET_LISTENER_CLOSED) ) {
		xrtThreadYield();
	}
	xrtNetStreamDestroy(pClient);
	xrtNetStreamDestroy(pServer);
	xrtNetListenerDestroy(pListener);
	return xrtNetEngineDestroy(pEngine) && bReceived ? 0 : 4;
}

What just happened. Six points deserve attention. (1) Workers = 2 — two Workers are enough for a utility; the Engine itself does no business work. (2) The listen address uses xrtNetAddrLoopback for the loopback, with port 0; afterwards xrtNetListenerLocal retrieves the real port assigned by the system — the standard posture for concurrency tests that must not collide. (3) The third argument 1 of the client's xrtNetStreamConnect is the address count; on the synchronous face the connection completes in the background on the Engine. (4) xrtNetListenerAcceptWait takes an relative milliseconds deadline; if no connection arrives in time it returns NULL and leaves a timeout category in the thread error slot (Chapter 4's model cashed in here). (5) xrtNetStreamRecv returns an owning xnetbytes; borrow a view with xrtNetBytesView to read it, and call xrtNetBytesDestroy when done. (6) The wind-down first Aborts both streams and the listener, spins until all three reach CLOSED, then Destroys them one by one, and finally calls xrtNetEngineDestroy.

Full event-face shape: a callback-driven echo service

The second program comes from examples/network/tcp/main.c — the standard skeleton of a production TCP service: event-table-driven reads and writes, the server echoing by moving the receive buffer directly (zero-copy), and a clean drain-and-close after the peer disconnects:

 */


#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 exampletcp {
	xnetstream* Client;
	xatomicptr Server;
	xatomic32 Accepted;
	xatomic32 Reply;
	xatomic32 Closed;
} exampletcp;



/* 接管服务端连接并设置回调数据。 */
static bool exampleAccept(
	xnetlistener* pListener,
	xnetstream* pStream,
	ptr pData
)
{
	exampletcp* pExample = (exampletcp*)pData;

	(void)pListener;
	(void)xrtNetStreamSetData(pStream, pExample);
	xrtAtomicPtrStore(
		&pExample->Server,
		pStream,
		XMEMORY_RELEASE
	);
	(void)xrtAtomic32FetchAdd(
		&pExample->Accepted,
		1,
		XMEMORY_RELEASE
	);
	return true;
}



/* 服务端直接回送字节，客户端输出回复。 */
static void exampleRead(
	xnetstream* pStream,
	xnetbuf* pBuffer,
	ptr pData
)
{
	exampletcp* pExample = (exampletcp*)pData;
	xnetstream* pServer = (xnetstream*)xrtAtomicPtrLoad(
		&pExample->Server,
		XMEMORY_ACQUIRE
	);
	char Data[64];
	size_t iSize;

	if ( pStream == pServer ) {
		(void)xrtNetStreamSendBuffer(pStream, pBuffer);
		return;
	}
	iSize = xrtNetBufRead(pBuffer, Data, sizeof(Data));
	printf("reply: %.*s\n", (int)iSize, Data);
	(void)xrtAtomic32FetchAdd(
		&pExample->Reply,
		1,
		XMEMORY_RELEASE
	);
}



/* 记录正常关闭。 */
static void exampleClose(
	xnetstream* pStream,
	xnetresult Result,
	const xerror* pError,
	ptr pData
)
{
	exampletcp* pExample = (exampletcp*)pData;

	(void)pStream;
	if ( (Result == XNET_RESULT_OK) && (pError == NULL) ) {
		(void)xrtAtomic32FetchAdd(
			&pExample->Closed,
			1,
			XMEMORY_RELEASE
		);
	}
}



/* 建立回环服务器和客户端，发送一次请求后完整回收。 */
int main(void)
{
	exampletcp Example;
	xnetengineconfig EngineConfig;
	xnetlistenconfig ListenConfig;
	xnetlistenerevents ListenerEvents;
	xnetstreamevents StreamEvents;
	xnetengine* pEngine;
	xnetlistener* pListener;
	xnetstream* pServer;
	xnetaddr Address;
	double iDeadline;
	str sEndpoint;
	int iResult = 1;

	memset(&Example, 0, sizeof(Example));
	xrtAtomicPtrInit(&Example.Server, NULL);
	memset(&ListenerEvents, 0, sizeof(ListenerEvents));
	memset(&StreamEvents, 0, sizeof(StreamEvents));
	ListenerEvents.Accept = exampleAccept;
	StreamEvents.Read = exampleRead;
	StreamEvents.Close = exampleClose;
	pEngine = NULL;
	pListener = NULL;
	pServer = NULL;
	sEndpoint = NULL;
	xrtNetEngineConfigInit(&EngineConfig);
	EngineConfig.Workers = 2;
	pEngine = xrtNetEngineCreate(&EngineConfig);
	if ( (pEngine == NULL) || !xrtNetEngineStart(pEngine) ) {
		goto Cleanup;
	}
	xrtNetListenConfigInit(&ListenConfig);
	(void)xrtNetAddrLoopback(
		&ListenConfig.Address,
		XNET_FAMILY_IPV4,
		0
	);
	pListener = xrtNetListen(
		pEngine,
		&ListenConfig,
		&ListenerEvents,
		&StreamEvents,
		&Example
	);
	if ( (pListener == NULL) ||
		 !xrtNetListenerLocal(pListener, &Address) ) {
		iResult = 2;
		goto Cleanup;
	}
	sEndpoint = xrtNetAddrEndpointString(&Address);
	if ( sEndpoint == NULL ) {
		iResult = 3;
		goto Cleanup;
	}
	printf("listening on %s\n", sEndpoint);
	xrtFree(sEndpoint);
	sEndpoint = NULL;
	Example.Client = xrtNetStreamConnect(
		pEngine,
		&Address,
		1,
		NULL,
		&StreamEvents,
		&Example
	);
	if ( Example.Client == NULL ) {
		iResult = 4;
		goto Cleanup;
	}
	iDeadline = exampleTimerLimit(3000);
	while ( xrtAtomic32Load(&Example.Accepted, XMEMORY_ACQUIRE) == 0 ) {
		if ( exampleTimerExpired(iDeadline) ) {
			iResult = 5;
			goto Cleanup;
		}
		xrtThreadYield();
	}
	pServer = (xnetstream*)xrtAtomicPtrLoad(
		&Example.Server,
		XMEMORY_ACQUIRE
	);
	if ( pServer == NULL ) {
		iResult = 5;
		goto Cleanup;
	}
	if ( xrtNetStreamSend(
		Example.Client,
		"hello TCP",
		9
	) != XNET_RESULT_OK ) {
		iResult = 6;
		goto Cleanup;
	}
	while ( xrtAtomic32Load(&Example.Reply, XMEMORY_ACQUIRE) == 0 ) {
		if ( exampleTimerExpired(iDeadline) ) {
			iResult = 7;
			goto Cleanup;
		}
		xrtThreadYield();
	}
	(void)xrtNetStreamClose(Example.Client);
	(void)xrtNetStreamClose(pServer);
	iDeadline = exampleTimerLimit(3000);
	while ( xrtAtomic32Load(&Example.Closed, XMEMORY_ACQUIRE) != 2 ) {
		if ( exampleTimerExpired(iDeadline) ) {
			iResult = 8;
			goto Cleanup;
		}
		xrtThreadYield();
	}
	(void)xrtNetListenerClose(pListener);
	iDeadline = exampleTimerLimit(3000);
	while ( xrtNetListenerState(pListener) != XNET_LISTENER_CLOSED ) {
		if ( exampleTimerExpired(iDeadline) ) {
			iResult = 8;
			goto Cleanup;
		}
		xrtThreadYield();
	}
	iResult = 0;

Cleanup:
	xrtFree(sEndpoint);
	pServer = (xnetstream*)xrtAtomicPtrLoad(
		&Example.Server,
		XMEMORY_ACQUIRE
	);
	if ( (Example.Client != NULL) &&
		 (xrtNetStreamState(Example.Client) != XNET_STREAM_CLOSED) ) {
		(void)xrtNetStreamAbort(Example.Client);
	}
	if ( (pServer != NULL) &&
		 (xrtNetStreamState(pServer) != XNET_STREAM_CLOSED) ) {
		(void)xrtNetStreamAbort(pServer);
	}
	if ( (pListener != NULL) &&
		 (xrtNetListenerState(pListener) != XNET_LISTENER_CLOSED) ) {
		(void)xrtNetListenerClose(pListener);
	}
	iDeadline = exampleTimerLimit(3000);
	while ( ((Example.Client != NULL) &&
		  (xrtNetStreamState(Example.Client) != XNET_STREAM_CLOSED)) ||
		 ((pServer != NULL) &&
		  (xrtNetStreamState(pServer) != XNET_STREAM_CLOSED)) ||
		 ((pListener != NULL) &&
		  (xrtNetListenerState(pListener) != XNET_LISTENER_CLOSED)) ) {
		if ( exampleTimerExpired(iDeadline) ) {
			if ( iResult == 0 ) {
				iResult = 8;
			}
			break;
		}
		xrtThreadYield();
	}
	xrtNetStreamDestroy(Example.Client);
	xrtNetStreamDestroy(pServer);
	xrtNetListenerDestroy(pListener);
	if ( (pEngine != NULL) && !xrtNetEngineDestroy(pEngine) &&
		 (iResult == 0) ) {
		iResult = 8;
	}
	return iResult;
}

Three key points of the event face. (1) The Accept callback takes over every new connection: it calls xrtNetStreamSetData to attach the business context to the stream, and every later callback can retrieve it — no hand-rolled "stream → context" map needed. (2) In the server's Read callback, xrtNetStreamSendBuffer hands the receive buffer back by reference — data moves from the receive buffer into the send queue with zero copies; this is the correct way to write echo-style services. The client's Read demonstrates plain reading (xrtNetBufRead copying into a stack buffer). (3) The Close callback is the sole final-state notification: the xnetresult parameter and the error object distinguish "clean close" from "error close"; the sample counts only clean closes. Callbacks run on the Engine's Worker threads, and the main thread observes progress with an atomic counter and a deadline — Chapter 68 upgrades this waiting into proper synchronization primitives and backpressure control.

67.4契约

• State: Stream and Listener states only move forward; CLOSED is the only final state; before entering CLOSED, the socket, buffers, and in-flight operations have all concluded.
• References: creation returns the caller reference; the runtime holds an internal reference until the Close callback ends; request the close first, wait for CLOSED, then Destroy; Ref is thread-safe but cannot resurrect.
• Closing: Close drains, Abort is immediate; error paths always Abort, the happy path lets the data finish.
• Threading: event callbacks execute on Engine Workers; the Close callback may still be winding down on a Worker, and xrtNetEngineDestroy waits for the Workers to exit normally.
• Errors: failures are declared by return values (NULL / non-XNET_RESULT_OK), with details in the thread error slot; categories and domains follow the Chapter 4 model.

Common Mistakes

xrtNetStreamClose(pStream);
xrtNetStreamDestroy(pStream);   /* state may still be CLOSING */

xrtNetStreamClose(pStream);
while ( xrtNetStreamState(pStream) != XNET_STREAM_CLOSED ) {
	xrtThreadYield();
}
xrtNetStreamDestroy(pStream);

xnetbytes* pBytes = xrtNetStreamRecv(pStream, 0, deadline, NULL);
printf("%.*s\n", (int)xrtNetBytesView(pBytes).Size,
	(cstr)xrtNetBytesView(pBytes).Data);
/* missing Destroy: leaks one result object per receive */

xnetbytes* pBytes = xrtNetStreamRecv(pStream, 0, deadline, NULL);
if ( pBytes != NULL ) {
	printf("%.*s\n", (int)xrtNetBytesView(pBytes).Size,
		(cstr)xrtNetBytesView(pBytes).Data);
	xrtNetBytesDestroy(pBytes);
}

Programming Exercises

Summary