Decorate SchedulerState methods with @ccall

Should allow faster C API calls to these methods in addition to their
existing Python APIs. Also disable exception checking when Python
objects are `return`ed since these already can and are checked for an
exception. Note that `@ccall` is not permitted on functions taking
`**kwargs` so those have been skipped.

Author: jakirkham

distributed/scheduler.py

@@ -1686,6 +1686,8 @@ def validate(self, v: bint):
     def workers(self):
         return self._workers
 
+    @ccall
+    @exceptval(check=False)
     def _remove_from_processing(self, ts: TaskState) -> str:
         """
         Remove *ts* from the set of processing tasks.
@@ -1771,6 +1773,8 @@ def _add_to_memory(
                 recommendations=recommendations,
             )
 
+    @ccall
+    @exceptval(check=False)
     def transition_released_waiting(self, key) -> tuple:
         try:
             ts: TaskState = self._tasks[key]
@@ -1826,6 +1830,8 @@ def transition_released_waiting(self, key) -> tuple:
                 pdb.set_trace()
             raise
 
+    @ccall
+    @exceptval(check=False)
     def transition_no_worker_waiting(self, key) -> tuple:
         try:
             ts: TaskState = self._tasks[key]
@@ -1873,6 +1879,8 @@ def transition_no_worker_waiting(self, key) -> tuple:
                 pdb.set_trace()
             raise
 
+    @ccall
+    @exceptval(check=False)
     def decide_worker(self, ts: TaskState) -> WorkerState:
         """
         Decide on a worker for task *ts*.  Return a WorkerState.
@@ -1916,6 +1924,8 @@ def decide_worker(self, ts: TaskState) -> WorkerState:
 
         return ws
 
+    @ccall
+    @exceptval(check=False)
     def transition_waiting_processing(self, key) -> tuple:
         try:
             ts: TaskState = self._tasks[key]
@@ -2121,6 +2131,8 @@ def transition_processing_memory(
                 pdb.set_trace()
             raise
 
+    @ccall
+    @exceptval(check=False)
     def transition_memory_released(self, key, safe: bint = False) -> tuple:
         ws: WorkerState
         try:
@@ -2194,6 +2206,8 @@ def transition_memory_released(self, key, safe: bint = False) -> tuple:
                 pdb.set_trace()
             raise
 
+    @ccall
+    @exceptval(check=False)
     def transition_released_erred(self, key) -> tuple:
         try:
             ts: TaskState = self._tasks[key]
@@ -2240,6 +2254,8 @@ def transition_released_erred(self, key) -> tuple:
                 pdb.set_trace()
             raise
 
+    @ccall
+    @exceptval(check=False)
     def transition_erred_released(self, key) -> tuple:
         try:
             ts: TaskState = self._tasks[key]
@@ -2281,6 +2297,8 @@ def transition_erred_released(self, key) -> tuple:
                 pdb.set_trace()
             raise
 
+    @ccall
+    @exceptval(check=False)
     def transition_waiting_released(self, key) -> tuple:
         try:
             ts: TaskState = self._tasks[key]
@@ -2320,6 +2338,8 @@ def transition_waiting_released(self, key) -> tuple:
                 pdb.set_trace()
             raise
 
+    @ccall
+    @exceptval(check=False)
     def transition_processing_released(self, key) -> tuple:
         try:
             ts: TaskState = self._tasks[key]
@@ -2446,6 +2466,8 @@ def transition_processing_erred(
                 pdb.set_trace()
             raise
 
+    @ccall
+    @exceptval(check=False)
     def transition_no_worker_released(self, key) -> tuple:
         try:
             ts: TaskState = self._tasks[key]
@@ -2475,6 +2497,8 @@ def transition_no_worker_released(self, key) -> tuple:
                 pdb.set_trace()
             raise
 
+    @ccall
+    @exceptval(check=False)
     def _propagate_forgotten(
         self, ts: TaskState, recommendations: dict, worker_msgs: dict
     ):
@@ -2514,6 +2538,8 @@ def _propagate_forgotten(
                 worker_msgs[w] = {"op": "delete-data", "keys": [key], "report": False}
         ts._who_has.clear()
 
+    @ccall
+    @exceptval(check=False)
     def transition_memory_forgotten(self, key) -> tuple:
         ws: WorkerState
         try:
@@ -2557,6 +2583,8 @@ def transition_memory_forgotten(self, key) -> tuple:
                 pdb.set_trace()
             raise
 
+    @ccall
+    @exceptval(check=False)
     def transition_released_forgotten(self, key) -> tuple:
         try:
             ts: TaskState = self._tasks[key]
@@ -2595,6 +2623,8 @@ def transition_released_forgotten(self, key) -> tuple:
                 pdb.set_trace()
             raise
 
+    @ccall
+    @exceptval(check=False)
     def check_idle_saturated(self, ws: WorkerState, occ: double = -1.0):
         """Update the status of the idle and saturated state
 
@@ -2636,6 +2666,8 @@ def check_idle_saturated(self, ws: WorkerState, occ: double = -1.0):
 
             saturated.discard(ws)
 
+    @ccall
+    @exceptval(check=False)
     def _client_releases_keys(
         self, keys=None, client=None, recommendations: dict = None
     ):
@@ -2660,6 +2692,8 @@ def _client_releases_keys(
             elif ts._state != "erred" and not ts._waiters:
                 recommendations[ts._key] = "released"
 
+    @ccall
+    @exceptval(check=False)
     def _task_to_msg(self, ts: TaskState, duration=None) -> dict:
         """ Convert a single computational task to a message """
         ws: WorkerState
@@ -2697,6 +2731,8 @@ def _task_to_msg(self, ts: TaskState, duration=None) -> dict:
 
         return msg
 
+    @ccall
+    @exceptval(check=False)
     def _task_to_report_msg(self, ts: TaskState) -> dict:
         if ts is None:
             return {"op": "cancelled-key", "key": ts._key}
@@ -2715,6 +2751,8 @@ def _task_to_report_msg(self, ts: TaskState) -> dict:
         else:
             return None
 
+    @ccall
+    @exceptval(check=False)
     def _task_to_client_msgs(self, ts: TaskState) -> dict:
         cs: ClientState
         clients: dict = self._clients
@@ -2734,6 +2772,8 @@ def _task_to_client_msgs(self, ts: TaskState) -> dict:
 
         return client_msgs
 
+    @ccall
+    @exceptval(check=False)
     def _reevaluate_occupancy_worker(self, ws: WorkerState):
         """ See reevaluate_occupancy """
         old = ws._occupancy
@@ -2758,6 +2798,7 @@ def _reevaluate_occupancy_worker(self, ws: WorkerState):
                 steal.remove_key_from_stealable(ts)
                 steal.put_key_in_stealable(ts)
 
+    @ccall
     def get_comm_cost(self, ts: TaskState, ws: WorkerState) -> double:
         """
         Get the estimated communication cost (in s.) to compute the task
@@ -2771,6 +2812,7 @@ def get_comm_cost(self, ts: TaskState, ws: WorkerState) -> double:
             nbytes += dts._nbytes
         return nbytes / bandwidth
 
+    @ccall
     def get_task_duration(self, ts: TaskState, default: double = -1) -> double:
         """
         Get the estimated computation cost of the given task
@@ -2787,6 +2829,8 @@ def get_task_duration(self, ts: TaskState, default: double = -1) -> double:
 
         return duration
 
+    @ccall
+    @exceptval(check=False)
     def valid_workers(self, ts: TaskState) -> set:
         """Return set of currently valid workers for key
 
@@ -2837,6 +2881,8 @@ def valid_workers(self, ts: TaskState) -> set:
 
         return s
 
+    @ccall
+    @exceptval(check=False)
     def worker_objective(self, ts: TaskState, ws: WorkerState) -> tuple:
         """
         Objective function to determine which worker should get the task