Drop oversubscription logic

Since `concurrent.futures` already has the logic necessary to avoid
oversubscription and our logic for it doesn't really apply, go ahead and
drop the oversubscription logic from the local scheduler and rely on
`concurrent.futures`.

Note it's still possible to configure pools with specific numbers of
workers. This doesn't affect that. Instead we just submit all tasks that
are ready until we are out of ready tasks.

dask/local.py

@@ -355,7 +355,6 @@ def identity(x):
 
 def get_async(
     apply_async,
-    num_workers,
     dsk,
     result,
     cache=None,
@@ -379,8 +378,6 @@ def get_async(
     ----------
     apply_async : function
         Asynchronous apply function as found on Pool or ThreadPool
-    num_workers : int
-        The number of active tasks we should have at any one time
     dsk : dict
         A dask dictionary specifying a workflow
     result : key or list of keys
@@ -475,8 +472,8 @@ def fire_task():
                     callback=queue.put,
                 )
 
-            # Seed initial tasks into the thread pool
-            while state["ready"] and len(state["running"]) < num_workers:
+            # Seed initial tasks into the pool
+            while state["ready"]:
                 fire_task()
 
             # Main loop, wait on tasks to finish, insert new ones
@@ -499,7 +496,7 @@ def fire_task():
                 for f in posttask_cbs:
                     f(key, res, dsk, state, worker_id)
 
-                while state["ready"] and len(state["running"]) < num_workers:
+                while state["ready"]:
                     fire_task()
 
             succeeded = True
@@ -543,8 +540,7 @@ def get_sync(dsk, keys, **kwargs):
 
     Can be useful for debugging.
     """
-    kwargs.pop("num_workers", None)  # if num_workers present, remove it
-    return get_async(apply_sync, 1, dsk, keys, **kwargs)
+    return get_async(apply_sync, dsk, keys, **kwargs)
 
 
 def sortkey(item):

dask/multiprocessing.py

@@ -245,7 +245,6 @@ def get(
         # Run
         result = get_async(
             partial(multiprocessing_apply_async, pool),
-            pool._max_workers,
             dsk3,
             keys,
             get_id=_process_get_id,

dask/threaded.py

@@ -76,7 +76,6 @@ def get(dsk, result, cache=None, num_workers=None, pool=None, **kwargs):
 
     results = get_async(
         partial(executor_apply_async, pool),
-        pool._max_workers,
         dsk,
         result,
         cache=cache,