Always return set from valid_workers

distributed/scheduler.py

@@ -2225,7 +2225,7 @@ async def add_worker(
 
             recommendations = {}
             for ts in list(self.unrunnable):
-                valid = self.valid_workers(ts)
+                valid: set = self.valid_workers(ts)
                 if valid is None or ws in valid:
                     recommendations[ts._key] = "waiting"
 
@@ -4655,7 +4655,7 @@ def decide_worker(self, ts: TaskState):
         """
         Decide on a worker for task *ts*.  Return a WorkerState.
         """
-        valid_workers = self.valid_workers(ts)
+        valid_workers: set = self.valid_workers(ts)
 
         if (
             valid_workers is not None
@@ -5520,7 +5520,7 @@ def check_idle_saturated(self, ws: WorkerState, occ: double = -1.0):
             else:
                 saturated.discard(ws)
 
-    def valid_workers(self, ts: TaskState):
+    def valid_workers(self, ts: TaskState) -> set:
         """Return set of currently valid workers for key
 
         If all workers are valid then this returns ``None``.
@@ -5530,7 +5530,7 @@ def valid_workers(self, ts: TaskState):
         *  host_restrictions
         *  resource_restrictions
         """
-        s = None
+        s: set = None
 
         if ts._worker_restrictions:
             s = {w for w in ts._worker_restrictions if w in self.workers}
@@ -6083,7 +6083,7 @@ def adaptive_target(self, comm=None, target_duration=None):
             return len(self.workers) - len(to_close)
 
 
-def decide_worker(ts: TaskState, all_workers, valid_workers, objective):
+def decide_worker(ts: TaskState, all_workers, valid_workers: set, objective):
     """
     Decide which worker should take task *ts*.