add process_list_in_threads

CHANGELOG.md

@@ -1,5 +1,12 @@
 # Versions
 
+## 0.3
+
+* PS
+  * `process_list_in_threads`
+
+----
+
 ## 0.2
 
 * Crypto
@@ -12,7 +19,7 @@
     * `valid_port`
     * `get_ipv`
 
-* SubPS
+* PS
   * `process` - Stdin handling
   * `wait_for_threads`
 
@@ -42,5 +49,8 @@
     * `valid_public_ip`
     * `valid_asn`
 
+* PS
+  * `process`
+
 * Django
   * `datetime_from_db`

README.md

@@ -90,6 +90,7 @@ from oxl_utils.ps import process_cache
 
 # thread handling
 from oxl_utils.ps import wait_for_threads
+from oxl_utils.ps import process_list_in_threads
 ```
 
 ----

src/oxl_utils/ps.py

@@ -4,6 +4,7 @@
 from os import environ, getcwd
 from functools import cache
 from threading import Thread
+from typing import Callable
 
 from .log import log
 
@@ -100,3 +101,28 @@ def wait_for_threads(threads: list[Thread], timeout: int = 600):
     while not threads_done and time() < break_time:
         threads_done = all(not t.is_alive() for t in threads)
         sleep(0.05)
+
+
+def process_list_in_threads(callback: Callable, to_process: list, key: str = 'item', parallel: int = 10):
+    next_idx = 0
+    last_idx = len(to_process) - 1
+    threads = []
+
+    while next_idx <= last_idx:
+        finished = [t for t in threads if not t.is_alive()]
+        for t in finished:
+            threads.remove(t)
+
+        if len(threads) >= parallel:
+            sleep(0.05)
+            continue
+
+        t = Thread(
+            target=callback,
+            kwargs={key: to_process[next_idx]},
+        )
+        threads.append(t)
+        t.start()
+        next_idx += 1
+
+    wait_for_threads(threads, timeout=3600)

src/oxl_utils/ps_test.py

@@ -1,5 +1,5 @@
-from time import sleep
-from threading import Thread
+from time import sleep, time
+from threading import Thread, Lock
 
 import pytest
 
@@ -107,3 +107,22 @@ def _dummy_workload(s: int):
 
     for t in threads:
         assert not t.is_alive()
+
+
+def test_process_list_in_threads():
+    from .ps import process_list_in_threads
+    lock = Lock()
+    results = []
+
+    def _dummy_workload(s: int):
+        sleep(s)
+        with lock:
+            results.append(s)
+
+    start_time = time()
+    workload = [1, 1, 1, 1]
+    process_list_in_threads(callback=_dummy_workload, to_process=workload, key='s', parallel=4)
+    elapsed = time() - start_time
+
+    assert 1 < elapsed < 1.2
+    assert len(results) == 4