remote/coordinator: run reservation scheduling under lock

Reservation scheduling mutates shared coordinator state (`reservations`,
`places`, and the derived `place.reservation` view), but the periodic
scheduler path was performing these updates without holding the
coordinator lock.

This can lead to races between the poll-based scheduler and concurrent
RPC handlers (e.g. acquire/release/reservation operations), potentially
resulting in inconsistent state or incorrect reservation transitions.

We now run reservation scheduling under the coordinator lock by making
it async and awaiting it in the poll path. This ensures that all
mutations of coordinator state are serialized consistently with the rest
of the coordinator operations.

Renamed `schedule_reservations` to `_schedule_reservations` to reflect
that it is now an internal async helper. All usages in the coordinator
were updated.  This method is not part of a public API, so renaming it
is not expected to impact external users.

Testing
=======

- Ran existing unit tests covering reservation and acquire/release
  flows.
- Exercised reservation creation/cancel and place acquire/release
  scenarios manually.
- No deadlocks were observed in these runs.
- This patch does not add a dedicated stress test, and the current tests
  do not prove the absence of deadlocks.

Performance
===========

- Measured `_schedule_reservations` runtime under load. With 100 places
  and 200 reservations created in parallel, observed runtime was
  ~0.0013s, with worst case ~0.0036 s.
- RPC paths were already invoking reservation scheduling while holding
  the coordinator lock; the primary change is that the poll-based
  scheduler path is now also serialized under the same lock.

This patch focuses on fixing the scheduler locking model and does not
change coordinator scheduling behavior.

Signed-off-by: Alex Tercete <alex.tercete@arm.com>
Reviewed-by: Alex Tercete <alex.tercete@arm.com> # gatekeeper
Co-authored-by: Idan Saadon <idan.saadon@arm.com>

Author: alextercete

labgrid/remote/coordinator.py

@@ -240,8 +240,9 @@ async def _poll_step_sync_resources(self):
 
     async def _poll_step_schedule(self):
         # update reservations
-        with warn_if_slow("schedule reservations"):
-            self.schedule_reservations()
+        async with self.lock:
+            with warn_if_slow("schedule reservations"):
+                await self._schedule_reservations()
 
     async def poll(self, step_func):
         while not self.loop.is_closed():
@@ -883,7 +884,7 @@ async def AcquirePlace(self, request, context):
         place.touch()
         self._publish_place(place)
         self.save_later()
-        self.schedule_reservations()
+        await self._schedule_reservations()
         print(f"{place.name}: place acquired by {place.acquired}")
         return labgrid_coordinator_pb2.AcquirePlaceResponse()
 
@@ -910,7 +911,7 @@ async def ReleasePlace(self, request, context):
         place.touch()
         self._publish_place(place)
         self.save_later()
-        self.schedule_reservations()
+        await self._schedule_reservations()
         print(f"{place.name}: place released")
         return labgrid_coordinator_pb2.ReleasePlaceResponse()
 
@@ -950,11 +951,12 @@ async def GetPlaces(self, unused_request, unused_context):
         except Exception:
             logging.exception("error during get places")
 
-    def schedule_reservations(self):
+    async def _schedule_reservations(self):
         # The primary information is stored in the reservations and the places
         # only have a copy for convenience.
 
         # expire reservations
+        assert self.lock.locked()
         for res in list(self.reservations.values()):
             if res.state is ReservationState.acquired:
                 # acquired reservations do not expire
@@ -1076,7 +1078,7 @@ async def CreateReservation(self, request: labgrid_coordinator_pb2.CreateReserva
         owner = self.clients[peer].name
         res = Reservation(owner=owner, prio=request.prio, filters=fltrs)
         self.reservations[res.token] = res
-        self.schedule_reservations()
+        await self._schedule_reservations()
         return labgrid_coordinator_pb2.CreateReservationResponse(reservation=res.as_pb2())
 
     @locked
@@ -1087,7 +1089,7 @@ async def CancelReservation(self, request: labgrid_coordinator_pb2.CancelReserva
         if token not in self.reservations:
             await context.abort(grpc.StatusCode.FAILED_PRECONDITION, f"Reservation {token} does not exist")
         del self.reservations[token]
-        self.schedule_reservations()
+        await self._schedule_reservations()
         return labgrid_coordinator_pb2.CancelReservationResponse()
 
     @locked