gh-39025: Add reseed_rng option to p_iter_fork
    
This PR implements a convenience feature for the `@parallel` decorator,
or more precisely, for `p_iter_fork`.
By setting `reseed_rng=True` the random number generator is reset in
each subprocess by running `set_random_seed(self.worker_seed)`.
It is ensured that the worker seeds are deterministically derived and
thus, the results are always reproducible.

The new behaviour is useful for running probabilistic experiments and
taking advantage of multiple cores.

The PR adds one "doc test". It would make sense to have additional
(randomized) tests to assert that future modifications don't break the
reproducibility. But I failed to locate a comprehensive test suite for
sage/parallel, in particular, no existing tests for reproducibility of
parallel computations. The `sage/tests` folder seems to contain other
non-specific tests?

Coding style w.r.t. <https://doc.sagemath.org/html/en/reference/misc/sag
e/misc/randstate.html>
-   The documentation said `with seed(worker_seed)` should be used. I
did not use it, because the current implementation reseeds the whole
subprocess once and for all.
-   The documentation says NTL does not reproduce. This is a very
surprising and unfortunate state of affairs. I expect that  `reseed_rng`
inherits this caveat. I did not make it explicit in the doc string,
because if the problem occurs, then also `reseed_rng=False` should have
it, but there is no mention of it.

### 📝 Checklist

- [x] The title is concise and informative.
- [x] The description explains in detail what this PR is about.
- [x] I have linked a relevant issue or discussion.
- [x] I have created tests covering the changes.
- [x] I have updated the documentation and checked the documentation
preview.

The doc tests passed for me, but preview via running `sage --docbuild
tutorial html` gives me `ImportError: cannot import name
count_all_local_good_types_normal_form' from
'sage.quadratic_forms.count_local_2'` so doc tests don't work although
that has probably nothing to with this PR.
    
URL: #39025
Reported by: mklss
Reviewer(s): Kwankyu Lee

Author: Release Manager

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 (random number generator) in each subprocess
 
     .. warning::
 
@@ -398,6 +399,15 @@ 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)) # random
+        [(((1000,), {}), 444), (((1000,), {}), 597), (((1000,), {}), 640)]
+
     .. warning::
 
        Currently, parallel methods do not work with the

src/sage/parallel/use_fork.py

@@ -19,6 +19,9 @@
 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 +71,8 @@ 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`` -- boolean (default: ``False``); whether or not to reseed
+      the rng in the subprocesses
 
     EXAMPLES::
 
@@ -80,7 +85,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 +108,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 +155,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 +163,29 @@ 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(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.
+                            if self.reseed_rng:
+                                self.worker_seed = seed0
                             self._subprocess(f, dir, *v0)
-
                     workers[pid] = WorkerData(v0)
 
                 if len(workers) > 0:
@@ -304,6 +317,10 @@ def _subprocess(self, f, dir, args, kwds={}):
             else:
                 invalidate_all()
 
+        # Reseed rng, if requested.
+        if self.reseed_rng:
+            set_random_seed(self.worker_seed)
+
         # Now evaluate the function f.
         value = f(*args, **kwds)