update37

app.py

@@ -13,6 +13,8 @@ import uuid
 import numpy as np
 from datetime import datetime
 from tqdm.auto import tqdm
+import redis
+import pickle
 from src.containerized_eval import eval_string_script
 
 # Add current directory and src directory to module search path
@@ -23,29 +25,80 @@ if current_dir not in sys.path:
 if src_dir not in sys.path:
     sys.path.append(src_dir)
 
-# Create message queue
-task_queue = queue.Queue()
-# Dictionary to store task status
-task_status = {}
-# List to store task history, max 200 tasks
-task_history = []
+# Initialize Redis connection (will use environment variables in Hugging Face Space)
+REDIS_URL = os.environ.get('REDIS_URL', 'redis://localhost:6379/0')
+redis_client = redis.from_url(REDIS_URL)
+
+# Keys for Redis
+QUEUE_KEY = 'eval_task_queue'
+STATUS_KEY = 'eval_task_status'
+HISTORY_KEY = 'eval_task_history'
+TASK_TIMES_KEY = 'eval_task_times'
+
+# Local queue for worker threads
+local_task_queue = queue.Queue()
 # Lock for shared resources
 lock = threading.Lock()
 # Number of worker threads
 worker_threads = max(1, multiprocessing.cpu_count() // 2)  # Using half the available cores for better stability
 # Flag for running background threads
 running = True
-# Mapping from task type to processing time
-task_type_times = {}
+
+def redis_queue_monitor():
+    """Monitor Redis queue and add tasks to local queue"""
+    last_check = 0
+    while running:
+        try:
+            # Check Redis queue every second
+            if time.time() - last_check >= 1:
+                last_check = time.time()
+                # Get all tasks in the queue
+                task_list = redis_client.lrange(QUEUE_KEY, 0, -1)
+                for task_data in task_list:
+                    task = pickle.loads(task_data)
+                    task_id = task['id']
+                    
+                    # Check if task is already in processing
+                    status_data = redis_client.hget(STATUS_KEY, task_id)
+                    if status_data:
+                        status = pickle.loads(status_data)
+                        if status['status'] == 'queued':
+                            # Add to local queue if not already processing
+                            local_task_queue.put((task_id, task['input_data'], task['request_time']))
+                            # Update status to processing
+                            with lock:
+                                status['status'] = 'processing'
+                                status['start_time'] = time.time()
+                                redis_client.hset(STATUS_KEY, task_id, pickle.dumps(status))
+                            # Remove from Redis queue
+                            redis_client.lrem(QUEUE_KEY, 1, task_data)
+            
+            time.sleep(0.1)
+        except Exception as e:
+            print(f"Redis queue monitor error: {e}")
+            time.sleep(1)
 
 def queue_processor():
-    """Process tasks in the queue"""
+    """Process tasks in the local queue"""
     while running:
         try:
-            task_id, input_data, request_time = task_queue.get(timeout=0.1)
-            with lock:
-                task_status[task_id]['status'] = 'processing'
-                task_status[task_id]['start_time'] = time.time()
+            task_id, input_data, request_time = local_task_queue.get(timeout=0.1)
+            
+            # Get current status
+            status_data = redis_client.hget(STATUS_KEY, task_id)
+            if status_data:
+                task_status = pickle.loads(status_data)
+            else:
+                task_status = {
+                    'status': 'processing',
+                    'queued_time': request_time,
+                    'start_time': time.time()
+                }
+            
+            # Update status
+            task_status['status'] = 'processing'
+            task_status['start_time'] = time.time()
+            redis_client.hset(STATUS_KEY, task_id, pickle.dumps(task_status))
             
             if isinstance(input_data, list) and len(input_data) > 0:
                 sample_task = input_data[0]
@@ -53,56 +106,83 @@ def queue_processor():
                 task_size = len(input_data)
                 task_complexity = _estimate_task_complexity(input_data)
                 
-                with lock:
-                    task_status[task_id]['estimated_factors'] = {
-                        'language': language,
-                        'size': task_size,
-                        'complexity': task_complexity
-                    }
+                estimated_factors = {
+                    'language': language,
+                    'size': task_size,
+                    'complexity': task_complexity
+                }
+                task_status['estimated_factors'] = estimated_factors
+                redis_client.hset(STATUS_KEY, task_id, pickle.dumps(task_status))
             
             result = evaluate(input_data)
             
             end_time = time.time()
-            process_time = end_time - task_status[task_id]['start_time']
+            process_time = end_time - task_status['start_time']
             
-            with lock:
-                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
+            # Update status
+            task_status['status'] = 'completed'
+            task_status['result'] = result
+            task_status['end_time'] = end_time
+            task_status['process_time'] = process_time
+            redis_client.hset(STATUS_KEY, task_id, pickle.dumps(task_status))
+            
+            # Update task type times
+            if 'estimated_factors' in task_status:
+                factors = task_status['estimated_factors']
+                key = f"{factors['language']}_{factors['complexity']}"
                 
-                if 'estimated_factors' in task_status[task_id]:
-                    factors = task_status[task_id]['estimated_factors']
-                    key = f"{factors['language']}_{factors['complexity']}"
-                    
-                    if key not in task_type_times:
-                        task_type_times[key] = []
-                    
-                    task_type_times[key].append(process_time / factors['size'])
-                    if len(task_type_times[key]) > 10:
-                        task_type_times[key] = task_type_times[key][-10:]
+                # Update task times in Redis
+                times_data = redis_client.hget(TASK_TIMES_KEY, key)
+                if times_data:
+                    times = pickle.loads(times_data)
+                else:
+                    times = []
                 
-                task_history.append({
-                    'task_id': task_id,
-                    'request_time': request_time,
-                    'process_time': process_time,
-                    'status': 'completed',
-                    'factors': task_status[task_id].get('estimated_factors', {})
-                })
-                while len(task_history) > 200:
-                    task_history.pop(0)
-                    
-            task_queue.task_done()
+                times.append(process_time / factors['size'])
+                if len(times) > 10:
+                    times = times[-10:]
+                
+                redis_client.hset(TASK_TIMES_KEY, key, pickle.dumps(times))
+            
+            # Add to history
+            history_item = {
+                'task_id': task_id,
+                'request_time': request_time,
+                'process_time': process_time,
+                'status': 'completed',
+                'factors': task_status.get('estimated_factors', {})
+            }
+            
+            # Get current history
+            history_data = redis_client.get(HISTORY_KEY)
+            if history_data:
+                history = pickle.loads(history_data)
+            else:
+                history = []
+            
+            history.append(history_item)
+            while len(history) > 200:
+                history.pop(0)
+            
+            redis_client.set(HISTORY_KEY, pickle.dumps(history))
+            
+            local_task_queue.task_done()
             
         except queue.Empty:
             continue
         except Exception as e:
             if 'task_id' in locals():
-                with lock:
-                    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()
+                status_data = redis_client.hget(STATUS_KEY, task_id)
+                if status_data:
+                    task_status = pickle.loads(status_data)
+                else:
+                    task_status = {}
+                
+                task_status['status'] = 'error'
+                task_status['error'] = str(e)
+                task_status['end_time'] = time.time()
+                redis_client.hset(STATUS_KEY, task_id, pickle.dumps(task_status))
+            local_task_queue.task_done()
 
 def _estimate_task_complexity(tasks):
     """Estimate task complexity
@@ -233,34 +313,41 @@ def synchronous_evaluate(input_data):
     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))
+    task_status = {
+        'status': 'queued',
+        'queued_time': request_time,
+        'queue_position': queue_info['queue_size'] + 1,
+        'synchronous': True,
+        'estimated_factors': {
+            'language': language,
+            'size': task_size,
+            'complexity': task_complexity
+        },
+        'estimated_time': estimated_total_time
+    }
+    
+    redis_client.hset(STATUS_KEY, task_id, pickle.dumps(task_status))
+    
+    # Add to queue
+    task = {
+        'id': task_id,
+        'input_data': input_data,
+        'request_time': request_time
+    }
+    redis_client.rpush(QUEUE_KEY, pickle.dumps(task))
     
     while True:
-        with lock:
-            if task_id in task_status:
-                status = task_status[task_id]['status']
-                if status == 'completed':
-                    result = task_status[task_id]['result']
-                    task_status.pop(task_id, None)
-                    return result
-                elif status == 'error':
-                    error = task_status[task_id].get('error', 'Unknown error')
-                    task_status.pop(task_id, None)
-                    return {"status": "Exception", "error": error}
+        status_data = redis_client.hget(STATUS_KEY, task_id)
+        if status_data:
+            status_info = pickle.loads(status_data)
+            if status_info['status'] == 'completed':
+                result = status_info.get('result', {"status": "Exception", "error": "No result found"})
+                redis_client.hdel(STATUS_KEY, task_id)
+                return result
+            elif status_info['status'] == 'error':
+                error = status_info.get('error', 'Unknown error')
+                redis_client.hdel(STATUS_KEY, task_id)
+                return {"status": "Exception", "error": error}
         
         time.sleep(0.1)
 
@@ -268,8 +355,11 @@ def _get_estimated_time_for_task(language, complexity):
     """Get estimated processing time for a specific task type"""
     key = f"{language}_{complexity}"
     
-    if key in task_type_times and len(task_type_times[key]) > 0:
-        return np.median(task_type_times[key])
+    times_data = redis_client.hget(TASK_TIMES_KEY, key)
+    if times_data:
+        times = pickle.loads(times_data)
+        if times:
+            return np.median(times)
     
     if complexity == 'simple':
         return 1.0
@@ -296,104 +386,128 @@ def enqueue_task(input_data):
     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,
-            'estimated_factors': {
-                'language': language,
-                'size': task_size,
-                'complexity': task_complexity
-            },
-            'estimated_time': estimated_total_time
-        }
-    
     queue_info = get_queue_status()
-    est_wait = queue_info['estimated_wait']
     
-    task_queue.put((task_id, input_data, request_time))
+    task_status = {
+        'status': 'queued',
+        'queued_time': request_time,
+        'queue_position': queue_info['queue_size'] + 1,
+        'estimated_factors': {
+            'language': language,
+            'size': task_size,
+            'complexity': task_complexity
+        },
+        'estimated_time': estimated_total_time
+    }
+    
+    redis_client.hset(STATUS_KEY, task_id, pickle.dumps(task_status))
+    
+    # Add to queue
+    task = {
+        'id': task_id,
+        'input_data': input_data,
+        'request_time': request_time
+    }
+    redis_client.rpush(QUEUE_KEY, pickle.dumps(task))
+    
+    est_wait = queue_info['estimated_wait']
     
     return {
         'task_id': task_id,
         'status': 'queued',
-        'queue_position': task_status[task_id]['queue_position'],
+        'queue_position': task_status['queue_position'],
         'estimated_wait': est_wait,
         'estimated_processing': estimated_total_time
     }
 
 def check_status(task_id):
     """Check task status"""
-    with lock:
-        if task_id not in task_status:
-            return {'status': 'not_found'}
-        
-        status_info = task_status[task_id].copy()
+    status_data = redis_client.hget(STATUS_KEY, task_id)
+    if not status_data:
+        return {'status': 'not_found'}
+    
+    status_info = pickle.loads(status_data)
+    
+    if status_info['status'] in ['completed', 'error'] and time.time() - status_info.get('end_time', 0) > 3600:
+        redis_client.hdel(STATUS_KEY, task_id)
         
-        if status_info['status'] in ['completed', 'error'] and time.time() - status_info.get('end_time', 0) > 3600:
-            task_status.pop(task_id, None)
-            
-        return status_info
+    return status_info
 
 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']
-        
-        queue_size = task_queue.qsize()
-        active_tasks = len(processing_tasks)
-        waiting_tasks = len(queued_tasks)
-        
-        remaining_processing_time = 0
-        for task in processing_tasks:
-            if 'start_time' in task and 'estimated_time' in task:
-                elapsed = time.time() - task['start_time']
-                remaining = max(0, task['estimated_time'] - elapsed)
-                remaining_processing_time += remaining
-            else:
-                remaining_processing_time += 2
-        
-        if active_tasks > 0:
-            remaining_processing_time = remaining_processing_time / min(active_tasks, worker_threads)
-        
-        queued_processing_time = 0
-        for task in queued_tasks:
-            if 'estimated_time' in task:
-                queued_processing_time += task['estimated_time']
-            else:
-                queued_processing_time += 5
-        
-        if worker_threads > 0 and queued_processing_time > 0:
-            queued_processing_time = queued_processing_time / worker_threads
+    # Get all task statuses
+    all_statuses = redis_client.hgetall(STATUS_KEY)
+    
+    queued_tasks = []
+    processing_tasks = []
+    
+    for task_id, status_data in all_statuses.items():
+        status_info = pickle.loads(status_data)
+        if status_info['status'] == 'queued':
+            queued_tasks.append(status_info)
+        elif status_info['status'] == 'processing':
+            processing_tasks.append(status_info)
+    
+    queue_size = redis_client.llen(QUEUE_KEY)
+    active_tasks = len(processing_tasks)
+    waiting_tasks = len(queued_tasks)
+    
+    remaining_processing_time = 0
+    for task in processing_tasks:
+        if 'start_time' in task and 'estimated_time' in task:
+            elapsed = time.time() - task['start_time']
+            remaining = max(0, task['estimated_time'] - elapsed)
+            remaining_processing_time += remaining
+        else:
+            remaining_processing_time += 2
+    
+    if active_tasks > 0:
+        remaining_processing_time = remaining_processing_time / min(active_tasks, worker_threads)
+    
+    queued_processing_time = 0
+    for task in queued_tasks:
+        if 'estimated_time' in task:
+            queued_processing_time += task['estimated_time']
+        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
+    
+    # Get task history
+    history_data = redis_client.get(HISTORY_KEY)
+    if history_data:
+        task_history = pickle.loads(history_data)
+    else:
+        task_history = []
+    
+    if task_history:
+        prediction_ratios = []
+        for task in task_history:
+            if 'factors' in task and 'estimated_time' in task:
+                prediction_ratios.append(task['process_time'] / task['estimated_time'])
         
-        estimated_wait = remaining_processing_time + queued_processing_time
+        if prediction_ratios:
+            correction_factor = np.median(prediction_ratios)
+            correction_factor = max(0.5, min(2.0, correction_factor))
+            estimated_wait *= correction_factor
+    
+    estimated_wait = max(0.1, estimated_wait)
+    if waiting_tasks == 0 and active_tasks == 0:
+        estimated_wait = 0
         
-        if task_history:
-            prediction_ratios = []
-            for task in task_history:
-                if 'factors' in task and 'estimated_time' in task:
-                    prediction_ratios.append(task['process_time'] / task['estimated_time'])
-            
-            if prediction_ratios:
-                correction_factor = np.median(prediction_ratios)
-                correction_factor = max(0.5, min(2.0, correction_factor))
-                estimated_wait *= correction_factor
+    recent_tasks = task_history[-5:] if task_history else []
         
-        estimated_wait = max(0.1, estimated_wait)
-        if waiting_tasks == 0 and active_tasks == 0:
-            estimated_wait = 0
-            
-        recent_tasks = task_history[-5:] if task_history else []
-            
-        return {
-            'queue_size': queue_size,
-            'active_tasks': active_tasks,
-            'waiting_tasks': waiting_tasks,
-            'worker_threads': worker_threads,
-            'estimated_wait': estimated_wait,
-            'recent_tasks': recent_tasks
-        }
+    return {
+        'queue_size': queue_size,
+        'active_tasks': active_tasks,
+        'waiting_tasks': waiting_tasks,
+        'worker_threads': worker_threads,
+        'estimated_wait': estimated_wait,
+        'recent_tasks': recent_tasks
+    }
 
 def format_time(seconds):
     """Format time into readable format"""
@@ -477,6 +591,12 @@ def launch_workers():
     global running
     running = True
     
+    # Start Redis queue monitor
+    monitor = threading.Thread(target=redis_queue_monitor)
+    monitor.daemon = True
+    monitor.start()
+    
+    # Start worker threads
     for _ in range(worker_threads):
         worker = threading.Thread(target=queue_processor)
         worker.daemon = True

requirements.txt

@@ -1 +1,2 @@
-gradio==4.44.1
+gradio==4.44.1
+redis==5.0.1