move around functions

Author: basnijholt

adaptive/runner.py

@@ -50,7 +50,6 @@
 except ImportError:
     from typing_extensions import Literal
 
-
 try:
     import ipyparallel
     from ipyparallel.client.asyncresult import AsyncResult
@@ -101,52 +100,6 @@
     _default_executor = loky.get_reusable_executor
 
 
-# -- Internal executor-related, things
-
-
-def _ensure_executor(
-    executor: ExecutorTypes | None,
-) -> concurrent.Executor:
-    if executor is None:
-        executor = concurrent.ProcessPoolExecutor()
-
-    if isinstance(executor, concurrent.Executor):
-        return executor
-    elif with_ipyparallel and isinstance(executor, ipyparallel.Client):
-        return executor.executor()
-    elif with_distributed and isinstance(executor, distributed.Client):
-        return executor.get_executor()
-    else:
-        raise TypeError(
-            # TODO: check if this is correct. Isn't MPI,loky supported?
-            "Only a concurrent.futures.Executor, distributed.Client,"
-            " or ipyparallel.Client can be used."
-        )
-
-
-def _get_ncores(
-    ex: (ExecutorTypes),
-) -> int:
-    """Return the maximum  number of cores that an executor can use."""
-    if with_ipyparallel and isinstance(ex, ipyparallel.client.view.ViewExecutor):
-        return len(ex.view)
-    elif isinstance(
-        ex, (concurrent.ProcessPoolExecutor, concurrent.ThreadPoolExecutor)
-    ):
-        return ex._max_workers  # not public API!
-    elif isinstance(ex, loky.reusable_executor._ReusablePoolExecutor):
-        return ex._max_workers  # not public API!
-    elif isinstance(ex, SequentialExecutor):
-        return 1
-    elif with_distributed and isinstance(ex, distributed.cfexecutor.ClientExecutor):
-        return sum(n for n in ex._client.ncores().values())
-    elif with_mpi4py and isinstance(ex, mpi4py.futures.MPIPoolExecutor):
-        ex.bootup()  # wait until all workers are up and running
-        return ex._pool.size  # not public API!
-    else:
-        raise TypeError(f"Cannot get number of cores for {ex.__class__}")
-
-
 class BaseRunner(metaclass=abc.ABCMeta):
     r"""Base class for runners that use `concurrent.futures.Executor`\'s.
 
@@ -1016,9 +969,6 @@ def stop_after(*, seconds=0, minutes=0, hours=0) -> Callable[[BaseLearner], bool
     return lambda _: time.time() > stop_time
 
 
-# -- Internal executor-related, things
-
-
 class _TimeGoal:
     def __init__(self, dt: timedelta | datetime | int | float):
         self.dt = dt if isinstance(dt, (timedelta, datetime)) else timedelta(seconds=dt)
@@ -1142,3 +1092,49 @@ def _goal(
         duration=duration_goal,
         allow_running_forever=allow_running_forever,
     )
+
+
+# -- Internal executor-related, things
+
+
+def _ensure_executor(
+    executor: ExecutorTypes | None,
+) -> concurrent.Executor:
+    if executor is None:
+        executor = concurrent.ProcessPoolExecutor()
+
+    if isinstance(executor, concurrent.Executor):
+        return executor
+    elif with_ipyparallel and isinstance(executor, ipyparallel.Client):
+        return executor.executor()
+    elif with_distributed and isinstance(executor, distributed.Client):
+        return executor.get_executor()
+    else:
+        raise TypeError(
+            # TODO: check if this is correct. Isn't MPI,loky supported?
+            "Only a concurrent.futures.Executor, distributed.Client,"
+            " or ipyparallel.Client can be used."
+        )
+
+
+def _get_ncores(
+    ex: (ExecutorTypes),
+) -> int:
+    """Return the maximum  number of cores that an executor can use."""
+    if with_ipyparallel and isinstance(ex, ipyparallel.client.view.ViewExecutor):
+        return len(ex.view)
+    elif isinstance(
+        ex, (concurrent.ProcessPoolExecutor, concurrent.ThreadPoolExecutor)
+    ):
+        return ex._max_workers  # not public API!
+    elif isinstance(ex, loky.reusable_executor._ReusablePoolExecutor):
+        return ex._max_workers  # not public API!
+    elif isinstance(ex, SequentialExecutor):
+        return 1
+    elif with_distributed and isinstance(ex, distributed.cfexecutor.ClientExecutor):
+        return sum(n for n in ex._client.ncores().values())
+    elif with_mpi4py and isinstance(ex, mpi4py.futures.MPIPoolExecutor):
+        ex.bootup()  # wait until all workers are up and running
+        return ex._pool.size  # not public API!
+    else:
+        raise TypeError(f"Cannot get number of cores for {ex.__class__}")

example-notebook.ipynb

@@ -2000,7 +2000,7 @@
     }
    ],
    "source": [
-    "type(ex.submit(lambda x: x, 1))"
+    "type(ex.submiddt(lambda x: x, 1))"
    ]
   },
   {