Skip to content

Add thread-based executor for concurrent task processing - #6

Closed
Fazel94 wants to merge 2 commits into
LaunchPlatform:masterfrom
Fazel94:claude/add-thread-executor-01ERqW1n37jWs9J4HFFtcvo7
Closed

Fazel94 wants to merge 2 commits into
LaunchPlatform:masterfrom
Fazel94:claude/add-thread-executor-01ERqW1n37jWs9J4HFFtcvo7

Conversation

@Fazel94

@Fazel94 Fazel94 commented Nov 19, 2025 •

Copy link
Copy Markdown
Contributor

No description provided.

This commit adds ThreadPoolExecutor support to enable concurrent task processing
within a single worker instance, addressing the TODO at line 348 in app.py.

Key changes:
- Added MAX_WORKER_THREADS configuration option (default: 1 for backward compatibility)
- Created _process_task_in_thread() method for thread-safe task processing
- Each thread uses its own database session to avoid session conflicts
- Modified process_tasks() to use ThreadPoolExecutor when MAX_WORKER_THREADS > 1
- Added graceful executor shutdown on worker exit
- Preserved sequential processing when MAX_WORKER_THREADS = 1

Benefits:
- Enables concurrent processing of multiple tasks within a single worker
- Improves throughput for I/O-bound tasks
- Maintains backward compatibility with sequential processing by default
- Thread-safe with proper database session management per thread

Configuration:
- Set MAX_WORKER_THREADS to desired concurrency level (e.g., 4, 8, 10)
- Set to 0 to use default (number of CPUs * 5)
- Set to 1 to disable thread pool (sequential processing)
This commit fixes critical issues for thread-based executor to work correctly:

1. **SQLAlchemy Session Context Issue**:
   - Changed _process_task_in_thread to accept task_id instead of task object
   - Each worker thread now reloads the task in its own session
   - This prevents "DetachedInstanceError" when task objects cross thread boundaries

2. **Thread-Safe Connection Pool**:
   - Added conditional pool selection based on MAX_WORKER_THREADS
   - Uses QueuePool (thread-safe) when MAX_WORKER_THREADS > 1
   - Keeps SingletonThreadPool for backwards compatibility when MAX_WORKER_THREADS = 1
   - Configures pool_size dynamically based on number of worker threads

The previous implementation used SingletonThreadPool which is NOT thread-safe,
causing "connection already closed" errors when multiple threads accessed the pool.

These fixes ensure that:
- Each thread has its own database session
- Task objects are not shared between threads
- Connection pool is thread-safe for concurrent access
- No session conflicts or detached instance errors occur
@Fazel94 Fazel94 closed this Nov 19, 2025
@Fazel94
Fazel94 deleted the claude/add-thread-executor-01ERqW1n37jWs9J4HFFtcvo7 branch November 19, 2025 12:07
Fazel94 pushed a commit to Fazel94/bq that referenced this pull request Nov 23, 2025
Identified 8 bugs in threading implementation:

CRITICAL (3):
- Bug LaunchPlatform#1: Database session never closed in worker heartbeat thread
- Bug LaunchPlatform#2: AttributeError when worker is None in HTTP health check
- Bug LaunchPlatform#3: Session leak in HTTP request handler

HIGH (2):
- Bug LaunchPlatform#4: No error handling in worker heartbeat thread
- Bug LaunchPlatform#5: Stale worker object in heartbeat thread

MODERATE (3):
- Bug LaunchPlatform#6: Race condition in metrics server shutdown
- Bug LaunchPlatform#7: Potential duplicate NOTIFY when transaction is None
- Bug LaunchPlatform#8: Thread join timeout may leave zombie threads

Report includes detailed descriptions, reproduction steps,
recommended fixes, and priority recommendations for each bug.
Fazel94 pushed a commit to Fazel94/bq that referenced this pull request Dec 30, 2025
CRITICAL FIXES (3):
- Bug LaunchPlatform#1: Fixed session management in update_workers thread
  * Create fresh session each iteration instead of reusing
  * Always close session in finally block
  * Prevents connection pool exhaustion and stale connections

- Bug LaunchPlatform#2: Fixed AttributeError in HTTP health check
  * Check if worker is None before accessing worker.state
  * Return proper error state ("NOT_FOUND") when worker is missing
  * Prevents HTTP server crashes

- Bug LaunchPlatform#3: Fixed session leak in HTTP request handler
  * Close database session in finally block for all requests
  * Prevents connection pool exhaustion from health checks

HIGH SEVERITY FIXES (2):
- Bug LaunchPlatform#4: Added error handling to update_workers thread
  * Wrap all database operations in try/except
  * Log errors with full traceback
  * Rollback failed transactions
  * Continue heartbeat loop after errors (no silent death)

- Bug LaunchPlatform#5: Fixed stale worker object in heartbeat thread
  * Refresh worker object from database each iteration
  * Check if worker exists (handle deletion case)
  * Prevents updates to detached/stale objects

MODERATE FIXES (3):
- Bug LaunchPlatform#6: Fixed metrics server shutdown race condition
  * Use threading.Event instead of callback assignment
  * Store server instance in thread-safe manner
  * Prevents race condition during early shutdown

- Bug LaunchPlatform#7: Added logging for duplicate NOTIFY
  * Log debug message when transaction is None
  * Explains why deduplication is not possible
  * Minor performance impact, not critical

- Bug LaunchPlatform#8: Fixed thread join timeout handling
  * Check if threads stopped after join timeout
  * Log errors if threads still alive
  * Only mark worker as SHUTDOWN if heartbeat thread stopped
  * Prevents zombie threads and state inconsistencies

All fixes tested with syntax check and module import.
Module instantiation successful.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants