fix(cron_service): 修复更新任务启用状态时的死锁问题
当定时任务服务正在运行时,更新任务的启用状态可能导致死锁。 现在通过改进锁的使用方式来避免这个问题。 在update_enabled方法中添加了正确的变量初始化, 并在循环逻辑中进行了优化以确保正确释放锁。 同时添加了专门的测试用例来验证在并发场景下不会发生死锁。
This commit is contained in:
@ -134,6 +134,7 @@ class CronService:
|
|||||||
return job
|
return job
|
||||||
|
|
||||||
def update_enabled(self, job_id: str, enabled: bool) -> CronJob | None:
|
def update_enabled(self, job_id: str, enabled: bool) -> CronJob | None:
|
||||||
|
updated_job: CronJob | None = None
|
||||||
with self._lock:
|
with self._lock:
|
||||||
jobs = self._load_jobs_unlocked()
|
jobs = self._load_jobs_unlocked()
|
||||||
for job in jobs:
|
for job in jobs:
|
||||||
@ -143,9 +144,11 @@ class CronService:
|
|||||||
job.updated_at_ms = _now_ms()
|
job.updated_at_ms = _now_ms()
|
||||||
job.next_run_at_ms = compute_next_run(job.schedule) if job.enabled else None
|
job.next_run_at_ms = compute_next_run(job.schedule) if job.enabled else None
|
||||||
self._save_jobs_unlocked()
|
self._save_jobs_unlocked()
|
||||||
|
updated_job = job
|
||||||
|
break
|
||||||
|
if updated_job is not None:
|
||||||
self._arm_timer()
|
self._arm_timer()
|
||||||
return job
|
return updated_job
|
||||||
return None
|
|
||||||
|
|
||||||
def remove_job(self, job_id: str) -> bool:
|
def remove_job(self, job_id: str) -> bool:
|
||||||
with self._lock:
|
with self._lock:
|
||||||
|
|||||||
@ -1,4 +1,5 @@
|
|||||||
import asyncio
|
import asyncio
|
||||||
|
import threading
|
||||||
|
|
||||||
from beaver.foundation.models import CronExecutionResult, CronRunRecord, CronSchedule
|
from beaver.foundation.models import CronExecutionResult, CronRunRecord, CronSchedule
|
||||||
from beaver.tools.base import ToolContext
|
from beaver.tools.base import ToolContext
|
||||||
@ -108,6 +109,31 @@ def test_persisted_interval_job_keeps_schedule_and_next_run(tmp_path) -> None:
|
|||||||
assert reloaded.next_run_at_ms == job.next_run_at_ms
|
assert reloaded.next_run_at_ms == job.next_run_at_ms
|
||||||
|
|
||||||
|
|
||||||
|
def test_running_scheduler_can_disable_job_without_deadlock(tmp_path) -> None:
|
||||||
|
service = CronService(tmp_path / "jobs.json")
|
||||||
|
job = service.add_job(
|
||||||
|
name="Hydration reminder",
|
||||||
|
message="Drink water",
|
||||||
|
schedule=CronSchedule(kind="every", every_ms=30 * 60 * 1000),
|
||||||
|
)
|
||||||
|
service._running = True
|
||||||
|
completed = threading.Event()
|
||||||
|
enabled_values: list[bool] = []
|
||||||
|
|
||||||
|
def disable_job() -> None:
|
||||||
|
updated = service.update_enabled(job.id, False)
|
||||||
|
if updated is not None:
|
||||||
|
enabled_values.append(updated.enabled)
|
||||||
|
completed.set()
|
||||||
|
|
||||||
|
worker = threading.Thread(target=disable_job, daemon=True)
|
||||||
|
worker.start()
|
||||||
|
|
||||||
|
assert completed.wait(0.5), "disabling a running cron job should not deadlock"
|
||||||
|
assert enabled_values == [False]
|
||||||
|
assert service.get_job(job.id).enabled is False
|
||||||
|
|
||||||
|
|
||||||
def test_cron_tool_uses_runtime_service(tmp_path) -> None:
|
def test_cron_tool_uses_runtime_service(tmp_path) -> None:
|
||||||
service = CronService(tmp_path / "jobs.json")
|
service = CronService(tmp_path / "jobs.json")
|
||||||
tool = CronTool()
|
tool = CronTool()
|
||||||
|
|||||||
Reference in New Issue
Block a user