Python Worker Dispatcher

https://github.com/yidas/python-worker-dispatcher
Admin
Python Worker Dispatcher is a flexible, lightweight concurrency framework for Python to orchestrate...
Tokens:3,668
Snippets:38
Trust Score:8.6
License:MIT
Update:2 months ago
Show doc for...
Tokens:
Raw
### Run Worker Dispatcher Demonstration

Source: https://github.com/yidas/python-worker-dispatcher/blob/main/README.md

Example output of the worker dispatcher execution process.

```bash
$ python3 main.py

Worker Dispatcher Configutation:
- Local CPU core: 10
- Tasks Count: 100
- Runtime: Unlimited
- Dispatch Mode: Fixed Workers (Default)
- Concurrency Info:
  ├─ Execution Type: Processing
  ├─ Configured Workers: 10 Worker(s)
  ├─ Pool Structure:
  │  └─ Main Pool : 10 Process(es)
  └─ Total Concurrency: 10 Active Worker(s)

--- Start to dispatch workers at 2024-06-14T17:46:30.996685+08:00 ---

...(User-defined output)...

--- End of worker dispatch at 2024-06-14T17:46:41.420888+08:00---

Spend Time: 10.424203 sec
Completed Tasks Count: 100
Uncompleted Tasks Count: 0
Undispatched Tasks Count: 0
```

--------------------------------

### Install Worker Dispatcher

Source: https://github.com/yidas/python-worker-dispatcher/blob/main/README.md

Command to install the library via pip.

```shell
$ pip install worker-dispatcher
```

--------------------------------

### Start Method Signature

Source: https://github.com/yidas/python-worker-dispatcher/blob/main/README.md

The primary method for initiating task dispatching with a configuration dictionary.

```python
def start(config: dict) -> list
```

--------------------------------

### TPS Report Output Example

Source: https://github.com/yidas/python-worker-dispatcher/blob/main/README.md

Example console output when running get_tps with debug and multiprocessing enabled.

```bash
--- Start calculating the TPS data ---
  - Average TPS: 0.83, Total Duration: 1202.3867809772491s, Success Count: 999
--- Start to compile intervals with an interval of 13 seconds ---
  - Interval - Start Time: 1734937209.851285, End Time: 1734937222.851285, TPS: 51.23
    * Peak detected above the current TPS threshold - Interval TPS: 51.23, Main TPS: 0.83
  - Interval - Start Time: 1734937222.851285, End Time: 1734937235.851285, TPS: 18.0
  - Interval - Start Time: 1734937235.851285, End Time: 1734937248.851285, TPS: 0.0
  ...
  - Interval - Start Time: 1734938405.851285, End Time: 1734938412.238066, TPS: 0.0
--- Start to find the peak TPS ---
  - Detecting from Start Time: 1734937210, Count: 67, Current TPS Threshold: 51.23, Worker: 104
    * Peak detected above the current TPS threshold - TPS: 53.5, Started at: 1734937210, Ended at: 1734937220
    * Peak detected above the current TPS threshold - TPS: 53.857142857142854, Started at: 1734937210, Ended at: 1734937224
    * Peak detected above the current TPS threshold - TPS: 55.13333333333333, Started at: 1734937210, Ended at: 1734937225
    * Peak detected above the current TPS threshold - TPS: 55.166666666666664, Started at: 1734937210, Ended at: 1734937228
  - Detecting from Start Time: 1734937224, Count: 73, Current TPS Threshold: 55.166666666666664, Worker: 105
  ...
  - Detecting from Start Time: 1734937212, Count: 82, Current TPS Threshold: 55.166666666666664, Worker: 102
    * Peak detected above the current TPS threshold - TPS: 55.53846153846154, Started at: 1734937212, Ended at: 1734937225
```

--------------------------------

### Retrieve Dispatcher Execution Metrics

Source: https://github.com/yidas/python-worker-dispatcher/blob/main/README.md

Example output format for dispatcher lifecycle metrics.

```python
{'started_at': 1782288809.0990121, 'ended_at': 1782288813.567204, 'duration': 4.468191862106323}
```

--------------------------------

### worker_dispatcher.start(config)

Source: https://github.com/yidas/python-worker-dispatcher/blob/main/README.md

Initiates the task dispatcher with the provided configuration dictionary. Returns a list of results from the executed tasks.

```APIDOC
## worker_dispatcher.start(config)

### Description
Starts the worker dispatcher to execute tasks based on the provided configuration. It manages worker pools, task distribution, and optional callbacks.

### Parameters
- **config** (dict) - Required - Configuration dictionary containing:
  - **debug** (bool) - Optional - Debug mode toggle.
  - **task** (dict) - Required - Task definitions including 'list' (tasks), 'function' (execution logic), 'config', and 'callback'.
  - **worker** (dict) - Optional - Worker settings including 'number', 'frequency_mode', 'use_processing', and 'parallel_processing'.
  - **runtime** (int) - Optional - Max runtime in seconds.
  - **verbose** (bool) - Optional - Verbose output toggle.

### Response
- **results** (list) - A list containing the return values of each executed task.
```

--------------------------------

### print()

Source: https://github.com/yidas/python-worker-dispatcher/blob/main/README.md

A custom print function wrapper.

```APIDOC
## print(*objects, sep=' ', end='\n', file=None, flush=True)

### Description
A custom print function that sets `flush=True` by default to ensure immediate output to stdout.
```

--------------------------------

### Initialize Worker Dispatcher

Source: https://github.com/yidas/python-worker-dispatcher/blob/main/README.md

Configures and executes the worker dispatcher with task functions, worker counts, and processing modes.

```python
import worker_dispatcher 

results = worker_dispatcher.start({
    'debug': False,
    'task': {
        'list': [],                         # Support list and integer. Integer represent the number of tasks to be generated.
        'function': task_function_sample,   # The main function to execute per task
        'config': {},
        'callback': {
            'on_done': False,               # Called with each task's result after each task completes; the return value will overwrite and define the task result
        }
    },
    'worker': {
        'number': 8,
        'frequency_mode': {             # Changing from assigning tasks to a fixed number of workers once, to assigning tasks and workers frequently.
            'enabled': False, 
            'interval': 1,              # The second(s) of interval
            'accumulated_workers': 0,   # Accumulate the number of workers for each interval for next dispatch. Can be set as a negative number.
            'max_workers': None,        # limit the maximum number of workers to prevent system exhaustion.
        },
        'use_processing': False,        # To break GIL, workers will be based on processing pool.
        'parallel_processing': {        # To break GIL and require a number of workers greater than the number of CPU cores.
            'enabled': False,           # `worker.use_processing` setting will be ignored when enabled. The actual number of workers will be adjusted to a multiple of the CPU core count.
            'use_queue': False,         # Enable a task queue to specify the number of workers without adjustment, though the maximum may be limited by your device.
        },   
    },
    'runtime': None,                    # Dispatcher max runtime in seconds
    'verbose': True
})
```

--------------------------------

### Enable immediate stdout flushing

Source: https://github.com/yidas/python-worker-dispatcher/blob/main/README.md

Use worker_dispatcher.print() to ensure output is flushed immediately when running in background processes.

```python
import worker_dispatcher

def each_task(id, config, task, metadata):
    # Print immediately to file (stdout is flushed)
    if id % 10 == 0:
        worker_dispatcher.print(f"TaskId: {id}")
    return True

responses = worker_dispatcher.start({
    'task': {
        'list': 600,
        'function': each_task,
        ...
})
```

--------------------------------

### task.function

Source: https://github.com/yidas/python-worker-dispatcher/blob/main/README.md

The main function executed per task. This function receives the task data and configuration, and its return value is used to determine task success or failure.

```APIDOC
## task.function

### Description
The main function to execute per task.

### Signature
`task_function (id: int, config, task, metadata: dict) -> Any`

### Arguments
- **id** (int) - The sequence number generated by each task starting from 1.
- **config** (multitype) - The custom variable to be passed to the callback function.
- **task** (multitype) - Each value from the task.list.
- **metadata** (dict) - A user-defined dictionary for custom metadata per task, saved in its log.
```

--------------------------------

### get_logs()

Source: https://github.com/yidas/python-worker-dispatcher/blob/main/README.md

Returns a list of all task logs containing execution details for each task.

```APIDOC
## get_logs()

### Description
Returns a list of all task logs after `start()` has completed. Each log is a dictionary containing task_id, started_at, ended_at, duration, result, and metadata.
```

--------------------------------

### Perform a stress test with worker_dispatcher

Source: https://github.com/yidas/python-worker-dispatcher/blob/main/README.md

Executes a task list with a defined number of workers and frequency control to simulate load.

```python
import worker_dispatcher

def each_task(id, config, task, metadata):
    response = None
    try:
        response = requests.get(config['my_endpoint'], timeout=(5, 10))
    except requests.exceptions.RequestException as e:
        print("An error occurred:", e)
    return response

responses = worker_dispatcher.start({
    'task': {
        'list': 600,
        'function': each_task,
        'config': {
            'my_endpoint': 'https://your.name/api'
        },
    },
    # Light Load with 10 RPS
    'worker': {
        'number': 10,
        'frequency_mode': {
            'enabled': True, 
            'interval': 1,
        },
    },
})

print(worker_dispatcher.get_logs())
print(worker_dispatcher.get_tps())
```

--------------------------------

### get_tps()

Source: https://github.com/yidas/python-worker-dispatcher/blob/main/README.md

Generates a Transactions Per Second (TPS) report based on task logs.

```APIDOC
## get_tps(logs=None, display_intervals=False, interval=0, reverse_interval=False, use_processing=False, verbose=False, debug=False)

### Description
Generates a TPS report as a dictionary based on task logs. The `logs` argument defaults to internal logs if not provided.
```

--------------------------------

### get_results()

Source: https://github.com/yidas/python-worker-dispatcher/blob/main/README.md

Retrieves a list of all task return values after the dispatcher has completed execution.

```APIDOC
## get_results()

### Description
Retrieves a list of all task return values after `start()` has completed. Returns an array/list containing only the custom return values from each executed task function.
```

--------------------------------

### task.callback.on_done

Source: https://github.com/yidas/python-worker-dispatcher/blob/main/README.md

The callback function invoked after each task completes. It allows for post-processing of task results.

```APIDOC
## task.callback.on_done

### Description
The callback function is called with each task's result after each task completes; the return value will overwrite and define the task result.

### Signature
`callback_on_done_function (id: int, config, result, log: dict) -> Any`

### Arguments
- **id** (int) - The sequence number generated by each task starting from 1.
- **config** (multitype) - The custom variable to be passed to the callback function.
- **result** (multitype) - Each value returned back from task.callback.
- **log** (dict) - Reference to the task log.
```

--------------------------------

### Compute Tasks Using All CPU Cores

Source: https://github.com/yidas/python-worker-dispatcher/blob/main/README.md

Utilizes multi-processing to distribute heavy computational tasks across all available CPU cores.

```python
import worker_dispatcher

def each_task(id: int, config, task, metadata):
    result = sum(id * i for i in range(10**9))
    return result

if __name__ == '__main__':
    results = worker_dispatcher.start({
        'task': {
            'list': 10,
            'function': each_task,
        },
        'worker': {   
            'use_processing': True
        }
    })
```

--------------------------------

### Define get_tps Method Signature

Source: https://github.com/yidas/python-worker-dispatcher/blob/main/README.md

The function signature for generating Transactions Per Second reports.

```python
def get_tps(logs: dict=None, display_intervals: bool=False, interval: float=0, reverse_interval: bool=False, use_processing: bool=False, verbose: bool=False, debug: bool=False,) -> dict:
```

--------------------------------

### get_result_info()

Source: https://github.com/yidas/python-worker-dispatcher/blob/main/README.md

Retrieves execution metrics for the dispatcher itself.

```APIDOC
## get_result_info()

### Description
Retrieves a dictionary containing the execution metrics of the dispatcher itself after `start()` has completed, including started_at, ended_at, and duration.
```

--------------------------------

### Run script as a background process

Source: https://github.com/yidas/python-worker-dispatcher/blob/main/README.md

Command to execute a Python script in the background while redirecting both stdout and stderr to a log file.

```bash
$ python3 main.py > log.txt 2>&1 &
```

--------------------------------

### Dispatch HTTP Requests Concurrently

Source: https://github.com/yidas/python-worker-dispatcher/blob/main/README.md

Uses 20 threads to dispatch HTTP requests defined in a task list.

```python
import worker_dispatcher
import requests

def each_task(id: int, config, task, metadata):
    response = requests.get(config['my_endpoint'] + task)
    return response

responses = worker_dispatcher.start({
    'task': {
        'list': ['ORD_AH001', 'ORD_KL502', '...' , 'ORD_GR393'],
        'function': each_task,
        'config': {
            'my_endpoint': 'https://your.name/order-handler/'
        },
    },
    'worker': {
        'number': 20,
    }
})
```

--------------------------------

### Define task.function signature

Source: https://github.com/yidas/python-worker-dispatcher/blob/main/README.md

The main function signature for executing individual tasks. The return value can be False or a requests.Response to signal task failure.

```python
task_function (id: int, config, task, metadata: dict) -> Any
```

--------------------------------

### Define task.callback.on_done signature

Source: https://github.com/yidas/python-worker-dispatcher/blob/main/README.md

The callback function signature triggered after each task completes. The return value overwrites the task result.

```python
callback_on_done_function (id: int, config, result, log: dict) -> Any
```
Tokens:
Raw