9.OpenClaw源码解析——Lane并发调度

上节课我们学习了三层洋葱重试通过该技术能够解决出错报错重试的问题今天我们来学习Lane并发调度系统使用多队列独立并发控制替代全局锁实现对话心跳定时任务的解耦和并行。1. 设计目标将不同类别的任务用户对话、心跳检查、定时作业分配到独立的命名队列Lane中。每个 Lane 拥有可配置的最大并发数max_concurrency互不阻塞。任务以Callable形式入队返回concurrent.futures.Future支持异步获取结果。通过generation计数器实现优雅的重置/重启旧任务不再触发后续调度。2. 核心组件实现2.1 LaneQueue – 单 Lane 的 FIFO 队列每个 Lane 内部维护一个双端队列和一个条件变量控制并发执行数量。importthreadingimportconcurrent.futuresfromcollectionsimportdequefromtypingimportAny,CallableclassLaneQueue:def__init__(self,name:str,max_concurrency:int1):self.namename self.max_concurrencymax_concurrency self._deque:deque[tuple[Callable,concurrent.futures.Future,int]]deque()self._conditionthreading.Condition()self._active_count0self._generation0# 生命周期标记defenqueue(self,fn:Callable[[],Any],generation:int|NoneNone)-concurrent.futures.Future:futureconcurrent.futures.Future()withself._condition:gengenerationifgenerationisnotNoneelseself._generation self._deque.append((fn,future,gen))self._pump()# 尝试启动任务returnfuturedef_pump(self):持有锁时调用从队列取出任务直到达到并发上限whileself._active_countself.max_concurrencyandself._deque:fn,future,genself._deque.popleft()self._active_count1tthreading.Thread(targetself._run_task,args(fn,future,gen),daemonTrue)t.start()def_run_task(self,fn,future,gen):try:resultfn()future.set_result(result)exceptExceptionase:future.set_exception(e)finally:self._task_done(gen)def_task_done(self,gen):withself._condition:self._active_count-1ifgenself._generation:# 仅当任务属于当前 generation 时才继续泵送self._pump()self._condition.notify_all()2.2 CommandQueue – 中央调度器管理所有 Lane提供统一入队接口。classCommandQueue:def__init__(self):self._lanes:dict[str,LaneQueue]{}self._lockthreading.Lock()defget_or_create_lane(self,name:str,max_concurrency:int1)-LaneQueue:withself._lock:ifnamenotinself._lanes:self._lanes[name]LaneQueue(name,max_concurrency)returnself._lanes[name]defenqueue(self,lane_name:str,fn:Callable[[],Any])-concurrent.futures.Future:laneself.get_or_create_lane(lane_name)returnlane.enqueue(fn)defreset_all(self)-dict[str,int]:递增所有 lane 的 generation使旧任务失效result{}withself._lock:forname,laneinself._lanes.items():withlane._condition:lane._generation1result[name]lane._generationreturnresult3. 后台任务集成3.1 HeartbeatRunner – 心跳检查周期性读取HEARTBEAT.md通过 LLM 执行检查结果输出到队列。classHeartbeatRunner:def__init__(self,workspace:Path,cmd_queue:CommandQueue,interval:float1800.0):self.cmd_queuecmd_queue self.intervalinterval self.last_run_at0.0# ...defheartbeat_tick(self):ifnotself.should_run():return# 非阻塞若 heartbeat lane 已有活跃任务则跳过本次 ticklane_statsself.cmd_queue.get_or_create_lane(heartbeat).stats()iflane_stats[active]0:returndef_do_heartbeat():# 构造 prompt调用 LLMreturnrun_agent_single_turn(...)futureself.cmd_queue.enqueue(heartbeat,_do_heartbeat)future.add_done_callback(self._on_heartbeat_done)3.2 CronService – 定时任务从CRON.json读取作业定义每秒检查到期任务并提交到cronlane。classCronService:defcron_tick(self):nowtime.time()forjobinself.jobs:ifjob[enabled]andnowjob[next_run_at]:self._enqueue_job(job,now)def_enqueue_job(self,job,now):def_do_cron():returnrun_agent_single_turn(job[payload][message])futureself.cmd_queue.enqueue(cron,_do_cron)future.add_done_callback(lambdaf:self._on_cron_done(f,job))job[next_run_at]nowjob[every_seconds]4. 主循环与 REPLagent_loop()启动全部组件处理用户输入和命令。defagent_loop():cmd_queueCommandQueue()cmd_queue.get_or_create_lane(main,max_concurrency1)cmd_queue.get_or_create_lane(cron,max_concurrency1)cmd_queue.get_or_create_lane(heartbeat,max_concurrency1)heartbeatHeartbeatRunner(WORKSPACE_DIR,cmd_queue)heartbeat.start()cron_svcCronService(WORKSPACE_DIR/CRON.json,cmd_queue)threading.Thread(targetcron_loop,daemonTrue).start()whileTrue:# 1. 输出心跳 / cron 结果formsginheartbeat.drain_output():print_lane(heartbeat,msg)# 2. 获取用户输入user_inputinput(You )# 3. 将对话回合包装为可调用对象并提交到 main lanedefuser_turn():# 执行 LLM 多轮对话含工具调用returnfinal_response futurecmd_queue.enqueue(main,user_turn)print_assistant(future.result(timeout120))5. 并发流程图5.1 整体架构与数据流入队阶段调用方通过CommandQueue.enqueue(lane_name, fn)将任务提交到指定 Lane。队列与并发控制每个LaneQueue独立维护 FIFO 队列与_active_count_pump()在持有锁时启动新任务直到达到该 Lane 的max_concurrency上限。任务执行每个任务在独立线程中运行完成后回调_task_done()递减活跃计数并再次触发_pump()从而持续消费队列。生命周期隔离_generation计数器在reset_all()时递增旧任务完成回调时若gen ! self._generation则不再触发后续_pump()实现优雅重置。后台服务集成心跳和定时任务分别通过各自的 Lane 提交工作主循环独立处理用户输入三者互不阻塞。