Add reseed_rng option to p_iter_fork

Author: mklss

src/sage/parallel/decorate.py

@@ -308,6 +308,7 @@ def parallel(p_iter='fork', ncpus=None, **kwds):
     - ``ncpus`` -- integer; maximal number of subprocesses to use at the same time
     - ``timeout`` -- number of seconds until each subprocess is killed (only supported
       by ``'fork'``; zero means not at all)
+    - ``reseed_rng``: reseed the rng in each subprocess
 
     .. warning::
 
@@ -398,6 +399,16 @@ def parallel(p_iter='fork', ncpus=None, **kwds):
         sage: Foo.square_classmethod(3)
         9
 
+    
+    By default, all subprocesses use the same random seed and therefore the same deterministic randomness.
+    For functions that should be randomized, we can reseed the random seed in each subprocess::
+
+        sage: @parallel(reseed_rng = True)
+        ....: def unif(n): return ZZ.random_element(x = 0, y = n)
+        sage: set_random_seed(42)
+        sage: sorted(unif([1000]*3))
+        [(((1000,), {}), 444), (((1000,), {}), 597), (((1000,), {}), 640)]
+
     .. warning::
 
        Currently, parallel methods do not work with the

src/sage/parallel/use_fork.py

@@ -19,6 +19,8 @@
 from sage.interfaces.process import ContainChildren
 from sage.misc.timing import walltime
 
+from sage.misc.randstate import set_random_seed
+from sage.misc.prandom import getrandbits
 
 class WorkerData:
     """
@@ -68,6 +70,7 @@ class p_iter_fork:
       about what the iterator does (e.g., killing subprocesses)
     - ``reset_interfaces`` -- boolean (default: ``True``); whether to reset all
       pexpect interfaces
+    - ``reseed_rng`` -- booolean (default: ``False``); whether or not to reseed the rng in the subprocesses
 
     EXAMPLES::
 
@@ -80,7 +83,7 @@ class p_iter_fork:
         sage: X.verbose
         False
     """
-    def __init__(self, ncpus, timeout=0, verbose=False, reset_interfaces=True):
+    def __init__(self, ncpus, timeout=0, verbose=False, reset_interfaces=True, reseed_rng=False):
         """
         Create a ``fork()``-based parallel iterator.
 
@@ -103,6 +106,8 @@ def __init__(self, ncpus, timeout=0, verbose=False, reset_interfaces=True):
         self.timeout = float(timeout)  # require a float
         self.verbose = verbose
         self.reset_interfaces = reset_interfaces
+        self.reseed_rng = reseed_rng
+        self.worker_seed = None
 
     def __call__(self, f, inputs):
         """
@@ -148,8 +153,6 @@ def __call__(self, f, inputs):
             sage: list(Polygen([QQ,QQ]))
             [(((Rational Field,), {}), x), (((Rational Field,), {}), x)]
         """
-        n = self.ncpus
-        v = list(inputs)
         import os
         import sys
         import signal
@@ -158,21 +161,28 @@ def __call__(self, f, inputs):
         dir = tmp_dir()
         timeout = self.timeout
 
+        n = self.ncpus
+        inputs = list(inputs)
+        if self.reseed_rng:
+            seeds = [getrandbits(512) for _ in range(0, len(inputs))]
+            vs = list(zip(inputs, seeds))
+        else:
+            vs = list(zip(inputs, [None]*len(inputs)))
         workers = {}
         try:
-            while v or workers:
+            while vs or workers:
                 # Spawn up to n subprocesses
-                while v and len(workers) < n:
-                    v0 = v.pop(0)  # Input value for the next subprocess
+                while vs and len(workers) < n:
+                    (v0, seed0) = vs.pop(0)  # Input value and seed for the next subprocess
                     with ContainChildren():
                         pid = os.fork()
                         # The way fork works is that pid returns the
                         # nonzero pid of the subprocess for the master
                         # process and returns 0 for the subprocess.
                         if not pid:
                             # This is the subprocess.
+                            self.worker_seed = seed0 if self.reseed_rng else None
                             self._subprocess(f, dir, *v0)
-
                     workers[pid] = WorkerData(v0)
 
                 if len(workers) > 0:
@@ -304,6 +314,11 @@ def _subprocess(self, f, dir, args, kwds={}):
             else:
                 invalidate_all()
 
+        # Reseed rng, if requested.
+        if self.reseed_rng == True:
+            set_random_seed(self.worker_seed)
+
+
         # Now evaluate the function f.
         value = f(*args, **kwds)