support scale engine for deploy and fix _process_dict in task_service

Author: LuckyYC

docker-compose.dev.yml

@@ -24,7 +24,6 @@ services:
     build:
       context: ./backend
       dockerfile: Dockerfile
-    container_name: lmeterx-backend
     restart: unless-stopped
     depends_on:
       mysql:
@@ -39,8 +38,6 @@ services:
     volumes:
       - ./logs:/logs
       - ./upload_files:/app/upload_files
-    ports:
-      - "5001:5001"
     healthcheck:
       test: [ "CMD", "curl", "-s", "-f", "http://localhost:5001/health" ]
       interval: 10s
@@ -53,7 +50,6 @@ services:
     build:
       context: ./st_engine
       dockerfile: Dockerfile
-    container_name: lmeterx-engine
     restart: unless-stopped
     depends_on:
       mysql:
@@ -80,14 +76,12 @@ services:
     volumes:
       - ./logs:/logs
       - ./upload_files:/app/upload_files
-    ports:
-      - "5002:5002"
     # Resource limits for the container
     deploy:
       resources:
         limits:
           cpus: '2.0'                           # Must match LOCUST_CPU_CORES
-          memory: 2G
+          memory: 4G
         reservations:
           cpus: '1.0'
           memory: 1G

docker-compose.yml

@@ -20,7 +20,6 @@ services:
   # backend service
   backend:
     image: charmy1220/lmeterx-be:latest
-    container_name: lmeterx-backend
     restart: unless-stopped
     depends_on:
       mysql:
@@ -35,8 +34,6 @@ services:
     volumes:
       - ./logs:/logs
       - ./upload_files:/app/upload_files
-    ports:
-      - "5001:5001"
     healthcheck:
       test: [ "CMD", "curl", "-s", "-f", "http://localhost:5001/health" ]
       interval: 10s
@@ -47,7 +44,6 @@ services:
   # engine service
   engine:
     image: charmy1220/lmeterx-eng:latest
-    container_name: lmeterx-engine
     restart: unless-stopped
     depends_on:
       mysql:
@@ -60,7 +56,7 @@ services:
       DB_NAME: lmeterx
       # Multi-process configuration
       ENABLE_MULTIPROCESS: auto
-      LOCUST_CPU_CORES: 4.0
+      LOCUST_CPU_CORES: 2.0
       MULTIPROCESS_THRESHOLD: 1000           # Enable multiprocess only for 1000+ users
       MIN_USERS_PER_PROCESS: 600             # Each process handles at least 600 users
       # Process management and stability
@@ -74,14 +70,12 @@ services:
     volumes:
       - ./logs:/logs
       - ./upload_files:/app/upload_files
-    ports:
-      - "5002:5002"
     # Resource limits for the container
     deploy:
       resources:
         limits:
           cpus: '2.0'                        # Multi-process mode requires at least 2 cores
-          memory: 2G
+          memory: 4G
         reservations:
           cpus: '1.0'
           memory: 1G

st_engine/service/task_service.py

@@ -463,7 +463,7 @@ def stop_task(self, task_id: str) -> bool:
             task_logger.info(f"Received stop request for task {task_id}.")
 
             # Check if the process is managed by this runner instance
-            process = self.runner.process_dict.get(task_id)
+            process = self.runner._process_dict.get(task_id)
             if not process:
                 task_logger.warning(
                     f"Task {task_id}: Process not found in runner's dictionary. "
@@ -477,7 +477,7 @@ def stop_task(self, task_id: str) -> bool:
                 task_logger.info(
                     f"Task {task_id}: Process with PID {process.pid} has already terminated. Cleaning up local reference."
                 )
-                self.runner.process_dict.pop(task_id, None)
+                self.runner._process_dict.pop(task_id, None)
                 return True
 
             # Delegate the complex termination logic to the process manager via the runner.
@@ -489,7 +489,7 @@ def stop_task(self, task_id: str) -> bool:
                     f"Successfully terminated process group for task {task_id}."
                 )
                 # Remove from local tracking after successful termination
-                self.runner.process_dict.pop(task_id, None)
+                self.runner._process_dict.pop(task_id, None)
                 return True
             else:
                 task_logger.error(
@@ -525,7 +525,7 @@ def stop_task_old(self, task_id: str) -> bool:
                 )
 
                 # Clean up local tracking
-                self.runner.process_dict.pop(task_id, None)
+                self.runner._process_dict.pop(task_id, None)
                 if hasattr(self.runner, "_terminating_processes"):
                     termination_key = f"{task_id}_terminating"
                     self.runner._terminating_processes.discard(termination_key)
@@ -535,13 +535,13 @@ def stop_task_old(self, task_id: str) -> bool:
                 return True
 
             # Step 2: Fallback to original process termination if multiprocess cleanup failed
-            process = self.runner.process_dict.get(task_id)
+            process = self.runner._process_dict.get(task_id)
 
             if not process:
                 task_logger.warning(
                     f"Task {task_id}, Process not found in runner's dictionary. It might have finished or be on another node."
                 )
-                self.runner.process_dict.pop(task_id, None)
+                self.runner._process_dict.pop(task_id, None)
                 # Clean up task resources (do not force cleanup orphaned processes here)
                 cleanup_task_resources(task_id)
                 return True
@@ -550,7 +550,7 @@ def stop_task_old(self, task_id: str) -> bool:
                 task_logger.info(
                     f"Task {task_id}, Process with PID {process.pid} has already terminated. Cleaning up."
                 )
-                self.runner.process_dict.pop(task_id, None)
+                self.runner._process_dict.pop(task_id, None)
                 cleanup_task_resources(task_id)
                 return True
 
@@ -577,7 +577,7 @@ def stop_task_old(self, task_id: str) -> bool:
                 task_logger.info(
                     f"Task {task_id}, Process with PID {process.pid} terminated naturally while preparing to stop it."
                 )
-                self.runner.process_dict.pop(task_id, None)
+                self.runner._process_dict.pop(task_id, None)
                 self.runner._terminating_processes.discard(termination_key)
                 cleanup_task_resources(task_id)
                 return True
@@ -587,7 +587,7 @@ def stop_task_old(self, task_id: str) -> bool:
             task_logger.info(
                 f"Task {task_id}, Process terminated successfully via SIGTERM."
             )
-            self.runner.process_dict.pop(task_id, None)
+            self.runner._process_dict.pop(task_id, None)
             self.runner._terminating_processes.discard(termination_key)
             cleanup_task_resources(task_id)
             return True
@@ -601,7 +601,7 @@ def stop_task_old(self, task_id: str) -> bool:
                     task_logger.info(
                         f"Task {task_id}, Process with PID {process.pid} terminated naturally during SIGTERM timeout."
                     )
-                    self.runner.process_dict.pop(task_id, None)
+                    self.runner._process_dict.pop(task_id, None)
                     self.runner._terminating_processes.discard(termination_key)
                     return True
 
@@ -610,7 +610,7 @@ def stop_task_old(self, task_id: str) -> bool:
                 task_logger.info(
                     f"Task {task_id}, Process killed successfully via SIGKILL."
                 )
-                self.runner.process_dict.pop(task_id, None)
+                self.runner._process_dict.pop(task_id, None)
                 self.runner._terminating_processes.discard(termination_key)
                 return True
             except subprocess.TimeoutExpired:
@@ -630,7 +630,7 @@ def stop_task_old(self, task_id: str) -> bool:
             task_logger.info(
                 f"Task {task_id}, Process with PID {process.pid} no longer exists (ProcessLookupError). Cleaning up."
             )
-            self.runner.process_dict.pop(task_id, None)
+            self.runner._process_dict.pop(task_id, None)
             self.runner._terminating_processes.discard(termination_key)
             return True
         except Exception as e:
@@ -646,7 +646,7 @@ def stop_task_old(self, task_id: str) -> bool:
                     task_logger.info(
                         f"Task {task_id}, Process completed its natural shutdown successfully."
                     )
-                    self.runner.process_dict.pop(task_id, None)
+                    self.runner._process_dict.pop(task_id, None)
                     self.runner._terminating_processes.discard(termination_key)
                     return True
                 except subprocess.TimeoutExpired:
@@ -659,7 +659,7 @@ def stop_task_old(self, task_id: str) -> bool:
                         task_logger.info(
                             f"Task {task_id}, Process force-killed successfully after natural shutdown timeout."
                         )
-                        self.runner.process_dict.pop(task_id, None)
+                        self.runner._process_dict.pop(task_id, None)
                         self.runner._terminating_processes.discard(termination_key)
                         return True
                     except Exception as kill_e: