fix: 抢占归还丢失——preempted_job 在重试循环内被重置为 None,关键字任务一旦重试(attempt≥2)被抢占的养号任务永不恢复;提升到循环外+已有值不覆盖+归还时被抢占任务停用/删除也打日志
This commit is contained in:
@@ -636,6 +636,9 @@ class TaskManager:
|
|||||||
# 错峰启动:在各自线程内等待,分摊批量启动的连接/占用压力
|
# 错峰启动:在各自线程内等待,分摊批量启动的连接/占用压力
|
||||||
if start_delay > 0:
|
if start_delay > 0:
|
||||||
time.sleep(start_delay)
|
time.sleep(start_delay)
|
||||||
|
# 被抢占的原任务 id(抢占结束后归还)——必须在循环外:
|
||||||
|
# 重试时若重置为 None,finally 归还逻辑会丢失信息,被抢占任务永不恢复
|
||||||
|
preempted_job = None
|
||||||
try:
|
try:
|
||||||
for attempt in range(1, max_attempts + 1):
|
for attempt in range(1, max_attempts + 1):
|
||||||
# 用户已请求停止 → 不再启动新 attempt
|
# 用户已请求停止 → 不再启动新 attempt
|
||||||
@@ -645,7 +648,6 @@ class TaskManager:
|
|||||||
return
|
return
|
||||||
# 同一 serial 同时只能一个 worker;不同任务可配置抢占
|
# 同一 serial 同时只能一个 worker;不同任务可配置抢占
|
||||||
preempt = False
|
preempt = False
|
||||||
preempted_job = None # 被抢占的原任务 id(抢占结束后归还)
|
|
||||||
with self._lock:
|
with self._lock:
|
||||||
if serial in self._running:
|
if serial in self._running:
|
||||||
cur = self._running[serial]
|
cur = self._running[serial]
|
||||||
@@ -658,6 +660,7 @@ class TaskManager:
|
|||||||
_log.warning(f"{serial} 已有任务在跑,跳过 (job={job.name})")
|
_log.warning(f"{serial} 已有任务在跑,跳过 (job={job.name})")
|
||||||
return
|
return
|
||||||
preempt = True
|
preempt = True
|
||||||
|
if not preempted_job:
|
||||||
preempted_job = cur.get("job_id")
|
preempted_job = cur.get("job_id")
|
||||||
if preempt:
|
if preempt:
|
||||||
# 抢占:锁外停止该设备上的其他任务(stop_device 内部拿同一把锁,
|
# 抢占:锁外停止该设备上的其他任务(stop_device 内部拿同一把锁,
|
||||||
@@ -749,6 +752,10 @@ class TaskManager:
|
|||||||
if j and j.enabled:
|
if j and j.enabled:
|
||||||
_log.info(f"{serial} 抢占任务 {job.name} 结束,归还设备给任务 {j.name}")
|
_log.info(f"{serial} 抢占任务 {job.name} 结束,归还设备给任务 {j.name}")
|
||||||
threading.Thread(target=self._run_job, args=(j,), daemon=True).start()
|
threading.Thread(target=self._run_job, args=(j,), daemon=True).start()
|
||||||
|
else:
|
||||||
|
# 打日志便于排查:被抢占任务已删除/停用时不会归还,但要知道原因
|
||||||
|
_log.info(f"{serial} 抢占任务 {job.name} 结束,"
|
||||||
|
f"被抢占任务 {preempted_job} {'已停用,不归还' if j else '已不存在,不归还'}")
|
||||||
|
|
||||||
# ---- 运行控制 ----
|
# ---- 运行控制 ----
|
||||||
def stop_device(self, serial):
|
def stop_device(self, serial):
|
||||||
|
|||||||
Reference in New Issue
Block a user