Spawn multiple processes for multi-process consumers

After doing some digging, I found that the --single-thread AND --multi-process options are solely used for making sure that no other process is running.
So if you run a single consumer, you add the --single-thread option. With this option, Magento will check if another process has a lock for that consumer.
The same thing goes for the --multi-process n option, where Magento will check the lock in combination with specified n.

Author: tdgroot

src/main.rs

@@ -75,9 +75,7 @@ fn main() {
     while !term.load(std::sync::atomic::Ordering::Relaxed) {
         // If any of the processes have exited, restart them
         for process in &mut processes {
-            if !process.is_running() {
-                process.restart(&context);
-            }
+            process.ensure_running(&context);
         }
         thread::sleep(Duration::from_secs(2));
     }

src/worker.rs

@@ -11,42 +11,66 @@ const RABBITMQ_CONSUMER_NAMES: [&str; 1] = ["async.operations.all"];
 pub struct WorkerProcess {
     // The consumer name
     consumer: String,
-    // The process handle
-    process: std::process::Child,
+    // The process handles
+    processes: Vec<std::process::Child>,
 }
 
 impl WorkerProcess {
     pub fn terminate(&mut self) {
         log::debug!("Terminating consumer: {}", self.consumer);
-        util::terminate_process_child(&self.process).unwrap();
-        if self.process.try_wait().unwrap().is_none() {
-            self.process.kill().unwrap();
+        for p in self.processes.iter_mut() {
+            util::terminate_process_child(&p).unwrap();
+            if p.is_running() {
+                log::debug!("Force killing consumer: {}", self.consumer);
+                p.kill().unwrap();
+            }
         }
     }
 
-    pub fn is_running(&mut self) -> bool {
-        match self.process.try_wait() {
-            Ok(Some(_)) => false,
-            Ok(None) => true,
-            Err(_) => false,
+    pub fn ensure_running(&mut self, context: &DaemonContext) {
+        if !self.processes.is_running() {
+            self.restart(context);
         }
     }
 
     pub fn restart(&mut self, context: &DaemonContext) {
-        if self.is_running() {
-            self.terminate();
+        self.terminate();
+        self.processes = run_worker(&context, &self.consumer).processes;
+    }
+}
+
+trait WorkerChildProcess {
+    fn is_running(&mut self) -> bool;
+}
+
+impl WorkerChildProcess for std::process::Child {
+    fn is_running(&mut self) -> bool {
+        match self.try_wait() {
+            Ok(Some(status)) => {
+                log::debug!("Process exited with status {:?}", status);
+                return false;
+            }
+            Ok(None) => true,
+            Err(err) => {
+                log::debug!("Process has error {:?}", err);
+                return false;
+            }
         }
-        self.process = run_worker(&context, &self.consumer).process;
+    }
+}
+
+impl WorkerChildProcess for Vec<std::process::Child> {
+    fn is_running(&mut self) -> bool {
+        self.iter_mut().all(|p| p.is_running())
     }
 }
 
 pub fn read_consumer_list(config: &DaemonConfig) -> Vec<String> {
-    // Read consumer list by running bin/magento queue:consumers:list
     let output = Command::new("bin/magento")
         .current_dir(&config.magento_dir)
         .arg("queue:consumers:list")
         .output()
-        .expect("failed to run bin/magento queue:consumers:list");
+        .expect("Failed to run bin/magento queue:consumers:list");
 
     // Split output by newline and convert from u8 sequences to String
     output
@@ -67,30 +91,41 @@ pub fn read_consumer_list(config: &DaemonConfig) -> Vec<String> {
 
 pub fn run_worker(context: &DaemonContext, consumer: &String) -> WorkerProcess {
     log::debug!("Running consumer: {}", consumer);
-    let mut command = Command::new("bin/magento");
-    let command = command
-        .current_dir(&context.daemon_config.magento_dir)
-        .arg("queue:consumers:start")
-        .arg(consumer)
-        .arg("--max-messages")
-        .arg(context.consumer_config.max_messages.to_string());
 
+    let mut number_of_processes = 1;
     if let Some(processes) = context.consumer_config.multiple_processes.get(consumer) {
-        if *processes > 1 {
-            command.arg("--multi-process").arg(processes.to_string());
+        number_of_processes = *processes;
+    }
+
+    let mut processes = Vec::<std::process::Child>::new();
+
+    for i in 0..number_of_processes {
+        let mut command = Command::new("bin/magento");
+        let command = command
+            .current_dir(&context.daemon_config.magento_dir)
+            .arg("queue:consumers:start")
+            .arg(consumer)
+            .arg("--max-messages")
+            .arg(context.consumer_config.max_messages.to_string());
+
+        // We could disable the --multi-process or --single-thread options with a --no-strict-mode flag,
+        // but not sure if users need that, so this is the default for now.
+        if number_of_processes > 1 {
+            command.arg("--multi-process");
+            command.arg(i.to_string());
         } else {
             command.arg("--single-thread");
         }
-    } else {
-        command.arg("--single-thread");
-    }
 
-    let process = command
-        .spawn()
-        .expect("failed to run bin/magento queue:consumers:start");
+        let process = command
+            .spawn()
+            .expect("Failed to run bin/magento queue:consumers:start");
+
+        processes.push(process);
+    }
 
     WorkerProcess {
         consumer: consumer.clone(),
-        process,
+        processes,
     }
 }