Added async lock and moved _up_to_date flag so it makes more sense

Author: ptsOSL

src/atip/simulator.py

@@ -140,6 +140,10 @@ class ATSimulator:
                                         for new changes to the AT
                                         lattice and recalculate the
                                         physics data upon a change.
+            _new_data_lock (asyncio.Lock): A lock which can be taken
+                                        to stop new caput callbacks
+                                        being added to the queue while
+                                        held.
     """
 
     _loop: asyncio.BaseEventLoop
@@ -148,6 +152,7 @@ class ATSimulator:
     _quit_thread: asyncio.Event
     _up_to_date: asyncio.Event
     _calculation_task: asyncio.Task
+    _new_data_lock: asyncio.Lock
 
     @classmethod
     async def create(
@@ -207,6 +212,7 @@ async def create(
         self._quit_thread = asyncio.Event()
         self._up_to_date = asyncio.Event()
         self._up_to_date.set()
+        self._new_data_lock = asyncio.Lock()
 
         self._calculation_task = asyncio.create_task(
             self._recalculate_phys_data(callback)
@@ -221,11 +227,13 @@ async def queue_set(self, func, field, value):
             field (str): The field to be changed.
             value (float): The value to be set.
         """
-        await self._queue.put((func, field, value))
-        # If this flag gets cleared while we are recalculating, then it can cause
-        # everything to lock, so we setup a lock between this function and the
-        # recalculate function
-        logging.debug(f"Added task to async queue. qsize={self._queue.qsize()}")
+        async with self._new_data_lock:
+            await self._queue.put((func, field, value))
+            # If this flag gets cleared while we are recalculating, then it can cause
+            # everything to lock, so we setup a lock between this function and the
+            # recalculate function
+            self._up_to_date.clear()
+            logging.debug(f"Added task to async queue. qsize={self._queue.qsize()}")
 
     async def _gather_one_sample(self):
         """If the queue is empty Wait() yields until an item is added. When the
@@ -306,14 +314,22 @@ async def _recalculate_phys_data(self, callback):
                 # callback checks the flag.
                 self._up_to_date.set()
                 logging.debug("Simulation up to date.")
-                if callback is not None:
-                    logging.debug(f"Executing callback function: {callback.__name__}")
-                    await callback()
-                    logging.debug("Callback completed.")
+                async with self._new_data_lock:
+                    if callback is not None:
+                        logging.debug(
+                            f"Executing callback function: {callback.__name__}"
+                        )
+                        # For Virtac this function calls update_pvs() which gets data
+                        # from the pytac datasource to update the softioc pvs with. The
+                        # data source is sim_data_sources.py and its get_value()
+                        # function waits on the wait_for_calculation() function which waits for the
+                        # up_to_date flag which currently will always be set, so this
+                        # process is pointless.
+                        await callback()
+                        logging.debug("Callback completed.")
                 # After this point we assume new setpoints have made the data stale. We
                 # cant clear this flag in queue_set() as the callbacks can depend on
                 # this being set.
-                self._up_to_date.clear()
 
     def toggle_calculations(self):
         """Pause or unpause the physics calculations by setting or clearing the