Start fewer but larger workers with LocalCluster

When we have a large number of available cores we should set the default number
of threads per process to be higher than one.

This implements a policy that aims for a number of processes equal to the
square root of the number of cores, at least above a certain amount of cores.

Partially addresses dask#2450

Author: mrocklin

distributed/deploy/local.py

@@ -8,6 +8,7 @@
 import weakref
 import toolz
 
+from dask.utils import factors
 from tornado import gen
 
 from .cluster import Cluster
@@ -99,8 +100,7 @@ def __init__(self, n_workers=None, threads_per_worker=None, processes=True,
             self._old_logging_level = silence_logging(level=silence_logs)
         if n_workers is None and threads_per_worker is None:
             if processes:
-                n_workers = _ncores
-                threads_per_worker = 1
+                n_workers, threads_per_worker = nprocesses_nthreads(_ncores)
             else:
                 n_workers = 1
                 threads_per_worker = _ncores
@@ -372,6 +372,34 @@ def scheduler_address(self):
             return '<unstarted>'
 
 
+def nprocesses_nthreads(n):
+    """
+    The default breakdown of processes and threads for a given number of cores
+
+    Parameters
+    ----------
+    n: int
+        Number of available cores
+
+    Examples
+    --------
+    >>> nprocesses_nthreads(4)
+    (4, 1)
+    >>> nprocesses_nthreads(32)
+    (8, 4)
+
+    Returns
+    -------
+    nprocesses, nthreads
+    """
+    if n <= 4:
+        processes = n
+    else:
+        processes = min(f for f in factors(n) if f >= math.sqrt(n))
+    threads = n // processes
+    return (processes, threads)
+
+
 clusters_to_close = weakref.WeakSet()
 
 

distributed/deploy/tests/test_local.py

@@ -14,7 +14,7 @@
 import pytest
 
 from distributed import Client, Worker, Nanny
-from distributed.deploy.local import LocalCluster
+from distributed.deploy.local import LocalCluster, nprocesses_nthreads
 from distributed.metrics import time
 from distributed.utils_test import (inc, gen_test, slowinc,
                                     assert_cannot_connect,
@@ -529,5 +529,18 @@ def test_local_tls_restart(loop):
             assert workers_before != workers_after
 
 
+def test_default_process_thread_breakdown():
+    assert nprocesses_nthreads(1) == (1, 1)
+    assert nprocesses_nthreads(4) == (4, 1)
+    assert nprocesses_nthreads(5) == (5, 1)
+    assert nprocesses_nthreads(8) == (4, 2)
+    assert nprocesses_nthreads(12) in ((6, 2), (4, 3))
+    assert nprocesses_nthreads(20) == (5, 4)
+    assert nprocesses_nthreads(24) in ((6, 4), (8, 3))
+    assert nprocesses_nthreads(32) == (8, 4)
+    assert nprocesses_nthreads(40) in ((8, 5), (10, 4))
+    assert nprocesses_nthreads(80) in ((10, 8), (16, 5))
+
+
 if sys.version_info >= (3, 5):
     from distributed.deploy.tests.py3_test_deploy import *  # noqa F401