ThreadManager概述一、概述Camera HAL 中存在两套独立的 ThreadManager 实现分别服务于不同的框架层特性CamX ThreadManagerCHI ThreadManager适用层CamX 核心框架CHI Feature2 框架最大线程数1516最大注册作业512256最大运行时作业4096-同步作业不支持支持Sync Thread顺序添加不支持支持addInOrder作业分割支持isSplitable不支持序列化执行支持isSerialize不支持二、CamX ThreadManager 架构2.1 组件层次图┌─────────────────────────────────────────────────────────────────┐ │ ThreadManager │ │ (对外 API 门面) │ ├─────────────────────────────────────────────────────────────────┤ │ ┌──────────────┐ ┌──────────────┐ ┌──────────────────────┐ │ │ │ JobRegistry │ │ JobList │ │ ThreadCore │ │ │ │ 作业注册表 │ │ 运行时作业池 │ │ 线程核心引擎 │ │ │ └──────────────┘ └──────────────┘ └──────────────────────┘ │ │ │ │ │ ┌─────────┐ │ │ │JobQueue │ │ │ │优先级队列│ │ │ └─────────┘ │ └─────────────────────────────────────────────────────────────────┘2.2 核心数据结构JobHandle 编码格式┌─────────────────────────────────────┬─────────────────────────────┐ │ 高 32 位: uniqueCounter │ 低 32 位: slot │ │ (唯一标识计数器) │ (注册表槽位索引) │ └─────────────────────────────────────┴─────────────────────────────┘RegisteredJob已注册作业族structRegisteredJob{JobFunc funcAddr;// 作业函数地址CHAR name[128];// 作业名称JobCb flushDoneCb;// Flush 完成回调JobPriority priority;// 优先级 (Critical/High/Normal)BOOL isSerial;// 是否序列化执行UINT32 uniqueCounter;// 唯一计数器UINT32 slot;// 注册表槽位UINT32 jobCount;// 挂起作业数UINT32 inflightCount;// 正在执行数UINT32 holdCount;// 等待中的作业数UINT32 submittedCount;// 已提交待执行数FlushStatus flushStatus;// Flush 状态RuntimeJob*pFirstJob;// 链表头RuntimeJob*pLastJob;// 链表尾UINT32 cameraId;// 相机 ID};RuntimeJob运行时作业structRuntimeJob{UINT64 hJob;// 作业族句柄CHAR name[256];// 作业名称JobCb stoppedCb;// 停止回调VOID*pData[MaxDataPartition];// 数据分区数组BOOL isSplitable;// 是否可分割BOOL isBlocking;// 是否阻塞执行UINT32 numPartitions;// 数据分区数JobStatus status;// 作业状态UINT32 slot;// JobList 槽位索引Semaphore*pJobSemaphore;// 阻塞信号量RuntimeJob*pNext;// 下一个串行作业};作业状态机┌───────────────────────────────────────────────┐ │ │ ┌─────────┐ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │Submitted│──┼─│ Ready │──│On Hold │──│Stopped │ │ └─────────┘ │ └─────────┘ └─────────┘ └─────────┘ │ 入队 │ 可执行 等待前置作业 Flush 停止 │ │ │ └───────────────────────────────────────────────┘2.3 各组件职责JobRegistry作业注册表职责:管理所有已注册的作业族提供注册/注销接口维护作业计数统计jobCount/inflightCount/holdCount/submittedCount实现 Handle 编解码classJobRegistry{RegisteredJob m_registeredJobs[512];// 作业族数组BOOL m_registeredSlots[512];// 槽位占用标记UINT32 m_counter;// 唯一计数器Mutex*m_pRegistryLock;// 注册表锁Mutex*m_pFlushJobSubmitLock;// Flush 与提交的同步锁};JobList运行时作业池职责:提供 RuntimeJob 对象的分配/回收使用空闲链表管理可用槽位避免频繁的内存分配/释放classJobList{RuntimeJob m_jobList[4096];// 运行时作业数组JobSlot m_freeList[4096];// 空闲链表UINT32 m_freeHead;// 链表头UINT32 m_freeTail;// 链表尾UINT32 m_jobOccupancy;// 当前占用数};ThreadCore线程核心职责:创建/销毁工作线程管理优先级队列分发作业执行处理 Flush/Sync 操作集成健康监控classThreadCore{JobQueue m_jobQueues[3];// Critical/High/Normal 三个优先级队列ThreadConfig m_workers[15];// 工作线程配置Condition*m_pReadOK;// 线程唤醒条件变量volatileBOOL m_stopped;// 停止标志volatileBOOL m_jobPending;// 新作业标志};JobQueue优先级队列职责:实现作业入队/出队支持条件出队队列加锁保护2.4 核心流程创建流程ThreadManager::Create() └─ new ThreadManager() └─ Initialize() ├─ new JobRegistry() - Initialize() ├─ new JobList() - Initialize() └─ new ThreadCore() - Initialize() └─ StartThreads() └─ 创建 numThreads 个工作线程注册作业族流程ThreadManager::RegisterJobFamily() └─ JobRegistry::RegisterNewJob() ├─ 检查是否已注册 ├─ 查找空闲槽位 ├─ 填充 RegisteredJob 结构 └─ 生成 JobHandle (counter 32 | slot)提交作业流程ThreadManager::PostJob(hJob, stoppedCb, ppData, isSplitable, isBlocking) └─ ThreadCore::AcceptNewJob() ├─ JobList::AcquireJobEntry() // 分配 RuntimeJob ├─ 处理作业分割isSplitable ├─ JobQueue::Enqueue() // 加入优先级队列 └─ Trigger() // 唤醒工作线程 └─ m_pReadOK-Broadcast()工作线程执行流程ThreadCore::WorkerThreadBody() └─ DoWork(threadIndex) while (!m_stopped) { // 等待新作业 m_pReadOK-Wait() // ProcessJobQueue ProcessJobQueue(threadIndex) ├─ 从高优先级队列开始取作业 ├─ JobQueue::CheckAndDequeue() │ ├─ 检查 Flush 状态 │ ├─ 检查 Serial 依赖 │ └─ 返回符合条件的作业 ├─ 健康监控处理 └─ DispatchJob(pJob) └─ func(pData) // 执行作业函数 }Flush/Sync 流程ThreadManager::FlushJobFamily(hJob, pUserData, isBlocking) └─ ThreadCore::SyncOrFlushJob(hJob, pUserData, isBlocking, FlushRequested) ├─ JobRegistry::StartSyncOrFlush() │ ├─ 设置 flushStatus FlushRequested │ └─ 设置 blockFlush TRUE ├─ 唤醒所有工作线程处理剩余作业 ├─ JobRegistry::WaitForSyncOrFlush() │ └─ pFlushSemaphore-Wait() └─ JobRegistry::CheckAllFlushDone() ├─ 检查 inflight count 0 ├─ 调用 flushDoneCb └─ Signal() 释放等待者2.5 CamX 特有特性作业分割Job Splitting// PostJob 时可以指定 isSplitable TRUE// 系统会将 ppData 数组中的每个分区作为独立子作业执行// 所有子作业完成后视为整个作业完成m_pCore-AcceptNewJob(hJob,stoppedCb,ppData,TRUE,// isSplitableisBlocking,maxExecTime,pJobName);序列化执行Serialization// 注册时指定 isSerialize TRUE// 同一家族的作业必须串行执行// JobQueue::CheckAndDequeue 会检查// - 如果上一个作业仍在执行则当前作业进入 OnHold 状态// - 只有前驱作业完成后后继作业才能进入 Ready 状态阻塞调用Blocking// PostJob 时指定 isBlocking TRUE// 调用方会阻塞等待作业完成// ThreadCore 会为该作业创建 SemaphorepJob-pJobSemaphoreSemaphore::Create();// 作业执行完成后 SignalpJob-pJobSemaphore-Signal();// 调用方 WaitpJob-pJobSemaphore-Wait();三、CHI ThreadManager 架构3.1 组件结构CHI ThreadManager没有将组件拆分为独立的类而是将所有逻辑集成在CHIThreadManager类中。classCHIThreadManager:publicIMonitorClient{ChiRegisteredJob m_registeredJobs[256];// 已注册作业数组ChiThreadConfig m_workers[16];// 工作线程配置数组std::dequeChiRuntimeJob*m_pq;// 全局异步作业队列std::array...,50m_inflightJobs;// HMS 监控的进行中作业};3.2 核心数据结构ChiRuntimeJob运行时作业structChiRuntimeJob{UINT64 hJob;// 作业句柄VOID*pData;// 作业数据UINT64 sequenceNum;// 序列号ChiJobStatus status;// Submitted/Stopped/InvalidBOOL isSync;// 是否同步作业ChiJobPriority priority;// Default/CriticalUINT32 threadId;// 执行线程 IDSync 作业UINT64 execStartTime;// 执行开始时间UINT64 maxExecTime;// 最大执行时间pid_t pid;// 进程 IDpid_t tid;// 线程 ID};ChiRegisteredJob已注册作业族structChiRegisteredJob{JobFunc funcAddr;// 作业函数地址CHAR name[255];// 作业名称BOOL isSync;// 是否同步作业BOOL isUsed;// 槽位是否占用UINT64 hRegister;// 注册句柄volatileUINT32 inFlight;// 进行中作业数volatileUINT32 jobPending;// 队列中是否有作业JobFlushStatus flushStatus;// Flush 状态Condition*pSyncJobCondn;// Sync 线程条件变量Condition*pFlushOK;// Flush 完成条件Condition*pSyncOK;// Sync 完成条件Mutex*pQueueLock;// 主队列锁Mutex*pHelperPQLock;// 辅助队列锁BOOL addInOrder;// 是否按序添加UINT64 sequenceCounter;// 序列号计数器UINT32 cameraId;// 相机 IDChiThreadConfig worker;// Sync 线程配置std::dequeChiRuntimeJob*pq;// 主作业队列std::dequeChiRuntimeJob*helperPQ;// 辅助作业队列乱序时使用};ChiThreadConfig线程配置structChiThreadConfig{UINT32 threadId;// 逻辑线程号OSThreadHandle hWorkThread;// 系统线程句柄JobFunc workThreadFunc;// 线程入口函数VOID*pContext;// 上下文指针Mutex*pQueueLock;// 队列锁ChiThreadStatus status;// 线程状态BOOL isUsed;// 是否被 Sync 作业占用Condition*pJobCondn;// 唤醒条件变量JobHandle hJob;// 关联的作业句柄std::dequeChiRuntimeJob**pq;// 关联的作业队列};3.3 CHI 特有特性同步作业Sync Job// 注册时指定 isSync TRUE// 系统会为该作业族创建独立的线程CHIThreadManager::CreateSyncThread(rRegJob)├─ 创建专用Condition(pSyncJobCondn)├─ 配置 ChiThreadConfig └─ThreadCreate()创建独立线程// 同步作业线程独立于线程池专门处理该作业族顺序添加addInOrder// 注册时指定 addInOrder TRUE// 作业必须按 sequenceNum 递增执行// 乱序到达的作业会暂存到 helperPQCHIThreadManager::AddToHelperQueue(pJob)├─ 如果 sequenceNumsequenceCounter1│ └─ 直接加入主队列 └─ 否则 └─ 加入 helperPQ 并排序CHIThreadManager::PostJob(hJob,pData,sequenceNum)├─AddToHelperQueue(pRuntimeJob)└─ 如果 helperPQ 头部可接受 └─ 移入主队列并触发执行双队列模型┌─────────────────────────────────────────────────────┐ │ CHIThreadManager │ │ │ │ ┌─────────────────────────────────────────────┐ │ │ │ 异步线程池 (m_workers) │ │ │ │ │ │ │ │ m_pAsyncCondn ── 唤醒所有异步线程 │ │ │ │ │ │ │ │ ┌────────────────────────────────────┐ │ │ │ │ │ 全局异步队列 (m_pq) │ │ │ │ │ │ ┌───┐ ┌───┐ ┌───┐ │ │ │ │ │ │ │Job│ │Job│ │Job│ ... │ │ │ │ │ │ └───┘ └───┘ └───┘ │ │ │ │ │ └────────────────────────────────────┘ │ │ │ └─────────────────────────────────────────────┘ │ │ │ │ ┌─────────────────────────────────────────────┐ │ │ │ Sync 线程每个 Sync 作业族独立 │ │ │ │ │ │ │ │ pSyncJobCondn ── 唤醒指定 Sync 线程 │ │ │ │ │ │ │ │ ┌────────────────────────────────────┐ │ │ │ │ │ 每个作业族独立队列 (regJob.pq) │ │ │ │ │ │ ┌───┐ ┌───┐ ┌───┐ │ │ │ │ │ │ │Job│ │Job│ │Job│ ... │ │ │ │ │ │ └───┘ └───┘ └───┘ │ │ │ │ │ └────────────────────────────────────┘ │ │ │ └─────────────────────────────────────────────┘ │ └─────────────────────────────────────────────────────┘3.4 CHI 核心流程创建与启动CHIThreadManager::Create(ppInstance, pName, numThreads) └─ new CHIThreadManager(numThreads) └─ Initialize(pName) ├─ 创建 Mutex/Condition 同步原语 └─ StartThreads() └─ 创建 numThreads 个异步工作线程注册流程CHIThreadManager::RegisterJobFamily(jobFunc, name, phJob, isSync, addInOrder) ├─ 加锁 m_pRegisteredJobLock ├─ 检查是否已注册 ├─ 查找空闲槽位 ├─ 初始化 ChiRegisteredJob │ ├─ 创建 Condition/Mutex │ ├─ 设置 isSync / addInOrder │ └─ 生成 hRegister (Handle) ├─ 如果 isSync TRUE │ └─ CreateSyncThread() // 创建专属线程 └─ 设置 flushStatus Noflush工作线程执行CHIThreadManager::DoWork(pConfig) while (status ! Stopped) { while (status ! Stopped (isUsed !jobPending || !isUsed !m_jobPending)) { pJobCondn-Wait(); // 等待新作业 } // ProcessJobQueue ProcessJobQueue(pConfig) ├─ 遍历 pq 中的作业 ├─ 检查 flushStatus │ ├─ FlushRequested - 弹出作业并释放 │ └─ Noflush/SyncRequested - DispatchJob ├─ MonitorAndDispatchJob (如果启用 HMS) │ ├─ 记录执行时间 │ ├─ 加入 inflight 集合 │ ├─ func(pData) │ └─ 从 inflight 集合移除 └─ DispatchJob (无 HMS) └─ func(pData) }Sync/Flush 流程CHIThreadManager::Sync(hJob, timeout) └─ SynchronizeBody(hJob, SyncRequested, timeout) ├─ 加锁 ├─ 如果 addInOrder处理 helperPQ ├─ 检查是否有进行中的作业 ├─ 设置 flushStatus SyncRequested/FlushRequested ├─ 触发所有相关线程处理剩余作业 │ ├─ AsyncTrigger(TRUE) // 广播所有异步线程 │ └─ SyncTrigger(rRegJob) // 唤醒 Sync 线程 ├─ 等待所有作业完成 │ └─ pFlushOK-TimedWait(timeout) └─ 重置 flushStatus Noflush四、两套实现对比4.1 架构差异维度CamX ThreadManagerCHI ThreadManager设计模式组件化Registry/List/Core/Queue 分离单体式所有逻辑在一个类中队列模型3个优先级队列 链表全局队列 每作业族独立队列线程模型纯线程池无专属线程线程池 Sync 专属线程作业调度优先级 序列号FIFO 可选顺序序列化支持isSerialize不支持作业分割支持isSplitable不支持4.2 适用场景CamX ThreadManager 适用于需要作业并行执行的场景如图像处理需要优先级调度的场景需要序列化保证的场景需要作业分割的场景CHI ThreadManager 适用于需要同步作业族独立线程的场景需要严格顺序执行的场景Feature2 层的算法处理4.3 API 映射功能CamXCHI创建Create(ppInstance, name, numThreads)Create(ppInstance, name, numThreads50)注册RegisterJobFamily(func, name, flushCb, priority, isSerialize, phJob, cameraId)RegisterJobFamily(func, name, phJob, isSync, addInOrder, cameraId)提交PostJob(hJob, stopCb, ppData, isSplitable, isBlocking, maxTime, name)PostJob(hJob, pData, seqNum, maxTime, priority)注销UnregisterJobFamily(hJob)UnregisterJobFamily(hJob)FlushFlushJobFamily(hJob, userData, blocking)FlushJob(hJob)(内部调用)SyncSyncJobFamily(hJob)Sync(hJob, timeout)恢复ResumeJobFamily(hJob)ResumeJobFamily(hJob)(需实现)五、关键设计总结5.1 线程安全所有共享数据访问都有 Mutex 保护状态标志使用原子操作Condition 变量用于线程间通知5.2 健康监控集成两套实现都继承IMonitorClient支持作业超时检测支持暂停/恢复监控通过OnScanAllJobs上报失败作业5.3 性能优化对象池模式JobList 预分配 RuntimeJob条件变量通知避免忙等待优先级队列重要作业优先执行批量唤醒Broadcast vs Signal5.4 容错设计Flush/Sync 超时机制作业停止回调状态机保护非法状态转换检查资源泄漏保护析构函数中清理所有资源