diff --git a/csaxs_bec/devices/epics/mcs_card/mcs_card_csaxs.py b/csaxs_bec/devices/epics/mcs_card/mcs_card_csaxs.py index e2b87184..70511295 100644 --- a/csaxs_bec/devices/epics/mcs_card/mcs_card_csaxs.py +++ b/csaxs_bec/devices/epics/mcs_card/mcs_card_csaxs.py @@ -70,11 +70,14 @@ def suppress_mca_callbacks(mcs_card: MCSCard, restore_after_timeout: None | floa """ with mcs_card._rlock: mcs_card._omit_mca_callbacks.set() # pylint: disable=protected-access - try: - yield - finally: - if restore_after_timeout is not None: - time.sleep(restore_after_timeout) + try: + # MCA callbacks must be able to acquire the data lock while suppression + # is active so they are discarded instead of queued until afterward. + yield + finally: + if restore_after_timeout is not None: + time.sleep(restore_after_timeout) + with mcs_card._rlock: mcs_card._omit_mca_callbacks.clear() # pylint: disable=protected-access @@ -467,7 +470,16 @@ class MCSCardCSAXS(PSIDeviceBase, MCSCard): logger.info( f"Software triggered scan: {self._current_data_index}/{self.scan_parameters.num_points} points received." ) - if self._current_data_index == self.scan_parameters.num_points: + if self._current_data_index > self.scan_parameters.num_points: + exception = RuntimeError( + f"MCS card {self.name} emitted {self._current_data_index} " + f"data points, but the scan requested " + f"{self.scan_parameters.num_points}." + ) + logger.error(str(exception)) + for callback in self._scan_done_callbacks: + callback(exception=exception) + elif self._current_data_index == self.scan_parameters.num_points: for callback in self._scan_done_callbacks: callback(exception=None) else: diff --git a/tests/tests_devices/test_mcs_card.py b/tests/tests_devices/test_mcs_card.py index 2f8f6b07..0bfa70aa 100644 --- a/tests/tests_devices/test_mcs_card.py +++ b/tests/tests_devices/test_mcs_card.py @@ -24,7 +24,7 @@ from csaxs_bec.devices.epics.mcs_card.mcs_card import ( READMODE, MCSCard, ) -from csaxs_bec.devices.epics.mcs_card.mcs_card_csaxs import MCSCardCSAXS +from csaxs_bec.devices.epics.mcs_card.mcs_card_csaxs import MCSCardCSAXS, suppress_mca_callbacks from csaxs_bec.devices.utils.utils import fetch_scan_info @@ -167,9 +167,10 @@ def test_mcs_card_csaxs_complete_and_stop(mock_mcs_csaxs: MCSCardCSAXS): """ Test complete method of MCSCarcCSAXS. - Two use cases: + Three use cases: I. Acquisition is stopped externally II. Acquisition completes normally + III. Acquisition emits more points than requested """ mcs = mock_mcs_csaxs mcs.scan_parameters = fetch_scan_info(mcs.scan_info) @@ -219,6 +220,38 @@ def test_mcs_card_csaxs_complete_and_stop(mock_mcs_csaxs: MCSCardCSAXS): mcs._start_monitor_async_data_emission.wait(2) assert not mcs._start_monitor_async_data_emission.is_set() + ############################## + # III. Extra data fails fast # + ############################## + mcs._current_data_index = 11 + st = mcs.complete() + with pytest.raises(RuntimeError, match="emitted 11 data points, but the scan requested 10"): + st.wait(timeout=1) + + +def test_suppress_mca_callbacks_discards_callbacks_without_blocking(mock_mcs_csaxs: MCSCardCSAXS): + """A complete MCA callback batch is discarded while suppression is active.""" + mcs = mock_mcs_csaxs + callback_finished = threading.Event() + + def emit_callbacks(): + for index in range(mcs.NUM_MCA_CHANNELS): + counter = getattr(mcs.counters, f"mca{index + 1}") + mcs._on_counter_update(index, obj=counter, timestamp=1.0) + callback_finished.set() + + worker = threading.Thread(target=emit_callbacks) + with mock.patch.object(mcs.mca, "put") as mca_put: + with suppress_mca_callbacks(mcs, restore_after_timeout=0): + worker.start() + assert callback_finished.wait(timeout=1) + + worker.join(timeout=1) + assert not worker.is_alive() + assert mcs._current_data == {} + assert mcs._current_data_index == 0 + mca_put.assert_not_called() + def test_mcs_on_stop(mock_mcs_csaxs: MCSCardCSAXS): """Test that on stop sets the omit_mca_callbacks flag. Also test that on stage clears the omit_mca_callbacks flag."""