Test macOS

README.rst

@@ -49,14 +49,13 @@ Features
 - Optionally utilizes dill_ as serialization backend through multiprocess_, enabling parallelizing more exotic objects,
   lambdas, and functions in iPython and Jupyter notebooks.
 
-MPIRE has been tested on both Linux and Windows. There are a few minor known caveats for Windows users, which can be
-found here_.
+MPIRE is tested on Linux and Windows. For Windows and macOS users, there are a few minor known caveats, which are documented in the Troubleshooting_ chapter.
 
 .. _benchmarks: https://towardsdatascience.com/mpire-for-python-multiprocessing-is-really-easy-d2ae7999a3e9
 .. _multiprocess: https://github.com/uqfoundation/multiprocess
 .. _dill: https://pypi.org/project/dill/
 .. _tqdm: https://tqdm.github.io/
-.. _here: https://sybrenjansen.github.io/mpire/troubleshooting.html#windows
+.. _Troubleshooting: https://sybrenjansen.github.io/mpire/troubleshooting.html
 
 
 Installation

docs/changelog.rst

@@ -5,8 +5,10 @@ Unreleased
 ----------
 
 * Excluded the ``tests`` folder from MPIRE distributions (`#89`_)
+* Added a workaround for semaphore leakage on macOS and fixed a bug when working in a fork context while the system default is spawn (`#92`_)
 
 .. _#89: https://github.com/sybrenjansen/mpire/issues/89
+.. _#92: https://github.com/sybrenjansen/mpire/issues/92
 
 
 2.8.0

docs/troubleshooting.rst

@@ -155,3 +155,15 @@ Windows support has some caveats:
 * Progress bar is not supported when using threading as start method;
 * When using ``dill`` and an exception occurs, or when the exception occurs in an exit function, it can print additional
   ``OSError`` messages in the terminal, but they can be safely ignored.
+
+
+.. _troubleshooting_macos:
+
+macOS
+-----
+
+macOS is not yet officially supported. It seems to work mostly fine, with these caveats:
+
+* Use ``ulimit -n <number>`` to increase the limit of the number of open files.
+  This is required because mpire uses a lot of file-descriptor based synchronization primitives.
+  `<Number>` should be large enough to accomodate all of mpire's file descriptors, eg. `<number> = 65536`.

mpire/comms.py

@@ -1,5 +1,7 @@
 import collections
+import gc
 import multiprocessing as mp
+import platform
 import queue
 import threading
 import time
@@ -231,7 +233,8 @@ def clear_progress_bar_shutdown(self) -> None:
         """
         Clears the progress bar shutdown signal
         """
-        self._progress_bar_shutdown.clear()
+        if self._progress_bar_shutdown is not None:
+            self._progress_bar_shutdown.clear()
 
     def signal_progress_bar_complete(self) -> None:
         """
@@ -243,13 +246,15 @@ def clear_progress_bar_complete(self) -> None:
         """
         Clear that the progress bar is complete
         """
-        self._progress_bar_complete.clear()
+        if self._progress_bar_complete is not None:
+            self._progress_bar_complete.clear()
 
     def wait_until_progress_bar_is_complete(self) -> None:
         """
         Waits until the progress bar is completed
         """
-        self._progress_bar_complete.wait()
+        if self._progress_bar_complete is not None:
+            self._progress_bar_complete.wait()
 
     ################
     # Order modifiers
@@ -635,6 +640,10 @@ def drain_results_queue_terminate_worker(self, dont_wait_event: threading.Event)
         except (queue.Empty, OSError):
             if got_results:
                 dont_wait_event.set()
+        if platform.system() == "Darwin":
+            # Force collection of semaphore objects.
+            # TODO: This is a workaround for semaphore leakage on macOS.
+            gc.collect()
 
     def drain_queues(self) -> None:
         """
@@ -656,7 +665,7 @@ def drain_and_join_queue(self, q: mp.JoinableQueue, join: bool = True) -> None:
         if RUNNING_WINDOWS:
             self._drain_and_join_queue(q, join)
         else:
-            process = mp.Process(target=self._drain_and_join_queue, args=(q, join))
+            process = self.ctx.Process(target=self._drain_and_join_queue, args=(q, join))
             process.start()
             process.join(timeout=5)
             if process.is_alive():

mpire/pool.py

@@ -1,5 +1,7 @@
+import gc
 import logging
 import os
+import platform
 import queue
 import signal
 import threading
@@ -999,6 +1001,10 @@ def terminate(self) -> None:
         send a sigkill.
         """
         if not self._workers:
+            if platform.system() == "Darwin":
+                # Force collection of semaphore objects.
+                # TODO: This is a workaround for semaphore leakage on macOS.
+                gc.collect()
             return
 
         # Set exception thrown so workers know to stop fetching new tasks