Repository: dask/dask

Issue:
Default `multiprocessing` context is broken and should never be used
The default `multiprocessing` context (on POSIX systems) is fork-without-exec. As a result, all process state is mirrored in the subprocesses, with often unexpected results.

1. Threads are not propagated. If the parent process started a thread and code relies on it working this can break the subprocess. This is probably fairly rare, though I do maintain a library where this is a problem.
2. More seriously, module-level globals are preserved. Common module-level state includes things like logging. Subprocesses inherit the parent process' logging, often with unexpected results. I have seen child processes trying to rotate the parent's log file, for example.
3. Even if dropping threads doesn't directly impact the subprocess, it can indirectly cause deadlocks. In particular, if it acquires a lock and a fork happens immediately after that, the lock will be acquired in the subprocess and *never released*.

The following example (on Python 3.5) demonstrates problems 2 and 3. Notice how merely using the `logging` module in a thread can break subsequent processes created with `multiprocessing`:

```python
import logging
import threading
import multiprocessing
import sys
import time

logging.basicConfig(stream=sys.stdout, level=logging.INFO)
logging.info("HELLO FROM PARENT")

# Emulate a thread that is holding the logging lock when fork() happens. In real
# code this would be a race condition, I'm just making it fail consistently for
# demonstration purposes.
def logging_thread():
    logging._acquireLock()
    for i in range(1, 10):
        print("THREAD COUNTDOWN TO RELEASE LOCK", i)
        time.sleep(1)
    logging._releaseLock()
    print("THREAD IS GOING AWAY")

t = threading.Thread(target=logging_thread)
t.start()


def child_process():
    logging.info("CHILD PROCESS 1")
    # Do something that acquires lock:
    logging.Handler().set_name("hello")
    logging.info("CHILD PROCESS 2")

p = multiprocessing.Process(target=child_process)
p.start()
```

Here is the output:

```console
$ python3.6 mp-fork.py 
INFO:root:HELLO FROM PARENT
THREAD COUNTDOWN TO RELEASE LOCK 1
INFO:root:CHILD PROCESS 1
THREAD COUNTDOWN TO RELEASE LOCK 2
THREAD COUNTDOWN TO RELEASE LOCK 3
THREAD COUNTDOWN TO RELEASE LOCK 4
THREAD COUNTDOWN TO RELEASE LOCK 5
THREAD COUNTDOWN TO RELEASE LOCK 6
THREAD COUNTDOWN TO RELEASE LOCK 7
THREAD COUNTDOWN TO RELEASE LOCK 8
THREAD COUNTDOWN TO RELEASE LOCK 9
THREAD IS GOING AWAY
```

"CHILD PROCESS 2" is never printed because the child process is deadlocked, and the process never exits.

---

Proposed solution: change default multiprocessing context to `spawn`.

Discussion:
mrocklin: I'm curious, how does this affect performance?  My guess is that it would pretty negatively affect things for modestly sized tasks.
itamarst: It only affects startup. Once the pool is started it doesn't make a difference. If pool startup is a bottleneck you probably shouldn't be using Dask, so that doesn't seem like a big deal to me.
itamarst: I wrote a little test program to check the above.

1. This change makes `if __name__ == __main__:` necessary for single module example to work, it seems.
2. The performance overhead is just small additional fixed cost at startup.

```python
import time
import sys
from dask.bag import from_sequence
from dask import compute, config

def expensive(x):
    time.sleep(1)
    return x * 2

if __name__ == '__main__':
    with config.set({"multiprocessing.context": sys.argv[1]}):
        start = time.time()
        bag = from_sequence(range(90)).map(expensive).sum()
        print(compute(bag, npartitions=8))
        print(f"Elapsed with method {sys.argv[1]}", time.time() - start)
```

```console
$ python3 benchmark.py fork
(8010,)
Elapsed with method fork 12.09099531173706
$ python3 benchmark.py spawn
(8010,)
Elapsed with method spawn 12.255864143371582
```
mrocklin: > The performance overhead is just small additional fixed cost at startup.

The tasks in your benchmark take 1s.  What if instead they take 10ms?  Is the cost still negligible?

I find myself not particularly excited about this change because I don't think that I've ever heard of a user running into a problem with the fork solution. I'm curious @itamarst have you run into issues with Dask's use of fork in practice, or are you promoting the use of good technology?  Are there common situations that you suspect occur frequently among dask users where this is an issue?
mrocklin: I'm not very well informed on this topic though.  It would be great if some other maintainers chimed in here.
itamarst: I have encountered—not in Dask, but in other uses of multiprocessing—random freezes that were caused by the default method, as well as messed up logging behavior. You won't always see them yourself because they're race conditions, but basically the experience is "my program is acting _wierd_ and I have no idea why." Most users will just give up, or just live with things freezing 1% of the time.

Basically, the default method is fundamentally broken. See the original issue, and a much longer writeup I did: https://codewithoutrules.com/2018/09/04/python-multiprocessing/

mrocklin: Dask.distributed workers have an intermittent failure like that as well in
Python 2.

However I've never run into anything in practice when using the
multiprocessing scheduler.  I wonder if that doesn't run into the same
issues somehow?  I would have expected it to have arisen in a bug report by
now.

On Tue, Sep 25, 2018 at 10:54 AM Itamar Turner-Trauring <
notifications@github.com> wrote:

> I have encountered—not in Dask, but in other uses of
> multiprocessing—random freezes that were caused by the default method, as
> well as messed up logging behavior. You won't always see them yourself
> because they're race conditions, but basically the experience is "my
> program is acting *wierd* and I have no idea why." Most users will just
> give up, or just live with things freezing 1% of the time.
>
> Basically, the default method is fundamentally broken. See the original
> issue, and a much longer writeup I did:
> https://codewithoutrules.com/2018/09/04/python-multiprocessing/
>
> —
> You are receiving this because you commented.
> Reply to this email directly, view it on GitHub
> <https://github.com/dask/dask/pull/4003#issuecomment-424375478>, or mute
> the thread
> <https://github.com/notifications/unsubscribe-auth/AASszCg4VFrT4E_PubrAbFNpDEF9Y-LGks5uekOOgaJpZM4W0taO>
> .
>

itamarst: This change might break your code... but your code was previously broken some percentage of the time already...

Would it be acceptable if lack of `if __name__ == '__main__'` resulted in a clear error indicating the fix? Might be possible to detect that (maybe).
jcrist: > but your code was previously broken some percentage of the time already

This isn't necessarily true. Perhaps I only care about posix, and only use modules with fork-safe behavior. I suspect a non-negligible portion of dask users don't care about writing code that deploys on every OS, only that their code works on their machine.

I'm not 100% against this, but in 3.5 years we've never had a user run into this issue when using dask. I'm hesitant to make changes that will break user's code without clear evidence that the existing behavior is bad (and I'm not currently convinced that it is).

That said, I suspect at this point that most users use either the threaded scheduler or the distributed scheduler - the distributed scheduler runs fine locally, and is more efficient in many cases than the multiprocessing scheduler. Breaking code for a few to save many from having future issues may be worth it still.
itamarst: Another alternative: warn in visible way if users are using current default (permanently, or as prelude to eventual merge of this PR).
mrocklin: > warn in visible way if users are using current default

Building on what @jcrist has said above, the use of fork doesn't really seem to be bothering users today or stopping them from being productive.  We get bug reports about *lots* of parts of the system, but this topic has never come up.  I hesitate to teach people and push them away from a system that, pragmatically at least, seems to work ok today.  There is a very real cost to raising a warning to every user that I don't think is matched by any perceived benefits here.  I would want to revisit this if we found that many people were having difficulty due to using the defaults.
itamarst: OK.
itamarst: @jakirkham said he is in favor of making this change, this time around (https://github.com/dask/dask/issues/3759#issuecomment-619169780).
jakirkham: To reiterate what I said there, we are already doing this in Distributed ( https://github.com/dask/distributed/pull/3461 ). This just does the same thing in Dask.
TomAugspurger: Thanks for reopening. IMO distributed using spawn by default tips the scales to also using `spawn` here, but we'll wait for feedback from others before merging..

TomAugspurger: Seems to be some related test failures https://github.com/dask/dask/pull/4003/checks?check_run_id=616399950#step:6:197

```
____________________ test_custom_context_ignored_elsewhere ____________________

self = <multiprocessing.context.DefaultContext object at 0x000001BE1AF94518>
method = 'forkserver'

    def get_context(self, method=None):
        if method is None:
            return self
        try:
>           ctx = _concrete_contexts[method]
E           KeyError: 'forkserver'
```

and https://travis-ci.org/github/dask/dask/jobs/679167905#L1057
jakirkham: Appears there is a linter failure. Could you please run `black` locally and push the changes?

Patch:
diff --git a/MANIFEST.in b/MANIFEST.in
index 43008f1a3..d7ba3f627 100644
--- a/MANIFEST.in
+++ b/MANIFEST.in
@@ -7,6 +7,7 @@ include README.rst
 include LICENSE.txt
 include MANIFEST.in
 include dask/dask.yaml
+include dask/dask-schema.yaml
 
 include versioneer.py
 include dask/_version.py
diff --git a/dask/multiprocessing.py b/dask/multiprocessing.py
index fe60909ab..481c55086 100644
--- a/dask/multiprocessing.py
+++ b/dask/multiprocessing.py
@@ -138,20 +138,16 @@ and on Windows, because they each only support a single context.
 
 def get_context():
     """ Return the current multiprocessing context."""
+    # fork context does fork()-without-exec(), which can lead to deadlocks,
+    # so default to "spawn".
+    context_name = config.get("multiprocessing.context", "spawn")
     if sys.platform == "win32":
-        # Just do the default, since we can't change it:
-        if config.get("multiprocessing.context", None) is not None:
+        if context_name != "spawn":
+            # Only spawn is supported on Win32, can't change it:
             warn(_CONTEXT_UNSUPPORTED, UserWarning)
         return multiprocessing
-    context_name = config.get("multiprocessing.context", None)
-    # The default context on OSX was switched from "fork" to "spawn" in
-    # Python 3.8. We keep "fork" as the default here for backwards
-    # compatibility and to avoid raising a RuntimeError when using the
-    # multiprocessing scheduler in a Python script that is not in a
-    # if __name__ == "__main__" block
-    if context_name is None and sys.platform == "darwin":
-        context_name = "fork"
-    return multiprocessing.get_context(context_name)
+    else:
+        return multiprocessing.get_context(context_name)
 
 
 def get(
diff --git a/docs/source/changelog.rst b/docs/source/changelog.rst
index 7ea4cf85d..f9c297083 100644
--- a/docs/source/changelog.rst
+++ b/docs/source/changelog.rst
@@ -1,6 +1,50 @@
 Changelog
 =========
 
+2.15.0 / 2020-04-24
+-------------------
+
+Array
++++++
+
+- Update ``dask.array.from_array`` to warn when passed a Dask collection (:pr:`6122`) `James Bourbeau`_
+- Un-numpy like behaviour in ``dask.array.pad`` (:pr:`6042`) `Mark Boer`_
+- Add support for ``repeats=0`` in ``da.repeat`` (:pr:`6080`) `James Bourbeau`_
+
+Core
+++++
+
+- Fix yaml layout for schema (:pr:`6132`) `Ben Zaitlen`_
+- Configuration Reference (:pr:`6069`) `Ben Zaitlen`_
+- Add configuration option to turn off task fusion (:pr:`6087`) `Matthew Rocklin`_
+- Skip pyarrow on windows (:pr:`6094`) `Tom Augspurger`_
+- Set limit to maximum length of fused key (:pr:`6057`) `Lucas Rademaker`_
+- Add test against #6062 (:pr:`6072`) `Martin Durant`_
+- Bump checkout action to v2 (:pr:`6065`) `James Bourbeau`_
+
+DataFrame
++++++++++
+
+- Generalize categorical calls to support cudf ``Categorical`` (:pr:`6113`) `GALI PREM SAGAR`_
+- Avoid reading ``_metadata`` on every worker (:pr:`6017`) `Richard (Rick) Zamora`_
+- Use ``group_split_dispatch`` and ``ignore_index`` in ``apply_concat_apply`` (:pr:`6119`) `Richard (Rick) Zamora`_
+- Handle new (dtype) pandas metadata with pyarrow (:pr:`6090`) `Richard (Rick) Zamora`_
+- Skip ``test_partition_on_cats_pyarrow`` if pyarrow is not installed (:pr:`6112`) `James Bourbeau`_
+- Update DataFrame len to handle columns with the same name (:pr:`6111`) `James Bourbeau`_
+- ``ArrowEngine`` bug fixes and test coverage (:pr:`6047`) `Richard (Rick) Zamora`_
+- Added mode (:pr:`5958`) `Adam Lewis`_
+
+Documentation
++++++++++++++
+
+- Update "helm install" for helm 3 usage (:pr:`6130`) `JulianWgs`_
+- Extend preload documentation (:pr:`6077`) `Matthew Rocklin`_
+- Fixed small typo in DataFrame ``map_partitions()`` docstring (:pr:`6115`) `Eugene Huang`_
+- Fix typo: "double" should be times, not plus (:pr:`6091`) `David Chudzicki`_
+- Fix first line of ``array.random.*`` docs (:pr:`6063`) `Martin Durant`_
+- Add section about ``Semaphore`` in distributed (:pr:`6053`) `Florian Jetter`_
+
+
 2.14.0 / 2020-04-03
 -------------------
 
@@ -3072,3 +3116,7 @@ Other
 .. _`mlondschien`: https://github.com/mlondschien
 .. _`petiop`: https://github.com/petiop
 .. _`Richard (Rick) Zamora`: https://github.com/rjzamora
+.. _`Mark Boer`: https://github.com/mark-boer
+.. _`Florian Jetter`: https://github.com/fjetter
+.. _`Adam Lewis`: https://github.com/balast
+.. _`David Chudzicki`: https://github.com/dchudz
diff --git a/docs/source/scheduler-overview.rst b/docs/source/scheduler-overview.rst
index b334cb41c..517806a3b 100644
--- a/docs/source/scheduler-overview.rst
+++ b/docs/source/scheduler-overview.rst
@@ -146,7 +146,8 @@ global pool set with ``dask.config.set``:
     ...     x.compute()
 
 The multiprocessing scheduler also supports `different contexts`_ ("spawn",
-"forkserver", "fork") which you can set with ``dask.config.set``:
+"forkserver", "fork") which you can set with ``dask.config.set``. The default
+context is "spawn", but you can set a different one:
 
 .. code-block:: python
 


Test patch:
diff --git a/dask/tests/test_multiprocessing.py b/dask/tests/test_multiprocessing.py
index 3c9c49627..78f4a66af 100644
--- a/dask/tests/test_multiprocessing.py
+++ b/dask/tests/test_multiprocessing.py
@@ -194,15 +194,22 @@ def check_for_pytest():
 def test_custom_context_used_python3_posix():
     """ The 'multiprocessing.context' config is used to create the pool.
 
-    We assume default is 'fork', and therefore test for 'spawn'.  If default
-    context is changed this test will need to be modified to be different than
-    that.
+    We assume default is 'spawn', and therefore test for 'fork'.
     """
+    # We check for 'fork' by ensuring subprocess doesn't have modules only
+    # parent process should have:
+    def check_for_pytest():
+        import sys
+
+        return "FAKE_MODULE_FOR_TEST" in sys.modules
+
+    import sys
+
     sys.modules["FAKE_MODULE_FOR_TEST"] = 1
     try:
-        with dask.config.set({"multiprocessing.context": "spawn"}):
+        with dask.config.set({"multiprocessing.context": "fork"}):
             result = get({"x": (check_for_pytest,)}, "x")
-        assert not result
+        assert result
     finally:
         del sys.modules["FAKE_MODULE_FOR_TEST"]
 
@@ -215,19 +222,18 @@ def test_get_context_using_python3_posix():
 
     If default context is changed this test will need to change too.
     """
-    default_context = None if sys.platform != "darwin" else "fork"
-    assert get_context() is multiprocessing.get_context(default_context)
+    assert get_context() is multiprocessing.get_context("spawn")
     with dask.config.set({"multiprocessing.context": "forkserver"}):
         assert get_context() is multiprocessing.get_context("forkserver")
-    with dask.config.set({"multiprocessing.context": "spawn"}):
-        assert get_context() is multiprocessing.get_context("spawn")
+    with dask.config.set({"multiprocessing.context": "fork"}):
+        assert get_context() is multiprocessing.get_context("fork")
 
 
 @pytest.mark.skipif(sys.platform != "win32", reason="POSIX supports different contexts")
 def test_custom_context_ignored_elsewhere():
     """ On Windows, setting 'multiprocessing.context' doesn't explode.
 
-    Presumption is it's not used since unsupported, but mostly we care about
+    Presumption is it's not used since it's unsupported, but mostly we care about
     not breaking anything.
     """
     assert get({"x": (inc, 1)}, "x") == 2