Handle constrained similar to ready

distributed/worker.py

@@ -1801,7 +1801,7 @@ def handle_cancel_compute(self, key, reason):
         Nothing will happen otherwise.
         """
         ts = self.tasks.get(key)
-        if ts and ts.state in ("waiting", "ready"):
+        if ts and ts.state in READY | {"waiting"}:
             self.log.append((key, "cancel-compute", reason))
             ts.scheduler_holds_ref = False
             # All possible dependents of TS should not be in state Processing on
@@ -1904,7 +1904,7 @@ def handle_compute_task(
             ts.dependencies.add(dep_ts)
             dep_ts.dependents.add(ts)
 
-        if ts.state in {"ready", "executing", "waiting", "resumed"}:
+        if ts.state in READY | {"executing", "waiting", "resumed"}:
             pass
         elif ts.state == "memory":
             recommendations[ts] = "memory"
@@ -2897,7 +2897,7 @@ def handle_steal_request(self, key, stimulus_id):
         }
         self.batched_stream.send(response)
 
-        if state in {"ready", "waiting", "constrained"}:
+        if state in READY | {"waiting"}:
             # If task is marked as "constrained" we haven't yet assigned it an
             # `available_resources` to run on, that happens in
             # `transition_constrained_executing`
@@ -3690,7 +3690,7 @@ def validate_state(self):
                     assert ts_wait.key in self.tasks
                     assert (
                         ts_wait.state
-                        in {"ready", "executing", "flight", "fetch", "missing"}
+                        in READY | {"executing", "flight", "fetch", "missing"}
                         or ts_wait.key in self._missing_dep_flight
                         or ts_wait.who_has.issubset(self.in_flight_workers)
                     ), (ts, ts_wait, self.story(ts), self.story(ts_wait))