Merge pull request #205 from jeromekelleher/fix-mac-progress

Fix mac progress

Author: jeromekelleher

bio2zarr/core.py

@@ -187,7 +187,7 @@ class ProgressConfig:
 # progressable thing happening per source process. This is
 # probably fine in practise, but there could be corner cases
 # where it's not. Something to watch out for.
-_progress_counter = multiprocessing.Value("Q", 0)
+_progress_counter = None
 
 
 def update_progress(inc):
@@ -201,23 +201,30 @@ def get_progress():
     return val
 
 
-def set_progress(value):
-    with _progress_counter.get_lock():
-        _progress_counter.value = value
+def setup_progress_counter(counter):
+    global _progress_counter
+    _progress_counter = counter
 
 
 class ParallelWorkManager(contextlib.AbstractContextManager):
     def __init__(self, worker_processes=1, progress_config=None):
+        # Need to specify this explicitly to suppport Macs and
+        # for future proofing.
+        ctx = multiprocessing.get_context("spawn")
+        global _progress_counter
+        _progress_counter = ctx.Value("Q", 0)
         if worker_processes <= 0:
             # NOTE: this is only for testing, not for production use!
             self.executor = SynchronousExecutor()
         else:
             self.executor = cf.ProcessPoolExecutor(
                 max_workers=worker_processes,
+                mp_context=ctx,
+                initializer=setup_progress_counter,
+                initargs=(_progress_counter,),
             )
         self.futures = set()
 
-        set_progress(0)
         if progress_config is None:
             progress_config = ProgressConfig()
         self.progress_config = progress_config
@@ -258,16 +265,6 @@ def submit(self, *args, **kwargs):
         self.futures.add(future)
         return future
 
-    def wait_for_completed(self, timeout=None):
-        done, not_done = cf.wait(self.futures, timeout, cf.FIRST_COMPLETED)
-        for future in done:
-            exception = future.exception()
-            # TODO do the check for BrokenProcessPool here
-            if exception is not None:
-                raise exception
-        self.futures = not_done
-        return done
-
     def results_as_completed(self):
         for future in cf.as_completed(self.futures):
             yield future.result()

tests/test_core.py

@@ -56,7 +56,6 @@ def test_bad_min_max(self, min_value, max_value):
             core.min_int_dtype(min_value, max_value)
 
 
-@pytest.mark.skipif(sys.platform == "darwin", reason="Issue #75")
 class TestParallelWorkManager:
     @pytest.mark.parametrize("total", [1, 10, 2**63])
     @pytest.mark.parametrize("workers", [0, 1])

validation.py

@@ -6,7 +6,7 @@
 
 import click
 
-from bio2zarr import icf, vcf, verification
+from bio2zarr import vcf2zarr
 
 # TODO add support here for split vcfs. Perhaps simplest to take a
 # directory provided as input as indicating this, and then having
@@ -40,11 +40,11 @@ def cli(vcfs, worker_processes, force):
         else:
             files = [f]
             source_file = f
-        exploded = tmp_path / (f.name + ".exploded")
+        exploded = tmp_path / (f.name + ".icf")
         if force and exploded.exists():
             shutil.rmtree(exploded)
         if not exploded.exists():
-            icf.explode(
+            vcf2zarr.explode(
                 exploded,
                 files,
                 worker_processes=worker_processes,
@@ -53,21 +53,21 @@ def cli(vcfs, worker_processes, force):
         spec = tmp_path / (f.name + ".schema")
         if force or not spec.exists():
             with open(spec, "w") as specfile:
-                vcf.mkschema(exploded, specfile)
+                vcf2zarr.mkschema(exploded, specfile)
 
-        zarr = tmp_path / (f.name + ".zarr")
+        zarr = tmp_path / (f.name + ".vcz")
         if force and zarr.exists():
             shutil.rmtree(zarr)
         if not zarr.exists():
-            vcf.encode(
+            vcf2zarr.encode(
                 exploded,
                 zarr,
                 spec,
                 worker_processes=worker_processes,
                 show_progress=True,
             )
 
-        verification.validate(source_file, zarr, show_progress=True)
+        vcf2zarr.verify(source_file, zarr, show_progress=True)
 
 
 if __name__ == "__main__":