RTU/mimo/工程/libtask模块分析.md

libtask 模块分析

日期: 2026-06-15（更新）
基于源码: src/public/libtask/src/myTask.c（1个文件，913行）

---

1. 模块定位

libtask 是 RTU 的线程间同步原语库，位于公共库层。它提供三个独立的子系统：事件标志、消息队列、定时器。全部基于 POSIX 标准 API（pthread、timer_create），零第三方依赖，纯 C 实现。

RTU 的 9 个应用线程（app_sys、app_cmd、app_comm_channel 等）全部通过 libtask 的原语进行同步和通信。

2. 三个子系统

2.1 事件标志（Task Event）

用途: 线程间事件通知，支持多事件同时等待。

核心结构:

typedef struct {
    char name[64];
    uint32_t events;           // 当前挂起的事件位掩码
    pthread_mutex_t mutex;
    pthread_cond_t cond;
    int ref_count;
} stru_task_event;

API:

函数	说明
task_event_create(name)	创建事件对象，ref_count=1
task_event_send(event, bits)	设置事件位（OR 操作），signal 唤醒等待者
task_event_recv(event, set, opt, timeout_ms, &recved)	等待事件。opt: AND/OR + CLEAR
task_event_clear(event, bits)	手动清除事件位
task_event_query(event)	新增 — 不等待，直接返回当前挂起的事件位
task_event_destroy(event)	引用计数减 1，归零则释放

等待逻辑:

task_event_recv(set, opt, timeout):
  lock mutex
  while 事件未满足:
    if timeout == 0: → 返回 -1 (非阻塞)
    if timeout == FOREVER: → pthread_cond_wait (永久阻塞)
    else: → pthread_cond_timedwait (超时返回 -1)
  if 满足:
    OR 模式: 检查 (events & mask) == mask
    AND 模式: 检查 events 包含所有 bits
  if CLEAR flag: events &= ~mask (自动清除已捕获的事件)
  unlock mutex

关键设计: 通过 TASK_EVENT_FLAG_CLEAR 实现边沿触发语义 —— 唤醒后自动清除，防止重复处理。

---

2.2 消息队列（Task Message Queue）

用途: 线程间数据传递，环形缓冲区 + 长度前缀编码。

核心结构:

typedef struct {
    char name[64];
    uint8_t *buffer;           // 环形缓冲区（msg_size × max_msgs）
    uint32_t size;             // 单条消息的最大大小
    uint32_t max_msgs;         // 队列容量
    uint32_t msg_count;        // 当前消息数
    uint32_t head;             // 读指针
    uint32_t tail;             // 写指针
    pthread_mutex_t mutex;
    pthread_cond_t cond;
    int ref_count;
} stru_task_msg_queue;

消息编码格式:

每条消息在缓冲区中的格式为：

[4字节 msg_len] [msg_len 字节 payload]

写入时先写长度再写数据，读时先读长度再读数据。这允许可变长度消息。

API:

函数	说明
task_msg_queue_create(name, msg_size, msg_num)	创建队列，分配 msg_size × msg_num 字节缓冲区
task_msg_queue_send(queue, msg, size)	非阻塞写入。队列满时返回 -1
task_msg_queue_send_timeout(queue, msg, size, timeout_ms)	新增 — 写入。队列满时阻塞等待（支持 FOREVER / 定时 / 0=非阻塞）
task_msg_queue_recv(queue, msg, size, timeout_ms)	阻塞读取（支持 FOREVER / 定时 / 0=try）
task_msg_queue_try_recv(queue, msg, size)	非阻塞读取（内部调用 recv(0)）
task_msg_queue_get_count(queue)	返回当前消息数
task_msg_queue_space(queue)	返回剩余空间
task_msg_queue_destroy(queue)	引用计数减 1，归零释放

关键行为:

• send 满时直接返回 -1，不阻塞，不覆盖旧数据。生产者需自行处理
• recv 队列空时 pthread_cond_wait 阻塞等待
• 头部 4 字节长度前缀确保可变长度消息的正确读取

---

2.3 定时器（Task Timer）

用途: 一次性或周期性定时任务。

核心结构:

typedef struct {
    char name[64];
    timer_t timerid;           // POSIX timer ID
    timer_func_cb fun;         // 回调函数
    void *arg;                 // 回调参数
    uint32_t timeout_ms;       // 超时毫秒
    int flags;                 // TASK_TIMER_FLAG_PERIODIC
    int active;                // 是否活跃
    pthread_mutex_t mutex;
    int ref_count;
} stru_task_timer;

定时器机制 ✅ 已优化 (2026-06-15): 使用 Linux timerfd + epoll + 单例管理器线程 替代原 SIGEV_THREAD

task_timer_create → timerfd_create(CLOCK_REALTIME, TFD_NONBLOCK) 创建 fd
  → task_timer_start → timerfd_settime 装备 → epoll_ctl(ADD) 注册
    → epoll_wait 就绪 → read(fd) 清除过期计数 → fun(arg) 调用回调
      → 周期模式: timerfd 自动重触发
      → 单次模式: active=0，回调不再执行

API:

函数	说明
task_timer_create(name, cb, arg, timeout_ms, flags)	创建定时器（未启动）
task_timer_start(timer)	启动定时器
task_timer_stop(timer)	停止定时器（设置 its={0,0}）
task_timer_restart(timer, new_timeout)	修改超时并重启
task_timer_is_active(timer)	查询是否活跃
task_timer_destroy(timer)	引用计数减 1，归零删除 timer + 释放

---

3. 引用计数机制

三个子系统都使用引用计数（ref_count）进行生命周期管理：

// 创建
p->ref_count = 1;

// 获取 → ref_count++
// 不需要显式 get API，通过外部指针共享隐式增加

// 销毁
lock mutex
p->ref_count--;
if (p->ref_count > 0) { unlock; return 0; }  // 仍有使用者，不释放
unlock mutex
free(p);  // 最后一个使用者，释放

特点: 多线程共享同一个事件/队列/定时器对象时，最后一个使用者释放。但没有显式的 retain/release API，ref_count 的实际增减依赖外部代码手动管理。

---

4. 辅助函数

void task_sleep_ms(uint32_t ms):
  // 使用 nanosleep 实现毫秒级休眠，支持 EINTR 中断后自动恢复
  struct timespec ts = {ms/1000, (ms%1000)*1000000};
  while(nanosleep(&ts, &ts) == -1 && errno == EINTR);

---

5. 在 RTU 中的应用

RTU 使用 libtask 实现线程调度框架 myTask.h（位于 release/inc/），定义：

// 在 myTask.h 中定义（不在 libtask 目录内）
#define EV_TIMER1  (1 << 0)   // 10ms
#define EV_TIMER2  (1 << 1)   // 100ms
#define EV_TIMER3  (1 << 2)   // 1000ms
#define EV_TIMER4  (1 << 3)

#define TASK_EVENT_WAIT_FOREVER  ~0
#define TASK_EVENT_FLAG_OR       0
#define TASK_EVENT_FLAG_AND      1
#define TASK_EVENT_FLAG_CLEAR    2

#define TASK_TIMER_FLAG_PERIODIC  1
#define TASK_TIMER_FLAG_ONE_SHOT  0

typedef void *(*task_thread_func)(void *arg);

每个应用线程的主循环模式：

while (1) {
    task_event_recv(p_event, EV_TIMER1 | EV_TIMER2 | EV_TIMER3,
                    TASK_EVENT_FLAG_OR | TASK_EVENT_FLAG_CLEAR,
                    TASK_EVENT_WAIT_FOREVER, &event);
    if (event & EV_TIMER1) { /* 10ms 任务 */ }
    if (event & EV_TIMER2) { /* 100ms 任务 */ }
    if (event & EV_TIMER3) { /* 1000ms 任务 */ }
}

---

6. 优点

优点	说明
纯 POSIX + Linux 原生	事件/消息队列用 pthread，定时器用 timerfd+epoll，无第三方库依赖，交叉编译无障碍
单线程管理所有定时器	✅ 已优化 (2026-06-15) — 一个持久化 epoll 线程管理全部 27 个定时器，消除原 SIGEV_THREAD 每秒 ~1000 次线程创建/销毁
stop/destroy 零 CPU 同步	✅ 已优化 (2026-06-15) — condvar 替代自旋，while(running) cond_wait 阻塞等待无 CPU 开销
完备的超时支持	事件和消息队列都支持三种模式：永久等待 / 指定超时 / 非阻塞立即返回
双条件事件	支持 AND（全部触发）和 OR（任一触发）两种等待模式，灵活匹配不同场景
边沿触发	TASK_EVENT_FLAG_CLEAR 自动在等待返回时清除事件位，防止重复处理
可变长度消息	消息队列通过 4 字节长度前缀编码，支持不同大小的消息共用同一队列
完备的队列查询	get_count 和 space 提供队列状态查询，生产者可据此做背压决策
引用计数生命周期	多线程共享对象时安全释放，不会出现 use-after-free
代码紧凑	全部功能在 913 行 C 代码中，易于审计和理解

7. 缺点

缺点	严重度	说明
定时器 destroy use-after-free	高	✅ 已修复 (2026-06-15): task_timer_destroy 中原 free(p) 后在 LOG 中访问 p->name，free 移到 LOG 之后
消息队列 destroy use-after-free	高	✅ 已修复 (2026-06-15): task_msg_queue_destroy 同上
事件 destroy use-after-free	高	✅ 已修复 (2026-06-15): task_event_destroy 同上
事件 recv AND 位测试陷阱	中	✅ 已修复 (2026-06-15): opt == TASK_EVENT_FLAG_AND 改为 opt & TASK_EVENT_FLAG_AND，避免 AND|CLEAR 组合被误识别
事件 send 用 signal 非 broadcast	低	✅ 已修复 (2026-06-15): pthread_cond_signal 改为 pthread_cond_broadcast，多等待者场景更稳健
无事件优先级	低	task_event_recv 按位掩码匹配，所有事件平等。多事件同时触发无优先级排序
ref_count 缺乏原子性	低	引用计数通过 mutex 保护无 CAS，极高频场景下瓶颈（本场景不适用）
头文件不在模块内	低	myTask.h 位于 release/inc/，新开发者可能找不到