Extract SchedulerState from Scheduler

Author: martindurant

distributed/active_memory_manager.py

@@ -222,10 +222,10 @@ def log_reject(msg: str) -> None:
             return None
 
         if candidates is None:
-            candidates = self.scheduler.running.copy()
+            candidates = self.scheduler.state.running.copy()
         else:
             # Don't modify orig_candidates
-            candidates = candidates & self.scheduler.running
+            candidates = candidates & self.scheduler.state.running
         if not candidates:
             log_reject("no running candidates")
             return None