Fix deadlock in delete_job by using unlocked save when lock is already held
This commit is contained in:
1 parent
f327f609e0
commit
7853229786
1 file changed
+10
-6
+10
-6
@@ -68,10 +68,9 @@ JOBS_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'jobs.json'
|
|||||||
job_queue = {}
|
job_queue = {}
|
||||||
job_lock = threading.Lock()
|
job_lock = threading.Lock()
|
||||||
|
|
||||||
def save_jobs():
|
def _save_jobs_unlocked():
|
||||||
"""Save jobs to persistent storage"""
|
"""Internal: Save jobs without acquiring lock (assumes lock is already held)"""
|
||||||
try:
|
try:
|
||||||
with job_lock:
|
|
||||||
# Convert datetime objects to strings for JSON serialization
|
# Convert datetime objects to strings for JSON serialization
|
||||||
jobs_to_save = {}
|
jobs_to_save = {}
|
||||||
for job_id, job in job_queue.items():
|
for job_id, job in job_queue.items():
|
||||||
@@ -85,6 +84,11 @@ def save_jobs():
|
|||||||
except Exception as e:
|
except Exception as e:
|
||||||
print(f"ERROR: Failed to save jobs: {e}")
|
print(f"ERROR: Failed to save jobs: {e}")
|
||||||
|
|
||||||
|
def save_jobs():
|
||||||
|
"""Save jobs to persistent storage"""
|
||||||
|
with job_lock:
|
||||||
|
_save_jobs_unlocked()
|
||||||
|
|
||||||
def load_jobs():
|
def load_jobs():
|
||||||
"""Load jobs from persistent storage"""
|
"""Load jobs from persistent storage"""
|
||||||
global job_queue
|
global job_queue
|
||||||
@@ -144,7 +148,7 @@ def update_job_status(job_id, status, **kwargs):
|
|||||||
|
|
||||||
# Only save for important status changes
|
# Only save for important status changes
|
||||||
if status in ['completed', 'error', 'processing']:
|
if status in ['completed', 'error', 'processing']:
|
||||||
save_jobs()
|
_save_jobs_unlocked() # Use unlocked version since we already hold the lock
|
||||||
|
|
||||||
def cleanup_old_jobs():
|
def cleanup_old_jobs():
|
||||||
"""Remove jobs older than JOB_RETENTION_DAYS"""
|
"""Remove jobs older than JOB_RETENTION_DAYS"""
|
||||||
@@ -182,7 +186,7 @@ def cleanup_old_jobs():
|
|||||||
print(f"Cleaned up old job: {job_id}")
|
print(f"Cleaned up old job: {job_id}")
|
||||||
|
|
||||||
if jobs_to_remove:
|
if jobs_to_remove:
|
||||||
save_jobs()
|
_save_jobs_unlocked() # Use unlocked version since we already hold the lock
|
||||||
print(f"Removed {len(jobs_to_remove)} old jobs")
|
print(f"Removed {len(jobs_to_remove)} old jobs")
|
||||||
|
|
||||||
def save_jobs_batched():
|
def save_jobs_batched():
|
||||||
@@ -1495,7 +1499,7 @@ def delete_job(job_id):
|
|||||||
print(f"DEBUG: Removing job from queue", file=sys.stderr, flush=True)
|
print(f"DEBUG: Removing job from queue", file=sys.stderr, flush=True)
|
||||||
del job_queue[job_id]
|
del job_queue[job_id]
|
||||||
print(f"DEBUG: Saving jobs to disk", file=sys.stderr, flush=True)
|
print(f"DEBUG: Saving jobs to disk", file=sys.stderr, flush=True)
|
||||||
save_jobs() # Persist job deletion
|
_save_jobs_unlocked() # Use unlocked version since we already hold the lock
|
||||||
print(f"DEBUG: Job deletion complete", file=sys.stderr, flush=True)
|
print(f"DEBUG: Job deletion complete", file=sys.stderr, flush=True)
|
||||||
else:
|
else:
|
||||||
print(f"DEBUG: Job status is not completed: {job.get('status')}", file=sys.stderr, flush=True)
|
print(f"DEBUG: Job status is not completed: {job.get('status')}", file=sys.stderr, flush=True)
|
||||||
|
|||||||
Reference in new issue
Block a user