diff options
Diffstat (limited to 'sec_certs/parallel_processing.py')
| -rw-r--r-- | sec_certs/parallel_processing.py | 17 |
1 files changed, 8 insertions, 9 deletions
diff --git a/sec_certs/parallel_processing.py b/sec_certs/parallel_processing.py index ae8a1221..0e2ac175 100644 --- a/sec_certs/parallel_processing.py +++ b/sec_certs/parallel_processing.py @@ -1,26 +1,25 @@ from sec_certs.helpers import tqdm -from multiprocessing.pool import Pool, ThreadPool -from typing import Callable, Iterable, Optional, Union +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] = ThreadPool(max_workers) if use_threading else 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] + 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] |
