Chapter 112 xws (Part 3): Connection Groups and Broadcast

Chapter Orientation

The basic operation of a push system is "send one message to every online connection" — xwsgroup makes "a set of connections" into a correct concurrent container: unique membership (the same connection joining twice is still one member), reference lifecycle (Add takes a reference, Remove/Clear/Destroy give it back — connections are neither closed nor aborted), hard capacity (when Limit is full, new members get a retryable capacity error), sealing (Seal permanently stops admissions — the standard move when a server stops taking sessions), stable snapshots (Snapshot copies members in join order and increments each reference — allocation outside the lock, so big snapshots don't block Add/Remove), and batch broadcast (synchronous and asynchronous forms; the async form returns a waitable operation object — cancellation of the wait does not break the ownership of data already accepted by each connection). Broadcast does not bypass per-connection limits: slow connections are skipped, waited on, or closed per your policy — never dragged into an unbounded shared queue.

112.1引入

"Maintaining the list of online connections" looks like an 数组+锁 (array + lock) beginner exercise — until the concurrency details queue up: the same connection's Open and Close callbacks run concurrently on different Workers, Add and Remove race; members join and leave during a broadcast iteration (iterator invalidation — Chapter 18's old problem); a slow connection holds up the whole batch during broadcast (one drags N); at server shutdown, "stop taking new ones first, close the old ones gracefully" (a half-closed state). Each is daily life for a real push service.

xwsgroup answers them one by one: unique-membership semantics absorb duplicate Adds; Snapshot's "linearize inside the lock, allocate outside it" gives the iteration a stable set (the snapshot holds its own references — even destroying the original group changes nothing; iterator invalidation becomes impossible); broadcast is per-connection independent acceptance (a slow connection doesn't drag others) + aggregate waiting (the async operation object); Seal provides the half-closed state. This chapter's value is not the API count (a dozen entrances) but that every concurrency semantic is thought through — read it as a "concurrent-container design lesson" with a WebSocket example.

112.2概念

Lifecycle and unique membership

The meaning of unique membership: Open-callback replays (an occasional network-layer event) cannot leak double references; Remove returns exactly the one share Add took. Capacity semantics: when Limit is full, Add fails with XERR_AGAIN / XWS_GROUP_ERROR_CAPACITY — retryable (after removing a member you can add again), not a protocol error; this error-class design separates "overloaded" from "illegal" (Chapter 4's category model applied at the container layer).

Sealing: the half-closed state

xrtWsGroupSeal is permanent and idempotent: after sealing no new members join, but re-adding an existing member still succeeds (idempotent semantics continue), and remove/clear/snapshot work as usual. The standard shutdown order: stop accepting new connections (the listener layer) → Seal the group (in-flight Opens no longer join) → close or abort each existing session gracefully — "close the entrance first, then clear the hall".

Snapshot: the correct way to iterate stably

xrtWsGroupSnapshotCreate: copies the current members in join order and increments a reference for each; the key engineering detail — the contiguous snapshot storage is allocated outside the group lock: first attempt the allocation lock-free; after retaking the lock, if membership grew past capacity, free and retry; only when capacity suffices does it increment references in order at the linearization point. The effect: a big snapshot's allocation work (which may trip the allocator lock) does not block the Add/Remove/Clear hot path. The snapshot does not hold the group — after the original group is destroyed the snapshot's members stay valid (their references live on the snapshot); release with SnapshotDestroy when done. This is exactly Chapter 18's "container invalidation rules" in solved form: when you need a stable iteration, copy a reference-holding snapshot — don't gamble your way across the original container.

Batch broadcast: independent acceptance and aggregate waiting

The synchronous form: iterate the snapshot and send per connection (each connection's AGAIN/failure handled independently — skip, note as slow, close per policy). The asynchronous form: entrances like xrtWsGroupTextAsync return an xwsgroupop operation object — GroupOpWait waits for the whole batch to finish, destroy releases; cancellation of the aggregate wait does not break the data ownership already accepted by each connection (the batch edition of Chapter 110's "cancel ≠ destroy"). Broadcast does not bypass per-connection send limits — a slow connection's queue stays bounded as always, and your policy decides its fate.

Pairing with the three send states

A broadcast's message payload is a natural fit for the ref form (Chapter 111): one static payload, references handed to N connections' queues — the release callback counts down (the true release only after the last write). The group's async broadcast entrances are exactly the batch application of this form; a hand-written iteration assembles the same with xrtWsConnBinaryRef (Chapter 111's exercise did this).

112.3示例

First complete program: the group lifecycle skeleton

The program below is from examples/websocket/group — the standard integration shape of Add/Remove:

#include <stdio.h>

#include <xws.h>



/* Upgrade 成功后把借用连接加入应用连接组。 */
static void connectionOpen(xwsgroup* pGroup, xwsconn* pConnection)
{
	(void)xrtWsGroupAdd(pGroup, pConnection);
}



/* 唯一 Close 回调中移除连接，归还连接组持有的引用。 */
static void connectionClose(xwsgroup* pGroup, xwsconn* pConnection)
{
	(void)xrtWsGroupRemove(pGroup, pConnection);
}



/* 示例保留回调形状，实际服务器可直接把组放入路由上下文。 */
int main(void)
{
	xwsgroup* pGroup = xrtWsGroupCreate(0);

	if ( pGroup == NULL ) {
		return 1;
	}
	(void)connectionOpen;
	(void)connectionClose;
	printf("WebSocket connection group is ready\n");
	xrtWsGroupDestroy(pGroup);
	return 0;
}

What just happened. (1) xrtWsGroupCreate(0) builds an unlimited group (production passes a Limit — capacity is the overload line). (2) The two functions are the integration template: connectionOpen (in the upgrade-success callback, GroupAdd — the connection joins), connectionClose (in the sole Close callback, GroupRemove — the reference returns). A real server puts the group into the routing context (Chapter 113 — where the Data field earns its keep) — one-line integration for Open/Close, and the group's concurrency semantics all inherited. (3) This sample's main only verifies readiness (template shape); the (void) references in connectionOpen/Close are compile-time self-proof of "keeping the callback shape".

Second complete program: a waitable broadcast on an empty group

The second program is from examples/websocket/group_future — the full lifecycle of the async broadcast operation object:

#include <stdio.h>

#include <xws.h>



/* 展示空组广播也会返回可等待、可检查的完整操作对象。 */
int main(void)
{
	xwsgroup* pGroup = xrtWsGroupCreate(0);
	xwsgroupop* pOperation;

	if ( pGroup == NULL ) {
		return 1;
	}
	pOperation = xrtWsGroupTextAsync(
		pGroup,
		XRT_STR_LITERAL("service-ready")
	);
	if ( (pOperation == NULL) ||
		(xrtWsGroupOpWait(pOperation) != XWAIT_OK) ) {
		xrtWsGroupOpDestroy(pOperation);
		xrtWsGroupDestroy(pGroup);
		return 1;
	}
	printf(
		"members=%zu accepted=%zu rejected=%zu\n",
		xrtWsGroupOpCount(pOperation),
		xrtWsGroupOpAccepted(pOperation),
		xrtWsGroupOpRejected(pOperation)
	);
	xrtWsGroupOpDestroy(pOperation);
	xrtWsGroupDestroy(pGroup);
	return 0;
}

What just happened. (1) xrtWsGroupTextAsync(组, 消息) (group, message) returns a complete operation object even for an empty group — the header comment spells out the design intent: "broadcast to an empty group also returns a full waitable, inspectable operation object" — the empty set is not an error; the aggregate semantics are self-consistent (zero connections, all succeed). (2) GroupOpWait waits for the whole batch (XWAIT_OK), GroupOpDestroy releases the operation object — the async broadcast's three-step lifecycle (initiate/wait/release). (3) Read the two samples together: after Open joins the group, TextAsync broadcasts — each member accepted independently; when the wait returns, the whole batch has its verdicts. The slow-connection policy (skip/close) lives in the configuration and your iteration code — the group does not decide for you.

112.4契约

• Unique membership: repeatedly Adding the same pointer succeeds without doubling; Remove/Clear/Destroy return exactly the references Add took; group operations neither close nor abort connections.
• Capacity: Limit zero = unlimited; a full Add returns XERR_AGAIN/CAPACITY (retryable, not a protocol error).
• Sealing: Seal is permanent and idempotent; after sealing no new members, re-adding existing ones still succeeds; remove/clear/snapshot as usual — at shutdown, Seal first, then clear.
• Snapshot: allocate outside the lock + linearized reference increments inside; order preserved by join order; does not hold the group (members stay valid after group destruction); Destroy when done.
• Queries: Has/Count/Limit/Sealed are locked instantaneous snapshots — no substitute for a stable iteration (use Snapshot).
• Broadcast independence: per-connection acceptance; per-connection limits not bypassed; the slow-connection policy (skip/wait/close) belongs to the caller.
• Async operation objects: the TextAsync family returns waitable, inspectable objects; cancelling a wait does not break accepted ownership.
• Lock discipline: Clear swaps in an empty set inside the lock and returns old references outside it — destruction and allocator work never hold the group lock.
• Opaque validation: group/snapshot/operation are all opaque; fixed headers and contiguous storage validated before dereference; wraparound rejected as a parameter error; Destroy(NULL) is a no-op.
• Trimming: WEBSOCKET_GROUP/_GROUP_FUTURE independent; the group depends on connections and set/mutex — the frame layer pulls in none of it.

Common Mistakes

size_t n = xrtWsGroupCount(pGroup);
for ( size_t i = 0; i < n; i++ ) {
	/* take member i - Remove may have changed the set meanwhile */
}

xwsgroupsnapshot* pSnap = xrtWsGroupSnapshotCreate(pGroup);
for ( size_t i = 0; i < xrtWsGroupSnapshotCount(pSnap); i++ ) {
	xwsconn* pConn = xrtWsGroupSnapshotGet(pSnap, i);
	send_or_skip(pConn);   /* stable set: joins/leaves during iteration don't affect this snapshot */
}
xrtWsGroupSnapshotDestroy(pSnap);

xrtWsGroupRemove(pGroup, pConn);
/* assumed the connection is closed - the peer stays online */

xrtWsGroupRemove(pGroup, pConn);       /* leave the group first */
if ( !xrtWsConnClose(pConn, XWS_CLOSE_NORMAL,
		XRT_STR_LITERAL("kicked")) ) {
	xrtWsConnAbort(pConn);              /* then close per policy */
}

stop_listener();
xrtWsGroupDestroy(pGroup);   /* in-flight Opens' Adds haven't finished */

stop_listener();             /* 1. stop taking new connections */
xrtWsGroupSeal(pGroup);      /* 2. seal: in-flight Opens' Adds rejected */
drain_workers();             /* 3. wait for callbacks to finish (Open/Close go quiet) */
close_all_members(pGroup);   /* 4. gracefully close existing sessions */
xrtWsGroupDestroy(pGroup);   /* 5. destroy the group last */

Programming Exercises

Summary