update52

app.py

@@ -31,74 +31,21 @@ task_status = {}
 task_history = []
 # Lock for shared resources
 lock = threading.Lock()
-# Number of worker threads - set to 1 to process one task at a time
-worker_threads = 1
+# Number of worker threads
+worker_threads = 1  # Process only one task at a time
 # Flag for running background threads
 running = True
 # Mapping from task type to processing time
 task_type_times = {}
-# Currently processing tasks counter
-processing_count = 0
-# Available CPU cores for task processing
-available_cores = multiprocessing.cpu_count()
-# Task ID counter for debugging
-task_counter = 0
-
-# Enable logging
-DEBUG_MODE = True
-
-def debug_log(message):
-    """Log debug messages if debug mode is enabled"""
-    if DEBUG_MODE:
-        print(f"[DEBUG] {datetime.now().strftime('%H:%M:%S')} - {message}")
 
 def queue_processor():
     """Process tasks in the queue"""
-    global processing_count
-    
     while running:
         try:
-            # Only process if we're not already processing a task
-            with lock:
-                if processing_count >= worker_threads:
-                    # Already processing a task, wait and try again
-                    time.sleep(0.5)
-                    continue
-            
-            # Check queue size before attempting to get a task
-            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
-            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
+            task_id, input_data, request_time = task_queue.get(timeout=0.1)
             with lock:
-                processing_count += 1
-                debug_log(f"Incremented processing count to {processing_count}")
-                
-                # Update task status
-                if task_id in task_status:
-                    task_status[task_id]['status'] = 'processing'
-                    task_status[task_id]['start_time'] = time.time()
-                    debug_log(f"Updated existing task {task_id} to processing state")
-                else:
-                    # Create task status entry if it doesn't exist
-                    task_status[task_id] = {
-                        'status': 'processing',
-                        'queued_time': request_time,
-                        'start_time': time.time()
-                    }
-                    debug_log(f"Created new task status entry for {task_id}")
+                task_status[task_id]['status'] = 'processing'
+                task_status[task_id]['start_time'] = time.time()
             
             if isinstance(input_data, list) and len(input_data) > 0:
                 sample_task = input_data[0]
@@ -113,24 +60,16 @@ def queue_processor():
                         'complexity': task_complexity
                     }
             
-            debug_log(f"Starting evaluation for task {task_id}")
             result = evaluate(input_data)
-            debug_log(f"Finished evaluation for task {task_id}")
             
             end_time = time.time()
             process_time = end_time - task_status[task_id]['start_time']
             
             with lock:
-                # Decrease processing count now that we're done
-                processing_count -= 1
-                debug_log(f"Decremented processing count to {processing_count}")
-                
-                # Update task status
                 task_status[task_id]['status'] = 'completed'
                 task_status[task_id]['result'] = result
                 task_status[task_id]['end_time'] = end_time
                 task_status[task_id]['process_time'] = process_time
-                debug_log(f"Updated task {task_id} to completed state")
                 
                 if 'estimated_factors' in task_status[task_id]:
                     factors = task_status[task_id]['estimated_factors']
@@ -154,32 +93,15 @@ def queue_processor():
                     task_history.pop(0)
                     
             task_queue.task_done()
-            debug_log(f"Task {task_id} completed and marked as done")
             
         except queue.Empty:
-            # Use a small timeout to avoid CPU spinning
-            time.sleep(0.1)
+            continue
         except Exception as e:
-            debug_log(f"Error in queue processor: {str(e)}")
             if 'task_id' in locals():
-                debug_log(f"Error occurred while processing task {task_id}")
                 with lock:
-                    # Decrease processing count on error
-                    processing_count -= 1
-                    debug_log(f"Decremented processing count to {processing_count} due to error")
-                    
-                    if task_id in task_status:
-                        task_status[task_id]['status'] = 'error'
-                        task_status[task_id]['error'] = str(e)
-                        task_status[task_id]['end_time'] = time.time()
-                        debug_log(f"Updated task {task_id} to error state")
-                    else:
-                        task_status[task_id] = {
-                            'status': 'error',
-                            'error': str(e),
-                            'end_time': time.time()
-                        }
-                        debug_log(f"Created new error entry for task {task_id}")
+                    task_status[task_id]['status'] = 'error'
+                    task_status[task_id]['error'] = str(e)
+                    task_status[task_id]['end_time'] = time.time()
             task_queue.task_done()
 
 def _estimate_task_complexity(tasks):
@@ -223,8 +145,9 @@ def evaluate(input_data):
             
         results = []
         
-        # Use all available cores for this single task but with a reasonable cap
-        max_workers = max(1, min(available_cores // 2, 8))
+        # Use a moderate number of workers for all language tests to ensure stability
+        # This prevents resource contention regardless of language
+        max_workers = max(1, min(multiprocessing.cpu_count() // 2, 4))
         
         with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
             future_to_item = {executor.submit(evaluate_single_case, item): item for item in input_data}
@@ -291,47 +214,53 @@ def evaluate_code(code, language):
 
 def synchronous_evaluate(input_data):
     """Synchronously evaluate code, compatible with original interface"""
-    debug_log(f"Received synchronous evaluation request")
+    if isinstance(input_data, list) and len(input_data) > 0:
+        sample_task = input_data[0]
+        language = sample_task.get('language', 'unknown') if isinstance(sample_task, dict) else 'unknown'
+        task_size = len(input_data)
+        task_complexity = _estimate_task_complexity(input_data)
+    else:
+        language = 'unknown'
+        task_size = 1
+        task_complexity = 'medium'
     
-    # 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'
+    estimated_time_per_task = _get_estimated_time_for_task(language, task_complexity)
+    estimated_total_time = estimated_time_per_task * task_size
     
-    # 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")
+    queue_info = get_queue_status()
+    waiting_tasks = queue_info['waiting_tasks']
     
-    # 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
+    task_id = str(uuid.uuid4())
+    request_time = time.time()
+    
+    with lock:
+        task_status[task_id] = {
+            'status': 'queued',
+            'queued_time': request_time,
+            'queue_position': task_queue.qsize() + 1,
+            'synchronous': True,
+            'estimated_factors': {
+                'language': language,
+                'size': task_size,
+                'complexity': task_complexity
+            },
+            'estimated_time': estimated_total_time
+        }
+    
+    task_queue.put((task_id, input_data, request_time))
     
-    # Wait for task to complete
     while True:
         with lock:
             if task_id in task_status:
                 status = task_status[task_id]['status']
                 if status == 'completed':
-                    debug_log(f"Task {task_id} completed, returning result")
                     result = task_status[task_id]['result']
-                    # Keep the result in status for a short time to ensure it shows in history
-                    if 'end_time' not in task_status[task_id]:
-                        task_status[task_id]['end_time'] = time.time()
-                    elif time.time() - task_status[task_id]['end_time'] > 5:
-                        task_status.pop(task_id, None)
+                    task_status.pop(task_id, None)
                     return result
                 elif status == 'error':
-                    debug_log(f"Task {task_id} failed with error")
                     error = task_status[task_id].get('error', 'Unknown error')
-                    # Keep the error in status for a short time to ensure it shows in history
-                    if 'end_time' not in task_status[task_id]:
-                        task_status[task_id]['end_time'] = time.time()
-                    elif time.time() - task_status[task_id]['end_time'] > 5:
-                        task_status.pop(task_id, None)
+                    task_status.pop(task_id, None)
                     return {"status": "Exception", "error": error}
-                else:
-                    debug_log(f"Task {task_id} still in status: {status}")
         
         time.sleep(0.1)
 
@@ -351,8 +280,6 @@ def _get_estimated_time_for_task(language, complexity):
 
 def enqueue_task(input_data):
     """Add task to queue"""
-    global task_counter
-    
     if isinstance(input_data, list) and len(input_data) > 0:
         sample_task = input_data[0]
         language = sample_task.get('language', 'unknown') if isinstance(sample_task, dict) else 'unknown'
@@ -366,36 +293,17 @@ 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
     
-    # 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]}"
+    task_id = str(uuid.uuid4())
     request_time = time.time()
     
-    debug_log(f"Creating new task: {task_id}")
+    # Get current queue size before adding new task
+    current_queue_size = task_queue.qsize()
     
-    # 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
-    
-    # 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': queue_position,
-            'is_async': is_async,
+            'queue_position': current_queue_size + 1,
             'estimated_factors': {
                 'language': language,
                 'size': task_size,
@@ -403,27 +311,11 @@ def enqueue_task(input_data):
             },
             'estimated_time': estimated_total_time
         }
-        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()
     est_wait = queue_info['estimated_wait']
-    debug_log(f"Estimated wait time for task {task_id}: {est_wait} seconds")
     
-    # Add to the task queue - this must be done AFTER adding to task_status
     task_queue.put((task_id, input_data, request_time))
-    debug_log(f"Added task {task_id} to task_queue")
-    
-    # Count queued tasks in task_status after adding
-    with lock:
-        queued_count = sum(1 for t in task_status.values() if t['status'] == 'queued')
-        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,
@@ -449,23 +341,13 @@ def check_status(task_id):
 def get_queue_status():
     """Get queue status"""
     with lock:
-        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)
+        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']
         
-        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 = [k for k, v in task_status.items() if v['status'] == 'queued']
-            debug_log(f"Queued task IDs: {task_ids}")
+        queue_size = len(queued_tasks)  # Use actual count of queued tasks
+        active_tasks = len(processing_tasks)
+        waiting_tasks = queue_size
         
-        # 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:
@@ -485,9 +367,6 @@ def get_queue_status():
             else:
                 queued_processing_time += 5
         
-        if worker_threads > 0 and queued_processing_time > 0:
-            queued_processing_time = queued_processing_time / worker_threads
-        
         estimated_wait = remaining_processing_time + queued_processing_time
         
         if task_history:
@@ -509,7 +388,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,
@@ -533,28 +412,6 @@ def ui_get_queue_info():
     """Get queue info for UI"""
     queue_info = get_queue_status()
     
-    # List queued tasks with details - make sure to use task_id as key
-    queued_tasks_html = ""
-    with lock:
-        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(queued_tasks):
-                task_id = task['task_id']
-                queued_time = datetime.fromtimestamp(task.get('queued_time', 0)).strftime('%H:%M:%S')
-                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 = ""
     for task in reversed(queue_info['recent_tasks']):
         tasks_html += f"""
@@ -572,46 +429,6 @@ def ui_get_queue_info():
         </tr>
         """
     
-    # Add more detailed queue information
-    queue_details = ""
-    if queue_info['waiting_tasks'] > 0:
-        queue_details = f"""
-        <div class="alert alert-info">
-            <p><strong>Currently {queue_info['waiting_tasks']} tasks in queue</strong></p>
-            <p>Estimated wait time: {format_time(queue_info['estimated_wait'])}</p>
-            {queued_tasks_html}
-        </div>
-        """
-    
-    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">
@@ -632,10 +449,7 @@ def ui_get_queue_info():
             </div>
             
             <div class="wait-time">
-                <p><b>Estimated Wait Time:</b> {format_time(queue_info['estimated_wait'])}</p>
-                {queue_details}
-                {processing_details}
-                {debug_details}
+                <p><b>Current Estimated Wait Time:</b> {format_time(queue_info['estimated_wait'])}</p>
                 <p class="last-update"><small>Last update: {datetime.now().strftime('%H:%M:%S')}</small></p>
             </div>
         </div>
@@ -820,63 +634,8 @@ button.primary {
 button.primary:hover {
     background-color: #3367d6;
 }
-
-.alert {
-    padding: 12px;
-    margin: 10px 0;
-    border-radius: 6px;
-}
-
-.alert-info {
-    background-color: #d1ecf1;
-    color: #0c5460;
-    border: 1px solid #bee5eb;
-}
-
-.alert-warning {
-    background-color: #fff3cd;
-    color: #856404;
-    border: 1px solid #ffeeba;
-}
-
-.queued-tasks {
-    text-align: left;
-    margin: 10px 0;
-    padding: 8px;
-    background: rgba(255, 255, 255, 0.5);
-    border-radius: 4px;
-}
-
-.queued-tasks ul {
-    margin: 5px 0;
-    padding-left: 20px;
-}
-
-.queued-tasks li {
-    margin-bottom: 3px;
-}
 """
 
-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()
 
@@ -895,41 +654,24 @@ with gr.Blocks(css=custom_css) as demo:
     with gr.Row(visible=False):
         api_input = gr.JSON()
         api_output = gr.JSON()
-        
-        async_api_input = gr.JSON()
-        async_api_output = gr.JSON()
-        
-        status_check_input = gr.Textbox()
-        status_check_output = gr.JSON()
     
     # Define update function
     def update_queue_info():
         return ui_get_queue_info()
     
-    # Update queue info more frequently
-    demo.load(update_queue_info, None, queue_info_html, every=0.5)
+    # Update queue info periodically
+    demo.load(update_queue_info, None, queue_info_html, every=3)
     
     # Refresh button event
     refresh_queue_btn.click(update_queue_info, None, queue_info_html)
     
-    # Force sync when handling API requests to prevent gradio's queue from interfering
-    # Use the correct queue configuration method for current Gradio version
-    
     # Add evaluation endpoint compatible with original interface
-    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=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)
+    demo.queue()
+    evaluate_endpoint = demo.load(fn=synchronous_evaluate, inputs=api_input, outputs=api_output, api_name="evaluate")
 
 if __name__ == "__main__":
-    debug_log("Starting application")
     try:
-        # Set max_threads for overall concurrency
-        demo.launch(max_threads=100)
+        demo.launch()
     finally:
         # Stop worker threads
-        running = False
-        debug_log("Shutting down application")
+        running = False