prepare API forkill_workers()

Author: denisri

capsul/database/__init__.py

@@ -166,6 +166,9 @@ def workers_command(self, engine_id):
         ]
         return workers_command
 
+    def kill_worker_command(self, engine_id, worker_id):
+        raise NotImplementedError
+
     def new_execution(
         self, executable, engine_id, execution_context, workflow, start_time
     ):
@@ -497,6 +500,12 @@ def worker_ended(self, engine_id, worker_id):
         """
         raise NotImplementedError
 
+    def get_workers(self, engine_id):
+        """
+        returns the workers IDs list
+        """
+        raise NotImplementedError
+
     def persistent(self, engine_id):
         """
         Return whether an engine is persistent or not.

capsul/engine/__init__.py

@@ -184,6 +184,8 @@ def start_workers(self):
         if start_count:
             for i in range(start_count):
                 workers_command = self.database.workers_command(self.engine_id)
+                # TODO: we should keep track of the running worker in order
+                # to be able to contact / kill him later.
                 try:
                     subprocess.run(
                         workers_command,
@@ -199,6 +201,28 @@ def quote(x):
                         f'Command failed: {" ".join(quote(i) for i in workers_command)}'
                     ) from e
 
+    def kill_workers(self, worker_ids=None):
+        if worker_ids is None:
+            # kill all workers
+            worker_ids = self.database_get_workers(self.engine_id)
+        for worker_id in worker_ids:
+            cmd = self.database.kill_worker_command(self.engine_id, worker_id)
+            try:
+                subprocess.run(
+                    cmd,
+                    capture_output=False,
+                    check=True,
+                )
+            except Exception as e:
+
+                def quote(x):
+                    return f"'{x}'"
+
+                raise RuntimeError(
+                    f'Command failed: {" ".join(quote(i) for i in cmd)}'
+                ) from e
+            self.database.worker_ended(self.engine_id, worker_id)
+
     def __exit__(self, exception_type, exception_value, exception_traceback):
         # exiting the engine disposes it from the database: executions will
         # be deleted from it, and later inspection will not be possible.