第 61 章 调度器实战：协程、通道与任务的联合作战

/* service_core.c —— 卷六骨架核心环：调度器 + 协程 + 池 + 通道 */
#define XRT_MODULE_ALL
#define XRT_IMPLEMENTATION
#include <xrt.h>
#include <stdio.h>

static xcosched* gSched;
static xtaskpool* gPool;
static xchannel* gStats;
static xatomic32 gDone;

static void aggregate(ptr pData)          /* 统计聚合协程体 */
{
	ptr Item;
	while ( xrtChannelRecv(gStats, &Item) == XWAIT_OK ) {
		(void)xrtAtomic32FetchAdd((xatomic32*)pData, 1, XMEMORY_RELAXED);
	}
}

static int computeHeavy(int Input)        /* 模拟重计算（池上跑） */
{
	int Sum = 0;
	for ( int i = 0; i < Input * 100; ++i ) {
		Sum += i % 7;
	}
	return Sum;
}

static ptr workerTask(ptr pData)          /* 池任务：算完发布 */
{
	int Input = (int)(uintptr_t)pData;
	int Result = computeHeavy(Input);
	return (ptr)(uintptr_t)Result;
}

static ptr connection(ptr pData)          /* 连接协程：直线逻辑 */
{
	int Input = (int)(uintptr_t)pData;
	xfuture* Job = xrtTaskSubmit(gPool, workerTask,
		(ptr)(uintptr_t)Input, NULL);      /* 外挂池（第 59 章） */
	if ( Job == NULL ) {
		return NULL;
	}
	if ( xrtFutureWait(Job) != XWAIT_OK ) {  /* 泵内不阻塞——此处演示用直等 */
		xrtFutureDestroy(Job);
		return NULL;
	}
	int Value = (int)(uintptr_t)xrtFutureValue(Job);  /* 借用读（第 58 章） */
	xrtFutureDestroy(Job);
	(void)xrtChannelTrySend(gStats, (ptr)(uintptr_t)Value);  /* 总线上报（第 57 章） */
	return (ptr)(uintptr_t)Value;
}

int main(void)
{
	xatomic32 Count;
	xcoro* Agg;
	xtaskpoolconfig PoolCfg = { 2, 32, 0 };

	xrtAtomic32Init(&Count, 0);
	xrtAtomic32Init(&gDone, 0);
	gSched = xrtCoSchedCreate();               /* 事件循环（第 56 章） */
	gPool = xrtTaskPoolCreate(&PoolCfg);       /* 计算外挂（第 59 章） */
	gStats = xrtChannelCreate(64);             /* 消息总线（第 57 章） */
	if ( (gSched == NULL) || (gPool == NULL) || (gStats == NULL) ) {
		return 1;
	}
	Agg = xrtCoGo(gSched, aggregate, &Count);  /* 聚合协程 */
	for ( int i = 1; i <= 3; ++i ) {
		(void)xrtCoGo(gSched, connection, (ptr)(uintptr_t)i);
	}
	(void)xrtCoSchedRun(gSched);              /* 泵跑到全部协程结束 */
	xrtChannelClose(gStats);                   /* 排空（第 57 章三方规则） */
	(void)xrtCoSchedRun(gSched);               /* 聚合协程消费存量 */
	xrtCoSchedDestroy(gSched);
	(void)xrtTaskPoolClose(gPool);             /* 池停机协议（第 59 章） */
	xrtTaskPoolWait(gPool);
	xrtTaskPoolDestroy(gPool);
	printf("aggregated=%d\n", (int)xrtAtomic32Load(&Count, XMEMORY_ACQUIRE));
	return 0;
}

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





















 */





 */
int main(void)
{
	int iResult = 1;
	xtaskgroup* pParent = NULL;
	xtaskgroup* pChild = NULL;
	xfuture* pLeaf = NULL;
	xfuture* pDone = NULL;
	xpromise* pLeafPromise = NULL;
	xcancel* pLeafCancel = NULL;
	xtaskgroupstats tStats;

	/* 子组作为父组中的一项，叶 Future 再由子组跟踪。 */
	pParent = xrtTaskGroupCreate(NULL);
	if ( pParent == NULL ) {
		goto cleanup;
	}
	pChild = xrtTaskGroupChild(pParent, NULL);
	pLeafPromise = xrtPromiseCreate(&pLeaf, NULL);
	if ( (pChild == NULL) || (pLeafPromise == NULL) ||
		!xrtTaskGroupAdd(pChild, pLeaf) ) {
		goto cleanup;
	}
	pDone = xrtTaskGroupFuture(pParent);
	pLeafCancel = xrtFutureCancelToken(pLeaf);
	if ( (pDone == NULL) || (pLeafCancel == NULL) ) {
		goto cleanup;
	}

	/* 父级只发出协作取消请求，叶生产端负责确认自己的终态。 */
	if ( !xrtTaskGroupCancel(pParent) ||
		!xrtCancelRequested(pLeafCancel) ||
		!xrtPromiseCancel(pLeafPromise) ||
		(xrtFutureWait(pDone) != XWAIT_OK) ) {
		goto cleanup;
	}
	if ( (xrtFutureState(pLeaf) != XFUTURE_CANCELLED) ||
		!xrtTaskGroupGet(pParent, &tStats) ) {
		goto cleanup;
	}
	printf(
		"parent completed = %llu, cancelled = %llu\n",
		(unsigned long long)tStats.Completed,
		(unsigned long long)tStats.Cancelled
	);
	iResult = 0;

cleanup:
	/* 子组和父组允许在仍有活动监听时延迟完成内部回收。 */
	xrtCancelDestroy(pLeafCancel);
	xrtFutureDestroy(pDone);
	xrtPromiseDestroy(pLeafPromise);
	xrtFutureDestroy(pLeaf);
	xrtTaskGroupDestroy(pChild);
	xrtTaskGroupDestroy(pParent);
	return iResult;
}

 */

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



/* 在 Engine Worker 上生成一个轻量结果。 */
static xtaskoutcome buildValue(
	xnetworker* pWorker,
	xcancel* pCancel,
	ptr pData,
	xtaskvalue* pResult
)
{
	(void)pCancel;
	printf("worker=%u\n", xrtNetWorkerIndex(pWorker));
	pResult->Value = pData;
	return XTASK_SUCCESS;
}





/* 演示立即网络任务与统一 Future 等待。 */
int main(void)
{
	xnetengine* pEngine = xrtNetEngineCreate(NULL);
	xfuture* pFuture;
	int iValue = 42;

	if ( (pEngine == NULL) || !xrtNetEngineStart(pEngine) ) {
		return 1;
	}
	pFuture = xrtTaskNet(
		pEngine,
		0,
		buildValue,
		&iValue,
		NULL
	);
	if ( (pFuture == NULL) ||
		(xrtFutureWaitFor(pFuture, 3000) != XWAIT_OK) ||
		(xrtFutureState(pFuture) != XFUTURE_RESOLVED) ) {
		xrtFutureDestroy(pFuture);
		(void)xrtNetEngineDestroy(pEngine);
		return 1;
	}
	printf("value=%d\n", *(int*)xrtFutureValue(pFuture));
	xrtFutureDestroy(pFuture);

	/* ---- After / Until：延迟与截止时间两种提交形态。 ---- */
	pFuture = xrtTaskNetAfter(pEngine, 0, buildValue, &iValue,
		NULL, 0);
	if ( (pFuture == NULL) ||
		(xrtFutureWaitFor(pFuture, 3000) != XWAIT_OK) ) {
		xrtFutureDestroy(pFuture);
		(void)xrtNetEngineDestroy(pEngine);
		return 2;
	}
	printf("after: value=%d\n", *(int*)xrtFutureValue(pFuture));
	xrtFutureDestroy(pFuture);
	pFuture = xrtTaskNetFor(pEngine, 0, buildValue, &iValue,
		NULL,0);
	if ( (pFuture == NULL) ||
		(xrtFutureWaitFor(pFuture, 3000) != XWAIT_OK) ) {
		xrtFutureDestroy(pFuture);
		(void)xrtNetEngineDestroy(pEngine);
		return 3;
	}
	printf("until: value=%d\n", *(int*)xrtFutureValue(pFuture));
	xrtFutureDestroy(pFuture);

	/* ---- GroupNetUntil：延迟任务原子纳入任务组。 ---- */
	{
		xtaskgroup* pGroup;

		pGroup = xrtTaskGroupCreate(NULL);
		if ( (pGroup == NULL) ||
			(xrtTaskGroupNetFor(pGroup, pEngine, 0,
				buildValue, &iValue, NULL,0) == NULL) ||
			!xrtTaskGroupClose(pGroup) ||
			(xrtTaskGroupWaitFor(pGroup, 3000) !=
				XWAIT_OK) ) {
			xrtTaskGroupDestroy(pGroup);
			(void)xrtNetEngineDestroy(pEngine);
			return 4;
		}
		printf("group-until: done\n");
		xrtTaskGroupDestroy(pGroup);
	}
	return xrtNetEngineDestroy(pEngine) ? 0 : 1;
}