Metadata-Version: 2.4
Name: proc-executor
Version: 0.1.4
Summary: Reusable multiprocessing execution support for controller classes
Author-email: Kengo NAKADA <kengo.nakada@email.com>
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: xlogger-py
Requires-Dist: dataclasses
Requires-Dist: coloredlogs
Requires-Dist: colorlog
Requires-Dist: pytest
Requires-Dist: pytest-asyncio
Requires-Dist: pyyaml
Requires-Dist: toml
Requires-Dist: nest_asyncio
Requires-Dist: sphinx
Requires-Dist: myst-parser
Requires-Dist: sphinx-autobuild
Requires-Dist: sphinx-rtd-theme
Requires-Dist: sphinxcontrib-napoleon
Requires-Dist: linkify-it-py
Dynamic: license-file

# process-executor

[![en](https://img.shields.io/badge/lang-en-red.svg)](https://github.com/ohara-lab-su/process_executor/blob/main/README.en.md)
[![ja](https://img.shields.io/badge/lang-ja-yellow.svg)](https://github.com/ohara-lab-su/process_executor/blob/main/README.md)

`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:

```bash
pip install -e .
```

### Minimal setup

Pass an existing class as `worker_class`:

```python
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

```python
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:

```python
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:

```text
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:

```python
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:

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

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

Creating dedicated classes solely for multiprocessing:

```text
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:

```python
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:

```text
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.

```text
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:

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

Conceptually, each worker process executes:

```python
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`:

```python
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()`:

```python
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:

```text
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:

```python
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`:

```python
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:

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

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

---

## 13. Logger

The `ProcessExecutor` constructor accepts a `logger`:

```python
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.

```text
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:

```text
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:

```python
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:

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

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

```python
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:

```bash
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.

```text
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:

```text
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`.

```text
                 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:

```text
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:

```text
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

```text
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:

```python
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/`:

```bash
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:

```bash
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:

```python
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:

```python
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.
