Repository: BQSKit/bqskit (language: python)

Issue:
Compiler Call raising an issue
Hi,

I have been calling the compiler call after passing a workflow to it and the error is as follows:

Exception ignored in: <function Compiler.__del__ at 0x7f966cdc8430>
Traceback (most recent call last):
  File "./conda/envs/quantum-venv/lib/python3.10/site-packages/bqskit/compiler/compiler.py", line 227, in __del__
TypeError: 'NoneType' object is not callable
Use `gc.freeze` during startup
We should evaluate if `gc.freeze` should be used at some point during runtime startup. Can potentially lead to a lot of savings: https://www.youtube.com/watch?v=Hgw_RlCaIds
Print warning during attached server connection issues
If the attached server has trouble connecting or takes a long time, we should print a warning to the user with helpful information, potentially nudging them to use detached mode.
Harmless Race Condition in AttachedServer Cleanup
When running the `compile` function, the following error can occur very rarely without any other issue:

```
Traceback (most recent call last):
  File ".../lib/python3.11/site-packages/bqskit/runtime/base.py", line 447, in __del__
    def __del__(self) -> None:

OSError: Signal 2 ignored due to race condition
```

It seems like a race condition in the destruction of an AttachedServer (or any runtime server).
Specifying permutation in logs in PAS
Currently, permutation aware synthesis runs all permutations in parallel, producing output like in the attached picture for 3 qubits.  

It would be really helpful if each line in the log specified which permutation it pertains to, eg 
`DEBUG    | [1,2,3] bqskit.passes ` ...
`DEBUG    | [1,3,2] bqskit.passes ` ...

This would allow generation of logs for each independent permutation, eg by using `grep [1,2,3] output.log`




Discussion / hints:
Hi, 

Can you please share the workflow you are trying to run?, or maybe even a minimal reproducible code?

Alon
@alonkukl Thanks for responding so quickly! This issue arose during a meeting, and I know the details. I will be looking into this shortly.
@mwalidcharrwi Are you importing or using `atexit` directly in your code at all? I know you might not want to share your code publically, but can you put it in a private repo and provide access to me?
@edyounis @alonkukl thanks for your response no there is no `atexit` imported in the code. Only bqskit.passes and bqskit.compiler (compiler and compilation task) is imported.
I will do that and share it with you.
@edyounis I sent you an invitation for the private repository with the script uploaded



![logs](https://github.com/BQSKit/bqskit/assets/29586514/77c02d3e-82c9-44fa-a4f4-3b308a029d48)


Fix patch:
diff --git a/bqskit/__init__.py b/bqskit/__init__.py
index dd6a87128..83ca808bb 100644
--- a/bqskit/__init__.py
+++ b/bqskit/__init__.py
@@ -1,91 +1,29 @@
 """The Berkeley Quantum Synthesis Toolkit Python Package."""
 from __future__ import annotations
 
-import logging
-from sys import stdout as _stdout
+from typing import Any
 
-from .version import __version__  # noqa: F401
-from .version import __version_info__  # noqa: F401
-from bqskit.compiler.compile import compile
-from bqskit.compiler.machine import MachineModel
-from bqskit.ir.circuit import Circuit
-from bqskit.ir.lang import register_language as _register_language
-from bqskit.ir.lang.qasm2 import OPENQASM2Language as _qasm
+from bqskit._logging import disable_logging
+from bqskit._logging import enable_logging
+from bqskit._version import __version__  # noqa: F401
+from bqskit._version import __version_info__  # noqa: F401
 
-# Initialize Logging
-_logging_initialized = False
 
+def __getattr__(name: str) -> Any:
+    # Lazy imports
+    if name == 'compile':
+        from bqskit.compiler.compile import compile
+        return compile
 
-def enable_logging(verbose: bool = False) -> None:
-    """
-    Enable logging for BQSKit.
+    if name == 'Circuit':
+        from bqskit.ir.circuit import Circuit
+        return Circuit
 
-    Args:
-        verbose (bool): If set to True, will print more verbose messages.
-            Defaults to False.
-    """
-    global _logging_initialized
-    if not _logging_initialized:
-        _logger = logging.getLogger('bqskit')
-        _handler = logging.StreamHandler(_stdout)
-        _handler.setLevel(0)
-        _fmt_header = '%(asctime)s.%(msecs)03d - %(levelname)-8s |'
-        _fmt_message = ' %(name)s: %(message)s'
-        _fmt = _fmt_header + _fmt_message
-        _formatter = logging.Formatter(_fmt, '%H:%M:%S')
-        _handler.setFormatter(_formatter)
-        _logger.addHandler(_handler)
-        _logging_initialized = True
+    if name == 'MachineModel':
+        from bqskit.compiler.machine import MachineModel
+        return MachineModel
 
-    level = logging.DEBUG if verbose else logging.INFO
-    logging.getLogger('bqskit').setLevel(level)
-
-
-def disable_logging() -> None:
-    """Disable logging for BQSKit."""
-    logging.getLogger('bqskit').setLevel(logging.CRITICAL)
-
-
-def enable_dashboard() -> None:
-    import warnings
-    warnings.warn(
-        'Dask has been removed from BQSKit. As a result, the'
-        ' enable_dashboard method has been removed.'
-        'This warning will turn into an error in a future update.',
-        DeprecationWarning,
-    )
-
-
-def disable_dashboard() -> None:
-    import warnings
-    warnings.warn(
-        'Dask has been removed from BQSKit. As a result, the'
-        ' disable_dashboard method has been removed.'
-        'This warning will turn into an error in a future update.',
-        DeprecationWarning,
-    )
-
-
-def disable_parallelism() -> None:
-    import warnings
-    warnings.warn(
-        'The disable_parallelism method has been removed.'
-        ' Instead, set the "num_workers" parameter to 1 during '
-        'Compiler construction. This warning will turn into'
-        'an error in a future update.',
-        DeprecationWarning,
-    )
-
-
-def enable_parallelism() -> None:
-    import warnings
-    warnings.warn(
-        'The enable_parallelism method has been removed.'
-        ' Instead, set the "num_workers" parameter to 1 during '
-        'Compiler construction. This warning will turn into'
-        'an error in a future update.',
-        DeprecationWarning,
-    )
+    raise AttributeError(f'module {__name__} has no attribute {name}')
 
 
 __all__ = [
@@ -95,6 +33,3 @@ def enable_parallelism() -> None:
     'enable_logging',
     'disable_logging',
 ]
-
-# Register supported languages
-_register_language('qasm', _qasm())
diff --git a/bqskit/_logging.py b/bqskit/_logging.py
new file mode 100644
index 000000000..59b079156
--- /dev/null
+++ b/bqskit/_logging.py
@@ -0,0 +1,38 @@
+"""This module contains the logging configuration and methods for BQSKit."""
+from __future__ import annotations
+
+import logging
+from sys import stdout as _stdout
+
+
+_logging_initialized = False
+
+
+def enable_logging(verbose: bool = False) -> None:
+    """
+    Enable logging for BQSKit.
+
+    Args:
+        verbose (bool): If set to True, will print more verbose messages.
+            Defaults to False.
+    """
+    global _logging_initialized
+    if not _logging_initialized:
+        _logger = logging.getLogger('bqskit')
+        _handler = logging.StreamHandler(_stdout)
+        _handler.setLevel(0)
+        _fmt_header = '%(asctime)s.%(msecs)03d - %(levelname)-8s |'
+        _fmt_message = ' %(name)s: %(message)s'
+        _fmt = _fmt_header + _fmt_message
+        _formatter = logging.Formatter(_fmt, '%H:%M:%S')
+        _handler.setFormatter(_formatter)
+        _logger.addHandler(_handler)
+        _logging_initialized = True
+
+    level = logging.DEBUG if verbose else logging.INFO
+    logging.getLogger('bqskit').setLevel(level)
+
+
+def disable_logging() -> None:
+    """Disable logging for BQSKit."""
+    logging.getLogger('bqskit').setLevel(logging.CRITICAL)
diff --git a/bqskit/version.py b/bqskit/_version.py
similarity index 100%
rename from bqskit/version.py
rename to bqskit/_version.py
diff --git a/bqskit/compiler/__init__.py b/bqskit/compiler/__init__.py
index f7048aa56..fde8f87ee 100644
--- a/bqskit/compiler/__init__.py
+++ b/bqskit/compiler/__init__.py
@@ -37,8 +37,9 @@
 """
 from __future__ import annotations
 
+from typing import Any
+
 from bqskit.compiler.basepass import BasePass
-from bqskit.compiler.compile import compile
 from bqskit.compiler.compiler import Compiler
 from bqskit.compiler.gateset import GateSet
 from bqskit.compiler.gateset import GateSetLike
@@ -49,6 +50,19 @@
 from bqskit.compiler.workflow import Workflow
 from bqskit.compiler.workflow import WorkflowLike
 
+
+def __getattr__(name: str) -> Any:
+    # Lazy imports
+    if name == 'compile':
+        # TODO: fix this (high-priority), overlap between module and function
+        from bqskit.compiler.compile import compile
+        return compile
+
+    # TODO: Move compile to a different subpackage and deprecate import
+
+    raise AttributeError(f'module {__name__} has no attribute {name}')
+
+
 __all__ = [
     'BasePass',
     'compile',
diff --git a/bqskit/compiler/compile.py b/bqskit/compiler/compile.py
index 4824c42ad..c38215c57 100644
--- a/bqskit/compiler/compile.py
+++ b/bqskit/compiler/compile.py
@@ -2,6 +2,7 @@
 from __future__ import annotations
 
 import logging
+import math
 import warnings
 from typing import Any
 from typing import Literal
@@ -10,8 +11,6 @@
 from typing import TYPE_CHECKING
 from typing import Union
 
-import numpy as np
-
 from bqskit.compiler.compiler import Compiler
 from bqskit.compiler.machine import MachineModel
 from bqskit.compiler.passdata import PassData
@@ -582,7 +581,7 @@ def type_and_check_input(input: CompilationInputLike) -> CompilationInput:
         if error_threshold is not None:
             for i, data in enumerate(datas):
                 error = data.error
-                nonsq_error = 1 - np.sqrt(max(1 - (error * error), 0))
+                nonsq_error = 1 - math.sqrt(max(1 - (error * error), 0))
                 if nonsq_error > error_threshold:
                     warnings.warn(
                         'Upper bound on error is greater than set threshold:'
@@ -631,7 +630,7 @@ def type_and_check_input(input: CompilationInputLike) -> CompilationInput:
         # Log error if necessary
         if error_threshold is not None:
             error = data.error
-            nonsq_error = 1 - np.sqrt(max(1 - (error * error), 0))
+            nonsq_error = 1 - math.sqrt(max(1 - (error * error), 0))
             if nonsq_error > error_threshold:
                 warnings.warn(
                     'Upper bound on error is greater than set threshold:'
diff --git a/bqskit/compiler/compiler.py b/bqskit/compiler/compiler.py
index 55bb50a1d..8f58f3c35 100644
--- a/bqskit/compiler/compiler.py
+++ b/bqskit/compiler/compiler.py
@@ -9,12 +9,12 @@
 import sys
 import time
 import uuid
-import warnings
 from multiprocessing.connection import Client
 from multiprocessing.connection import Connection
 from subprocess import Popen
 from types import FrameType
 from typing import Literal
+from typing import MutableMapping
 from typing import overload
 from typing import TYPE_CHECKING
 
@@ -71,6 +71,7 @@ def __init__(
         num_workers: int = -1,
         runtime_log_level: int = logging.WARNING,
         worker_port: int = default_worker_port,
+        num_blas_threads: int = 1,
     ) -> None:
         """
         Construct a Compiler object.
@@ -99,30 +100,42 @@ def __init__(
             worker_port (int): The optional port to pass to an attached
                 runtime. See :obj:`~bqskit.runtime.attached.AttachedServer`
                 for more info.
+
+            num_blas_threads (int): The number of threads to use in the
+                BLAS libraries on the worker nodes. (Defaults to 1)
         """
         self.p: Popen | None = None  # type: ignore
         self.conn: Connection | None = None
 
-        atexit.register(self.close)
+        _compiler_instances.add(self)
 
         if ip is None:
             ip = 'localhost'
-            self._start_server(num_workers, runtime_log_level, worker_port)
+            self._start_server(
+                num_workers,
+                runtime_log_level,
+                worker_port,
+                num_blas_threads,
+            )
 
-        self._connect_to_server(ip, port)
+        self._connect_to_server(ip, port, self.p is not None)
 
     def _start_server(
         self,
         num_workers: int,
         runtime_log_level: int,
         worker_port: int,
+        num_blas_threads: int,
     ) -> None:
         """
         Start an attached serer with `num_workers` workers.
 
         See :obj:`~bqskit.runtime.attached.AttachedServer` for more info.
         """
-        params = f'{num_workers}, {runtime_log_level}, {worker_port=}'
+        params = f'{num_workers}, '
+        params += f'log_level={runtime_log_level}, '
+        params += f'{worker_port=}, '
+        params += f'{num_blas_threads=}, '
         import_str = 'from bqskit.runtime.attached import start_attached_server'
         launch_str = f'{import_str}; start_attached_server({params})'
         if sys.platform == 'win32':
@@ -132,24 +145,45 @@ def _start_server(
         self.p = Popen([sys.executable, '-c', launch_str], creationflags=flags)
         _logger.debug('Starting runtime server process.')
 
-    def _connect_to_server(self, ip: str, port: int) -> None:
+    def _connect_to_server(self, ip: str, port: int, attached: bool) -> None:
         """Connect to a runtime server at `ip` and `port`."""
         max_retries = 8
         wait_time = .25
-        for _ in range(max_retries):
+        current_retry = 0
+        while current_retry < max_retries or attached:
             try:
                 family = 'AF_INET' if sys.platform == 'win32' else None
                 conn = Client((ip, port), family)
             except ConnectionRefusedError:
+                if wait_time > 4:
+                    _logger.warning(
+                        'Connection refused by runtime server.'
+                        ' Retrying in %s seconds.', wait_time,
+                    )
+                if wait_time > 16 and attached:
+                    _logger.warning(
+                        'Connection is still refused by runtime server.'
+                        ' This may be due to the server not being started.'
+                        ' You may want to check the server logs, by starting'
+                        ' the compiler with "runtime_log_level" set. You'
+                        ' can also try launching the bqskit runtime in'
+                        ' detached mode. See the bqskit runtime documentation'
+                        ' for more information:'
+                        ' https://bqskit.readthedocs.io/en/latest/guides/'
+                        'distributing.html',
+                    )
                 time.sleep(wait_time)
                 wait_time *= 2
+                current_retry += 1
             else:
                 self.conn = conn
                 handle = functools.partial(sigint_handler, compiler=self)
                 self.old_signal = signal.signal(signal.SIGINT, handle)
                 if self.conn is None:
                     raise RuntimeError('Connection unexpectedly none.')
-                self.conn.send((RuntimeMessage.CONNECT, None))
+                msg, payload = self._send_recv(RuntimeMessage.CONNECT, sys.path)
+                if msg != RuntimeMessage.READY:
+                    raise RuntimeError(f'Unexpected message type: {msg}.')
                 _logger.debug('Successfully connected to runtime server.')
                 return
         raise RuntimeError('Client connection refused')
@@ -221,28 +255,25 @@ def close(self) -> None:
         # Reset interrupt signal handler and remove exit handler
         if hasattr(self, 'old_signal'):
             signal.signal(signal.SIGINT, self.old_signal)
+            del self.old_signal
 
-    def __del__(self) -> None:
-        self.close()
-        atexit.unregister(self.close)
-        _logger.debug('Compiler successfully shutdown.')
+        _compiler_instances.discard(self)
+        _logger.debug('Compiler has been closed.')
 
     def submit(
         self,
-        task_or_circuit: CompilationTask | Circuit,
-        workflow: WorkflowLike | None = None,
+        circuit: Circuit,
+        workflow: WorkflowLike,
         request_data: bool = False,
         logging_level: int | None = None,
         max_logging_depth: int = -1,
+        data: MutableMapping[str, Any] | None = None,
     ) -> uuid.UUID:
         """
         Submit a compilation job to the Compiler.
 
         Args:
-            task_or_circuit (CompilationTask | Circuit): The task to compile,
-                or the input circuit. If a task is specified, no other
-                argument should be specified. If a task is not specified,
-                the circuit must be paired with a workflow argument.
+            circuit (Circuit): The input circuit to be compiled.
 
             workflow (WorkflowLike): The compilation job submitted
                 is defined by executing this workflow on the input circuit.
@@ -262,91 +293,40 @@ def submit(
                 tasks equal opportunity to log.
 
         Returns:
-            (uuid.UUID): The ID of the generated task in the system. This
+            uuid.UUID: The ID of the generated task in the system. This
                 ID can be used to check the status of, cancel, and request
                 the result of the task.
         """
         # Build CompilationTask
-        if isinstance(task_or_circuit, CompilationTask):
-            if workflow is not None:
-                raise ValueError(
-                    'Cannot specify workflow and task.'
-                    ' Either specify a workflow and circuit or a task alone.',
-                )
-
-            task = task_or_circuit
-
-        else:
-            if workflow is None:
-                m = 'Must specify workflow when providing a circuit to submit.'
-                raise TypeError(m)
-
-            task = CompilationTask(task_or_circuit, Workflow(workflow))
+        task = CompilationTask(circuit, Workflow(workflow))
 
         # Set task configuration
         task.request_data = request_data
         task.logging_level = logging_level or self._discover_lowest_log_level()
         task.max_logging_depth = max_logging_depth
+        if data is not None:
+            task.data.update(data)
 
         # Submit task to runtime
         self._send(RuntimeMessage.SUBMIT, task)
         return task.task_id
 
-    def status(self, task_id: CompilationTask | uuid.UUID) -> CompilationStatus:
+    def status(self, task_id: uuid.UUID) -> CompilationStatus:
         """Retrieve the status of the specified task."""
-        if isinstance(task_id, CompilationTask):
-            warnings.warn(
-                'Request a status from a CompilationTask is deprecated.\n'
-                ' Instead, pass a task ID to request a status.\n'
-                ' `compiler.submit` returns a task id, and you can get an\n'
-                ' ID from a task via `task.task_id`.\n'
-                ' This warning will turn into an error in a future update.',
-                DeprecationWarning,
-            )
-            task_id = task_id.task_id
-        assert isinstance(task_id, uuid.UUID)
-
         msg, payload = self._send_recv(RuntimeMessage.STATUS, task_id)
         if msg != RuntimeMessage.STATUS:
             raise RuntimeError(f'Unexpected message type: {msg}.')
         return payload
 
-    def result(
-        self,
-        task_id: CompilationTask | uuid.UUID,
-    ) -> Circuit | tuple[Circuit, PassData]:
+    def result(self, task_id: uuid.UUID) -> Circuit | tuple[Circuit, PassData]:
         """Block until the task is finished, return its result."""
-        if isinstance(task_id, CompilationTask):
-            warnings.warn(
-                'Request a result from a CompilationTask is deprecated.'
-                ' Instead, pass a task ID to request a result.\n'
-                ' `compiler.submit` returns a task id, and you can get an\n'
-                ' ID from a task via `task.task_id`.\n'
-                ' This warning will turn into an error in a future update.',
-                DeprecationWarning,
-            )
-            task_id = task_id.task_id
-        assert isinstance(task_id, uuid.UUID)
-
         msg, payload = self._send_recv(RuntimeMessage.REQUEST, task_id)
         if msg != RuntimeMessage.RESULT:
             raise RuntimeError(f'Unexpected message type: {msg}.')
         return payload
 
-    def cancel(self, task_id: CompilationTask | uuid.UUID) -> bool:
+    def cancel(self, task_id: uuid.UUID) -> bool:
         """Cancel the execution of a task in the system."""
-        if isinstance(task_id, CompilationTask):
-            warnings.warn(
-                'Cancelling a CompilationTask is deprecated. Instead,'
-                ' Instead, pass a task ID to cancel a task.\n'
-                ' `compiler.submit` returns a task id, and you can get an\n'
-                ' ID from a task via `task.task_id`.\n'
-                ' This warning will turn into an error in a future update.',
-                DeprecationWarning,
-            )
-            task_id = task_id.task_id
-        assert isinstance(task_id, uuid.UUID)
-
         msg, _ = self._send_recv(RuntimeMessage.CANCEL, task_id)
         if msg != RuntimeMessage.CANCEL:
             raise RuntimeError(f'Unexpected message type: {msg}.')
@@ -355,63 +335,51 @@ def cancel(self, task_id: CompilationTask | uuid.UUID) -> bool:
     @overload
     def compile(
         self,
-        task_or_circuit: CompilationTask,
-    ) -> Circuit | tuple[Circuit, PassData]:
-        ...
-
-    @overload
-    def compile(
-        self,
-        task_or_circuit: Circuit,
+        circuit: Circuit,
         workflow: WorkflowLike,
         request_data: Literal[False] = ...,
         logging_level: int | None = ...,
         max_logging_depth: int = ...,
+        data: MutableMapping[str, Any] | None = ...,
     ) -> Circuit:
         ...
 
     @overload
     def compile(
         self,
-        task_or_circuit: Circuit,
+        circuit: Circuit,
         workflow: WorkflowLike,
         request_data: Literal[True],
         logging_level: int | None = ...,
         max_logging_depth: int = ...,
+        data: MutableMapping[str, Any] | None = ...,
     ) -> tuple[Circuit, PassData]:
         ...
 
     @overload
     def compile(
         self,
-        task_or_circuit: Circuit,
+        circuit: Circuit,
         workflow: WorkflowLike,
         request_data: bool,
         logging_level: int | None = ...,
         max_logging_depth: int = ...,
+        data: MutableMapping[str, Any] | None = ...,
     ) -> Circuit | tuple[Circuit, PassData]:
         ...
 
     def compile(
         self,
-        task_or_circuit: CompilationTask | Circuit,
-        workflow: WorkflowLike | None = None,
+        circuit: Circuit,
+        workflow: WorkflowLike,
         request_data: bool = False,
         logging_level: int | None = None,
         max_logging_depth: int = -1,
+        data: MutableMapping[str, Any] | None = None,
     ) -> Circuit | tuple[Circuit, PassData]:
         """Submit a task, wait for its results; see :func:`submit` for more."""
-        if isinstance(task_or_circuit, CompilationTask):
-            warnings.warn(
-                'Manually constructing and compiling CompilationTasks'
-                ' is deprecated. Instead, call compile directly with'
-                ' your input circuit and workflow. This warning will'
-                ' turn into an error in a future update.',
-                DeprecationWarning,
-            )
-
         task_id = self.submit(
-            task_or_circuit,
+            circuit,
             workflow,
             request_data,
             logging_level,
@@ -530,3 +498,12 @@ def sigint_handler(signum: int, frame: FrameType, compiler: Compiler) -> None:
     _logger.critical('Compiler interrupted.')
     compiler.close()
     raise KeyboardInterrupt
+
+
+_compiler_instances: set[Compiler] = set()
+
+
+@atexit.register
+def _cleanup_compiler_instances() -> None:
+    for compiler in list(_compiler_instances):
+        compiler.close()
diff --git a/bqskit/compiler/passdata.py b/bqskit/compiler/passdata.py
index 160d4f44e..23d584712 100644
--- a/bqskit/compiler/passdata.py
+++ b/bqskit/compiler/passdata.py
@@ -252,6 +252,24 @@ def __contains__(self, _o: object) -> bool:
         in_data = self._data.__contains__(_o)
         return in_resv or in_data
 
+    def update(self, other: Any = (), /, **kwds: Any) -> None:
+        """Update the data with key-values pairs from `other` and `kwds`."""
+        if isinstance(other, PassData):
+            for key in other:
+                # Handle target specially to avoid circuit evaluation
+                if key == 'target':
+                    self._target = other._target
+                    continue
+
+                self[key] = other[key]
+
+            for key, value in kwds.items():
+                self[key] = value
+
+            return
+
+        super().update(other, **kwds)
+
     def copy(self) -> PassData:
         """Returns a deep copy of the data."""
         return copy.deepcopy(self)
diff --git a/bqskit/compiler/workflow.py b/bqskit/compiler/workflow.py
index 6134d07aa..9a0ad2677 100644
--- a/bqskit/compiler/workflow.py
+++ b/bqskit/compiler/workflow.py
@@ -10,6 +10,8 @@
 from typing import TYPE_CHECKING
 from typing import Union
 
+import dill
+
 from bqskit.compiler.basepass import BasePass
 from bqskit.utils.random import seed_random_sources
 from bqskit.utils.typing import is_iterable
@@ -39,6 +41,7 @@ def __init__(self, passes: WorkflowLike, name: str = '') -> None:
         """
         if isinstance(passes, Workflow):
             self._passes: list[BasePass] = copy.deepcopy(passes._passes)
+            self._name: str = name if name else copy.deepcopy(passes._name)
             return
 
         if isinstance(passes, BasePass):
@@ -118,5 +121,11 @@ def __getitem__(self, _key: slice, /) -> list[BasePass]:
     def __getitem__(self, _key: int | slice) -> BasePass | list[BasePass]:
         return self._passes.__getitem__(_key)
 
+    def __getstate__(self) -> bytes:
+        return dill.dumps(self.__dict__, recurse=True)
+
+    def __setstate__(self, state: bytes) -> None:
+        self.__dict__.update(dill.loads(state))
+
 
 WorkflowLike = Union[Workflow, Iterable[BasePass], BasePass]
diff --git a/bqskit/ext/__init__.py b/bqskit/ext/__init__.py
index a6a4ee456..131eec208 100644
--- a/bqskit/ext/__init__.py
+++ b/bqskit/ext/__init__.py
@@ -73,6 +73,7 @@
 from bqskit.ext.supermarq import supermarq_liveness
 from bqskit.ext.supermarq import supermarq_parallelism
 from bqskit.ext.supermarq import supermarq_program_communication
+# TODO: Deprecate imports from __init__, use lazy import to deprecate
 
 
 __all__ = [
diff --git a/bqskit/ir/__init__.py b/bqskit/ir/__init__.py
index 10d0e4342..9959e3f04 100644
--- a/bqskit/ir/__init__.py
+++ b/bqskit/ir/__init__.py
@@ -62,6 +62,8 @@
 from bqskit.ir.interval import CycleInterval
 from bqskit.ir.interval import IntervalLike
 from bqskit.ir.iterator import CircuitIterator
+from bqskit.ir.lang import register_language as _register_language
+from bqskit.ir.lang.qasm2 import OPENQASM2Language as _qasm
 from bqskit.ir.location import CircuitLocation
 from bqskit.ir.location import CircuitLocationLike
 from bqskit.ir.operation import Operation
@@ -71,6 +73,11 @@
 from bqskit.ir.region import CircuitRegionLike
 from bqskit.ir.structure import CircuitStructure
 
+
+# Register supported languages
+_register_language('qasm', _qasm())
+
+
 __all__ = [
     'Operation',
     'Circuit',
diff --git a/bqskit/ir/circuit.py b/bqskit/ir/circuit.py
index 38eb40d1d..3c7fed2b7 100644
--- a/bqskit/ir/circuit.py
+++ b/bqskit/ir/circuit.py
@@ -3,6 +3,7 @@
 
 import copy
 import logging
+import pickle
 import warnings
 from typing import Any
 from typing import Callable
@@ -16,6 +17,7 @@
 from typing import Sequence
 from typing import TYPE_CHECKING
 
+import dill
 import numpy as np
 import numpy.typing as npt
 
@@ -1074,33 +1076,8 @@ def point(
 
         raise ValueError('No such operation exists in the circuit.')
 
-    def append(self, op: Operation) -> int:
-        """
-        Append `op` to the end of the circuit and return its cycle index.
-
-        Args:
-            op (Operation): The operation to append.
-
-        Returns:
-            int: The cycle index of the appended operation.
-
-        Raises:
-            ValueError: If `op` cannot be placed on the circuit due to
-                either an invalid location or gate radix mismatch.
-
-        Notes:
-            Due to the circuit being represented as a matrix,
-            `circuit.append(op)` does not imply `op` is last in simulation
-            order but it implies `op` is in the last cycle of circuit.
-
-        Examples:
-            >>> from bqskit.ir.gates import HGate
-            >>> circ = Circuit(1)
-            >>> op = Operation(HGate(), [0])
-            >>> circ.append(op) # Appends a Hadamard gate to qudit 0.
-        """
-        self.check_valid_operation(op)
-        cycle_index = self._find_available_or_append_cycle(op.location)
+    def _append(self, op: Operation, cycle_index: int) -> None:
+        """Append the operation to the circuit at the specified cycle."""
         point = CircuitPoint(cycle_index, op.location[0])
 
         prevs: dict[int, CircuitPoint | None] = {i: None for i in op.location}
@@ -1135,6 +1112,34 @@ def append(self, op: Operation) -> int:
             self._gate_info[op.gate] = 0
         self._gate_info[op.gate] += 1
 
+    def append(self, op: Operation) -> int:
+        """
+        Append `op` to the end of the circuit and return its cycle index.
+
+        Args:
+            op (Operation): The operation to append.
+
+        Returns:
+            int: The cycle index of the appended operation.
+
+        Raises:
+            ValueError: If `op` cannot be placed on the circuit due to
+                either an invalid location or gate radix mismatch.
+
+        Notes:
+            Due to the circuit being represented as a matrix,
+            `circuit.append(op)` does not imply `op` is last in simulation
+            order but it implies `op` is in the last cycle of circuit.
+
+        Examples:
+            >>> from bqskit.ir.gates import HGate
+            >>> circ = Circuit(1)
+            >>> op = Operation(HGate(), [0])
+            >>> circ.append(op) # Appends a Hadamard gate to qudit 0.
+        """
+        self.check_valid_operation(op)
+        cycle_index = self._find_available_or_append_cycle(op.location)
+        self._append(op, cycle_index)
         return cycle_index
 
     def append_gate(
@@ -2753,17 +2758,9 @@ def perform(
             :class:`~bqskit.compiler.compiler.Compiler` directly.
         """
         from bqskit.compiler.compiler import Compiler
-        from bqskit.compiler.passdata import PassData
-        from bqskit.compiler.task import CompilationTask
-
-        pass_data = PassData(self)
-        if data is not None:
-            pass_data.update(data)
 
         with Compiler() as compiler:
-            task = CompilationTask(self, [compiler_pass])
-            task.data = pass_data
-            task_id = compiler.submit(task)
+            task_id = compiler.submit(self, [compiler_pass], data=data)
             self.become(compiler.result(task_id))  # type: ignore
 
     def instantiate(
@@ -3277,4 +3274,73 @@ def from_operation(op: Operation) -> Circuit:
         circuit.append_gate(op.gate, list(range(circuit.num_qudits)), op.params)
         return circuit
 
+    def __reduce__(self) -> tuple[
+        Callable[
+            [int, tuple[int, ...], list[tuple[bool, bytes]], bytes],
+            Circuit,
+        ],
+        tuple[int, tuple[int, ...], list[tuple[bool, bytes]], bytes],
+    ]:
+        """Return the pickle state of the circuit."""
+        serialized_gates: list[tuple[bool, bytes]] = []
+        gate_table = {}
+        for gate in self.gate_set:
+            gate_table[gate] = len(serialized_gates)
+            if gate.__class__.__module__.startswith('bqskit'):
+                serialized_gates.append((False, pickle.dumps(gate)))
+            else:
+                serialized_gates.append((True, dill.dumps(gate, recurse=True)))
+
+        cycles: list[list[tuple[int, tuple[int, ...], list[float]]]] = []
+        last_cycle = -1
+        for cycle, op in self.operations_with_cycles():
+
+            if cycle != last_cycle:
+                last_cycle = cycle
+                cycles.append([])
+
+            marshalled_op = (
+                gate_table[op.gate],
+                op.location._location,
+                op.params,
+            )
+            cycles[-1].append(marshalled_op)
+
+        data = (
+            self.num_qudits,
+            self.radixes,
+            serialized_gates,
+            pickle.dumps(cycles),
+        )
+        return (rebuild_circuit, data)
+
     # endregion
+
+
+def rebuild_circuit(
+    num_qudits: int,
+    radixes: tuple[int, ...],
+    serialized_gates: list[tuple[bool, bytes]],
+    serialized_cycles: bytes,
+) -> Circuit:
+    """Rebuild a circuit from a pickle state."""
+    circuit = Circuit(num_qudits, radixes)
+
+    gate_table = {}
+    for i, (is_dill, serialized_gate) in enumerate(serialized_gates):
+        if is_dill:
+            gate = dill.loads(serialized_gate)
+        else:
+            gate = pickle.loads(serialized_gate)
+        gate_table[i] = gate
+
+    cycles = pickle.loads(serialized_cycles)
+    for i, cycle in enumerate(cycles):
+        circuit._append_cycle()
+        for marshalled_op in cycle:
+            gate = gate_table[marshalled_op[0]]
+            location = marshalled_op[1]
+            params = marshalled_op[2]
+            circuit._append(Operation(gate, location, params), i)
+
+    return circuit
diff --git a/bqskit/passes/control/paralleldo.py b/bqskit/passes/control/paralleldo.py
index 42b9bbeee..a95168058 100644
--- a/bqskit/passes/control/paralleldo.py
+++ b/bqskit/passes/control/paralleldo.py
@@ -34,7 +34,7 @@ def __init__(
         self,
         pass_sequences: Iterable[WorkflowLike],
         less_than: Callable[[Circuit, Circuit], bool],
-        pick_fisrt: bool = False,
+        pick_first: bool = False,
     ) -> None:
         """
         Construct a ParallelDo.
@@ -63,7 +63,7 @@ def __init__(
 
         self.workflows = [Workflow(p) for p in pass_sequences]
         self.less_than = less_than
-        self.pick_first = pick_fisrt
+        self.pick_first = pick_first
 
         if len(self.workflows) == 0:
             raise ValueError('Must specify at least one workflow.')
diff --git a/bqskit/passes/synthesis/pas.py b/bqskit/passes/synthesis/pas.py
index 2cad278d2..9a109f15a 100644
--- a/bqskit/passes/synthesis/pas.py
+++ b/bqskit/passes/synthesis/pas.py
@@ -113,6 +113,7 @@ async def synthesize(
             self.inner_synthesis.synthesize,
             targets,
             [data] * len(targets),
+            log_context=[{'perm': str(perm)} for perm in permsbyperms],
         )
 
         # Return best circuit
diff --git a/bqskit/runtime/__init__.py b/bqskit/runtime/__init__.py
index 31764cb46..75e3691c7 100644
--- a/bqskit/runtime/__init__.py
+++ b/bqskit/runtime/__init__.py
@@ -70,10 +70,8 @@
 :class:`RuntimeHandle`, which you can use to submit, map, wait on, and
 cancel tasks in the execution environment.
 
-For more information on how to design a custom pass, see this (TODO, sorry,
-you can look at the source code of existing
-`passes <https://github.com/BQSKit/bqskit/blob/master/bqskit/passes>`_
-for a good example for the time being).
+For more information on how to design a custom pass, see the following
+guide: :doc:`guides/custompass.md`.
 
 .. autosummary::
     :toctree: autogen
@@ -98,23 +96,30 @@
 from typing import Any
 from typing import Callable
 from typing import Protocol
+from typing import Sequence
 from typing import TYPE_CHECKING
 
-# Enable low-level fault handling: system crashes print a minimal trace.
-faulthandler.enable()
+if TYPE_CHECKING:
+    from bqskit.runtime.future import RuntimeFuture
 
 
-# Disable multi-threading in BLAS libraries.
-os.environ['OMP_NUM_THREADS'] = '1'
-os.environ['OPENBLAS_NUM_THREADS'] = '1'
-os.environ['MKL_NUM_THREADS'] = '1'
-os.environ['NUMEXPR_NUM_THREADS'] = '1'
-os.environ['VECLIB_MAXIMUM_THREADS'] = '1'
+# Enable low-level fault handling: system crashes print a minimal trace.
+faulthandler.enable()
 os.environ['RUST_BACKTRACE'] = '1'
 
 
-if TYPE_CHECKING:
-    from bqskit.runtime.future import RuntimeFuture
+# Control multi-threading in BLAS libraries.
+def set_blas_thread_counts(i: int = 1) -> None:
+    """
+    Control number of threads used by numpy and others.
+
+    Must be called before any numpy or other BLAS libraries are loaded.
+    """
+    os.environ['OMP_NUM_THREADS'] = str(i)
+    os.environ['OPENBLAS_NUM_THREADS'] = str(i)
+    os.environ['MKL_NUM_THREADS'] = str(i)
+    os.environ['NUMEXPR_NUM_THREADS'] = str(i)
+    os.environ['VECLIB_MAXIMUM_THREADS'] = str(i)
 
 
 class RuntimeHandle(Protocol):
@@ -137,18 +142,129 @@ def submit(
         self,
         fn: Callable[..., Any],
         *args: Any,
+        task_name: str | None = None,
+        log_context: dict[str, str] = {},
         **kwargs: Any,
     ) -> RuntimeFuture:
-        """Submit a `fn` to the runtime."""
+        """
+        Submit a function to the runtime for execution.
+
+        This method schedules the function `fn` to be executed by the
+        runtime with the provided arguments `args` and keyword arguments
+        `kwargs`. The execution may happen asynchronously.
+
+        Args:
+            fn (Callable[..., Any]): The function to be executed.
+
+            *args (Any): Variable length argument list to be passed to
+                the function `fn`.
+
+            task_name (str | None): An optional name for the task, which
+                can be used for logging or tracking purposes. Defaults to
+                None, which will use the function name as the task name.
+
+            log_context (dict[str, str]): A dictionary containing logging
+                context information. All log messages produced by the fn
+                and any children tasks will contain this context if the
+                appropriate level (logging.DEBUG) is set on the logger.
+                Defaults to an empty dictionary for no added context.
+
+            **kwargs (Any): Arbitrary keyword arguments to be passed to
+                the function `fn`.
+
+        Returns:
+            RuntimeFuture: An object representing the future result of
+            the function execution. This can be used to retrieve the
+            result by `await`ing it.
+
+        Example:
+            >>> from bqskit.runtime import get_runtime
+            >>>
+            >>> def add(x, y):
+            ...     return x + y
+            >>>
+            >>> future = get_runtime().submit(add, 1, 2)
+            >>> result = await future
+            >>> print(result)
+            3
+
+        See Also:
+            - :func:`map` for submitting multiple tasks in parallel.
+            - :func:`cancel` for cancelling tasks.
+            - :class:`~bqskit.runtime.future.RuntimeFuture` for more
+                information on how to interact with the future object.
+        """
         ...
 
     def map(
         self,
         fn: Callable[..., Any],
         *args: Any,
+        task_name: Sequence[str | None] | str | None = None,
+        log_context: Sequence[dict[str, str]] | dict[str, str] = {},
         **kwargs: Any,
     ) -> RuntimeFuture:
-        """Map `fn` over the input arguments distributed across the runtime."""
+        """
+        Map a function over a sequence of arguments and execute in parallel.
+
+        This method schedules the function `fn` to be executed by the runtime
+        for each set of arguments provided in `args`. Each invocation of `fn`
+        will be executed potentially in parallel, depending on the runtime's
+        capabilities and current load.
+
+        Args:
+            fn (Callable[..., Any]): The function to be executed.
+
+            *args (Any): Variable length argument list to be passed to
+                the function `fn`. Each argument is expected to be a
+                sequence of arguments to be passed to a separate
+                invocation. The sequences should be of equal length.
+
+            task_name (Sequence[str | None] | str | None): An optional
+                name for the task group, which can be used for logging
+                or tracking purposes. Defaults to None, which will use
+                the function name as the task name. If a string is
+                provided, it will be used as the prefix for all task
+                names. If a sequence of strings is provided, each task
+                will be named with the corresponding string in the
+                sequence.
+
+            log_context (Sequence[dict[str, str]]) | dict[str, str]): A
+                dictionary containing logging context information. All
+                log messages produced by the `fn` and any children tasks
+                will contain this context if the appropriate level
+                (logging.DEBUG) is set on the logger. Defaults to an
+                empty dictionary for no added context. Can be a sequence
+                of contexts, one for each task, or a single context to be
+                used for all tasks.
+
+            **kwargs (Any): Arbitrary keyword arguments to be passed to
+                each invocation of the function `fn`.
+
+        Returns:
+            RuntimeFuture: An object representing the future result of
+            the function executions. This can be used to retrieve the
+            results by `await`ing it, which will return a list.
+
+        Example:
+            >>> from bqskit.runtime import get_runtime
+            >>>
+            >>> def add(x, y):
+            ...     return x + y
+            >>>
+            >>> args_list = [(1, 2, 3), (4, 5, 6)]
+            >>> future = get_runtime().map(add, *args_list)
+            >>> results = await future
+            >>> print(results)
+            [5, 7, 9]
+
+        See Also:
+            - :func:`submit` for submitting a single task.
+            - :func:`cancel` for cancelling tasks.
+            - :func:`next` for retrieving results incrementally.
+            - :class:`~bqskit.runtime.future.RuntimeFuture` for more
+                information on how to interact with the future object.
+        """
         ...
 
     def cancel(self, future: RuntimeFuture) -> None:
diff --git a/bqskit/runtime/attached.py b/bqskit/runtime/attached.py
index 6cd6a6a98..f68be0ee3 100644
--- a/bqskit/runtime/attached.py
+++ b/bqskit/runtime/attached.py
@@ -15,6 +15,9 @@
 from bqskit.runtime.direction import MessageDirection
 
 
+_logger = logging.getLogger(__name__)
+
+
 class AttachedServer(DetachedServer):
     """
     BQSKit Runtime Server in attached mode.
@@ -33,6 +36,8 @@ def __init__(
         num_workers: int = -1,
         port: int = default_server_port,
         worker_port: int = default_worker_port,
+        log_level: int = logging.WARNING,
+        num_blas_threads: int = 1,
     ) -> None:
         """
         Create a server with `num_workers` workers.
@@ -49,7 +54,24 @@ def __init__(
             worker_port (int): The port this server will listen for workers
                 on. Default can be found in the
                 :obj:`~bqskit.runtime.default_worker_port` global variable.
+
+            log_level (int): The logging level for the server and workers.
+                (Default: logging.WARNING).
+
+            num_blas_threads (int): The number of threads to use in BLAS
+                libraries. (Default: 1).
         """
+        # Initialize runtime logging
+        logging.getLogger().setLevel(log_level)
+        _handler = logging.StreamHandler()
+        _handler.setLevel(0)
+        _fmt_header = '%(asctime)s.%(msecs)03d - %(levelname)-8s |'
+        _fmt_message = ' %(module)s: %(message)s'
+        _fmt = _fmt_header + _fmt_message
+        _formatter = logging.Formatter(_fmt, '%H:%M:%S')
+        _handler.setFormatter(_formatter)
+        logging.getLogger().addHandler(_handler)
+
         ServerBase.__init__(self)
 
         # See DetachedServer for more info on the following fields:
@@ -59,9 +81,6 @@ def __init__(
         self.mailboxes: dict[int, ServerMailbox] = {}
         self.mailbox_counter = 0
 
-        # Start workers
-        self.spawn_workers(num_workers, worker_port)
-
         # Connect to client
         client_conn = self.listen_once('localhost', port)
         self.clients[client_conn] = set()
@@ -70,24 +89,23 @@ def __init__(
             selectors.EVENT_READ,
             MessageDirection.CLIENT,
         )
-        self.logger.info('Connected to client.')
+        _logger.info('Connected to client.')
+
+        # Start workers
+        self.spawn_workers(
+            num_workers,
+            worker_port,
+            log_level,
+            num_blas_threads,
+        )
 
     def handle_disconnect(self, conn: Connection) -> None:
         """A client disconnect in attached mode is equal to a shutdown."""
         self.handle_shutdown()
 
 
-def start_attached_server(
-    num_workers: int,
-    log_level: int,
-    **kwargs: Any,
-) -> None:
+def start_attached_server(num_workers: int, **kwargs: Any) -> None:
     """Start a runtime server in attached mode."""
-    # Initialize runtime logging
-    _logger = logging.getLogger('bqskit-runtime')
-    _logger.setLevel(log_level)
-    _logger.addHandler(logging.StreamHandler())
-
     # Initialize the server
     server = AttachedServer(num_workers, **kwargs)
 
diff --git a/bqskit/runtime/base.py b/bqskit/runtime/base.py
index 17cdf2747..d10996e49 100644
--- a/bqskit/runtime/base.py
+++ b/bqskit/runtime/base.py
@@ -25,6 +25,7 @@
 
 from bqskit.runtime import default_manager_port
 from bqskit.runtime import default_worker_port
+from bqskit.runtime import set_blas_thread_counts
 from bqskit.runtime.address import RuntimeAddress
 from bqskit.runtime.direction import MessageDirection
 from bqskit.runtime.message import RuntimeMessage
@@ -33,56 +34,88 @@
 from bqskit.runtime.worker import start_worker
 
 
+_logger = logging.getLogger(__name__)
+
+
 class RuntimeEmployee:
     """Data structure for a boss's view of an employee."""
 
     def __init__(
         self,
+        id: int,
         conn: Connection,
         total_workers: int,
         process: Process | None = None,
-        num_tasks: int = 0,
+        is_manager: bool = False,
     ) -> None:
         """Construct an employee with all resources idle."""
+
+        self.id = id
+        """
+        The ID of the employee.
+
+        If this is a worker, then their unique worker id. If this is a manager,
+        then their local id.
+        """
+
         self.conn: Connection = conn
         self.total_workers = total_workers
         self.process = process
-        self.num_tasks = num_tasks
+        self.num_tasks = 0
         self.num_idle_workers = total_workers
+        self.is_manager = is_manager
 
-    def shutdown(self) -> None:
-        """Shutdown the employee."""
+        self.submit_cache: list[tuple[RuntimeAddress, int]] = []
+        """
+        Tracks recently submitted tasks by id and count.
+
+        This is used to adjust the idle worker count when the employee sends a
+        waiting message.
+        """
+
+    def initiate_shutdown(self) -> None:
+        """Instruct employee to shutdown."""
         try:
             self.conn.send((RuntimeMessage.SHUTDOWN, None))
         except Exception:
             pass
 
+    def complete_shutdown(self) -> None:
+        """Ensure employee is shutdown and clean up resources."""
         if self.process is not None:
             self.process.join()
 
         self.process = None
         self.conn.close()
 
+    def shutdown(self) -> None:
+        """Initiate and complete shutdown."""
+        self.initiate_shutdown()
+        self.complete_shutdown()
+
+    @property
+    def recipient_string(self) -> str:
+        """Return a string representation of the employee."""
+        return f'{"Manager" if self.is_manager else "Worker"} {self.id}'
+
     @property
     def has_idle_resources(self) -> bool:
         return self.num_idle_workers > 0
 
+    def get_num_of_tasks_sent_since(
+        self,
+        read_receipt: RuntimeAddress | None,
+    ) -> int:
+        """Return the number of tasks sent since the read receipt."""
+        if read_receipt is None:
+            return sum(count for _, count in self.submit_cache)
 
-def send_outgoing(node: ServerBase) -> None:
-    """Outgoing thread forwards messages as they are created."""
-    while True:
-        outgoing = node.outgoing.get()
-
-        if not node.running:
-            # NodeBase's handle_shutdown will put a dummy value in the
-            # queue to wake the thread up so it can exit safely.
-            # Hence the node.running check now rather than in the
-            # while condition.
-            break
+        for i, (addr, _) in enumerate(self.submit_cache):
+            if addr == read_receipt:
+                self.submit_cache = self.submit_cache[i:]
+                return sum(count for _, count in self.submit_cache[1:])
 
-        outgoing[0].send((outgoing[1], outgoing[2]))
-        node.logger.debug(f'Sent message {outgoing[1].name}.')
-        node.logger.log(1, f'{outgoing[2]}\n')
+        raise RuntimeError('Read receipt not found in submit cache.')
 
 
 def sigint_handler(signum: int, _: FrameType | None, node: ServerBase) -> None:
@@ -92,7 +125,7 @@ def sigint_handler(signum: int, _: FrameType | None, node: ServerBase) -> None:
 
     node.running = False
     node.terminate_hotline.send(b'\0')
-    node.logger.info('Server interrupted.')
+    _logger.info('Server interrupted.')
 
 
 class ServerBase:
@@ -122,24 +155,24 @@ def __init__(self) -> None:
         self.sel.register(p, selectors.EVENT_READ, MessageDirection.SIGNAL)
         """Terminate hotline is used to unblock select while running."""
 
-        self.logger = logging.getLogger('bqskit-runtime')
-        """Logger used to print operational log messages."""
-
         self.employees: list[RuntimeEmployee] = []
         """Tracks this node's employees, which are managers or workers."""
 
         self.conn_to_employee_dict: dict[Connection, RuntimeEmployee] = {}
         """Used to find the employee associated with a message."""
 
+        # Servers do not need blas threads
+        set_blas_thread_counts(1)
+
         # Safely and immediately exit on interrupt signals
         handle = functools.partial(sigint_handler, node=self)
         signal.signal(signal.SIGINT, handle)
 
         # Start outgoing thread
         self.outgoing: Queue[tuple[Connection, RuntimeMessage, Any]] = Queue()
-        self.outgoing_thread = Thread(target=send_outgoing, args=(self,))
+        self.outgoing_thread = Thread(target=self.send_outgoing, daemon=True)
         self.outgoing_thread.start()
-        self.logger.info('Started outgoing thread.')
+        _logger.info('Started outgoing thread.')
 
     def connect_to_managers(self, ipports: Sequence[tuple[str, int]]) -> None:
         """Connect to all managers given by endpoints in `ipports`."""
@@ -155,26 +188,33 @@ def connect_to_managers(self, ipports: Sequence[tuple[str, int]]) -> None:
                 self.upper_id_bound,
             )
             manager_conns.append(self.connect_to_manager(ip, port, lb, ub))
-            self.logger.info(f'Connected to manager {i} at {ip}:{port}.')
-            self.logger.debug(f'Gave bounds {lb=} and {ub=} to manager {i}.')
+            _logger.info(f'Connected to manager {i} at {ip}:{port}.')
+            _logger.debug(f'Gave bounds {lb=} and {ub=} to manager {i}.')
 
         # Wait for started messages from all managers and register them
         self.total_workers = 0
         for i, conn in enumerate(manager_conns):
             msg, num_workers = conn.recv()
             assert msg == RuntimeMessage.STARTED
-            self.employees.append(RuntimeEmployee(conn, num_workers))
+            self.employees.append(
+                RuntimeEmployee(
+                    i,
+                    conn,
+                    num_workers,
+                    is_manager=True,
+                ),
+            )
             self.conn_to_employee_dict[conn] = self.employees[-1]
             self.sel.register(
                 conn,
                 selectors.EVENT_READ,
                 MessageDirection.BELOW,
             )
-            self.logger.info(f'Registered manager {i} with {num_workers=}.')
+            _logger.info(f'Registered manager {i} with {num_workers=}.')
             self.total_workers += num_workers
         self.num_idle_workers = self.total_workers
 
-        self.logger.info(f'Node has {self.total_workers} total workers.')
+        _logger.info(f'Node has {self.total_workers} total workers.')
 
     def connect_to_manager(
         self,
@@ -216,6 +256,8 @@ def spawn_workers(
         self,
         num_workers: int = -1,
         port: int = default_worker_port,
+        logging_level: int = logging.WARNING,
+        num_blas_threads: int = 1,
     ) -> None:
         """
         Spawn worker processes.
@@ -228,6 +270,11 @@ def spawn_workers(
             port (int): The port this server will listen for workers on.
                 Default can be found in the
                 :obj:`~bqskit.runtime.default_worker_port` global variable.
+
+            logging_level (int): The logging level for the workers.
+
+            num_blas_threads (int): The number of threads to use in BLAS
+                libraries. (Default: 1).
         """
         if num_workers == -1:
             oscount = os.cpu_count()
@@ -240,9 +287,17 @@ def spawn_workers(
         procs = {}
         for i in range(num_workers):
             w_id = self.lower_id_bound + i
-            procs[w_id] = Process(target=start_worker, args=(w_id, port))
+            procs[w_id] = Process(
+                target=start_worker,
+                args=(w_id, port),
+                kwargs={
+                    'logging_level': logging_level,
+                    'num_blas_threads': num_blas_threads,
+                },
+            )
+            procs[w_id].daemon = True
             procs[w_id].start()
-            self.logger.debug(f'Stated worker process {i}.')
+            _logger.debug(f'Stated worker process {i}.')
 
         # Listen for the worker connections
         family = 'AF_INET' if sys.platform == 'win32' else None
@@ -255,7 +310,7 @@ def spawn_workers(
         for i, conn in enumerate(conns):
             msg, w_id = conn.recv()
             assert msg == RuntimeMessage.STARTED
-            employee = RuntimeEmployee(conn, 1, procs[w_id])
+            employee = RuntimeEmployee(w_id, conn, 1, procs[w_id])
             temp_reorder[w_id - self.lower_id_bound] = employee
             self.conn_to_employee_dict[conn] = employee
 
@@ -264,18 +319,18 @@ def spawn_workers(
             self.employees.append(temp_reorder[i])
 
         # Register employee communication
-        for i, employee in enumerate(self.employees):
+        for employee in self.employees:
             self.sel.register(
                 employee.conn,
                 selectors.EVENT_READ,
                 MessageDirection.BELOW,
             )
-            self.logger.info(f'Registered worker {i}.')
+            _logger.debug(f'Registered worker {employee.id}.')
 
         self.step_size = 1
         self.total_workers = num_workers
         self.num_idle_workers = num_workers
-        self.logger.info(f'Node has spawned {num_workers} workers.')
+        _logger.info(f'Node has spawned {num_workers} workers.')
 
     def connect_to_workers(
         self,
@@ -298,7 +353,7 @@ def connect_to_workers(
             oscount = os.cpu_count()
             num_workers = oscount if oscount else 1
 
-        self.logger.info(f'Expecting {num_workers} worker connections.')
+        _logger.info(f'Expecting {num_workers} worker connections.')
 
         if self.lower_id_bound + num_workers >= self.upper_id_bound:
             raise RuntimeError('Insufficient id range for workers.')
@@ -312,25 +367,25 @@ def connect_to_workers(
         for i, conn in enumerate(conns):
             w_id = self.lower_id_bound + i
             self.outgoing.put((conn, RuntimeMessage.STARTED, w_id))
-            employee = RuntimeEmployee(conn, 1)
+            employee = RuntimeEmployee(w_id, conn, 1)
             self.employees.append(employee)
             self.conn_to_employee_dict[conn] = employee
 
         # Register employee communication
-        for i, employee in enumerate(self.employees):
-            w_id = self.lower_id_bound + i
+        for employee in self.employees:
+            w_id = employee.id
             assert employee.conn.recv() == (RuntimeMessage.STARTED, w_id)
             self.sel.register(
                 employee.conn,
                 selectors.EVENT_READ,
                 MessageDirection.BELOW,
             )
-            self.logger.info(f'Registered worker {i}.')
+            _logger.info(f'Registered worker {w_id}.')
 
         self.step_size = 1
         self.total_workers = num_workers
         self.num_idle_workers = num_workers
-        self.logger.info(f'Node has connected to {num_workers} workers.')
+        _logger.info(f'Node has connected to {num_workers} workers.')
 
     def listen_once(self, ip: str, port: int) -> Connection:
         """Listen on `ip`:`port` for a connection and return on first one."""
@@ -340,9 +395,36 @@ def listen_once(self, ip: str, port: int) -> Connection:
         listener.close()
         return conn
 
+    def send_outgoing(self) -> None:
+        """Outgoing thread forwards messages as they are created."""
+        while True:
+            outgoing = self.outgoing.get()
+
+            if not self.running:
+                # NodeBase's handle_shutdown will put a dummy value in the
+                # queue to wake the thread up so it can exit safely.
+                # Hence the node.running check now rather than in the
+                # while condition.
+                break
+
+            if outgoing[0].closed:
+                continue
+
+            outgoing[0].send((outgoing[1], outgoing[2]))
+            if _logger.isEnabledFor(logging.DEBUG):
+                to = self.get_to_string(outgoing[0])
+                _logger.debug(f'Sent message {outgoing[1].name} to {to}.')
+
+            if outgoing[1] == RuntimeMessage.SUBMIT_BATCH:
+                _logger.log(1, f'[{outgoing[2][0]}] * {len(outgoing[2])}\n')
+            else:
+                _logger.log(1, f'{outgoing[2]}\n')
+
+            self.outgoing.task_done()
+
     def run(self) -> None:
         """Main loop."""
-        self.logger.info(f'{self.__class__.__name__} running...')
+        _logger.info(f'{self.__class__.__name__} running...')
 
         try:
             while self.running:
@@ -356,7 +438,7 @@ def run(self) -> None:
 
                     # If interrupted by signal, shutdown and exit
                     if direction == MessageDirection.SIGNAL:
-                        self.logger.debug('Received interrupt signal.')
+                        _logger.debug('Received interrupt signal.')
                         self.handle_shutdown()
                         return
 
@@ -367,8 +449,11 @@ def run(self) -> None:
                         self.handle_disconnect(conn)
                         continue
                     log = f'Received message {msg.name} from {direction.name}.'
-                    self.logger.debug(log)
-                    self.logger.log(1, f'{payload}\n')
+                    _logger.debug(log)
+                    if msg == RuntimeMessage.SUBMIT_BATCH:
+                        _logger.log(1, f'[{payload[0]}] * {len(payload)}\n')
+                    else:
+                        _logger.log(1, f'{payload}\n')
 
                     # Handle message
                     self.handle_message(msg, direction, conn, payload)
@@ -376,7 +461,7 @@ def run(self) -> None:
         except Exception:
             exc_info = sys.exc_info()
             error_str = ''.join(traceback.format_exception(*exc_info))
-            self.logger.error(error_str)
+            _logger.error(error_str)
             self.handle_system_error(error_str)
 
         finally:
@@ -412,27 +497,35 @@ def handle_system_error(self, error_str: str) -> None:
         RuntimeTask's coroutine code.
         """
 
+    @abc.abstractmethod
+    def get_to_string(self, conn: Connection) -> str:
+        """Return a string representation of the connection."""
+
     def handle_shutdown(self) -> None:
         """Shutdown the node and release resources."""
         # Stop running
-        self.logger.info('Shutting down node.')
+        _logger.info('Shutting down node.')
         self.running = False
 
         # Instruct employees to shutdown
         for employee in self.employees:
-            employee.shutdown()
+            employee.initiate_shutdown()
+
+        for employee in self.employees:
+            employee.complete_shutdown()
+
         self.employees.clear()
-        self.logger.debug('Shutdown employees.')
+        _logger.debug('Shutdown employees.')
 
         # Close selector
         self.sel.close()
-        self.logger.debug('Cleared selector.')
+        _logger.debug('Cleared selector.')
 
         # Close outgoing thread
         if self.outgoing_thread.is_alive():
             self.outgoing.put(b'\0')  # type: ignore
             self.outgoing_thread.join()
-            self.logger.debug('Joined outgoing thread.')
+            _logger.debug('Joined outgoing thread.')
             assert not self.outgoing_thread.is_alive()
 
     def handle_disconnect(self, conn: Connection) -> None:
@@ -444,10 +537,6 @@ def handle_disconnect(self, conn: Connection) -> None:
         if conn in self.conn_to_employee_dict:
             self.handle_shutdown()
 
-    def __del__(self) -> None:
-        """Ensure resources are cleaned up."""
-        self.handle_shutdown()
-
     def assign_tasks(
         self,
         tasks: Sequence[RuntimeTask],
@@ -513,10 +602,13 @@ def schedule_tasks(self, tasks: Sequence[RuntimeTask]) -> None:
         """Schedule tasks between this node's employees."""
         if len(tasks) == 0:
             return
-
-        assignments = self.assign_tasks(tasks)
-
-        for e, assignment in zip(self.employees, assignments):
+        assignments = zip(self.employees, self.assign_tasks(tasks))
+        sorted_assignments = sorted(
+            assignments,
+            key=lambda x: x[0].num_idle_workers,
+            reverse=True,
+        )  # Employees with the most idle workers get assignments first
+        for e, assignment in sorted_assignments:
             num_tasks = len(assignment)
 
             if num_tasks == 0:
@@ -526,6 +618,7 @@ def schedule_tasks(self, tasks: Sequence[RuntimeTask]) -> None:
 
             e.num_tasks += num_tasks
             e.num_idle_workers -= min(num_tasks, e.num_idle_workers)
+            e.submit_cache.append((assignment[0].unique_id, num_tasks))
 
         self.num_idle_workers = sum(e.num_idle_workers for e in self.employees)
 
@@ -549,28 +642,41 @@ def get_employee_responsible_for(self, worker_id: int) -> RuntimeEmployee:
         employee_id = (worker_id - self.lower_id_bound) // self.step_size
         return self.employees[employee_id]
 
-    def broadcast_cancel(self, addr: RuntimeAddress) -> None:
+    def broadcast(self, msg: RuntimeMessage, payload: Any) -> None:
         """Broadcast a cancel message to my employees."""
         for employee in self.employees:
-            self.outgoing.put((employee.conn, RuntimeMessage.CANCEL, addr))
+            self.outgoing.put((employee.conn, msg, payload))
+
+    def handle_importpath(self, paths: list[str]) -> None:
+        """Update the system path with the given paths."""
+        for path in paths:
+            if path not in sys.path:
+                sys.path.append(path)
+        self.broadcast(RuntimeMessage.IMPORTPATH, paths)
 
-    def handle_waiting(self, conn: Connection, new_idle_count: int) -> None:
+    def handle_waiting(
+        self,
+        conn: Connection,
+        new_idle_count: int,
+        read_receipt: RuntimeAddress | None,
+    ) -> None:
         """
         Record that an employee is idle with nothing to do.
 
-        There is a race condition here that is allowed. If an employee
-        sends a waiting message at the same time that this sends it a
-        task, it will still be marked waiting even though it is running
-        a task. We allow this for two reasons. First, the consequences are
-        minimal: this situation can only lead to one extra task assigned
-        to the worker that could otherwise go to a truly idle worker.
-        Second, it is unlikely in the common BQSKit workflows, which have
-        wide and shallow task graphs and each leaf task can require seconds
-        of runtime.
+        There is a race condition that is corrected here. If an employee sends a
+        waiting message at the same time that its boss sends it a task, the
+        boss's idle count will eventually be incorrect. To fix this, every
+        waiting message sent by an employee is accompanied by a read receipt of
+        the latest batch of tasks it has processed. The boss can then adjust the
+        idle count by the number of tasks sent since the read receipt.
         """
-        old_count = self.conn_to_employee_dict[conn].num_idle_workers
-        self.conn_to_employee_dict[conn].num_idle_workers = new_idle_count
-        self.num_idle_workers += (new_idle_count - old_count)
+        employee = self.conn_to_employee_dict[conn]
+        unaccounted_task = employee.get_num_of_tasks_sent_since(read_receipt)
+        adjusted_idle_count = max(new_idle_count - unaccounted_task, 0)
+
+        old_count = employee.num_idle_workers
+        employee.num_idle_workers = adjusted_idle_count
+        self.num_idle_workers += (adjusted_idle_count - old_count)
         assert 0 <= self.num_idle_workers <= self.total_workers
 
 
diff --git a/bqskit/runtime/detached.py b/bqskit/runtime/detached.py
index 3b817ef1a..8740c7170 100644
--- a/bqskit/runtime/detached.py
+++ b/bqskit/runtime/detached.py
@@ -15,10 +15,10 @@
 from typing import Any
 from typing import cast
 from typing import List
+from typing import Optional
 from typing import Sequence
+from typing import Tuple
 
-from bqskit.compiler.status import CompilationStatus
-from bqskit.compiler.task import CompilationTask
 from bqskit.runtime import default_server_port
 from bqskit.runtime.address import RuntimeAddress
 from bqskit.runtime.base import import_tests_package
@@ -30,25 +30,7 @@
 from bqskit.runtime.task import RuntimeTask
 
 
-def listen(server: DetachedServer, port: int) -> None:
-    """Listening thread listens for client connections."""
-    listener = Listener(('0.0.0.0', port))
-    while server.running:
-        client = listener.accept()
-
-        if server.running:
-            # We check again that the server is running before registering
-            # the client because dummy data is sent to unblock
-            # listener.accept() during server shutdown
-            server.clients[client] = set()
-            server.sel.register(
-                client,
-                selectors.EVENT_READ,
-                MessageDirection.CLIENT,
-            )
-            server.logger.debug('Connected and registered new client.')
-
-    listener.close()
+_logger = logging.getLogger(__name__)
 
 
 @dataclass
@@ -115,9 +97,30 @@ def __init__(
 
         # Start client listener
         self.port = port
-        self.listen_thread = Thread(target=listen, args=(self, port))
+        self.listen_thread = Thread(target=self.listen, args=(port,))
+        self.listen_thread.daemon = True
         self.listen_thread.start()
-        self.logger.info(f'Started client listener on port {self.port}.')
+        _logger.info(f'Started client listener on port {self.port}.')
+
+    def listen(self, port: int) -> None:
+        """Listening thread listens for client connections."""
+        listener = Listener(('0.0.0.0', port))
+        while self.running:
+            client = listener.accept()
+
+            if self.running:
+                # We check again that the server is running before registering
+                # the client because dummy data is sent to unblock
+                # listener.accept() during server shutdown
+                self.clients[client] = set()
+                self.sel.register(
+                    client,
+                    selectors.EVENT_READ,
+                    MessageDirection.CLIENT,
+                )
+                _logger.debug('Connected and registered new client.')
+
+        listener.close()
 
     def handle_message(
         self,
@@ -130,14 +133,14 @@ def handle_message(
         if direction == MessageDirection.CLIENT:
 
             if msg == RuntimeMessage.CONNECT:
-                pass
+                paths = cast(List[str], payload)
+                self.handle_connect(conn, paths)
 
             elif msg == RuntimeMessage.DISCONNECT:
                 self.handle_disconnect(conn)
 
             elif msg == RuntimeMessage.SUBMIT:
-                ctask = cast(CompilationTask, payload)
-                self.handle_new_comp_task(conn, ctask)
+                self.handle_new_comp_task(conn, payload)
 
             elif msg == RuntimeMessage.REQUEST:
                 request = cast(uuid.UUID, payload)
@@ -176,25 +179,34 @@ def handle_message(
                 self.handle_log(payload)
 
             elif msg == RuntimeMessage.CANCEL:
-                self.broadcast_cancel(payload)
+                self.broadcast(msg, payload)
 
             elif msg == RuntimeMessage.SHUTDOWN:
                 self.handle_shutdown()
 
             elif msg == RuntimeMessage.WAITING:
-                num_idle = cast(int, payload)
-                self.handle_waiting(conn, num_idle)
+                p = cast(Tuple[int, Optional[RuntimeAddress]], payload)
+                num_idle, read_receipt = p
+                self.handle_waiting(conn, num_idle, read_receipt)
 
             elif msg == RuntimeMessage.UPDATE:
                 task_diff = cast(int, payload)
                 self.conn_to_employee_dict[conn].num_tasks += task_diff
 
+            elif msg == RuntimeMessage.COMMUNICATE:
+                self.broadcast(msg, payload)
+
             else:
                 raise RuntimeError(f'Unexpected message type: {msg.name}')
 
         else:
             raise RuntimeError(f'Unexpected message from {direction.name}.')
 
+    def handle_connect(self, conn: Connection, paths: list[str]) -> None:
+        """Handle a client connection request."""
+        self.handle_importpath(paths)
+        self.outgoing.put((conn, RuntimeMessage.READY, None))
+
     def handle_system_error(self, error_str: str) -> None:
         """
         Handle an error in runtime code as opposed to client code.
@@ -208,6 +220,13 @@ def handle_system_error(self, error_str: str) -> None:
         # Sleep to ensure clients receive error message before shutdown
         time.sleep(1)
 
+    def get_to_string(self, conn: Connection) -> str:
+        """Return a string representation of the connection."""
+        if conn in self.clients:
+            return 'CLIENT'
+
+        return self.conn_to_employee_dict[conn].recipient_string
+
     def handle_shutdown(self) -> None:
         """Shutdown the runtime."""
         super().handle_shutdown()
@@ -219,7 +238,7 @@ def handle_shutdown(self) -> None:
             except Exception:
                 pass
         self.clients.clear()
-        self.logger.debug('Cleared clients.')
+        _logger.debug('Cleared clients.')
 
         # Close listener (hasattr checked for attachedserver shutdown)
         if hasattr(self, 'listen_thread') and self.listen_thread.is_alive():
@@ -231,7 +250,7 @@ def handle_shutdown(self) -> None:
             dummy_socket.connect(('localhost', self.port))
             dummy_socket.close()
             self.listen_thread.join()
-            self.logger.debug('Joined listening thread.')
+            _logger.debug('Joined listening thread.')
 
     def handle_disconnect(self, conn: Connection) -> None:
         """Disconnect a client connection from the runtime."""
@@ -239,19 +258,20 @@ def handle_disconnect(self, conn: Connection) -> None:
         tasks = self.clients.pop(conn)
         for task_id in tasks:
             self.handle_cancel_comp_task(task_id)
-        self.logger.info('Unregistered client.')
+        _logger.info('Unregistered client.')
 
     def handle_new_comp_task(
         self,
         conn: Connection,
-        task: CompilationTask,
+        task: Any,  # Explicitly not CompilationTask to avoid early import
     ) -> None:
         """Convert a :class:`CompilationTask` into an internal one."""
+        from bqskit.compiler.task import CompilationTask
         mailbox_id = self._get_new_mailbox_id()
         self.tasks[task.task_id] = (mailbox_id, conn)
         self.mailbox_to_task_dict[mailbox_id] = task.task_id
         self.mailboxes[mailbox_id] = ServerMailbox()
-        self.logger.info(f'New CompilationTask: {task.task_id}.')
+        _logger.info(f'New CompilationTask: {task.task_id}.')
 
         self.clients[conn].add(task.task_id)
 
@@ -279,7 +299,7 @@ def handle_request(self, conn: Connection, request: uuid.UUID) -> None:
 
         if box.ready:
             # If the result has already arrived, ship it to the client.
-            self.logger.info(f'Responding to request for task {request}.')
+            _logger.info(f'Responding to request for task {request}.')
             self.outgoing.put((conn, RuntimeMessage.RESULT, box.result))
             self.mailboxes.pop(mailbox_id)
             self.clients[conn].remove(request)
@@ -294,6 +314,7 @@ def handle_request(self, conn: Connection, request: uuid.UUID) -> None:
 
     def handle_status(self, conn: Connection, request: uuid.UUID) -> None:
         """Inform the client if the task is finished or not."""
+        from bqskit.compiler.status import CompilationStatus
         if request not in self.clients[conn] or request not in self.tasks:
             # This task is unknown to the system
             m = (conn, RuntimeMessage.STATUS, CompilationStatus.UNKNOWN)
@@ -309,7 +330,7 @@ def handle_status(self, conn: Connection, request: uuid.UUID) -> None:
 
     def handle_cancel_comp_task(self, request: uuid.UUID) -> None:
         """Cancel a compilation task in the system."""
-        self.logger.info(f'Cancelling: {request}.')
+        _logger.info(f'Cancelling: {request}.')
 
         # Remove task from server data
         mailbox_id, client_conn = self.tasks[request]
@@ -319,7 +340,7 @@ def handle_cancel_comp_task(self, request: uuid.UUID) -> None:
 
         # Forward internal cancel messages
         addr = RuntimeAddress(-1, mailbox_id, 0)
-        self.broadcast_cancel(addr)
+        self.broadcast(RuntimeMessage.CANCEL, addr)
 
         # Acknowledge the client's cancel request
         if not client_conn.closed:
@@ -340,10 +361,10 @@ def handle_result(self, result: RuntimeResult) -> None:
             box = self.mailboxes[mailbox_id]
             box.result = result.result
             t_id = self.mailbox_to_task_dict[mailbox_id]
-            self.logger.info(f'Finished: {t_id}.')
+            _logger.info(f'Finished: {t_id}.')
 
             if box.client_waiting:
-                self.logger.info(f'Responding to request for task {t_id}.')
+                _logger.info(f'Responding to request for task {t_id}.')
                 m = (self.tasks[t_id][1], RuntimeMessage.RESULT, box.result)
                 self.outgoing.put(m)
                 self.clients[self.tasks[t_id][1]].remove(t_id)
@@ -367,6 +388,12 @@ def handle_error(self, error_payload: tuple[int, str]) -> None:
         tid = error_payload[0]
         conn = self.tasks[self.mailbox_to_task_dict[tid]][1]
         self.outgoing.put((conn, RuntimeMessage.ERROR, error_payload[1]))
+        # TODO: Broadcast cancel to all tasks with compilation task id tid
+        # But avoid double broadcasting it. If the client crashes due to
+        # this error, which it may not, then we will quickly process
+        # a handle_disconnect and call the cancel anyways. We should
+        # still cancel here incase the client catches the error and
+        # resubmits a job.
 
     def handle_log(self, log_payload: tuple[int, LogRecord]) -> None:
         """Forward logs to appropriate client."""
@@ -413,9 +440,16 @@ def start_server() -> None:
     ipports = parse_ipports(args.managers)
 
     # Set up logging
-    _logger = logging.getLogger('bqskit-runtime')
-    _logger.setLevel([30, 20, 10, 1][min(args.verbose, 3)])
-    _logger.addHandler(logging.StreamHandler())
+    log_level = [30, 20, 10, 1][min(args.verbose, 3)]
+    logging.getLogger().setLevel(log_level)
+    _handler = logging.StreamHandler()
+    _handler.setLevel(0)
+    _fmt_header = '%(asctime)s.%(msecs)03d - %(levelname)-8s |'
+    _fmt_message = ' %(module)s: %(message)s'
+    _fmt = _fmt_header + _fmt_message
+    _formatter = logging.Formatter(_fmt, '%H:%M:%S')
+    _handler.setFormatter(_formatter)
+    logging.getLogger().addHandler(_handler)
 
     # Import tests package recursively
     if args.import_tests:
diff --git a/bqskit/runtime/future.py b/bqskit/runtime/future.py
index 70f6ac2cc..ab69d5911 100644
--- a/bqskit/runtime/future.py
+++ b/bqskit/runtime/future.py
@@ -27,8 +27,8 @@ def __await__(self) -> Any:
 
         Informs the event loop which mailbox this is waiting on.
         """
-        if self._next_flag:
-            return (yield self)
+        # if self._next_flag:
+        #     return (yield self)
 
         return (yield self)
 
diff --git a/bqskit/runtime/manager.py b/bqskit/runtime/manager.py
index 7779f47ca..14827af9e 100644
--- a/bqskit/runtime/manager.py
+++ b/bqskit/runtime/manager.py
@@ -9,7 +9,9 @@
 from typing import Any
 from typing import cast
 from typing import List
+from typing import Optional
 from typing import Sequence
+from typing import Tuple
 
 from bqskit.runtime import default_manager_port
 from bqskit.runtime import default_worker_port
@@ -23,6 +25,9 @@
 from bqskit.runtime.task import RuntimeTask
 
 
+_logger = logging.getLogger(__name__)
+
+
 class Manager(ServerBase):
     """
     BQSKit Runtime Manager.
@@ -42,6 +47,8 @@ def __init__(
         ipports: list[tuple[str, int]] | None = None,
         worker_port: int = default_worker_port,
         only_connect: bool = False,
+        log_level: int = logging.WARNING,
+        num_blas_threads: int = 1,
     ) -> None:
         """
         Create a manager instance in one of two ways:
@@ -78,6 +85,16 @@ def __init__(
 
             only_connect (bool): If true, do not spawn workers, only connect
                 to already spawned workers.
+
+            log_level (int): The logging level for the manager and workers.
+                If `only_connect` is True, doesn't set worker's log level.
+                In that case, set the worker's log level when spawning them.
+                (Default: logging.WARNING).
+
+            num_blas_threads (int): The number of threads to use in BLAS
+                libraries. If `only_connect` is True this is ignored. In
+                that case, set the thread count when spawning workers.
+                (Default: 1).
         """
         super().__init__()
 
@@ -95,24 +112,32 @@ def __init__(
             MessageDirection.ABOVE,
         )
 
-        # Case 1: spawn and manage workers
+        # Case 1: spawn and/or manage workers
         if ipports is None:
             if only_connect:
                 self.connect_to_workers(num_workers, worker_port)
             else:
-                self.spawn_workers(num_workers, worker_port)
-
-        # Case 2: Connect to managers at ipports
+                self.spawn_workers(
+                    num_workers,
+                    worker_port,
+                    log_level,
+                    num_blas_threads,
+                )
+
+        # Case 2: Connect to detached managers at ipports
         else:
             self.connect_to_managers(ipports)
 
         # Track info on sent messages to reduce redundant messages:
         self.last_num_idle_sent_up = self.total_workers
 
+        # Track info on received messages to report read receipts:
+        self.most_recent_read_submit: RuntimeAddress | None = None
+
         # Inform upstream we are starting
         msg = (self.upstream, RuntimeMessage.STARTED, self.total_workers)
         self.outgoing.put(msg)
-        self.logger.info('Sent start message upstream.')
+        _logger.info('Sent start message upstream.')
 
     def handle_message(
         self,
@@ -126,25 +151,33 @@ def handle_message(
 
             if msg == RuntimeMessage.SUBMIT:
                 rtask = cast(RuntimeTask, payload)
+                self.most_recent_read_submit = rtask.unique_id
                 self.schedule_tasks([rtask])
-                self.update_upstream_idle_workers()
+                # self.update_upstream_idle_workers()
 
             elif msg == RuntimeMessage.SUBMIT_BATCH:
                 rtasks = cast(List[RuntimeTask], payload)
+                self.most_recent_read_submit = rtasks[0].unique_id
                 self.schedule_tasks(rtasks)
-                self.update_upstream_idle_workers()
+                # self.update_upstream_idle_workers()
 
             elif msg == RuntimeMessage.RESULT:
                 result = cast(RuntimeResult, payload)
                 self.send_result_down(result)
 
             elif msg == RuntimeMessage.CANCEL:
-                addr = cast(RuntimeAddress, payload)
-                self.broadcast_cancel(addr)
+                self.broadcast(RuntimeMessage.CANCEL, payload)
 
             elif msg == RuntimeMessage.SHUTDOWN:
                 self.handle_shutdown()
 
+            elif msg == RuntimeMessage.IMPORTPATH:
+                paths = cast(List[str], payload)
+                self.handle_importpath(paths)
+
+            elif msg == RuntimeMessage.COMMUNICATE:
+                self.broadcast(RuntimeMessage.COMMUNICATE, payload)
+
             else:
                 raise RuntimeError(f'Unexpected message type: {msg.name}')
 
@@ -153,20 +186,19 @@ def handle_message(
             if msg == RuntimeMessage.SUBMIT:
                 rtask = cast(RuntimeTask, payload)
                 self.send_up_or_schedule_tasks([rtask])
-                self.update_upstream_idle_workers()
 
             elif msg == RuntimeMessage.SUBMIT_BATCH:
                 rtasks = cast(List[RuntimeTask], payload)
                 self.send_up_or_schedule_tasks(rtasks)
-                self.update_upstream_idle_workers()
 
             elif msg == RuntimeMessage.RESULT:
                 result = cast(RuntimeResult, payload)
                 self.handle_result_from_below(result)
 
             elif msg == RuntimeMessage.WAITING:
-                num_idle = cast(int, payload)
-                self.handle_waiting(conn, num_idle)
+                p = cast(Tuple[int, Optional[RuntimeAddress]], payload)
+                num_idle, read_receipt = p
+                self.handle_waiting(conn, num_idle, read_receipt)
                 self.update_upstream_idle_workers()
 
             elif msg == RuntimeMessage.UPDATE:
@@ -197,6 +229,13 @@ def handle_system_error(self, error_str: str) -> None:
             # If server has crashed then just exit
             pass
 
+    def get_to_string(self, conn: Connection) -> str:
+        """Return a string representation of the connection."""
+        if conn == self.upstream:
+            return 'BOSS'
+
+        return self.conn_to_employee_dict[conn].recipient_string
+
     def handle_shutdown(self) -> None:
         """Shutdown the manager and clean up spawned processes."""
         super().handle_shutdown()
@@ -217,6 +256,7 @@ def send_up_or_schedule_tasks(self, tasks: Sequence[RuntimeTask]) -> None:
         if num_idle != 0:
             self.outgoing.put((self.upstream, RuntimeMessage.UPDATE, num_idle))
             self.schedule_tasks(tasks[:num_idle])
+            self.update_upstream_idle_workers()
 
         if len(tasks) > num_idle:
             self.outgoing.put((
@@ -244,7 +284,8 @@ def update_upstream_idle_workers(self) -> None:
         """Update the total number of idle workers upstream."""
         if self.num_idle_workers != self.last_num_idle_sent_up:
             self.last_num_idle_sent_up = self.num_idle_workers
-            m = (self.upstream, RuntimeMessage.WAITING, self.num_idle_workers)
+            payload = (self.num_idle_workers, self.most_recent_read_submit)
+            m = (self.upstream, RuntimeMessage.WAITING, payload)
             self.outgoing.put(m)
 
     def handle_update(self, conn: Connection, task_diff: int) -> None:
@@ -305,9 +346,16 @@ def start_manager() -> None:
     ipports = None if args.managers is None else parse_ipports(args.managers)
 
     # Set up logging
-    _logger = logging.getLogger('bqskit-runtime')
-    _logger.setLevel([30, 20, 10, 1][min(args.verbose, 3)])
-    _logger.addHandler(logging.StreamHandler())
+    log_level = [30, 20, 10, 1][min(args.verbose, 3)]
+    logging.getLogger().setLevel(log_level)
+    _handler = logging.StreamHandler()
+    _handler.setLevel(0)
+    _fmt_header = '%(asctime)s.%(msecs)03d - %(levelname)-8s |'
+    _fmt_message = ' %(module)s: %(message)s'
+    _fmt = _fmt_header + _fmt_message
+    _formatter = logging.Formatter(_fmt, '%H:%M:%S')
+    _handler.setFormatter(_formatter)
+    logging.getLogger().addHandler(_handler)
 
     # Import tests package recursively
     if args.import_tests:
diff --git a/bqskit/runtime/message.py b/bqskit/runtime/message.py
index 63f687048..d2585aef2 100644
--- a/bqskit/runtime/message.py
+++ b/bqskit/runtime/message.py
@@ -20,3 +20,6 @@ class RuntimeMessage(IntEnum):
     CANCEL = 11
     WAITING = 12
     UPDATE = 13
+    IMPORTPATH = 14
+    READY = 15
+    COMMUNICATE = 16
diff --git a/bqskit/runtime/task.py b/bqskit/runtime/task.py
index bd413374a..d8cef7855 100644
--- a/bqskit/runtime/task.py
+++ b/bqskit/runtime/task.py
@@ -6,6 +6,8 @@
 from typing import Any
 from typing import Coroutine
 
+import dill
+
 from bqskit.runtime.address import RuntimeAddress
 
 
@@ -34,18 +36,26 @@ def __init__(
         breadcrumbs: tuple[RuntimeAddress, ...],
         logging_level: int | None = None,
         max_logging_depth: int = -1,
+        task_name: str | None = None,
+        log_context: dict[str, str] = {},
     ) -> None:
         """Create the task with a new id and return address."""
         RuntimeTask.task_counter += 1
         self.task_id = RuntimeTask.task_counter
 
-        self.fnargs = fnargs
+        self.serialized_fnargs = dill.dumps(fnargs)
+        self._fnargs: tuple[Any, Any, Any] | None = None
+        self._name = fnargs[0].__name__ if task_name is None else task_name
         """Tuple of function pointer, arguments, and keyword arguments."""
 
         self.return_address = return_address
-        """Where the result of this task should be sent."""
+        """
+        Where the result of this task should be sent.
+
+        This doubles as a unique system-wide id for the task.
+        """
 
-        self.logging_level = logging_level
+        self.logging_level = logging_level or 0
         """Logs with levels >= to this get emitted, if None always emit."""
 
         self.comp_task_id = comp_task_id
@@ -60,9 +70,6 @@ def __init__(
         self.coro: Coroutine[Any, Any, Any] | None = None
         """The coroutine containing this tasks code."""
 
-        # self.send: Any = None
-        # """A register that both the coroutine and task have access to."""
-
         self.desired_box_id: int | None = None
         """When waiting on a mailbox, this stores that mailbox's id."""
 
@@ -72,6 +79,19 @@ def __init__(
         self.wake_on_next: bool = False
         """Set to true if this task should wake immediately on a result."""
 
+        self.log_context: dict[str, str] = log_context
+        """Additional context to be logged with this task."""
+
+        self.msg_buffer: list[Any] = []
+
+    @property
+    def fnargs(self) -> tuple[Any, Any, Any]:
+        """Return the function pointer, arguments, and keyword arguments."""
+        if self._fnargs is None:
+            self._fnargs = dill.loads(self.serialized_fnargs)
+        assert self._fnargs is not None  # for type checker
+        return self._fnargs
+
     def step(self, send_val: Any = None) -> Any:
         """Execute one step of the task."""
         if self.coro is None:
@@ -87,7 +107,9 @@ def step(self, send_val: Any = None) -> Any:
             self.max_logging_depth < 0
             or len(self.breadcrumbs) <= self.max_logging_depth
         ):
-            logging.getLogger().setLevel(0)
+            logging.getLogger().setLevel(self.logging_level)
+        else:
+            logging.getLogger().setLevel(100)
 
         # Execute a task step
         to_return = self.coro.send(send_val)
@@ -97,6 +119,11 @@ def step(self, send_val: Any = None) -> Any:
 
         return to_return
 
+    @property
+    def unique_id(self) -> RuntimeAddress:
+        """Return the task's system-wide unique id."""
+        return self.return_address
+
     def start(self) -> None:
         """Initialize the task."""
         self.coro = self.run()
@@ -108,7 +135,11 @@ def cancel(self) -> None:
             # it is likely a blanket try/accept catching the
             # error used to stop the coroutine, preventing
             # it from stopping correctly.
-            self.coro.close()
+            try:
+                self.coro.close()
+            except ValueError:
+                # Coroutine is running and cannot be closed.
+                pass
         else:
             raise RuntimeError('Task was cancelled with None coroutine.')
 
@@ -121,3 +152,11 @@ async def run(self) -> Any:
     def is_descendant_of(self, addr: RuntimeAddress) -> bool:
         """Return true if `addr` identifies a parent (or this) task."""
         return addr == self.return_address or addr in self.breadcrumbs
+
+    def __str__(self) -> str:
+        """Return a string representation of the task."""
+        return f'{self._name}'
+
+    def __repr__(self) -> str:
+        """Return a string representation of the task."""
+        return f'<RuntimeTask {self.unique_id} {self._name}>'
diff --git a/bqskit/runtime/worker.py b/bqskit/runtime/worker.py
index 00961d8de..1684f7dbc 100644
--- a/bqskit/runtime/worker.py
+++ b/bqskit/runtime/worker.py
@@ -8,18 +8,22 @@
 import sys
 import time
 import traceback
-from collections import OrderedDict
 from dataclasses import dataclass
 from multiprocessing import Process
 from multiprocessing.connection import Client
 from multiprocessing.connection import Connection
-from multiprocessing.connection import wait
+from queue import Empty
+from queue import Queue
+from threading import Lock
+from threading import Thread
 from typing import Any
 from typing import Callable
 from typing import cast
 from typing import List
+from typing import Sequence
 
 from bqskit.runtime import default_worker_port
+from bqskit.runtime import set_blas_thread_counts
 from bqskit.runtime.address import RuntimeAddress
 from bqskit.runtime.future import RuntimeFuture
 from bqskit.runtime.message import RuntimeMessage
@@ -27,31 +31,7 @@
 from bqskit.runtime.task import RuntimeTask
 
 
-class WorkerQueue():
-    """The worker's task FIFO queue."""
-
-    def __init__(self) -> None:
-        """
-        Initialize the worker queue.
-
-        An OrderedDict is used to internally store the task. This prevents the
-        same task appearing multiple times in the queue, while also ensuring
-        O(1) operations.
-        """
-        self._queue: OrderedDict[RuntimeAddress, None] = OrderedDict()
-
-    def put(self, addr: RuntimeAddress) -> None:
-        """Enqueue a task by its address."""
-        if addr not in self._queue:
-            self._queue[addr] = None
-
-    def get(self) -> RuntimeAddress:
-        """Get the next task to run."""
-        return self._queue.popitem(last=False)[0]
-
-    def empty(self) -> bool:
-        """Check if the queue is empty."""
-        return len(self._queue) == 0
+_logger = logging.getLogger(__name__)
 
 
 @dataclass
@@ -126,11 +106,13 @@ class Worker:
     """
     BQSKit Runtime's Worker.
 
-    BQSKit Runtime utilizes a single-threaded worker to accept, execute,
+    BQSKit Runtime utilizes a dual-threaded worker to accept, execute,
     pause, spawn, resume, and complete tasks in a custom event loop built
     with python's async await mechanisms. Each worker receives and sends
     tasks and results to the greater system through a single duplex
-    connection with a runtime server or manager.
+    connection with a runtime server or manager. One thread performs
+    work and sends outgoing messages, while the other thread handles
+    incoming messages.
 
     At start-up, the worker receives an ID and waits for its first task.
     An executing task may use the `submit` and `map` methods to spawn child
@@ -178,16 +160,20 @@ def __init__(self, id: int, conn: Connection) -> None:
         self._id = id
         self._conn = conn
 
-        self._outgoing: list[tuple[RuntimeMessage, Any]] = []
-        """Stores outgoing messages to be handled by the event loop."""
-
         self._tasks: dict[RuntimeAddress, RuntimeTask] = {}
         """Tracks all started, unfinished tasks on this worker."""
 
         self._delayed_tasks: list[RuntimeTask] = []
-        """Store all delayed tasks in LIFO order."""
+        """
+        Store all delayed tasks in LIFO order.
+
+        Delayed tasks have no context and are stored (more-or-less) as a
+        function pointer together with the arguments. When it gets started, it
+        consumes much more memory, so we delay the task start until necessary
+        (at no cost)
+        """
 
-        self._ready_task_ids: WorkerQueue = WorkerQueue()
+        self._ready_task_ids: Queue[RuntimeAddress] = Queue()
         """Tasks queued up for execution."""
 
         self._cancelled_task_ids: set[RuntimeAddress] = set()
@@ -196,7 +182,7 @@ def __init__(self, id: int, conn: Connection) -> None:
         self._active_task: RuntimeTask | None = None
         """The currently executing task if one is running."""
 
-        self._running = False
+        self._running = True
         """Controls if the event loop is running."""
 
         self._mailboxes: dict[int, WorkerMailbox] = {}
@@ -208,71 +194,104 @@ def __init__(self, id: int, conn: Connection) -> None:
         self._cache: dict[str, Any] = {}
         """Local worker cache."""
 
-        # Send out every emitted log message upstream
+        self.most_recent_read_submit: RuntimeAddress | None = None
+        """Tracks the most recently processed submit message from above."""
+
+        self.read_receipt_mutex = Lock()
+        """
+        A lock to ensure waiting messages's read receipt is correct.
+
+        This lock enforces atomic update of `most_recent_read_submit` and
+        task addition/enqueueing. This is necessary to ensure that the
+        idle status is always correct.
+        """
+
+        # Send out every client emitted log message upstream
         old_factory = logging.getLogRecordFactory()
 
         def record_factory(*args: Any, **kwargs: Any) -> logging.LogRecord:
             record = old_factory(*args, **kwargs)
-            active_task = get_worker()._active_task
-            if active_task is not None:
-                lvl = active_task.logging_level
-                if lvl is None or lvl <= record.levelno:
-                    tid = active_task.comp_task_id
-                    self._outgoing.append((RuntimeMessage.LOG, (tid, record)))
+            active_task = self._active_task
+            if not record.name.startswith('bqskit.runtime'):
+                if active_task is not None:
+                    lvl = active_task.logging_level
+                    if lvl is None or lvl <= record.levelno:
+                        if lvl <= logging.DEBUG:
+                            record.msg += f' [wid={self._id}'
+                            items = active_task.log_context.items()
+                            if len(items) > 0:
+                                record.msg += ', '
+                            con_str = ', '.join(f'{k}={v}' for k, v in items)
+                            record.msg += con_str
+                            record.msg += ']'
+                        tid = active_task.comp_task_id
+                        self._conn.send((RuntimeMessage.LOG, (tid, record)))
             return record
 
         logging.setLogRecordFactory(record_factory)
 
+        # Start incoming thread
+        self.incoming_thread = Thread(target=self.recv_incoming)
+        self.incoming_thread.daemon = True
+        self.incoming_thread.start()
+        _logger.debug('Started incoming thread.')
+
         # Communicate that this worker is ready
         self._conn.send((RuntimeMessage.STARTED, self._id))
 
     def _loop(self) -> None:
         """Main worker event loop."""
-        self._running = True
         while self._running:
-            self._try_step_next_ready_task()
-            self._try_idle()
-            self._handle_comms()
-
-    def _try_idle(self) -> None:
-        """If there is nothing to do, wait until we receive a message."""
-        empty_outgoing = len(self._outgoing) == 0
-        no_ready_tasks = self._ready_task_ids.empty()
-        no_delayed_tasks = len(self._delayed_tasks) == 0
-
-        if empty_outgoing and no_ready_tasks and no_delayed_tasks:
-            self._conn.send((RuntimeMessage.WAITING, 1))
-            wait([self._conn])
-
-    def _handle_comms(self) -> None:
-        """Handle all incoming and outgoing messages."""
-
-        # Handle outgoing communication
-        for out_msg in self._outgoing:
-            self._conn.send(out_msg)
-        self._outgoing.clear()
-
-        # Handle incomming communication
-        while self._conn.poll():
-            msg, payload = self._conn.recv()
+            try:
+                self._try_step_next_ready_task()
+            except Exception:
+                self._running = False
+                exc_info = sys.exc_info()
+                error_str = ''.join(traceback.format_exception(*exc_info))
+                _logger.error(error_str)
+                try:
+                    self._conn.send((RuntimeMessage.ERROR, error_str))
+                except Exception:
+                    pass
+
+    def recv_incoming(self) -> None:
+        """Continuously receive all incoming messages."""
+        while self._running:
+            # Receive message
+            try:
+                msg, payload = self._conn.recv()
+            except Exception:
+                _logger.debug('Crashed due to lost connection')
+                if sys.platform == 'win32':
+                    os.kill(os.getpid(), 9)
+                else:
+                    os.kill(os.getpid(), signal.SIGKILL)
+                exit()
+
+            _logger.debug(f'Received message {msg.name}.')
+            _logger.log(1, f'Payload: {payload}')
 
             # Process message
             if msg == RuntimeMessage.SHUTDOWN:
-                self._running = False
-                return
+                if sys.platform == 'win32':
+                    os.kill(os.getpid(), 9)
+                else:
+                    os.kill(os.getpid(), signal.SIGKILL)
 
             elif msg == RuntimeMessage.SUBMIT:
+                self.read_receipt_mutex.acquire()
                 task = cast(RuntimeTask, payload)
+                self.most_recent_read_submit = task.unique_id
                 self._add_task(task)
+                self.read_receipt_mutex.release()
 
             elif msg == RuntimeMessage.SUBMIT_BATCH:
+                self.read_receipt_mutex.acquire()
                 tasks = cast(List[RuntimeTask], payload)
+                self.most_recent_read_submit = tasks[0].unique_id
                 self._add_task(tasks.pop())  # Submit one task
                 self._delayed_tasks.extend(tasks)  # Delay rest
-                # Delayed tasks have no context and are stored (more-or-less)
-                # as a function pointer together with the arguments.
-                # When it gets started, it consumes much more memory,
-                # so we delay the task start until necessary (at no cost)
+                self.read_receipt_mutex.release()
 
             elif msg == RuntimeMessage.RESULT:
                 result = cast(RuntimeResult, payload)
@@ -281,6 +300,17 @@ def _handle_comms(self) -> None:
             elif msg == RuntimeMessage.CANCEL:
                 addr = cast(RuntimeAddress, payload)
                 self._handle_cancel(addr)
+                # TODO: preempt?
+
+            elif msg == RuntimeMessage.COMMUNICATE:
+                addrs, msg = cast(tuple[list[RuntimeAddress], Any], payload)
+                self._handle_communicate(addrs, msg)
+
+            elif msg == RuntimeMessage.IMPORTPATH:
+                paths = cast(List[str], payload)
+                for path in paths:
+                    if path not in sys.path:
+                        sys.path.append(path)
 
     def _add_task(self, task: RuntimeTask) -> None:
         """Start a task and add it to the loop."""
@@ -290,8 +320,9 @@ def _add_task(self, task: RuntimeTask) -> None:
 
     def _handle_result(self, result: RuntimeResult) -> None:
         """Insert result into appropriate mailbox and wake waiting task."""
-        mailbox_id = result.return_address.mailbox_index
         assert result.return_address.worker_id == self._id
+
+        mailbox_id = result.return_address.mailbox_index
         if mailbox_id not in self._mailboxes:
             # If the mailbox has been dropped due to a cancel, ignore result
             return
@@ -304,7 +335,11 @@ def _handle_result(self, result: RuntimeResult) -> None:
             task = self._tasks[box.dest_addr]
 
             if task.wake_on_next or box.ready:
+                # print(f'Worker {self._id} is waking task
+                # {task.return_address}, with {task.wake_on_next=},
+                # {box.ready=}')
                 self._ready_task_ids.put(box.dest_addr)  # Wake it
+                box.dest_addr = None  # Prevent double wake
 
     def _handle_cancel(self, addr: RuntimeAddress) -> None:
         """
@@ -323,6 +358,7 @@ def _handle_cancel(self, addr: RuntimeAddress) -> None:
             for themselves using breadcrumbs and the original `addr` cancel
             message.
         """
+        # TODO: Send update message?
         self._cancelled_task_ids.add(addr)
 
         # Remove all tasks that are children of `addr` from initialized tasks
@@ -342,16 +378,48 @@ def _handle_cancel(self, addr: RuntimeAddress) -> None:
             if not t.is_descendant_of(addr)
         ]
 
+    def _handle_communicate(
+        self,
+        addrs: list[RuntimeAddress],
+        msg: Any,
+    ) -> None:
+        for task_addr in addrs:
+            if task_addr not in self._tasks:
+                continue
+
+            self._tasks[task_addr].msg_buffer.append(msg)
+
     def _get_next_ready_task(self) -> RuntimeTask | None:
-        """Return the next ready task if one exists, otherwise None."""
+        """Return the next ready task if one exists, otherwise block."""
         while True:
-            if self._ready_task_ids.empty():
-                if len(self._delayed_tasks) > 0:
-                    self._add_task(self._delayed_tasks.pop())
-                    continue
-                return None
+            if self._ready_task_ids.empty() and len(self._delayed_tasks) > 0:
+                self._add_task(self._delayed_tasks.pop())
+                continue
 
-            addr = self._ready_task_ids.get()
+            # Critical section
+            # Attempt to get a ready task. If none are available, message
+            # the manager/server with a waiting message letting them
+            # know the worker is idle. This needs to be atomic to prevent
+            # the self.more_recent_read_submit from being updated after
+            # catching the Empty exception, but before forming the payload.
+            self.read_receipt_mutex.acquire()
+            try:
+                addr = self._ready_task_ids.get_nowait()
+
+            except Empty:
+                payload = (1, self.most_recent_read_submit)
+                self._conn.send((RuntimeMessage.WAITING, payload))
+                self.read_receipt_mutex.release()
+                # Block for new message. Can release lock here since the
+                # the `self.most_recent_read_submit` has been used.
+                addr = self._ready_task_ids.get()
+
+            else:
+                self.read_receipt_mutex.release()
+
+            # Handle a shutdown request that occured while waiting
+            if not self._running:
+                return None
 
             if addr in self._cancelled_task_ids or addr not in self._tasks:
                 # When a task is cancelled on the worker it is not removed
@@ -366,6 +434,7 @@ def _get_next_ready_task(self) -> RuntimeTask | None:
                 # then discard this one too. Each breadcrumb (bcb) is a
                 # task address (unique system-wide task id) of an ancestor
                 # task.
+                # TODO: do I need to manually remove addr from self._tasks?
                 continue
 
             return task
@@ -375,7 +444,6 @@ def _try_step_next_ready_task(self) -> None:
         task = self._get_next_ready_task()
 
         if task is None:
-            # Nothing to do
             return
 
         try:
@@ -396,7 +464,7 @@ def _try_step_next_ready_task(self) -> None:
             exc_info = sys.exc_info()
             error_str = ''.join(traceback.format_exception(*exc_info))
             error_payload = (self._active_task.comp_task_id, error_str)
-            self._outgoing.append((RuntimeMessage.ERROR, error_payload))
+            self._conn.send((RuntimeMessage.ERROR, error_payload))
 
         finally:
             self._active_task = None
@@ -415,14 +483,17 @@ def _process_await(self, task: RuntimeTask, future: RuntimeFuture) -> None:
         box.dest_addr = task.return_address
         task.desired_box_id = future.mailbox_id
 
-        if future._next_flag:
-            # Set from Worker.next, implies the task wants the next result
-            if box.ready:
-                m = 'Cannot wait for next results on a complete task.'
-                raise RuntimeError(m)
-            task.wake_on_next = True
-
-        elif box.ready:
+        # if future._next_flag:
+        #     # Set from Worker.next, implies the task wants the next result
+        #     # if box.ready:
+        #     #     m = 'Cannot wait for next results on a complete task.'
+        #     #     raise RuntimeError(m)
+        #     task.wake_on_next = True
+        task.wake_on_next = future._next_flag
+        # print(f'Worker {self._id} is waiting on task
+        # {task.return_address}, with {task.wake_on_next=}')
+
+        if box.ready:
             self._ready_task_ids.put(task.return_address)
 
     def _process_task_completion(self, task: RuntimeTask, result: Any) -> None:
@@ -430,13 +501,18 @@ def _process_task_completion(self, task: RuntimeTask, result: Any) -> None:
         assert task is self._active_task
         packaged_result = RuntimeResult(task.return_address, result, self._id)
 
+        if task.return_address not in self._tasks:
+            # print(f'Task was cancelled: {task.return_address},
+            # {task.fnargs[0].__name__}')
+            return
+
         if task.return_address.worker_id == self._id:
             self._handle_result(packaged_result)
-            self._outgoing.append((RuntimeMessage.UPDATE, -1))
+            self._conn.send((RuntimeMessage.UPDATE, -1))
             # Let manager know this worker has one less task
             # without sending a result
         else:
-            self._outgoing.append((RuntimeMessage.RESULT, packaged_result))
+            self._conn.send((RuntimeMessage.RESULT, packaged_result))
 
         # Remove task
         self._tasks.pop(task.return_address)
@@ -452,10 +528,6 @@ def _process_task_completion(self, task: RuntimeTask, result: Any) -> None:
             # Otherwise send a cancel message
             self.cancel(RuntimeFuture(mailbox_id))
 
-        # Start delayed task
-        if self._ready_task_ids.empty() and len(self._delayed_tasks) > 0:
-            self._add_task(self._delayed_tasks.pop())
-
     def _get_desired_result(self, task: RuntimeTask) -> Any:
         """Retrieve the task's desired result from the mailboxes."""
         if task.desired_box_id is None:
@@ -465,7 +537,7 @@ def _get_desired_result(self, task: RuntimeTask) -> Any:
 
         if task.wake_on_next:
             fresh_results = box.get_new_results()
-            assert len(fresh_results) > 0
+            # assert len(fresh_results) > 0
             return fresh_results
 
         assert box.ready
@@ -482,10 +554,25 @@ def submit(
         self,
         fn: Callable[..., Any],
         *args: Any,
+        task_name: str | None = None,
+        log_context: dict[str, str] = {},
         **kwargs: Any,
     ) -> RuntimeFuture:
         """Submit `fn` as a task to the runtime."""
         assert self._active_task is not None
+
+        if task_name is not None and not isinstance(task_name, str):
+            raise RuntimeError('task_name must be a string.')
+
+        if not isinstance(log_context, dict):
+            raise RuntimeError('log_context must be a dictionary.')
+
+        for k, v in log_context.items():
+            if not isinstance(k, str) or not isinstance(v, str):
+                raise RuntimeError(
+                    'log_context must be a map from strings to strings.',
+                )
+
         # Group fnargs together
         fnarg = (fn, args, kwargs)
 
@@ -503,10 +590,12 @@ def submit(
             + (self._active_task.return_address,),
             self._active_task.logging_level,
             self._active_task.max_logging_depth,
+            task_name,
+            {**self._active_task.log_context, **log_context},
         )
 
         # Submit the task (on the next cycle)
-        self._outgoing.append((RuntimeMessage.SUBMIT, task))
+        self._conn.send((RuntimeMessage.SUBMIT, task))
 
         # Return future pointing to the mailbox
         return RuntimeFuture(mailbox_id)
@@ -515,10 +604,38 @@ def map(
         self,
         fn: Callable[..., Any],
         *args: Any,
+        task_name: Sequence[str | None] | str | None = None,
+        log_context: Sequence[dict[str, str]] | dict[str, str] = {},
         **kwargs: Any,
     ) -> RuntimeFuture:
         """Map `fn` over the input arguments distributed across the runtime."""
         assert self._active_task is not None
+
+        if task_name is None or isinstance(task_name, str):
+            task_name = [task_name] * len(args[0])
+
+        if len(task_name) != len(args[0]):
+            raise RuntimeError(
+                'task_name must be a string or a list of strings equal'
+                'in length to the number of tasks.',
+            )
+
+        if isinstance(log_context, dict):
+            log_context = [log_context] * len(args[0])
+
+        if len(log_context) != len(args[0]):
+            raise RuntimeError(
+                'log_context must be a dictionary or a list of dictionaries'
+                ' equal in length to the number of tasks.',
+            )
+
+        for context in log_context:
+            for k, v in context.items():
+                if not isinstance(k, str) or not isinstance(v, str):
+                    raise RuntimeError(
+                        'log_context must be a map from strings to strings.',
+                    )
+
         # Group fnargs together
         fnargs = []
         if len(args) == 1:
@@ -548,16 +665,37 @@ def map(
                 breadcrumbs,
                 self._active_task.logging_level,
                 self._active_task.max_logging_depth,
+                task_name[i],
+                {**self._active_task.log_context, **log_context[i]},
             )
             for i, fnarg in enumerate(fnargs)
         ]
 
         # Submit the tasks
-        self._outgoing.append((RuntimeMessage.SUBMIT_BATCH, tasks))
+        self._conn.send((RuntimeMessage.SUBMIT_BATCH, tasks))
 
         # Return future pointing to the mailbox
         return RuntimeFuture(mailbox_id)
 
+    def communicate(self, future: RuntimeFuture, msg: Any) -> None:
+        """Send a message to the task associated with `future`."""
+        assert self._active_task is not None
+        assert future.mailbox_id in self._mailboxes
+
+        num_slots = self._mailboxes[future.mailbox_id].expected_num_results
+        addrs = [
+            RuntimeAddress(self._id, future.mailbox_id, slot_id)
+            for slot_id in range(num_slots)
+        ]
+        self._conn.send((RuntimeMessage.COMMUNICATE, (addrs, msg)))
+
+    def get_messages(self) -> list[Any]:
+        """Return all messages received by the worker for this task."""
+        assert self._active_task is not None
+        x = self._active_task.msg_buffer
+        self._active_task.msg_buffer = []
+        return x
+
     def cancel(self, future: RuntimeFuture) -> None:
         """Cancel all tasks associated with `future`."""
         assert self._active_task is not None
@@ -568,8 +706,8 @@ def cancel(self, future: RuntimeFuture) -> None:
             RuntimeAddress(self._id, future.mailbox_id, slot_id)
             for slot_id in range(num_slots)
         ]
-        msgs = [(RuntimeMessage.CANCEL, addr) for addr in addrs]
-        self._outgoing.extend(msgs)
+        for addr in addrs:
+            self._conn.send((RuntimeMessage.CANCEL, addr))
 
     def get_cache(self) -> dict[str, Any]:
         """
@@ -596,7 +734,8 @@ async def next(self, future: RuntimeFuture) -> list[tuple[int, Any]]:
                 returned. Each result is paired with the index of its
                 arguments in the original map call.
         """
-        if future._done:
+        # if future._done:
+        if future.mailbox_id not in self._mailboxes:
             raise RuntimeError('Cannot wait on an already completed result.')
 
         future._next_flag = True
@@ -609,25 +748,35 @@ async def next(self, future: RuntimeFuture) -> list[tuple[int, Any]]:
 _worker = None
 
 
-def start_worker(w_id: int | None, port: int, cpu: int | None = None) -> None:
+def start_worker(
+    w_id: int | None,
+    port: int,
+    cpu: int | None = None,
+    logging_level: int = logging.WARNING,
+    num_blas_threads: int = 1,
+    log_client: bool = False,
+) -> None:
     """Start this process's worker."""
     if w_id is not None:
         # Ignore interrupt signals on workers, boss will handle it for us
         # If w_id is None, then we are being spawned separately.
         signal.signal(signal.SIGINT, signal.SIG_IGN)
+        # TODO: check what needs to be done on win
 
-    # Purge all standard python logging configurations
-    for _, logger in logging.Logger.manager.loggerDict.items():
-        if isinstance(logger, logging.PlaceHolder):
-            continue
-        logger.handlers.clear()
-    logging.Logger.manager.loggerDict = {}
+    # Set number of BLAS threads
+    set_blas_thread_counts(num_blas_threads)
 
+    # Enforce no default logging
+    logging.lastResort = logging.NullHandler()
+    logging.getLogger().handlers.clear()
+
+    # Pin worker to cpu
     if cpu is not None:
         if sys.platform == 'win32':
             raise RuntimeError('Cannot pin worker to cpu on windows.')
         os.sched_setaffinity(0, [cpu])
 
+    # Connect to manager
     max_retries = 7
     wait_time = .1
     conn: Connection | None = None
@@ -644,10 +793,29 @@ def start_worker(w_id: int | None, port: int, cpu: int | None = None) -> None:
     if conn is None:
         raise RuntimeError('Unable to establish connection with manager.')
 
+    # If id isn't provided, wait for assignment
     if w_id is None:
         msg, w_id = conn.recv()
+        assert isinstance(w_id, int)
         assert msg == RuntimeMessage.STARTED
 
+    # Set up runtime logging
+    if not log_client:
+        _runtime_logger = logging.getLogger('bqskit.runtime')
+    else:
+        _runtime_logger = logging.getLogger()
+    _runtime_logger.propagate = False
+    _runtime_logger.setLevel(logging_level)
+    _handler = logging.StreamHandler()
+    _handler.setLevel(0)
+    _fmt_header = '%(asctime)s.%(msecs)03d - %(levelname)-8s |'
+    _fmt_message = f' [wid={w_id}]: %(message)s'
+    _fmt = _fmt_header + _fmt_message
+    _formatter = logging.Formatter(_fmt, '%H:%M:%S')
+    _handler.setFormatter(_formatter)
+    _runtime_logger.addHandler(_handler)
+
+    # Build and start worker
     global _worker
     _worker = Worker(w_id, conn)
     _worker._loop()
@@ -695,6 +863,23 @@ def start_worker_rank() -> None:
         default=default_worker_port,
         help='The port the workers will try to connect to a manager on.',
     )
+    parser.add_argument(
+        '-v', '--verbose',
+        action='count',
+        default=0,
+        help='Enable logging of increasing verbosity, either -v, -vv, or -vvv.',
+    )
+    parser.add_argument(
+        '-l', '--log-client',
+        action='store_true',
+        help='Log messages from the client process.',
+    )
+    parser.add_argument(
+        '-t', '--num_blas_threads',
+        type=int,
+        default=1,
+        help='The number of threads to use in BLAS libraries.',
+    )
     args = parser.parse_args()
 
     if args.cpus is not None:
@@ -714,10 +899,23 @@ def start_worker_rank() -> None:
     else:
         cpus = [None for _ in range(args.num_workers)]
 
+    logging_level = [30, 20, 10, 1][min(args.verbose, 3)]
+
+    if args.log_client and logging_level > 10:
+        raise RuntimeError('Cannot log client messages without at least -vv.')
+
     # Spawn worker process
     procs = []
     for cpu in cpus:
-        procs.append(Process(target=start_worker, args=(None, args.port, cpu)))
+        pargs = (
+            None,
+            args.port,
+            cpu,
+            logging_level,
+            args.num_blas_threads,
+            args.log_client,
+        )
+        procs.append(Process(target=start_worker, args=pargs))
         procs[-1].start()
 
     # Join them
diff --git a/bqskit/utils/cachedclass.py b/bqskit/utils/cachedclass.py
index 751a361b6..7edac47b9 100644
--- a/bqskit/utils/cachedclass.py
+++ b/bqskit/utils/cachedclass.py
@@ -63,7 +63,8 @@ def __new__(cls: type[T], *args: Any, **kwargs: Any) -> T:
         _instances = cls._instances  # type: ignore
 
         if _instances.get(key, None) is None:
-            _logger.debug(
+            _logger.log(
+                1,
                 (
                     'Creating cached instance for class: %s,'
                     ' with args %s, and kwargs %s'
diff --git a/docs/conf.py b/docs/conf.py
index cfd238319..f48262b93 100644
--- a/docs/conf.py
+++ b/docs/conf.py
@@ -100,7 +100,7 @@
     'pytket',
     'cirq',
     'qutip',
-    'qiskit',
+    'dill',
 ]
 nbsphinx_allow_errors = True
 nbsphinx_execute = 'never'
diff --git a/setup.py b/setup.py
index 005681ffd..8edcd43de 100644
--- a/setup.py
+++ b/setup.py
@@ -9,7 +9,7 @@
 root_dir_path = os.path.abspath(os.path.dirname(__file__))
 pkg_dir_path = os.path.join(root_dir_path, 'bqskit')
 readme_path = os.path.join(root_dir_path, 'README.md')
-version_path = os.path.join(pkg_dir_path, 'version.py')
+version_path = os.path.join(pkg_dir_path, '_version.py')
 
 # Load Version Number
 with open(version_path) as version_file:
@@ -71,6 +71,7 @@
         'numpy>=1.22.0',
         'scipy>=1.8.0',
         'typing-extensions>=4.0.0',
+        'dill>=0.3.8',
     ],
     python_requires='>=3.8, <4',
     entry_points={


Test patch:
diff --git a/tests/compiler/test_data.py b/tests/compiler/test_data.py
index 075a95c82..934215db6 100644
--- a/tests/compiler/test_data.py
+++ b/tests/compiler/test_data.py
@@ -26,3 +26,9 @@ def test_update_error_mul() -> None:
     assert data.error == 0.75
     data.update_error_mul(0.5)
     assert data.error == 0.875
+
+
+def test_target_doesnt_get_expanded_on_update() -> None:
+    data = PassData(Circuit(64))
+    data2 = PassData(Circuit(64))
+    data.update(data2)  # Should not crash
diff --git a/tests/passes/control/test_paralleldo.py b/tests/passes/control/test_paralleldo.py
index 2d69b5e70..91a5c4f72 100644
--- a/tests/passes/control/test_paralleldo.py
+++ b/tests/passes/control/test_paralleldo.py
@@ -38,11 +38,11 @@ async def run(self, circuit: Circuit, data: PassData) -> None:
         data['key'] = '1'
 
 
-class Sleep3Pass(BasePass):
+class Sleep9Pass(BasePass):
     async def run(self, circuit: Circuit, data: PassData) -> None:
         circuit.append_gate(ZGate(), 0)
-        time.sleep(0.3)
-        data['key'] = '3'
+        time.sleep(0.9)
+        data['key'] = '9'
 
 
 def pick_z(c1: Circuit, c2: Circuit) -> bool:
@@ -66,7 +66,7 @@ def test_parallel_do_no_passes() -> None:
 
 
 def test_parallel_do_pick_first(compiler: Compiler) -> None:
-    passes: list[list[BasePass]] = [[Sleep3Pass()], [Sleep1Pass()]]
+    passes: list[list[BasePass]] = [[Sleep9Pass()], [Sleep1Pass()]]
     pd_pass = ParallelDo(passes, pick_z, True)
     _, data = compiler.compile(Circuit(1), pd_pass, True)
     assert data['key'] == '1'
diff --git a/tests/runtime/test_attached.py b/tests/runtime/test_attached.py
index 0c2ecb67e..aa155dc3c 100644
--- a/tests/runtime/test_attached.py
+++ b/tests/runtime/test_attached.py
@@ -17,16 +17,16 @@
 from bqskit.runtime import get_runtime
 
 
-@pytest.mark.parametrize('num_workers', [1, -1])
-def test_startup_shutdown_transparently(num_workers: int) -> None:
-    in_num_childs = len(psutil.Process(os.getpid()).children(recursive=True))
-    compiler = Compiler(num_workers=num_workers)
-    assert compiler.p is not None
-    compiler.__del__()
-    if sys.platform == 'win32':
-        time.sleep(1)
-    out_num_childs = len(psutil.Process(os.getpid()).children(recursive=True))
-    assert in_num_childs == out_num_childs
+# @pytest.mark.parametrize('num_workers', [1, -1])
+# def test_startup_shutdown_transparently(num_workers: int) -> None:
+#     in_num_childs = len(psutil.Process(os.getpid()).children(recursive=True))
+#     compiler = Compiler(num_workers=num_workers)
+#     assert compiler.p is not None
+#     compiler.__del__()
+#     if sys.platform == 'win32':
+#         time.sleep(1)
+#     out_num_childs = len(psutil.Process(os.getpid()).children(recursive=True))
+#     assert in_num_childs == out_num_childs
 
 
 @pytest.mark.parametrize('num_workers', [1, -1])
@@ -60,15 +60,17 @@ def test_create_workers(num_workers: int) -> None:
     compiler.close()
 
 
-def test_one_thread_per_worker() -> None:
-    # On windows we aren't sure how the threads are handeled
+def test_two_thread_per_worker() -> None:
     if sys.platform == 'win32':
-        return
+        pytest.skip('Not sure how to count threads on Windows.')
+
+    if sys.platform == 'darwin':
+        pytest.skip('MacOS requires permissions to count threads.')
 
     compiler = Compiler(num_workers=1)
     assert compiler.p is not None
     assert len(psutil.Process(compiler.p.pid).children()) in [1, 2]
-    assert psutil.Process(compiler.p.pid).children()[0].num_threads() == 1
+    assert psutil.Process(compiler.p.pid).children()[0].num_threads() == 2
     compiler.close()
 
 
diff --git a/tests/runtime/test_next.py b/tests/runtime/test_next.py
index 30642a7d7..81c45da5a 100644
--- a/tests/runtime/test_next.py
+++ b/tests/runtime/test_next.py
@@ -29,7 +29,7 @@ async def run(self, circuit: Circuit, data: PassData) -> None:
 
 class TestNoDuplicateResultsInTwoNexts(BasePass):
     async def run(self, circuit: Circuit, data: PassData) -> None:
-        future = get_runtime().map(sleepi, [0.3, 0.4, 0.1, 0.2])
+        future = get_runtime().map(sleepi, [0.3, 0.4, 0.1, 0.2, 5])
         seen = [0]
         int_ids = await get_runtime().next(future)