process-executor

process-executor is a general-purpose execution framework for running ordinary Python classes in multiprocessing worker processes. Its primary goal is to reuse existing, independently developed device-control (Ctrl) classes without rewriting them solely for multiprocessing.

The code, including its design and underlying policies, is still under evaluation and will be refined through practical use.

This README serves as both a specification and a tutorial. In addition to installation instructions, it documents the current design rationale, separation of responsibilities, internal architecture, public API, limitations, and working examples.

---

Quick Start

Installation

Run from the repository root:

pip install -e .

Minimal setup

Pass an existing class as worker_class:

from process_executor import ProcessExecutor


class Calculator:
    def add(self, a: int, b: int) -> int:
        return a + b


def main() -> None:
    with ProcessExecutor(
        worker_class=Calculator,
        worker_count=2,
        start_method="spawn",
    ) as executor:
        job_id = executor.submit("add", 10, 20)
        result = executor.wait(job_id, timeout_sec=5.0)
        print(result)


if __name__ == "__main__":
    main()

No multiprocessing-specific implementation is added to Calculator. Each worker process creates its own Calculator() instance. Calling submit("add", 10, 20) executes add(10, 20) on that worker-local instance.

Handling jobs asynchronously

job_id = executor.submit("add", 10, 20)

print(executor.state(job_id))
print(executor.job(job_id))
print(executor.workers())

result() does not wait. Use wait() when you need to wait for completion.

---

How to Read This README

The recommended reading order is:

1. Quick Start — Run the minimal example.
2. Why This Project Was Created — Understand its purpose.
3. Core Design — Understand why existing Ctrl classes are separated from multiprocessing.
4. Workers / Jobs / SharedContext — Understand the execution model.
5. Examples — See concrete applications to generic classes, existing controllers, and PM2CD.
6. Cases Requiring Additional Design / Responsibility Boundaries — Understand what ProcessExecutor deliberately does not handle.

In addition to this manually maintained documentation, GitHub Pages provides an API reference generated by Sphinx from src/process_executor.

---

1. Why This Project Was Created

Device-control code is naturally implemented first as an ordinary synchronous class:

class MotorCtrl:
    def move(
        self,
        motor: int,
        pulse: int,
    ) -> bool:
        ...

    def get_position(
        self,
        motor: int,
    ) -> int:
        ...

This class is responsible for device protocols, commands, status retrieval, motion, stopping, and related operations.

As a control system grows, requirements arise that are separate from the device protocol:

• Decouple long-running operations from their callers.
• Submit operations as jobs and receive job IDs.
• Track job states: queued, running, succeeded, and failed.
• Prestart and reuse multiple worker processes.
• Inspect worker status from the parent process.
• Preserve exceptions raised inside workers as job failures.
• Separate job submission from result retrieval in higher-level servers such as FastAPI.
• Avoid reimplementing Process, Queue, Manager, and job management for each device.

These concerns are not part of a Ctrl class's primary responsibility of operating a device.

Embedding multiprocessing directly into each Ctrl class tightly couples device-specific logic to process management. It also leads to duplicated worker loops, job queues, and state management for every new device.

process-executor was created to extract these shared multiprocessing concerns into an independent component.

---

2. Core Design

The central design principle of the current implementation is:

Rather than rewriting a Ctrl class for multiprocessing, instantiate the existing Ctrl class inside a worker process and execute its public methods as jobs.

Conceptually:

Application / Server
        |
        v
Multi Ctrl / Facade
        |
        v
ProcessExecutor
        |
        +-- job queue
        +-- job metadata
        +-- worker metadata
        +-- shared state
        +-- shared lock
        |
        +-- Worker Process 1
        |       |
        |       +-- Ctrl instance
        |
        +-- Worker Process 2
                |
                +-- Ctrl instance

ProcessExecutor has no knowledge of device protocols, TCP, serial communication, motors, stages, or similar hardware.

Inside a worker process, the essential operation is simply:

handler = getattr(instance, handler_name)
result = handler(*args, **kwargs)

Thus, ProcessExecutor is device-independent.

---

3. Why Not Create Dedicated Worker Ctrl Classes?

Suppose an existing controller already looks like this:

class ExistingCtrl:
    def __init__(
        self,
        host: str,
    ) -> None:
        ...

    def move(
        self,
        position: int,
    ) -> bool:
        ...

Creating dedicated classes solely for multiprocessing:

ExistingCtrl
    ↓
ExistingWorkerCtrl
    ↓
ExistingWorkerFactory

would duplicate much of the original constructor and delegation logic.

To avoid this duplication, the current implementation provides the worker_class interface:

executor = ProcessExecutor(
    worker_class=ExistingCtrl,
    worker_kwargs={
        "host": "192.168.0.10",
    },
    worker_count=2,
    start_method="spawn",
)

Each worker process instantiates ExistingCtrl(...). The existing Ctrl class does not need to be aware of multiprocessing.

---

4. Why Composition / Delegation Instead of Inheritance?

Device-side Ctrl classes do not inherit from ProcessExecutor.

The relationship is:

Multi Ctrl
    has-a
ProcessExecutor

In other words, composition and delegation are used. This preserves the following properties:

• Multiprocessing does not dictate the device Ctrl class hierarchy.
• Existing Ctrl classes can be reused without modification.
• ProcessExecutor remains device-independent.
• Each device can define its own multi-controller public API.
• ProcessExecutor can be reused across different control systems.

---

5. Worker Model

A worker is not a process dedicated to a particular handler.

The design does not assign separate processes to move(), get_position(), query(), and so on.

Instead, each worker retains one worker-local instance and executes different methods according to jobs received from the queue.

Worker 1
  Ctrl instance
    |- query()
    |- get()
    |- move()
    `- ...

Worker 2
  Ctrl instance
    |- query()
    |- get()
    |- move()
    `- ...

Handler types and worker lifetimes are independent.

---

6. Creating Worker Instances

There are two ways to create worker-local instances.

6.1 worker_class

Use this to run an ordinary existing class directly in workers:

executor = ProcessExecutor(
    worker_class=MyCtrl,
    worker_args=(arg1, arg2),
    worker_kwargs={
        "option": value,
    },
    worker_count=2,
    start_method="spawn",
)

Conceptually, each worker process executes:

instance = MyCtrl(
    *worker_args,
    **worker_kwargs,
)

This is the default approach when an existing Ctrl class can be used directly as a worker-local instance.

6.2 worker_factory

Use this when worker creation requires SharedContext:

def create_ctrl(
    context: SharedContext,
) -> MyCtrl:
    return MyCtrl(
        shared_state=context.state,
        lock=context.lock,
    )


executor = ProcessExecutor(
    worker_factory=create_ctrl,
    worker_count=2,
    start_method="spawn",
)

The worker_factory(context) interface allows shared state and a process-safe lock to be passed into a worker instance.

6.3 Relationship Between the Two

worker_factory and worker_class cannot be specified together.

worker_args and worker_kwargs are supported only with worker_class.

---

7. Job Execution Model

Submit a job with submit():

job_id = executor.submit(
    "move",
    motor,
    pulse,
)

submit() does not wait for the handler to complete. It returns a job ID, not the handler's return value.

State transitions are:

queued
  |
  v
running
  |
  +--> succeeded
  |
  `--> failed

Job metadata contains:

• job_id
• handler
• args
• kwargs
• state
• worker_id
• result
• error
• created_at
• started_at
• finished_at

Retrieve one job with job(job_id) or all jobs with jobs().

---

8. Roles of submit(), result(), and wait()

submit(handler, *args, **kwargs) -> str

Enqueues a job and returns its job ID.

result(job_id)

Retrieves the result of a completed job.

This method does not wait. It raises RuntimeError if the job has not reached succeeded.

wait(job_id, timeout_sec=None, interval_sec=0.01)

Explicitly waits for job completion.

If the job succeeded, it returns the handler's return value.

If the job failed, it raises RuntimeError.

If timeout_sec is exceeded, it raises TimeoutError.

This separation allows higher-level Ctrl classes to expose asynchronous or synchronous APIs as appropriate.

---

9. Worker Metadata

Worker metadata includes:

• worker_id
• pid
• state
• current_job_id
• current_handler

Worker states are:

• idle
• running
• stopped
• dead

Retrieve one worker with worker(worker_id) or all workers with workers().

Calls to submit(), worker(), and workers() check for dead workers and replenish them if the number of workers falls below the configured worker_count.

---

10. SharedContext

SharedContext is a container for these shared objects:

context.jobs
context.workers
context.state
context.lock

jobs

A shared mapping of job metadata managed by ProcessExecutor.

workers

A shared mapping of worker metadata managed by ProcessExecutor.

state

A shared mapping available for application-specific use.

The parent process can access the same SharedContext through ProcessExecutor.shared_context:

context = executor.shared_context
context.state["key"] = value

lock

A process-safe lock created using multiprocessing.Manager().RLock().

This lock is deliberately not given a device-specific name such as motion_lock or communication_lock. The consuming Ctrl class decides which handlers need to acquire it.

---

11. Exception Handling

If an Exception occurs while executing a handler on a worker-local instance, the job becomes failed.

Error metadata records:

• Exception type
• Message
• Traceback

Calling result() or wait() on a failed job raises RuntimeError.

The current implementation does not reconstruct and re-raise the original exception class from the worker in the parent process.

Importantly, this exception handling covers handler execution only. If an exception occurs while creating a worker-local instance through worker_factory(context) or worker_class(...), the current implementation has no mechanism to report it to the parent as a job failure. The worker process exits if instance creation fails.

Likewise, if a worker disappears without passing through ordinary Python Exception handling—for example, if the process is forcibly terminated during handler execution—the job that was running is not automatically changed to failed. Calling wait() without a timeout for such a job may therefore continue waiting indefinitely.

---

12. Worker Lifecycle

The constructor starts the configured number of worker processes.

The current implementation does not wait for every worker to finish registering as idle before the constructor returns. After startup, each worker creates its worker-local instance and then registers its metadata as idle.

Jobs may be queued before workers are ready and are processed once a worker becomes available.

close() places stop sentinels for live workers, calls join() on the workers, and then shuts down the Queue and Manager.

Currently, join() has no timeout. If a worker is executing a handler that never returns, close() may also continue waiting for that worker indefinitely.

Workers are started with daemon=True. Therefore, the design does not assume that worker-local handlers will create additional multiprocessing child processes.

Calling close() multiple times has no effect after the first call.

The executor can also be used as a context manager:

with ProcessExecutor(
    worker_class=MyCtrl,
) as executor:
    ...

The termination API is close(), not shutdown().

---

13. Logger

The ProcessExecutor constructor accepts a logger:

executor = ProcessExecutor(
    worker_class=MyCtrl,
    logger=logger,
)

In the current implementation, the logger is retained only in the parent process; ProcessExecutor core does not automatically inject it into worker processes.

Aggregating rotating log files written directly by multiple processes is outside the responsibility of ProcessExecutor core.

---

14. Directory Structure

The main source and example layout is shown below. Generated artifacts such as build/, *.egg-info, and __pycache__/ are omitted.

process-executor/
|- pyproject.toml
|- README.md
|- CHANGELOG.md
|- src/
|  `- process_executor/
|     |- __init__.py
|     |- executor.py
|     `- shared_context.py
|- tests/
|  |- test_process_executor.py
|  `- test_example.py
`- example/
   |- dummy_single.py
   |- dummy_ctrl.py
   |- example.py
   `- pm2cd/
      |- single/
      |  |- client_sync.py
      |  `- server.py
      |- multi/
      |  |- client_sync.py
      |  |- pm2cd_multi.py
      |  `- server.py
      |- multi_shared/
      |  |- client_sync.py
      |  |- pm2cd_multi.py
      |  `- server.py
      |- try_move_single_local.py
      |- try_move_single_client.py
      |- try_move_multi_local.py
      |- try_move_multi_client.py
      |- try_move_multi_shared_local.py
      |- try_move_multi_shared_client.py
      `- x1_multi/

tests/ contains the pytest test suite.

example/ is not an installed package; it contains samples and tutorials intended for direct execution.

example/pm2cd/x1_multi/ preserves the previous PM2CD multi example replaced in v0.1.1. The current examples use multi/ and multi_shared/.

---

15. Example 1: Running a Custom Class with Multiple Workers

The relevant files are:

example/dummy_single.py
example/dummy_ctrl.py
example/example.py

15.1 dummy_single.py

DummySingle is an ordinary synchronous class with no knowledge of multiprocessing:

class DummySingle:
    def query(
        self,
        name: str,
    ) -> str:
        return "dummy:{}".format(name)

    def get(
        self,
        value: int,
        delay_sec: float = 0.5,
    ) -> int:
        time.sleep(delay_sec)
        return value

15.2 dummy_ctrl.py

DummyCtrl composes a ProcessExecutor.

This example demonstrates the worker_factory interface:

def create_dummy_single(
    context: SharedContext,
) -> DummySingle:
    _ = context
    return DummySingle()

DummyCtrl.get() and DummyCtrl.query() return job IDs rather than results:

job_id = self._executor.submit(
    "get",
    value,
    delay_sec,
)

15.3 example.py

Creates two workers and submits a time-consuming get() and a query() as separate jobs:

cd example
python example.py

This example illustrates the basic separation between a single-process class and a multiprocessing facade.

---

16. Example 2: Running an Existing Control Class with Multiple Workers

The PM2CD examples progressively demonstrate single, basic multi, and shared multi configurations.

example/pm2cd/
|- single/
|  |- client_sync.py
|  `- server.py
|- multi/
|  |- client_sync.py
|  |- pm2cd_multi.py
|  `- server.py
|- multi_shared/
|  |- client_sync.py
|  |- pm2cd_multi.py
|  `- server.py
|- try_move_single_local.py
|- try_move_single_client.py
|- try_move_multi_local.py
|- try_move_multi_client.py
|- try_move_multi_shared_local.py
`- try_move_multi_shared_client.py

16.1 Single

try_move_single_local.py is a baseline example that uses the existing PM2CD single class directly.

single/server.py and single/client_sync.py demonstrate the same single class through a FastAPI server/client.

The single class itself is not rewritten for process-executor.

16.2 multi

multi/pm2cd_multi.py is the basic multi example using multiple worker processes.

Each worker creates an existing PM2CDsingle instance. However, to prevent each worker from owning an independent TCP connection, physical communication is consolidated into a single Communication instance in the manager process.

Conceptually:

PM2CDsingle worker 1 --+
PM2CDsingle worker 2 --+--> Communication proxy
PM2CDsingle worker N --+          |
                                  v
                           Communication x 1
                                  |
                                  v
                           TCP connection x 1
                                  |
                                  v
                                PM2CD

The Communication dependency is injected into each worker through the PM2CDsingle(..., communication=...) injection point, using the same proxy for all workers.

Thus, there are multiple worker processes but only one physical communication path to the PM2CD.

No PM2CD-specific worker class is introduced; controller logic continues to use the existing PM2CDsingle.

The number of workers is specified by the multi facade constructor's worker_count.

Example scripts use the in-code WORKER_COUNT setting rather than command-line options.

The basic multi configuration does not share controller-local motion context. Accordingly, the sample wait_for_complete() avoids reliance on worker-local state: it checks a move explicitly issued with backlash=False and an explicit target_position.

16.3 multi_shared

multi_shared/pm2cd_multi.py extends the single-Communication configuration of multi by sharing PM2CD motion context among worker processes.

It creates the existing PM2CDsingle via ProcessExecutor's worker_factory(context) and uses the worker's SharedContext.state as the PM2CDsingle _move_context.

                 SharedContext.state
                         ^
                         |
        +----------------+----------------+
        |                |                |
   PM2CDsingle      PM2CDsingle      PM2CDsingle
    worker 1         worker 2         worker N
        \                |                /
         +------ Communication proxy -----+
                         |
                         v
                  Communication x 1
                         |
                         v
                       PM2CD

PM2CD motion context stores information carried from move() / move_relative() to wait_for_complete(), including target_position, whether backlash is enabled, and the backlash value.

Sharing this context lets wait_for_complete() use the same motion information even when it runs on a different worker from the one that executed move().

The sharing mechanism is internal to the multi facade; it does not add a shared-specific API for callers.

16.4 Local and FastAPI Server/Client

Each configuration can be tested locally or through a FastAPI server/client:

single
  try_move_single_local.py
  try_move_single_client.py

multi
  try_move_multi_local.py
  try_move_multi_client.py

multi_shared
  try_move_multi_shared_local.py
  try_move_multi_shared_client.py

The multi facade's _execute() internally combines submit() and wait(), so the public API in these PM2CD examples appears synchronous.

This is a design decision of the PM2CD multi facade, not a synchronous API requirement imposed by ProcessExecutor itself.

---

17. Multiple-Worker PM2CD Configuration

ProcessExecutor creates one worker-local instance per worker.

Therefore, with worker_count=N, the PM2CD examples create N PM2CDsingle instances.

However, in the v0.1.1 PM2CD multi / multi_shared examples, each worker does not create its own Communication.

Instead, a single Communication instance is created in the manager process, and its proxy is injected into every PM2CDsingle.

The resulting configuration is:

controller instance: N
worker process:      N
Communication:       1
TCP connection:      1

Crucially, this communication-consolidation feature is not built into ProcessExecutor core.

ProcessExecutor remains unaware of device protocols and TCP connections. The PM2CD example manages physical resource ownership using the PM2CD dependency-injection point and BaseManager.

The previous worker_count=1 example is preserved in example/pm2cd/x1_multi/.

---

18. When Is It Useful?

18.1 When Existing Ctrl Classes Must Remain Unchanged

When a class already works independently and only its execution location needs to move into worker processes.

18.2 When Multiprocessing Infrastructure Should Not Be Reimplemented per Device

When job queues, worker loops, job metadata, worker metadata, and result management should be shared.

18.3 When Separating Higher-Level Servers from Actual Execution

It can serve as the execution layer beneath FastAPI or another server that submits jobs and delegates work to worker processes.

18.4 When Managing State by Job ID

submit
  ↓
job ID
  ↓
state
  ↓
result / wait

18.5 When Processes Should Not Be Created for Each Handler

When prestarted generic workers should be reused.

---

19. Cases Requiring Additional Design

19.1 Sharing a Physical Resource Across Multiple Workers

If multiple worker processes need access to one TCP socket, serial port, USB handle, or similar resource, device-specific architecture is required, such as a resource-owner process and an IPC proxy.

ProcessExecutor core does not provide this broker functionality.

As a concrete example, the v0.1.1 PM2CD multi / multi_shared configurations place a manager-owned Communication on the PM2CD example side and inject its proxy into the PM2CDsingle instances of multiple workers.

19.2 Sharing the Same Python Object as Threads Do

Processes have separate address spaces. This executor does not share ordinary object references in the same way as threads.

When controller state must be shared, use SharedContext.state and SharedContext.lock as needed.

19.3 Ultra-Low-Latency Shared-Memory Parallelism

The current implementation uses multiprocessing.Manager and Queue. It is not intended as a shared-memory executor for high-frequency numerical computing.

---

20. Requirements When Using spawn

When using spawn, for example on macOS, structure the entry-point script as follows:

def main() -> None:
    ...


if __name__ == "__main__":
    main()

Worker classes, worker factories, constructor arguments, and other objects must be usable across process boundaries.

The examples in this repository follow this pattern.

---

21. Tests

The pytest suite is in tests/:

pytest

tests/test_process_executor.py verifies:

• submit() is non-blocking.
• Multiple workers execute independent jobs concurrently.
• Workers can access SharedContext.state and lock.
• result() does not wait implicitly.
• An Exception in a worker-local handler causes a failed job.
• Arguments and return values cross process boundaries.
• The configured worker count is maintained.
• submit() is rejected after close().
• worker_class can instantiate existing classes directly.
• worker_factory and worker_class are mutually exclusive.

tests/test_example.py verifies the generic example.

The PM2CD examples depend on physical hardware, so pytest does not operate the actual device.

---

22. Installation

Run from the repository root:

pip install -e .

The version in pyproject.toml is 0.1.1.

Python 3.10 or later is supported.

---

23. Minimal Usage Example

Suppose the existing class is:

class MyCtrl:
    def __init__(
        self,
        base: int,
    ) -> None:
        self._base = base

    def run(
        self,
        value: int,
    ) -> int:
        return self._base + value

It can be executed without modifying the class itself:

from process_executor import ProcessExecutor


def main() -> None:
    executor = ProcessExecutor(
        worker_class=MyCtrl,
        worker_kwargs={
            "base": 10,
        },
        worker_count=2,
        start_method="spawn",
    )

    try:
        job_id = executor.submit(
            "run",
            5,
        )

        result = executor.wait(
            job_id,
            timeout_sec=5.0,
        )

        print(result)
    finally:
        executor.close()


if __name__ == "__main__":
    main()

The result is 15.

---

24. Current Responsibility Boundaries

ProcessExecutor is responsible for:

• Creating worker processes
• Creating worker-local instances
• Job queue
• Job IDs
• Job states
• Job results and errors
• Worker metadata
• Shared state
• Generic process-safe lock
• Detecting and replenishing dead workers
• Worker lifecycle
• Shutting down the Queue and Manager

It is not responsible for:

• Device protocols
• TCP / serial commands
• Device-specific mutual-exclusion rules
• Determining when physical device motion is complete
• Device-specific error recovery
• Implementing device-specific brokers that share one Communication instance across multiple workers
• External servers such as FastAPI / gRPC
• Process-safe rotating-log aggregation

Maintaining this separation of responsibilities is central to the current design.