update50

app.py

@@ -32,7 +32,7 @@ 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  # 只处理一个任务，避免资源竞争
+worker_threads = 1
 # Flag for running background threads
 running = True
 # Mapping from task type to processing time
@@ -41,6 +41,16 @@ task_type_times = {}
 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"""
@@ -54,18 +64,26 @@ def queue_processor():
                     # 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")
+                
             # 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}")
             
             # Increment processing count to track active tasks
             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] = {
@@ -73,6 +91,7 @@ def queue_processor():
                         'queued_time': request_time,
                         'start_time': time.time()
                     }
+                    debug_log(f"Created new task status entry for {task_id}")
             
             if isinstance(input_data, list) and len(input_data) > 0:
                 sample_task = input_data[0]
@@ -87,7 +106,9 @@ 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']
@@ -95,12 +116,14 @@ def queue_processor():
             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']
@@ -124,25 +147,31 @@ 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:
             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_queue.task_done()
 
 def _estimate_task_complexity(tasks):
@@ -254,9 +283,12 @@ def evaluate_code(code, language):
 
 def synchronous_evaluate(input_data):
     """Synchronously evaluate code, compatible with original interface"""
+    debug_log(f"Received synchronous evaluation request")
+    
     # 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")
     
     # Wait for task to complete
     while True:
@@ -264,6 +296,7 @@ def synchronous_evaluate(input_data):
             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]:
@@ -272,6 +305,7 @@ def synchronous_evaluate(input_data):
                         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]:
@@ -279,6 +313,8 @@ def synchronous_evaluate(input_data):
                     elif time.time() - task_status[task_id]['end_time'] > 5:
                         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)
 
@@ -298,6 +334,8 @@ 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'
@@ -311,15 +349,22 @@ 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_id = str(uuid.uuid4())
+    task_counter += 1
+    task_id = f"task_{task_counter}_{str(uuid.uuid4())}"
     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}")
+    
     # Add to task_status with 'queued' status first
     with lock:
         task_status[task_id] = {
             'status': 'queued',
             'queued_time': request_time,
-            'queue_position': task_queue.qsize() + 1,
+            'queue_position': current_queue_size + 1,
             'estimated_factors': {
                 'language': language,
                 'size': task_size,
@@ -327,13 +372,25 @@ 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}")
     
     # 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")
+    
+    # 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
+    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}")
     
     return {
         'task_id': task_id,
@@ -366,6 +423,12 @@ def get_queue_status():
         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 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]
+            debug_log(f"Queued task IDs: {task_ids}")
+        
         remaining_processing_time = 0
         for task in processing_tasks:
             if 'start_time' in task and 'estimated_time' in task:
@@ -433,6 +496,18 @@ def ui_get_queue_info():
     """Get queue info for UI"""
     queue_info = get_queue_status()
     
+    # List queued tasks with details
+    queued_tasks_html = ""
+    with lock:
+        queued_tasks = [t for t in task_status.values() if t['status'] == 'queued']
+        if queued_tasks:
+            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")
+                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>"
+            queued_tasks_html += "</ul></div>"
+    
     tasks_html = ""
     for task in reversed(queue_info['recent_tasks']):
         tasks_html += f"""
@@ -450,50 +525,60 @@ def ui_get_queue_info():
         </tr>
         """
     
-    # 增加更多详细信息到UI
+    # Add more detailed queue information
     queue_details = ""
     if queue_info['waiting_tasks'] > 0:
-        queue_details = f"<p>当前有 {queue_info['waiting_tasks']} 个任务在等待队列中，预计等待时间：{format_time(queue_info['estimated_wait'])}</p>"
+        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:
-        processing_details = f"<p>当前有 {queue_info['active_tasks']} 个任务正在处理中</p>"
+        processing_details = f"""
+        <div class="alert alert-warning">
+            <p><strong>Currently {queue_info['active_tasks']} tasks being processed</strong></p>
+        </div>
+        """
     
     return f"""
     <div class="dashboard">
         <div class="queue-info-card main-card">
-            <h3 class="card-title">队列状态监控</h3>
+            <h3 class="card-title">Queue Status Monitor</h3>
             <div class="queue-stats">
                 <div class="stat-item">
                     <div class="stat-value">{queue_info['waiting_tasks']}</div>
-                    <div class="stat-label">等待中</div>
+                    <div class="stat-label">Waiting</div>
                 </div>
                 <div class="stat-item">
                     <div class="stat-value">{queue_info['active_tasks']}</div>
-                    <div class="stat-label">处理中</div>
+                    <div class="stat-label">Processing</div>
                 </div>
                 <div class="stat-item">
                     <div class="stat-value">{queue_info['worker_threads']}</div>
-                    <div class="stat-label">工作线程</div>
+                    <div class="stat-label">Worker Threads</div>
                 </div>
             </div>
             
             <div class="wait-time">
-                <p><b>预计等待时间:</b> {format_time(queue_info['estimated_wait'])}</p>
+                <p><b>Estimated Wait Time:</b> {format_time(queue_info['estimated_wait'])}</p>
                 {queue_details}
                 {processing_details}
-                <p class="last-update"><small>最后更新时间: {datetime.now().strftime('%H:%M:%S')}</small></p>
+                <p class="last-update"><small>Last update: {datetime.now().strftime('%H:%M:%S')}</small></p>
             </div>
         </div>
             
         <div class="queue-info-card history-card">
-            <h3 class="card-title">最近处理的任务</h3>
+            <h3 class="card-title">Recently Processed Tasks</h3>
             <table class="recent-tasks">
                 <thead>
                     <tr>
-                        <th>任务ID</th>
-                        <th>请求时间</th>
-                        <th>处理时间</th>
+                        <th>Task ID</th>
+                        <th>Request Time</th>
+                        <th>Processing Time</th>
                     </tr>
                 </thead>
                 <tbody>
@@ -666,6 +751,41 @@ 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;
+}
 """
 
 # Initialize and launch worker threads
@@ -673,14 +793,14 @@ launch_workers()
 
 # Create Gradio interface
 with gr.Blocks(css=custom_css) as demo:
-    gr.Markdown("# 代码评估服务")
-    gr.Markdown("支持多种编程语言的代码评估服务，使用队列机制处理请求")
+    gr.Markdown("# Code Evaluation Service")
+    gr.Markdown("Code evaluation service supporting multiple programming languages, using queue mechanism to process requests")
     
     with gr.Row():
         with gr.Column(scale=3):
             # Queue status info card
             queue_info_html = gr.HTML()
-            refresh_queue_btn = gr.Button("刷新队列状态", variant="primary")
+            refresh_queue_btn = gr.Button("Refresh Queue Status", variant="primary")
     
     # Hidden API interface components
     with gr.Row(visible=False):
@@ -697,14 +817,16 @@ with gr.Blocks(css=custom_css) as demo:
     def update_queue_info():
         return ui_get_queue_info()
     
-    # Update queue info periodically
-    demo.load(update_queue_info, None, queue_info_html, every=1)
+    # Update queue info more frequently
+    demo.load(update_queue_info, None, queue_info_html, every=0.5)
     
     # 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
+    demo.queue(concurrency_count=1, max_size=100)
+    
     # Add evaluation endpoint compatible with original interface
-    demo.queue()
     evaluate_endpoint = demo.load(fn=synchronous_evaluate, inputs=api_input, outputs=api_output, api_name="evaluate")
     
     # Add async evaluation endpoint
@@ -714,8 +836,10 @@ with gr.Blocks(css=custom_css) as demo:
     status_endpoint = demo.load(fn=check_status, inputs=status_check_input, outputs=status_check_output, api_name="status")
 
 if __name__ == "__main__":
+    debug_log("Starting application")
     try:
         demo.launch()
     finally:
         # Stop worker threads
-        running = False
+        running = False
+        debug_log("Shutting down application")