### 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 ``` === COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.