Repository: dask/dask
Pull request #5562 (2019-11-05 17:40:47)

Update minimum pyarrow version on CI
This PR sets `pyarrow>=0.14.0` on the CI, which has recently started failing due to pulling in older, unsupported versions of `pyarrow` (ref https://travis-ci.org/dask/dask/jobs/607318590)


Patch:
diff --git a/continuous_integration/hdfs/install.sh b/continuous_integration/hdfs/install.sh
index 393ce2a1a..d832818d5 100644
--- a/continuous_integration/hdfs/install.sh
+++ b/continuous_integration/hdfs/install.sh
@@ -1,6 +1,6 @@
 set -xe
 
-docker exec -it $CONTAINER_ID conda install -y -q dask pyarrow fsspec -c conda-forge
+docker exec -it $CONTAINER_ID conda install -y -q dask pyarrow">=0.14.0" fsspec -c conda-forge
 docker exec -it $CONTAINER_ID pip install -e .
 
 set +xe
diff --git a/continuous_integration/travis/travis-36.yaml b/continuous_integration/travis/travis-36.yaml
index 25f23c1ec..a42513bc7 100644
--- a/continuous_integration/travis/travis-36.yaml
+++ b/continuous_integration/travis/travis-36.yaml
@@ -22,7 +22,7 @@ dependencies:
   - zarr
   - tiledb-py
   - sqlalchemy
-  - pyarrow
+  - pyarrow>=0.14.0
   # other -- IO
   - bcolz
   - blosc
diff --git a/continuous_integration/travis/travis-37-dev.yaml b/continuous_integration/travis/travis-37-dev.yaml
index 6741c5a12..1978d4396 100644
--- a/continuous_integration/travis/travis-37-dev.yaml
+++ b/continuous_integration/travis/travis-37-dev.yaml
@@ -18,7 +18,7 @@ dependencies:
   - zarr
   - tiledb-py
   - sqlalchemy
-  - pyarrow
+  - pyarrow>=0.14.0
   # other -- IO
   - bcolz
   - blosc
diff --git a/continuous_integration/travis/travis-37.yaml b/continuous_integration/travis/travis-37.yaml
index 3818380ac..b558baf7b 100644
--- a/continuous_integration/travis/travis-37.yaml
+++ b/continuous_integration/travis/travis-37.yaml
@@ -17,7 +17,7 @@ dependencies:
   - tiledb-py
   - fsspec>=0.5.1
   - sqlalchemy
-  - pyarrow
+  - pyarrow>=0.14.0
   # other -- IO
   - bcolz
   - blosc
diff --git a/dask/multiprocessing.py b/dask/multiprocessing.py
index 9883599bb..99e129ae9 100644
--- a/dask/multiprocessing.py
+++ b/dask/multiprocessing.py
@@ -8,6 +8,7 @@ from warnings import warn
 import cloudpickle
 
 from . import config
+from .system import CPU_COUNT
 from .local import reraise, get_async  # TODO: get better get
 from .optimization import fuse, cull
 
@@ -166,7 +167,7 @@ def get(
         If True [default], `fuse` is applied to the graph before computation.
     """
     pool = pool or config.get("pool", None)
-    num_workers = num_workers or config.get("num_workers", None)
+    num_workers = num_workers or config.get("num_workers", None) or CPU_COUNT
     if pool is None:
         context = get_context()
         pool = context.Pool(num_workers, initializer=initialize_worker_process)
diff --git a/dask/system.py b/dask/system.py
new file mode 100644
index 000000000..9d9a693a5
--- /dev/null
+++ b/dask/system.py
@@ -0,0 +1,53 @@
+import math
+import os
+import sys
+
+try:
+    import psutil
+except ImportError:
+    psutil = None
+
+__all__ = ("cpu_count", "CPU_COUNT")
+
+
+def cpu_count():
+    """Get the available CPU count for this system.
+
+    Takes the minimum value from the following locations:
+
+    - Total system cpus available on the host.
+    - CPU Affinity (if set)
+    - Cgroups limit (if set)
+    """
+    count = os.cpu_count()
+
+    # Check CPU affinity if available
+    if psutil is not None:
+        try:
+            affinity_count = len(psutil.Process().cpu_affinity())
+            if affinity_count > 0:
+                count = min(count, affinity_count)
+        except Exception:
+            pass
+
+    # Check cgroups if available
+    if sys.platform == "linux":
+        # The directory name isn't standardized across linux distros, check both
+        for dirname in ["cpuacct,cpu", "cpu,cpuacct"]:
+            try:
+                with open("/sys/fs/cgroup/%s/cpu.cfs_quota_us" % dirname) as f:
+                    quota = int(f.read())
+                with open("/sys/fs/cgroup/%s/cpu.cfs_period_us" % dirname) as f:
+                    period = int(f.read())
+                # We round up on fractional CPUs
+                cgroups_count = math.ceil(quota / period)
+                if cgroups_count > 0:
+                    count = min(count, cgroups_count)
+                break
+            except Exception:
+                pass
+
+    return count
+
+
+CPU_COUNT = cpu_count()
diff --git a/dask/threaded.py b/dask/threaded.py
index 622bb8434..a8f83dd1d 100644
--- a/dask/threaded.py
+++ b/dask/threaded.py
@@ -10,6 +10,7 @@ import threading
 from threading import current_thread, Lock
 
 from . import config
+from .system import CPU_COUNT
 from .local import get_async
 from .utils_test import inc, add  # noqa: F401
 
@@ -61,7 +62,7 @@ def get(dsk, result, cache=None, num_workers=None, pool=None, **kwargs):
         if pool is None:
             if num_workers is None and thread is main_thread:
                 if default_pool is None:
-                    default_pool = ThreadPool()
+                    default_pool = ThreadPool(CPU_COUNT)
                 pool = default_pool
             elif thread in pools and num_workers in pools[thread]:
                 pool = pools[thread][num_workers]
diff --git a/docs/source/install.rst b/docs/source/install.rst
index a43dd565f..62ccef30e 100644
--- a/docs/source/install.rst
+++ b/docs/source/install.rst
@@ -110,7 +110,7 @@ These optional dependencies and their minimum supported versions are listed belo
 +-------------+----------+--------------------------------------------------------------+
 |    psutil   |          |             Enables a more accurate CPU count                |
 +-------------+----------+--------------------------------------------------------------+
-|    pyarrow  | >=0.9.0  |               Python library for Apache Arrow                |
+|    pyarrow  | >=0.14.0 |               Python library for Apache Arrow                |
 +-------------+----------+--------------------------------------------------------------+
 |    s3fs     |          |                    Reading from Amazon S3                    |
 +-------------+----------+--------------------------------------------------------------+


Test patch:
diff --git a/dask/tests/test_system.py b/dask/tests/test_system.py
new file mode 100644
index 000000000..575a0bfd4
--- /dev/null
+++ b/dask/tests/test_system.py
@@ -0,0 +1,54 @@
+import builtins
+import io
+import os
+import sys
+
+import psutil
+import pytest
+
+from dask.system import cpu_count
+
+
+def test_cpu_count():
+    count = cpu_count()
+    assert isinstance(count, int)
+    assert count <= os.cpu_count()
+    assert count >= 1
+
+
+@pytest.mark.parametrize("dirname", ["cpuacct,cpu", "cpu,cpuacct", None])
+def test_cpu_count_cgroups(dirname, monkeypatch):
+    def mycpu_count():
+        # Absurdly high, unlikely to match real value
+        return 250
+
+    monkeypatch.setattr(os, "cpu_count", mycpu_count)
+
+    class MyProcess(object):
+        def cpu_affinity(self):
+            # No affinity set
+            return []
+
+    monkeypatch.setattr(psutil, "Process", MyProcess)
+
+    if dirname:
+        paths = {
+            "/sys/fs/cgroup/%s/cpu.cfs_quota_us" % dirname: io.StringIO("2005"),
+            "/sys/fs/cgroup/%s/cpu.cfs_period_us" % dirname: io.StringIO("10"),
+        }
+        builtin_open = builtins.open
+
+        def myopen(path, *args, **kwargs):
+            if path in paths:
+                return paths.get(path)
+            return builtin_open(path, *args, **kwargs)
+
+        monkeypatch.setattr(builtins, "open", myopen)
+        monkeypatch.setattr(sys, "platform", "linux")
+
+    count = cpu_count()
+    if dirname:
+        # Rounds up
+        assert count == 201
+    else:
+        assert count == 250
diff --git a/dask/tests/test_threaded.py b/dask/tests/test_threaded.py
index 4b45ed67b..f824d9be5 100644
--- a/dask/tests/test_threaded.py
+++ b/dask/tests/test_threaded.py
@@ -8,6 +8,7 @@ from time import time, sleep
 import pytest
 
 import dask
+from dask.system import CPU_COUNT
 from dask.threaded import get
 from dask.utils_test import inc, add
 
@@ -109,6 +110,19 @@ def test_dont_spawn_too_many_threads():
     assert after <= before + 8
 
 
+def test_dont_spawn_too_many_threads_CPU_COUNT():
+    before = threading.active_count()
+
+    dsk = {("x", i): (lambda: i,) for i in range(10)}
+    dsk["x"] = (sum, list(dsk))
+    for i in range(20):
+        get(dsk, "x")
+
+    after = threading.active_count()
+
+    assert after <= before + CPU_COUNT * 2
+
+
 def test_thread_safety():
     def f(x):
         return 1