Repository: SneaksAndData/adapta (language: python)

Issue:
[FEATURE] Allow parallel read from multiple DataSockets/JobSockets
**Is your feature request related to a problem? Please describe.**
Currently we can only read data from sockets with `load` or similar methods sequentially. For time-bound algorithms like Omnichannel this increases overall duration by >50%.

**Describe the solution you'd like**
A method that allows to run a set of functions in parallel and in lazy mode, so we don't have to join threads in advance.


Discussion / hints:

Fix patch:
diff --git a/proteus/utils/concurrent_task_runner.py b/proteus/utils/concurrent_task_runner.py
new file mode 100644
index 00000000..3bd92f1f
--- /dev/null
+++ b/proteus/utils/concurrent_task_runner.py
@@ -0,0 +1,89 @@
+"""
+ Models used by utility methods
+"""
+import concurrent
+import os
+import sys
+from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
+from dataclasses import dataclass
+from typing import Callable, Any, List, TypeVar, Generic, Optional, Dict
+
+T = TypeVar('T')
+
+
+@dataclass
+class Executable(Generic[T]):
+    """
+     A single executable function with arguments, ready to run if invoked.
+    """
+    func: Callable[[...], T]
+    args: List[Any]
+    alias: str
+
+
+class ConcurrentTaskRunner(Generic[T]):
+    """
+     Provides parallel execution capability for a list of given functions.
+
+      my_funcs = [Executable(func=my_func1, args=[arg1, .., argN], alias='task1'),..]
+      threads = len(os.sched_getaffinity(0))
+      runner = ConcurrentTaskRunner(my_funcs, threads, True)
+
+      # get results lazily
+      tasks = runner.run()['taskN'].result()
+
+      # get results as they come
+      task_results = runner.run(lazy=False)
+
+     NB: For CPU bound work, set use_processes to True, otherwise overall run time will not improve due to GIL.
+
+    :param func_list: A list of (function, arguments) tuples to parallelise.
+    :param num_threads: Maximum number of threads to use. On Linux platforms use len(os.sched_getaffinity(0))
+          to get number of threads available to current process
+    :param use_processes: Use processes instead of thread for parallelisation. Preferrable for work that depends on GIL release.
+    """
+
+    def __init__(
+            self,
+            func_list: List[Executable[T]],
+            num_threads: Optional[int] = None,
+            use_processes: bool = False
+    ):
+        self._func_list = func_list
+        self._num_threads = num_threads
+        self._use_processes = use_processes
+
+    def _run_tasks(self) -> Dict[str, concurrent.futures.Future]:
+        """
+                 Executes a list of functions in parallel using threads or processes.
+
+                :param lazy: Whether to collect results right away or leave this to the caller.
+                :return: A dictionary of (callable_name, callable_future)
+                """
+        worker_count = self._num_threads or (
+            len(os.sched_getaffinity(0)) if sys.platform != "win32" else os.cpu_count())
+        runner_pool = ProcessPoolExecutor(max_workers=worker_count)  if self._use_processes else ThreadPoolExecutor(max_workers=worker_count)
+        with runner_pool:
+            return {
+                executable.alias: runner_pool.submit(executable.func, *executable.args) for executable in
+                self._func_list
+            }
+
+    def lazy(self) -> Dict[str, concurrent.futures.Future]:
+        """
+         Executes the function list without explicitly collecting the results, allowing them to be retrieved by the client
+         when needed.
+
+        :return: A dictionary of (task_alias, task_future)
+        """
+        return self._run_tasks()
+
+    def eager(self) -> Dict[str, T]:
+        """
+         Executes the function list and wait for all threads to complete execution before returning
+
+        :return: A dictionary of (task_alias, task_result)
+        """
+        return {
+            task_name: task.result() for task_name, task in self._run_tasks().items()
+        }


Test patch:
diff --git a/tests/test_utils.py b/tests/test_utils.py
index 61764a66..bf9af0be 100644
--- a/tests/test_utils.py
+++ b/tests/test_utils.py
@@ -1,8 +1,9 @@
 import time
-from typing import List, Any
+from typing import List, Any, Dict
 
 import pytest
 from proteus.utils import doze, operation_time, chunk_list
+from proteus.utils.concurrent_task_runner import Executable, ConcurrentTaskRunner
 
 
 @pytest.mark.parametrize('sleep_period,doze_interval', [(1, 50), (2, 10)])
@@ -29,3 +30,122 @@ def custom_method():
 ])
 def test_chunk_list(list_to_chunk: List[Any], num_chunks: int, expected_list):
     assert chunk_list(list_to_chunk, num_chunks) == expected_list
+
+
+def mock_func(a: float, b: str, c: bool) -> Dict:
+    time.sleep(a)
+    return {
+        'a': a,
+        'b': b,
+        'c': c
+    }
+
+
+@pytest.mark.parametrize('func_list,num_threads,use_processes,expectations,expected_wait', [
+    (
+            # Run 3 functions in 3 threads
+            # Each function sleeps for `a` seconds before returning
+            # Since we do lazy result fetch, we should expect to wait LESS than max(a0,.. aN), because all tasks effectively start at the same time
+            # however since time.sleep effectively blocks the main thread if using ThreadPoolExecutor, subsequent submissions will delay each other
+            # thus we should expect at most 0.5s + small time to get results of each future.
+            [
+                Executable[Dict](mock_func, [0.1, 'test', True], 'case1'),
+                Executable[Dict](mock_func, [0.3, 'test1', True], 'case2'),
+                Executable[Dict](mock_func, [0.5, 'test2', False], 'case3')
+            ],
+            3,
+            False,
+            {
+                'case1': {
+                    'a': 0.1,
+                    'b': 'test',
+                    'c': True
+                },
+                'case2': {
+                    'a': 0.3,
+                    'b': 'test1',
+                    'c': True
+                },
+                'case3': {
+                    'a': 0.5,
+                    'b': 'test2',
+                    'c': False
+                }
+            },
+            0.55
+    ),
+    # Runs 1 thread for each function
+    # Expected to see 1s + 2s + 3s + result process time ~ slightly above 6s
+    (
+            [
+                Executable[Dict](mock_func, [1, 'test', True], 'case1'),
+                Executable[Dict](mock_func, [2, 'test1', True], 'case2'),
+                Executable[Dict](mock_func, [3, 'test2', False], 'case3')
+            ],
+            1,
+            False,
+            {
+                'case1': {
+                    'a': 1,
+                    'b': 'test',
+                    'c': True
+                },
+                'case2': {
+                    'a': 2,
+                    'b': 'test1',
+                    'c': True
+                },
+                'case3': {
+                    'a': 3,
+                    'b': 'test2',
+                    'c': False
+                }
+            },
+            6.1
+    ),
+    # Runs 3 processes for 3 functions
+    # Same as the second test case, but now we use ProcessPoolExecutor, so we should expect 3s + process start time overhead
+    (
+            [
+                Executable[Dict](mock_func, [1, 'test', True], 'case1'),
+                Executable[Dict](mock_func, [2, 'test1', True], 'case2'),
+                Executable[Dict](mock_func, [3, 'test2', False], 'case3')
+            ],
+            3,
+            True,
+            {
+                'case1': {
+                    'a': 1,
+                    'b': 'test',
+                    'c': True
+                },
+                'case2': {
+                    'a': 2,
+                    'b': 'test1',
+                    'c': True
+                },
+                'case3': {
+                    'a': 3,
+                    'b': 'test2',
+                    'c': False
+                }
+            },
+            4
+    )
+])
+def test_concurrent_task_runner(
+        func_list: List[Executable[Dict]],
+        num_threads: int,
+        use_processes: bool,
+        expectations: Dict[str, Dict],
+        expected_wait: float
+):
+    start = time.monotonic_ns()
+    runner = ConcurrentTaskRunner(func_list, num_threads, use_processes)
+    tasks = runner.lazy()
+    results = {}
+    for task_name, task_future in tasks.items():
+        results[task_name] = task_future.result()
+    total_wait = (time.monotonic_ns() - start) / 1e9
+
+    assert results == expectations and total_wait < expected_wait