executor API 参考 - XRT 程序设计

模块契约Executor

所有权

xrtExecutorSubmit 和 xrtExecutorSubmitBatch 成功后接管每项 Destroy。工作执行完成，或 xrtExecutorCancel 丢弃尚未开始的工作时，析构恰好执行一次。提交失败时执行器不会调用析构。

性能契约

创建时按 Threads * QueueLimit 预分配全部作业槽。
提交、批量提交、窃取和完成不申请通用堆内存。
每个 Worker 独立维护双端队列，本地从新端执行，空闲 Worker 从旧端窃取。
QueueLimit 是每个 Worker 的硬上限；满载立即返回 XERR_AGAIN。
批量提交在一个 Worker 队列中全成或全败，成功后才整体转移数据析构责任。

示例

完整示例位于 examples/concurrency/executor/main.c 与 examples/concurrency/executor_tour/main.c。

xexecutoritem类型

批量描述符只在提交调用期间借用，成功后各项数据责任转移给执行器。

字段	类型	语义
`Proc`	`xexecutorproc`	过程指针
`Data`	`ptr`	数据
`Destroy`	`xexecutorfreeproc`	Destroy
`DestroyContext`	`ptr`	DestroyContext

typedef struct xexecutoritem { xexecutorproc Proc; ptr Data; xexecutorfreeproc Destroy; ptr DestroyContext; } xexecutoritem;

xexecutorconfig类型

QueueLimit 是每个 Worker 的硬队列上限；总排队容量等于 Threads 乘 QueueLimit。

字段	类型	语义
`Threads`	`uint32`	线程数
`QueueLimit`	`size_t`	QueueLimit
`StackSize`	`size_t`	栈大小

typedef struct xexecutorconfig { uint32 Threads; size_t QueueLimit; size_t StackSize; } xexecutorconfig;

xexecutorstats类型

统计区分瞬时负载、累计受理、执行、窃取、取消和拒绝数量。

字段	类型	语义
`Threads`	`uint32`	线程数
`QueueLimit`	`size_t`	QueueLimit
`Queued`	`size_t`	Queued
`Running`	`size_t`	Running
`Submitted`	`uint64`	Submitted
`Completed`	`uint64`	Completed
`Executed`	`uint64`	Executed
`Stolen`	`uint64`	Stolen
`Cancelled`	`uint64`	Cancelled
`Rejected`	`uint64`	Rejected
`Closed`	`bool`	Closed
`Cancelling`	`bool`	Cancelling

typedef struct xexecutorstats { uint32 Threads; size_t QueueLimit; size_t Queued; size_t Running; uint64 Submitted; uint64 Completed; uint64 Executed; uint64 Stolen; uint64 Cancelled; uint64 Rejected; bool Closed; bool Cancelling; } xexecutorstats;

xexecutor类型

Executor 是无 Future 的有界高吞吐执行器，与传播结果和取消的 TaskPool 分工。不透明句柄或别名；生命周期与所有权见各使用方 API 节。

typedef struct xexecutor xexecutor;

xexecutorproc类型

工作过程不返回结果；错误上下文只在当前工作内有效，返回后由执行器清理。回调类型；参数与返回语义见签名及各使用方 API 节。

typedef void(*xexecutorproc)(ptr pData)

xexecutorfreeproc类型

工作数据析构在过程返回或排队工作被取消后执行一次。回调类型；参数与返回语义见签名及各使用方 API 节。

typedef void(*xexecutorfreeproc)(ptr pData, ptr pContext)

xrtExecutorCreate函数

创建预分配作业槽、Worker 本地队列和工作窃取线程。

xexecutor* xrtExecutorCreate(const xexecutorconfig* pConfig)

参数

参数	方向	约束	说明
`pConfig`	输入	允许空	空 = 默认配置；`Threads` 与 `QueueLimit` 零值自动取默认

返回值

返回	含义	失败时状态
非空	运行中的执行器	—
`NULL`	配置非法、溢出或分配失败	错误经 `xrtGetError()` 报告

错误

XERR_ARGUMENT — 配置字段非法
XERR_RANGE — Threads * QueueLimit 槽位计算溢出
XERR_MEMORY — 作业槽或线程分配失败

范例已编译验证

concurrency/executor_tour · 批量 · 显式 2 线程 8 槽

xexecutorconfig Config = { 2, 8, 0 };

xrtExecutorSubmit函数

提交一个 detached 工作；成功后接管数据析构，失败时所有权仍属于调用方。

bool xrtExecutorSubmit(xexecutor* pExecutor, xexecutorproc pProc, ptr pData, xexecutorfreeproc pDestroy, ptr pDestroyContext)

参数

参数	方向	约束	说明
`pExecutor`	输入/输出	非空、未关闭	目标执行器
`pProc`	输入	非空	工作过程
`pData`	输入	任意值	过程数据；成功后析构责任移交
`pDestroy`	输入	允许空	数据析构过程；空 = 无析构
`pDestroyContext`	输入	任意值	原样传给析构过程

返回值

返回	含义	失败时状态
`true`	已受理；执行完成或被取消丢弃时析构恰好一次	—
`false`	队列满、已关闭或参数非法	数据所有权不变

错误

XERR_ARGUMENT — 执行器或过程为空
XERR_AGAIN — 目标 Worker 队列达到 QueueLimit（满载背压）
XERR_CLOSED — 执行器已 Close/Cancel

范例已编译验证

concurrency/executor · 提交 · 千次提交后销毁收口

if ( !xrtExecutorSubmit(
	pExecutor,
	executorExampleRun,
	&Count,
	NULL,
	NULL
) ) {
	(void)xrtExecutorCancel(pExecutor);

xrtExecutorSubmitBatch函数

原子提交一组 detached 工作；整组成功或整组失败，不保证执行顺序。

bool xrtExecutorSubmitBatch(xexecutor* pExecutor, const xexecutoritem* pItems, size_t iCount)

参数

参数	方向	约束	说明
`pExecutor`	输入/输出	非空、未关闭	目标执行器
`pItems`	输入	非空	工作数组（每项含 Proc/Data/Destroy/DestroyContext）
`iCount`	输入	`> 0`	工作数量

返回值

返回	含义	失败时状态
`true`	整组受理；析构责任整体移交	—
`false`	容量不足、已关闭或参数非法	整组所有权不变（全成或全败）

错误

XERR_ARGUMENT — 指针为空或计数为零
XERR_AGAIN — 单 Worker 队列容不下整组
XERR_CLOSED — 已停止受理

范例已编译验证

concurrency/executor_tour · 批量 · 三项原子受理

if ( !xrtExecutorSubmitBatch(pExecutor, Items, 3u) ) {
	goto Cleanup;
}

xrtExecutorClose函数

停止受理新工作，并让已经受理的工作自然排空。

bool xrtExecutorClose(xexecutor* pExecutor)

参数

参数	方向	约束	说明
`pExecutor`	输入/输出	非空	目标执行器；幂等

返回值

返回	含义	失败时状态
`true`	已停止受理；已受理工作继续执行	—
`false`	指针为空	`XERR_ARGUMENT`

错误

XERR_ARGUMENT — 指针为空

范例已编译验证

concurrency/executor_tour · 收口 · Close 后 Wait 族收口

if ( !xrtExecutorClose(pExecutor) ||
	(xrtExecutorWaitFor(pExecutor,
		UINT64_C(3000000)) != XWAIT_OK) ||

xrtExecutorCancel函数

停止受理，丢弃尚未开始的工作并执行其数据析构；运行中工作不会被强停。

bool xrtExecutorCancel(xexecutor* pExecutor)

参数

参数	方向	约束	说明
`pExecutor`	输入/输出	非空	目标执行器

返回值

返回	含义	失败时状态
`true`	已停止受理；排队工作的析构被立即执行	—
`false`	指针为空	`XERR_ARGUMENT`

错误

XERR_ARGUMENT — 指针为空

范例已编译验证

concurrency/executor · 失败路径 · 提交失败即取消销毁

	(void)xrtExecutorCancel(pExecutor);
	(void)xrtExecutorDestroy(pExecutor);
	return 2;

xrtExecutorWait函数

永久等待已经关闭的执行器排空；Executor Worker 不能等待自身。

xwaitresult xrtExecutorWait(xexecutor* pExecutor)

参数

参数	方向	约束	说明
`pExecutor`	输入	非空、已关闭	目标执行器

返回值

返回	含义	失败时状态
`XWAIT_OK`	已排空	—
`XWAIT_ERROR`	未关闭、参数非法或本执行器 Worker 调用	错误经 `xrtGetError()` 报告

错误

XERR_ARGUMENT — 指针为空
XERR_STATE — 未先 Close，或从本执行器 Worker 线程等待自身

范例已编译验证

concurrency/executor_tour · 收口 · Close 后三形态依次核对

	(xrtExecutorWait(pExecutor) != XWAIT_OK) ||
	!xrtExecutorGet(pExecutor, &Stats) ||
	!Stats.Closed ||

xrtExecutorWaitFor函数

在相对毫秒数内等待已经关闭的执行器排空。

xwaitresult xrtExecutorWaitFor(xexecutor* pExecutor, int64 iTimeout)

参数

参数	方向	约束	说明
`pExecutor`	输入	非空	目标执行器
`iTimeout`	输入	毫秒	相对时限

返回值

返回	含义	失败时状态
`XWAIT_OK`	已排空（排空优先于超时）	—
`XWAIT_TIMEOUT`	到期未排空	不设置错误
`XWAIT_ERROR`	同 `xrtExecutorWait` 的失败条件	错误经 `xrtGetError()` 报告

错误

同 xrtExecutorWait（XERR_ARGUMENT / XERR_STATE）

范例已编译验证

concurrency/executor_tour · 收口 · 关闭后的限期等待

(xrtExecutorWaitFor(pExecutor,
	UINT64_C(3000000)) != XWAIT_OK) ||

xrtExecutorGet函数

复制当前负载和累计统计快照。

bool xrtExecutorGet(const xexecutor* pExecutor, xexecutorstats* pStats)

参数

参数	方向	约束	说明
`pExecutor`	输入	非空	目标执行器
`pStats`	输出	非空	接收 Threads/QueueLimit/Queued/Running 与 Submitted/Completed/Executed/Stolen/Cancelled/Rejected

返回值

返回	含义	失败时状态
`true`	并发一致快照已写入	—
`false`	指针为空	`XERR_ARGUMENT`

错误

XERR_ARGUMENT — 任一指针为空

范例已编译验证

concurrency/executor_tour · 统计 · 受理/完成/线程数核对

if ( !xrtExecutorGet(pExecutor, &Stats) ||
	(Stats.Submitted < 3u) ||
	(Stats.Completed != 3u) ||
	(Stats.Threads != 2u) ) {

xrtExecutorDestroy函数

关闭、排空、停止 Worker 并释放执行器；Worker 不能销毁自身执行器。

bool xrtExecutorDestroy(xexecutor* pExecutor)

参数

参数	方向	约束	说明
`pExecutor`	输入	允许空	空 = 空操作

返回值

返回	含义	失败时状态
`true`	已关闭、排空并释放	—
`false`	参数非法或 Worker 自销毁	对象保留；错误经 `xrtGetError()` 报告

错误

XERR_ARGUMENT — 指针为空（非空操作路径）
XERR_STATE — 从本执行器 Worker 线程销毁自身

范例已编译验证

concurrency/executor · 收口 · 内部执行 Close+排空+释放

if ( !xrtExecutorDestroy(pExecutor) ) {
	return 3;
}
printf("completed: %u\n", xrtAtomic32Load(&Count, XMEMORY_ACQUIRE));

常量索引常量

以下常量在本页所属头文件中定义，语义见公共头注释。

常量	说明
XRT_EXECUTOR_QUEUE_LIMIT_DEFAULT	见头文件注释
XRT_EXECUTOR_THREAD_LIMIT	见头文件注释