update51

app.py

@@ -69,10 +69,17 @@ def queue_processor():
             queue_size = task_queue.qsize()
             if queue_size > 0:
                 debug_log(f"Queue processor found {queue_size} tasks waiting")
+            else:
+                # No tasks waiting, sleep briefly to avoid CPU spinning
+                time.sleep(0.1)
+                continue
                 
             # Get a task from the queue with small timeout to prevent blocking
-            task_id, input_data, request_time = task_queue.get(timeout=0.1)
-            debug_log(f"Processing task {task_id}")
+            try:
+                task_id, input_data, request_time = task_queue.get(timeout=0.1)
+                debug_log(f"Processing task {task_id}")
+            except queue.Empty:
+                continue
             
             # Increment processing count to track active tasks
             with lock:
@@ -150,7 +157,8 @@ def queue_processor():
             debug_log(f"Task {task_id} completed and marked as done")
             
         except queue.Empty:
-            continue
+            # Use a small timeout to avoid CPU spinning
+            time.sleep(0.1)
         except Exception as e:
             debug_log(f"Error in queue processor: {str(e)}")
             if 'task_id' in locals():
@@ -285,11 +293,20 @@ def synchronous_evaluate(input_data):
     """Synchronously evaluate code, compatible with original interface"""
     debug_log(f"Received synchronous evaluation request")
     
+    # Add metadata to identify sync requests
+    if isinstance(input_data, list) and len(input_data) > 0 and isinstance(input_data[0], dict):
+        if 'metadata' not in input_data[0]:
+            input_data[0]['metadata'] = {}
+        input_data[0]['metadata']['source'] = 'sync_api'
+    
     # Create a task and queue it
     task_info = enqueue_task(input_data)
     task_id = task_info['task_id']
     debug_log(f"Created task {task_id} for synchronous evaluation")
     
+    # Ensure the task appears in the queue UI, add artificial delay if needed
+    time.sleep(0.1)  # Small delay to make sure the task is visible in queue
+    
     # Wait for task to complete
     while True:
         with lock:
@@ -349,22 +366,36 @@ def enqueue_task(input_data):
     estimated_time_per_task = _get_estimated_time_for_task(language, task_complexity)
     estimated_total_time = estimated_time_per_task * task_size
     
-    task_counter += 1
-    task_id = f"task_{task_counter}_{str(uuid.uuid4())}"
+    # Generate task ID in a thread-safe way
+    with lock:
+        task_counter += 1
+        local_counter = task_counter
+    
+    task_id = f"task_{local_counter}_{str(uuid.uuid4())[:8]}"
     request_time = time.time()
     
     debug_log(f"Creating new task: {task_id}")
     
-    # Get current queue status before adding to task_status
-    current_queue_size = task_queue.qsize()
-    debug_log(f"Current queue size before adding: {current_queue_size}")
+    # Track if this is a synchronous or asynchronous submission
+    is_async = 'async_submission' in str(threading.current_thread().name).lower() or 'async' in input_data[0].get('metadata', {}).get('source', '') if isinstance(input_data, list) and input_data and isinstance(input_data[0], dict) and 'metadata' in input_data[0] else False
     
-    # Add to task_status with 'queued' status first
+    # Get current queue status before adding to task_status
     with lock:
+        # Count actual queue status - both in queue AND waiting in task_status
+        current_queue_size = task_queue.qsize()
+        actual_waiting = sum(1 for t in task_status.values() if t['status'] == 'queued')
+        total_waiting = actual_waiting  # Use the actual count from task_status
+        
+        debug_log(f"Current queue metrics: queue_size={current_queue_size}, task_status_waiting={actual_waiting}, total={total_waiting}")
+        
+        queue_position = total_waiting + 1
+        
+        # Add to task_status with 'queued' status first
         task_status[task_id] = {
             'status': 'queued',
             'queued_time': request_time,
-            'queue_position': current_queue_size + 1,
+            'queue_position': queue_position,
+            'is_async': is_async,
             'estimated_factors': {
                 'language': language,
                 'size': task_size,
@@ -372,7 +403,7 @@ def enqueue_task(input_data):
             },
             'estimated_time': estimated_total_time
         }
-        debug_log(f"Added task {task_id} to task_status with queue position {current_queue_size + 1}")
+        debug_log(f"Added task {task_id} to task_status with queue position {queue_position}")
     
     # Get queue info for wait time estimation
     queue_info = get_queue_status()
@@ -383,14 +414,16 @@ def enqueue_task(input_data):
     task_queue.put((task_id, input_data, request_time))
     debug_log(f"Added task {task_id} to task_queue")
     
-    # Double-check queue status after adding
-    new_queue_size = task_queue.qsize()
-    debug_log(f"New queue size after adding: {new_queue_size}")
-    
-    # Count queued tasks in task_status
+    # Count queued tasks in task_status after adding
     with lock:
         queued_count = sum(1 for t in task_status.values() if t['status'] == 'queued')
-        debug_log(f"Total tasks with 'queued' status: {queued_count}")
+        processing_tasks = sum(1 for t in task_status.values() if t['status'] == 'processing')
+        debug_log(f"Queue status after adding: {task_queue.qsize()} in queue, {queued_count} with 'queued' status, {processing_tasks} processing")
+        
+        # Display all task IDs currently in queue
+        task_ids = [(k, v['status']) for k, v in task_status.items() if v['status'] in ('queued', 'processing')]
+        if task_ids:
+            debug_log(f"Current tasks: {task_ids}")
     
     return {
         'task_id': task_id,
@@ -416,19 +449,23 @@ def check_status(task_id):
 def get_queue_status():
     """Get queue status"""
     with lock:
-        queued_tasks = [t for t in task_status.values() if t['status'] == 'queued']
-        processing_tasks = [t for t in task_status.values() if t['status'] == 'processing']
+        queued_tasks = [v for k, v in task_status.items() if v['status'] == 'queued']
+        processing_tasks = [v for k, v in task_status.items() if v['status'] == 'processing']
         
         queue_size = task_queue.qsize()
         active_tasks = processing_count
         waiting_tasks = len(queued_tasks)
         
-        debug_log(f"Queue status: size={queue_size}, active={active_tasks}, waiting={waiting_tasks}")
+        debug_log(f"Queue status check: size={queue_size}, active={active_tasks}, waiting={waiting_tasks}")
+        if waiting_tasks != queue_size and abs(waiting_tasks - queue_size) > 1:
+            debug_log(f"WARNING: Queue size mismatch - task_queue has {queue_size} items but task_status has {waiting_tasks} queued items")
+            
         debug_log(f"Queue status details: {len(queued_tasks)} queued tasks found in task_status")
         if queued_tasks:
-            task_ids = [t.get('task_id', 'unknown') for t in queued_tasks if 'task_id' in t]
+            task_ids = [k for k, v in task_status.items() if v['status'] == 'queued']
             debug_log(f"Queued task IDs: {task_ids}")
         
+        # Calculate remaining processing time for active tasks
         remaining_processing_time = 0
         for task in processing_tasks:
             if 'start_time' in task and 'estimated_time' in task:
@@ -472,7 +509,7 @@ def get_queue_status():
             
         return {
             'queue_size': queue_size,
-            'active_tasks': active_tasks,
+            'active_tasks': active_tasks, 
             'waiting_tasks': waiting_tasks,
             'worker_threads': worker_threads,
             'estimated_wait': estimated_wait,
@@ -496,16 +533,26 @@ def ui_get_queue_info():
     """Get queue info for UI"""
     queue_info = get_queue_status()
     
-    # List queued tasks with details
+    # List queued tasks with details - make sure to use task_id as key
     queued_tasks_html = ""
     with lock:
-        queued_tasks = [t for t in task_status.values() if t['status'] == 'queued']
+        queued_tasks = []
+        for task_id, task in task_status.items():
+            if task['status'] == 'queued':
+                task_with_id = task.copy()
+                task_with_id['task_id'] = task_id
+                queued_tasks.append(task_with_id)
+        
         if queued_tasks:
+            # Sort by queue position
+            queued_tasks.sort(key=lambda x: x.get('queue_position', 999999))
             queued_tasks_html = "<div class='queued-tasks'><h4>Tasks in Queue:</h4><ul>"
-            for idx, task in enumerate(sorted(queued_tasks, key=lambda x: x.get('queued_time', 0))):
-                task_id = next((k for k, v in task_status.items() if v == task), "unknown")
+            for idx, task in enumerate(queued_tasks):
+                task_id = task['task_id']
                 queued_time = datetime.fromtimestamp(task.get('queued_time', 0)).strftime('%H:%M:%S')
-                queued_tasks_html += f"<li>Task {task_id[:8]}... - Queued at {queued_time} - Position {idx+1}</li>"
+                source = "async" if task.get('is_async', False) else "sync"
+                time_in_queue = time.time() - task.get('queued_time', time.time())
+                queued_tasks_html += f"<li>Task {task_id[:8]}... - Queued at {queued_time} ({time_in_queue:.1f}s ago) - Position {idx+1} ({source})</li>"
             queued_tasks_html += "</ul></div>"
     
     tasks_html = ""
@@ -538,12 +585,33 @@ def ui_get_queue_info():
     
     processing_details = ""
     if queue_info['active_tasks'] > 0:
+        # Display which tasks are being processed
+        processing_tasks_html = ""
+        with lock:
+            processing_task_ids = [k for k, v in task_status.items() if v['status'] == 'processing']
+            if processing_task_ids:
+                processing_tasks_html = "<ul>"
+                for task_id in processing_task_ids:
+                    task = task_status[task_id]
+                    start_time = datetime.fromtimestamp(task.get('start_time', 0)).strftime('%H:%M:%S')
+                    time_processing = time.time() - task.get('start_time', time.time())
+                    processing_tasks_html += f"<li>Task {task_id[:8]}... - Started at {start_time} ({time_processing:.1f}s ago)</li>"
+                processing_tasks_html += "</ul>"
+        
         processing_details = f"""
         <div class="alert alert-warning">
             <p><strong>Currently {queue_info['active_tasks']} tasks being processed</strong></p>
+            {processing_tasks_html}
         </div>
         """
     
+    # Add debug info
+    debug_details = f"""
+    <div class="debug-info">
+        <p><small>Queue: {queue_info['queue_size']} in queue, {queue_info['waiting_tasks']} waiting, {queue_info['active_tasks']} processing</small></p>
+    </div>
+    """
+    
     return f"""
     <div class="dashboard">
         <div class="queue-info-card main-card">
@@ -567,6 +635,7 @@ def ui_get_queue_info():
                 <p><b>Estimated Wait Time:</b> {format_time(queue_info['estimated_wait'])}</p>
                 {queue_details}
                 {processing_details}
+                {debug_details}
                 <p class="last-update"><small>Last update: {datetime.now().strftime('%H:%M:%S')}</small></p>
             </div>
         </div>
@@ -788,6 +857,26 @@ button.primary:hover {
 }
 """
 
+def async_enqueue(input_data):
+    """Async version of enqueue_task - specifically for async API calls"""
+    # Add metadata to identify async requests
+    if isinstance(input_data, list) and len(input_data) > 0 and isinstance(input_data[0], dict):
+        if 'metadata' not in input_data[0]:
+            input_data[0]['metadata'] = {}
+        input_data[0]['metadata']['source'] = 'async_api'
+    
+    # Just call enqueue_task but set thread name to identify as async
+    current_thread = threading.current_thread()
+    original_name = current_thread.name
+    current_thread.name = f"async_submission_{original_name}"
+    
+    result = enqueue_task(input_data)
+    
+    # Reset thread name
+    current_thread.name = original_name
+    
+    return result
+
 # Initialize and launch worker threads
 launch_workers()
 
@@ -830,7 +919,7 @@ with gr.Blocks(css=custom_css) as demo:
     evaluate_endpoint = demo.load(fn=synchronous_evaluate, inputs=api_input, outputs=api_output, api_name="evaluate", concurrency_limit=1)
     
     # Add async evaluation endpoint
-    enqueue_endpoint = demo.load(fn=enqueue_task, inputs=async_api_input, outputs=async_api_output, api_name="enqueue", concurrency_limit=1)
+    enqueue_endpoint = demo.load(fn=async_enqueue, inputs=async_api_input, outputs=async_api_output, api_name="enqueue", concurrency_limit=1)
     
     # Add status check endpoint
     status_endpoint = demo.load(fn=check_status, inputs=status_check_input, outputs=status_check_output, api_name="status", concurrency_limit=1)
@@ -838,7 +927,8 @@ with gr.Blocks(css=custom_css) as demo:
 if __name__ == "__main__":
     debug_log("Starting application")
     try:
-        demo.launch()
+        # Set max_threads for overall concurrency
+        demo.launch(max_threads=100)
     finally:
         # Stop worker threads
         running = False