Raise runtime error when submitting after executor shutdown (#851)

* Raise runtime error when submitting after executor shutdown

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

---------

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>

Author: jan-janssen

src/executorlib/executor/base.py

@@ -21,6 +21,7 @@ class BaseExecutor(FutureExecutor, ABC):
 
     def __init__(self, executor: TaskSchedulerBase):
         self._task_scheduler = executor
+        self._is_active = True
 
     @property
     def max_workers(self) -> Optional[int]:
@@ -99,9 +100,12 @@ def submit(  # type: ignore
         Returns:
             Future: A Future representing the given call.
         """
-        return self._task_scheduler.submit(
-            *([fn] + list(args)), resource_dict=resource_dict, **kwargs
-        )
+        if self._is_active:
+            return self._task_scheduler.submit(
+                *([fn] + list(args)), resource_dict=resource_dict, **kwargs
+            )
+        else:
+            raise RuntimeError("cannot schedule new futures after shutdown")
 
     def shutdown(self, wait: bool = True, *, cancel_futures: bool = False):
         """
@@ -119,6 +123,7 @@ def shutdown(self, wait: bool = True, *, cancel_futures: bool = False):
                 cancelled.
         """
         self._task_scheduler.shutdown(wait=wait, cancel_futures=cancel_futures)
+        self._is_active = False
 
     def __len__(self) -> int:
         """
@@ -143,3 +148,4 @@ def __exit__(self, *args, **kwargs) -> None:
         Exit method called when exiting the context manager.
         """
         self._task_scheduler.__exit__(*args, **kwargs)
+        self._is_active = False

tests/test_singlenodeexecutor_noblock.py

@@ -166,3 +166,10 @@ def test_single_node_executor_init_function(self):
             with SingleNodeExecutor(max_workers=2, init_function=exit_funct, block_allocation=True) as exe:
                 f = exe.submit(sum, [1, 1])
                 print(f.result())
+
+    def test_single_node_executor_exit(self):
+        exe = SingleNodeExecutor(max_workers=2)
+        self.assertEqual(exe.submit(sum, [1,2,3]).result(), 6)
+        exe.shutdown()
+        with self.assertRaises(RuntimeError):
+            exe.submit(sum, [1, 2, 3])