Round up saturation-factor

distributed/distributed-schema.yaml

@@ -125,6 +125,8 @@ properties:
 
               Up to worker-saturation * nthreads root tasks are sent to a
               worker at a time. If `.inf`, all runnable tasks are immediately sent to workers.
+              The target number is rounded up, so any `worker-saturation` value > 1.0 guarantees
+              at least one extra task will be sent to workers.
 
               Allowing oversaturation (> 1.0) means a worker may start running a new root task as
               soon as it completes the previous, even if there is a higher-priority downstream task
@@ -360,7 +362,7 @@ properties:
                   The maximum amount of data for a worker to request from another in a single gather operation
 
                   Tasks are gathered in batches, and if the first task in a batch is larger than this value,
-                  the task will still be gathered to ensure progress. Hence, this limit is not absolute. 
+                  the task will still be gathered to ensure progress. Hence, this limit is not absolute.
                   Note that this limit applies to a single gather operation and a worker may gather data from
                   multiple workers in parallel.
           connections:

distributed/scheduler.py

@@ -8199,7 +8199,7 @@ def _task_slots_available(ws: WorkerState, saturation_factor: float) -> int:
     "Number of tasks that can be sent to this worker without oversaturating it"
     assert not math.isinf(saturation_factor)
     nthreads = ws.nthreads
-    return max(int(saturation_factor * nthreads), 1) - (
+    return max(math.ceil(saturation_factor * nthreads), 1) - (
         len(ws.processing) - len(ws.long_running)
     )
 

distributed/tests/test_scheduler.py

@@ -413,9 +413,10 @@ async def test_queued_remove_add_worker(c, s, a, b):
 @pytest.mark.parametrize(
     "saturation_config, expected_task_counts",
     [
-        (2.5, (5, 2)),
-        ("2.5", (5, 2)),
+        (2.5, (5, 3)),
+        ("2.5", (5, 3)),
         (2.0, (4, 2)),
+        (1.1, (3, 2)),
         (1.0, (2, 1)),
         (-1.0, (1, 1)),
         (float("inf"), (6, 4))