Merge pull request #88 from ploxiln/parallel_fork

Fix parallel mode on macOS with python-3.8+ (which change default
multiprocessing mode from "fork" to "spawn" due to macOS system
libraries using threads ...)

This situation may affect Linux too starting with Python-3.14 because
"spawn" will be the default *if any threads are created*. (In practice
Fabric / fab-classic have worked OK with "fork" mode, even with some
threads running ...)

Also fix-up support for Threads in job_queue.py
(even though fab-classic doesn't use it)

Author: ploxiln

fabric/job_queue.py

@@ -7,12 +7,14 @@
 
 import time
 import queue as Queue
-from multiprocessing import Process
+from collections import namedtuple
 
 from fabric.network import ssh
 from fabric.context_managers import settings
 
 
+DoneProc = namedtuple('DoneProc', ['name', 'exitcode'])
+
 class JobQueue(object):
     """
     The goal of this class is to make a queue of processes to run, and go
@@ -73,7 +75,7 @@ def close(self):
 
     def append(self, process):
         """
-        Add the Process() to the queue, so that later it can be checked up on.
+        Add the Process to the queue, so that later it can be checked up on.
         That is if the JobQueue is still open.
 
         If the queue is closed, this will just silently do nothing.
@@ -145,12 +147,15 @@ def _advance_the_queue():
                         if self._debug:
                             print("Job queue found finished proc: %s." % job.name)
                         done = self._running.pop(id)
-                        self._completed.append((done.name, done.exitcode))
-                        if hasattr(done, 'close') and callable(done.close):
-                            done.close()
-                        # multiprocessing.Process.close() added in Python-3.7
-                        # for older versions of python, GC will have to do
-                        del done
+                        # might be a Process or a Thread
+                        if hasattr(done, 'exitcode'):
+                            proc = done
+                            done = DoneProc(proc.name, proc.exitcode)
+                            # multiprocessing.Process.close() added in Python-3.7
+                            if hasattr(proc, 'close'):
+                                proc.close()
+                            del proc
+                        self._completed.append(done)
 
                 if self._debug:
                     print("Job queue has %d running." % len(self._running))
@@ -174,9 +179,9 @@ def _advance_the_queue():
         self._fill_results(results)
 
         # Attach exit codes now that we're all done & have joined all jobs
-        for job_name, exit_code in self._completed:
-            if isinstance(job, Process):
-                results[job_name]['exit_code'] = exit_code
+        for job in self._completed:
+            if hasattr(job, 'exitcode'):
+                results[job.name]['exit_code'] = job.exitcode
 
         return results
 

fabric/tasks.py

@@ -325,13 +325,15 @@ def execute(task, *args, **kwargs):
     parallel = requires_parallel(task)
     if parallel:
         import multiprocessing
+        ctx = multiprocessing.get_context('fork')
+        # Set up job queue for parallel cases
+        queue = ctx.Queue()
     else:
-        multiprocessing = None
+        ctx = None
+        queue = None
 
     # Get pool size for this task
     pool_size = task.get_pool_size(my_env['all_hosts'], state.env.pool_size)
-    # Set up job queue in case parallel is needed
-    queue = multiprocessing.Queue() if parallel else None
     jobs = JobQueue(pool_size, queue)
     if state.output.debug:
         jobs._debug = True
@@ -342,8 +344,7 @@ def execute(task, *args, **kwargs):
         for host in my_env['all_hosts']:
             try:
                 results[host] = _execute(
-                    task, host, my_env, args, new_kwargs, jobs, queue,
-                    multiprocessing
+                    task, host, my_env, args, new_kwargs, jobs, queue, ctx,
                 )
             except NetworkError as e:
                 results[host] = e