aboutsummaryrefslogtreecommitdiffhomepage
path: root/sec_certs/parallel_processing.py
diff options
context:
space:
mode:
authorAdam Janovsky2021-12-15 09:39:44 +0100
committerAdam Janovsky2021-12-15 09:39:44 +0100
commit94414e8e040e162883d1d0182a198ca1533cd42c (patch)
treefdb1790275d9db86cc75676180bba3a865ba33ce /sec_certs/parallel_processing.py
parent80a39c9a279a7ed20e0deec004d02744c914f581 (diff)
parent9d7e8aebdbba5978d62ec8beca079b4db5b3fdc9 (diff)
downloadsec-certs-94414e8e040e162883d1d0182a198ca1533cd42c.tar.gz
sec-certs-94414e8e040e162883d1d0182a198ca1533cd42c.tar.zst
sec-certs-94414e8e040e162883d1d0182a198ca1533cd42c.zip
Merge branch 'dev' into mypy-ci-cd
Diffstat (limited to 'sec_certs/parallel_processing.py')
-rw-r--r--sec_certs/parallel_processing.py25
1 files changed, 10 insertions, 15 deletions
diff --git a/sec_certs/parallel_processing.py b/sec_certs/parallel_processing.py
index c0a4ea63..0e2ac175 100644
--- a/sec_certs/parallel_processing.py
+++ b/sec_certs/parallel_processing.py
@@ -1,30 +1,25 @@
-from tqdm import tqdm
-from multiprocessing.pool import Pool, ThreadPool
-from typing import Callable, Iterable, Optional, Union
+from sec_certs.helpers import tqdm
+from concurrent.futures import ProcessPoolExecutor as ProcessPool, ThreadPoolExecutor as ThreadPool
+from typing import Callable, Iterable, Optional
import time
-def process_parallel(func: Callable, items: Iterable, max_workers: int, callback: Optional[Callable] = None,
+def process_parallel(func: Callable, items: Iterable, max_workers: int,
use_threading: bool = True, progress_bar: bool = True, unpack: bool = False,
progress_bar_desc: Optional[str] = None):
- pool: Union[Pool, ThreadPool]
- if use_threading is True:
- pool = ThreadPool(max_workers)
- else:
- pool = Pool(max_workers)
- results = [pool.apply_async(func, (*i,), callback=callback) for i in items] if unpack else [pool.apply_async(func, (i, ), callback=callback) for i in items]
+ pool: Union[Pool, ThreadPool] = ThreadPool(max_workers) if use_threading else Pool(max_workers)
+ results = [pool.submit(func, *i) for i in items] if unpack else [pool.submit(func, i) for i in items]
if progress_bar is True and items:
bar = tqdm(total=len(results), desc=progress_bar_desc)
- while not all([x.ready() for x in results]):
- done_count = len([x.ready() for x in results if x.ready()])
+ while not all(all_done := [x.done() for x in results]):
+ done_count = len(list(filter(lambda x: x, all_done)))
bar.update(done_count - bar.n)
time.sleep(1)
bar.update(len(results) - bar.n)
bar.close()
- pool.close()
- pool.join()
+ pool.shutdown()
- return [r.get() for r in results]
+ return [r.result() for r in results]