manage

controller.py

@@ -5,11 +5,13 @@ import time
 import uuid
 import os
 import re
-import asyncio
-import multiprocessing
-import hashlib
-import json
-from typing import List, Optional, Dict, Any, TypeVar, Generic
+import asyncio  # Added for parallel batching
+import multiprocessing  # Added for worker calculation
+import hashlib # Added for Coalescing
+import json # Added for Coalescing
+import psutil # Added for Dynamic RAM calculation
+from typing import List, Optional, Dict, Any, TypeVar, Generic  # Added Generic/TypeVar
+from urllib.parse import urlparse, parse_qs, urlencode, urlunparse
 
 # --- FastAPI & Core ---
 from fastapi import FastAPI, HTTPException, Depends
@@ -17,20 +19,18 @@ from fastapi.encoders import jsonable_encoder
 from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
 from fastapi.middleware.cors import CORSMiddleware
 from starlette.concurrency import run_in_threadpool
-from anyio import to_thread
+from anyio import to_thread # Added for AnyIO 4.x concurrency tuning
 from dotenv import load_dotenv
 from pydantic import BaseModel
 
-# --- Database Drivers ---
+# --- Database Drivers & Pooling ---
 from bson import ObjectId
+import mysql.connector
+from mysql.connector import pooling # For MySQL Pooling
+import psycopg2
+from psycopg2 import pool as pg_pool # For Postgres Pooling
 from psycopg2.extras import RealDictCursor
-from urllib.parse import urlparse, parse_qs, urlencode, urlunparse
-
-# --- Database Pool ---
-from mysql.connector import pooling
-from psycopg2 import pool as pg_pool
-from threading import Lock
-
+from threading import Lock # To handle concurrency safely
 import uvicorn
 
 # --- Existing Services ---
@@ -53,12 +53,49 @@ logging.basicConfig(
 )
 logger = logging.getLogger("API_Controller")
 
+def get_dynamic_thread_limit():
+    """
+    Calculates a safe thread limit based on available RAM and CPU cores.
+    Returns an integer safe for to_thread.current_default_thread_limiter().total_tokens
+    """
+    try:
+        # 1. Get System Resources
+        total_ram_bytes = psutil.virtual_memory().total
+        total_cores = multiprocessing.cpu_count()
+        
+        # 2. Identify how many Uvicorn Workers are running
+        # Assuming production setup uses: max(1, total_cores - 2)
+        num_workers = max(1, total_cores - 2)
+        
+        # 3. Calculate RAM available PER WORKER
+        ram_per_worker = total_ram_bytes / num_workers
+        
+        # 4. Reserve Buffer (Keep 30% for OS/Python overhead, use 70% for threads)
+        safe_ram_pool = ram_per_worker * 0.70
+        
+        # 5. Estimate Thread Cost (~8MB conservative safety margin)
+        BYTES_PER_THREAD = 8 * 1024 * 1024 
+        
+        calculated_limit = int(safe_ram_pool / BYTES_PER_THREAD)
+        
+        # 6. Apply Reasonable Hard Caps (Min 100, Max 3000)
+        final_limit = max(100, min(calculated_limit, 3000))
+        
+        logger.info(f"Dynamic Limit Config: {total_ram_bytes/(1024**3):.2f}GB RAM / {num_workers} Workers. Limit: {final_limit}")
+        return final_limit
+
+    except Exception as e:
+        logger.warning(f"Failed to calculate dynamic threads ({e}). Fallback to 1000.")
+        return 1000
+
 @asynccontextmanager
 async def lifespan(app: FastAPI):
-    # Startup logic
-    to_thread.current_default_thread_limiter().total_tokens = 2000
-    logger.info("Worker Process Started: Thread pool capacity set to 2000.")
+    # Startup logic: Calculate and set dynamic thread limit
+    safe_limit = get_dynamic_thread_limit()
+    to_thread.current_default_thread_limiter().total_tokens = safe_limit
+    logger.info(f"Worker Process Started: Thread pool capacity set to {safe_limit}.")
     yield
+    # Shutdown logic (if any) goes here
 
 app = FastAPI(
     title="Unified Data Executor API (Mongo, SQL, CSV)",