Test macOS

.github/workflows/python-package.yml

@@ -15,7 +15,7 @@ jobs:
     runs-on: ${{ matrix.os }}
     strategy:
       matrix:
-        os: [ubuntu-20.04, windows-latest]
+        os: [ubuntu-20.04, windows-latest, macos-11]
         python-version: ["3.6", "3.7", "3.8", "3.9", "3.10", "3.11"]
 
     steps:
@@ -40,4 +40,6 @@ jobs:
     - name: Test with pytest
       timeout-minutes: 30
       run: |
+        # mpire needs a lot of file descriptors, too many for macOS' default settings
+        ulimit -n 5000
         pytest -v -o log_cli=true -s

mpire/comms.py

@@ -1,4 +1,5 @@
 import collections
+import gc
 import multiprocessing as mp
 import queue
 import threading
@@ -231,7 +232,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 +245,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 +639,8 @@ def drain_results_queue_terminate_worker(self, dont_wait_event: threading.Event)
         except (queue.Empty, OSError):
             if got_results:
                 dont_wait_event.set()
+        # Force collection of semaphore objects
+        gc.collect()
 
     def drain_queues(self) -> None:
         """
@@ -656,7 +662,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,3 +1,4 @@
+import gc
 import logging
 import os
 import queue
@@ -999,6 +1000,8 @@ def terminate(self) -> None:
         send a sigkill.
         """
         if not self._workers:
+            # Force collection of semaphore objects
+            gc.collect()
             return
 
         # Set exception thrown so workers know to stop fetching new tasks