Revert changes made in #3358 (#3411)

* Revert #3358

* Revision

* Code review

* Merge from master

* Obsolescence note

doc/whats-new.rst

@@ -31,6 +31,11 @@ New Features
     ``pip install git+https://github.com/andrewgsavage/pint.git@refs/pull/6/head)``.
     Even with it, interaction with non-numpy array libraries, e.g. dask or sparse, is broken.
 
+Bug fixes
+~~~~~~~~~
+- Fix regression introduced in v0.14.0 that would cause a crash if dask is installed
+  but cloudpickle isn't (:issue:`3401`) by `Rhys Doyle <https://github.com/rdoyle45>`_
+
 Documentation
 ~~~~~~~~~~~~~
 
@@ -39,6 +44,7 @@ Documentation
   datetime-like dimension is required. (:pull:`3400`)
   By `Justus Magin <https://github.com/keewis>`_.
 
+
 .. _whats-new.0.14.0:
 
 v0.14.0 (14 Oct 2019)

xarray/backends/locks.py

@@ -1,7 +1,7 @@
 import multiprocessing
 import threading
 import weakref
-from typing import Any, MutableMapping
+from typing import Any, MutableMapping, Optional
 
 try:
     from dask.utils import SerializableLock
@@ -62,7 +62,7 @@ def _get_lock_maker(scheduler=None):
     return _LOCK_MAKERS[scheduler]
 
 
-def _get_scheduler(get=None, collection=None):
+def _get_scheduler(get=None, collection=None) -> Optional[str]:
     """Determine the dask scheduler that is being used.
 
     None is returned if no dask scheduler is active.
@@ -86,10 +86,15 @@ def _get_scheduler(get=None, collection=None):
     except (ImportError, AttributeError):
         pass
 
-    if actual_get is dask.multiprocessing.get:
-        return "multiprocessing"
-    else:
-        return "threaded"
+    try:
+        # As of dask=2.6, dask.multiprocessing requires cloudpickle to be installed
+        # Dependency removed in https://github.com/dask/dask/pull/5511
+        if actual_get is dask.multiprocessing.get:
+            return "multiprocessing"
+    except AttributeError:
+        pass
+
+    return "threaded"
 
 
 def get_write_lock(key):