util.py 1.9 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556
  1. """Utility functions.
  2. """
  3. import contextlib
  4. import multiprocessing
  5. from milc import cli
  6. @contextlib.contextmanager
  7. def parallelize():
  8. """Returns a function that can be used in place of a map() call.
  9. Attempts to use `mpire`, falling back to `multiprocessing` if it's not
  10. available. If parallelization is not requested, returns the original map()
  11. function.
  12. """
  13. # Work out if we've already got a config value for parallel searching
  14. if cli.config.user.parallel_search is None:
  15. parallel_search = True
  16. else:
  17. parallel_search = cli.config.user.parallel_search
  18. # Non-parallel searches use `map()`
  19. if not parallel_search:
  20. yield map
  21. return
  22. # Prefer mpire's `WorkerPool` if it's available
  23. with contextlib.suppress(ImportError):
  24. from mpire import WorkerPool
  25. from mpire.utils import make_single_arguments
  26. with WorkerPool() as pool:
  27. def _worker(func, *args):
  28. # Ensure we don't unpack tuples -- mpire's `WorkerPool` tries to do so normally so we tell it not to.
  29. for r in pool.imap_unordered(func, make_single_arguments(*args, generator=False), progress_bar=True):
  30. yield r
  31. yield _worker
  32. return
  33. # Otherwise fall back to multiprocessing's `Pool`
  34. with multiprocessing.Pool() as pool:
  35. yield pool.imap_unordered
  36. def parallel_map(*args, **kwargs):
  37. """Effectively runs `map()` but executes it in parallel if necessary.
  38. """
  39. with parallelize() as map_fn:
  40. # This needs to be enclosed in a `list()` as some implementations return
  41. # a generator function, which means the scope of the pool is closed off
  42. # before the results are returned. Returning a list ensures results are
  43. # materialised before any worker pool is shut down.
  44. return list(map_fn(*args, **kwargs))