(task: Task)
| 496 | |
| 497 | |
| 498 | def enqueue_task(task: Task): |
| 499 | current_thread_count = is_alive() |
| 500 | if current_thread_count <= 0: # Render thread is dead |
| 501 | raise ChildProcessError("Rendering thread has died.") |
| 502 | |
| 503 | # Alive, check if task in cache |
| 504 | session = get_cached_session(task.session_id, update_ttl=True) |
| 505 | pending_tasks = list(filter(lambda t: t.is_pending, session.tasks)) |
| 506 | if len(pending_tasks) > current_thread_count * MAX_OVERLOAD_ALLOWED_RATIO: |
| 507 | raise ConnectionRefusedError( |
| 508 | f"Session {task.session_id} already has {len(pending_tasks)} pending tasks, with {current_thread_count} workers." |
| 509 | ) |
| 510 | |
| 511 | if session.put(task, TASK_TTL): |
| 512 | # Use twice the normal timeout for adding user requests. |
| 513 | # Tries to force session.put to fail before tasks_queue.put would. |
| 514 | if manager_lock.acquire(blocking=True, timeout=LOCK_TIMEOUT * 2): |
| 515 | try: |
| 516 | tasks_queue.append(task) |
| 517 | idle_event.set() |
| 518 | return task |
| 519 | finally: |
| 520 | manager_lock.release() |
| 521 | raise RuntimeError("Failed to add task to cache.") |
nothing calls this directly
no test coverage detected