diff --git a/.github/workflows/pytest.yaml b/.github/workflows/pytest.yaml index 54aae58d02f..84e0f8e9f84 100644 --- a/.github/workflows/pytest.yaml +++ b/.github/workflows/pytest.yaml @@ -120,21 +120,13 @@ jobs: if: ${{ runner.os == 'Windows' }} run: echo "PYTEST_BASETEMP=--basetemp=D:\\tmp" >> $GITHUB_ENV shell: bash - - name: Run parallel tests + - name: Run tests if: ${{ always() }} shell: bash run: | set -x - pytest -m "not serial" --cov --cov-report xml --hypothesis-profile ci --durations=20 $PYTEST_OPT $PYTEST_BASETEMP tests - id: parallel_tests -# a subset of the tests fails when run in parallel on Windows so run those in serial here - - name: Run serial tests - if: ${{ always() }} - shell: bash - run: | - set -x - pytest -m "serial" -n 0 --dist no --cov --cov-report xml --cov-append --hypothesis-profile ci $PYTEST_OPT $PYTEST_BASETEMP tests - id: serial_tests + pytest --cov --cov-report xml --hypothesis-profile ci --durations=20 $PYTEST_OPT $PYTEST_BASETEMP tests + id: tests - name: Upload coverage to Codecov uses: codecov/codecov-action@303a32d7a59b442fa8d48b6a1cc6825c09c847a5 # v7.1.1 with: diff --git a/docs/changes/newsfragments/8758.improved b/docs/changes/newsfragments/8758.improved new file mode 100644 index 00000000000..a069836c209 --- /dev/null +++ b/docs/changes/newsfragments/8758.improved @@ -0,0 +1,5 @@ +Fixed two race conditions in ``DataSet`` subscribers. A result inserted while a subscriber's +callback was running could be held back until the next insert, and when a dataset was +completed, its final callback could run concurrently with one already in progress on the +subscriber thread, delivering results out of order. Subscriber callbacks now never overlap +and always receive results in the order they were inserted. diff --git a/docs/changes/newsfragments/8758.improved.1 b/docs/changes/newsfragments/8758.improved.1 new file mode 100644 index 00000000000..541a3c8b108 --- /dev/null +++ b/docs/changes/newsfragments/8758.improved.1 @@ -0,0 +1,5 @@ +Subscribing to a ``DataSet`` that is written in the background (``write_in_background=True``) +now raises a ``NotImplementedError``. This combination never worked: the background writer +failed to insert the first results and the measurement then hung. ``Measurement.run`` raises +the error before the run is created if subscribers are set up via ``Measurement.add_subscriber``, +the ``subscription.default_subscribers`` config option or ``DataSaver.default_callback``. diff --git a/pyproject.toml b/pyproject.toml index 700f2bde59d..173b964e476 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -76,7 +76,6 @@ test = [ "pytest-asyncio>=0.24.0", "pytest-cov>=5.0.0", "pytest-mock>=3.14.0", - "pytest-rerunfailures>=14.0", "pytest-xdist>=3.6.1", "PyVisa-sim>=0.6.0", "sphinx>=4.5.0", # sphinx extension tests @@ -215,7 +214,6 @@ minversion = "7.2" testpaths = "tests" addopts = "-n auto --dist=loadfile --cov-config=pyproject.toml" asyncio_default_fixture_loop_scope = "function" -markers = "serial" # we ignore warnings # triggered by third party packages diff --git a/requirements.txt b/requirements.txt index dfc2cd655ea..66a5201cdce 100644 --- a/requirements.txt +++ b/requirements.txt @@ -243,7 +243,6 @@ packaging==26.3 # matplotlib # nbconvert # pytest - # pytest-rerunfailures # qcodes # sphinx # versioningit @@ -306,7 +305,6 @@ pytest==9.2.0 # pytest-asyncio # pytest-cov # pytest-mock - # pytest-rerunfailures # pytest-xdist pytest-asyncio==1.4.1.post1 # via qcodes (pyproject.toml) @@ -314,8 +312,6 @@ pytest-cov==7.1.0 # via qcodes (pyproject.toml) pytest-mock==3.17.0 # via qcodes (pyproject.toml) -pytest-rerunfailures==16.7 - # via qcodes (pyproject.toml) pytest-xdist==3.8.0 # via qcodes (pyproject.toml) python-dateutil==4.0.0 diff --git a/src/qcodes/dataset/data_set.py b/src/qcodes/dataset/data_set.py index 08536a0855d..e9888fc7e68 100644 --- a/src/qcodes/dataset/data_set.py +++ b/src/qcodes/dataset/data_set.py @@ -207,6 +207,15 @@ class _WriterStatus: _WRITERS: dict[str, _WriterStatus] = {} +# The SQL function that a subscriber's trigger calls only exists on the +# DataSet's own connection, so inserts made by the background writer, which +# uses its own connection, would fail. +_SUBSCRIBERS_WITH_BACKGROUND_WRITING_ERROR = ( + "Subscribers are not supported for a DataSet that is written in the " + "background. Write the DataSet in the main thread " + "(write_in_background=False) to subscribe to it." +) + class DataSet(BaseDataSet): # the "persistent traits" are the attributes/properties of the DataSet @@ -770,6 +779,8 @@ def _perform_start_actions( "be written either in the background or in the " "main thread. You cannot mix." ) + if start_bg_writer and (self.subscribers or self._pending_subscribers): + raise NotImplementedError(_SUBSCRIBERS_WITH_BACKGROUND_WRITING_ERROR) with atomic(self.conn) as conn: for spec in paramspecs: @@ -1248,6 +1259,15 @@ def subscribe( state: Any | None = None, callback_kwargs: Mapping[str, Any] | None = None, ) -> str: + writer_status = self._writer_status + # Background writing is tracked per database, so also check that it is + # this run that is being written. A dataset that has not been started + # can be subscribed to; starting it in the background then raises. + if ( + writer_status.write_in_background + and self.run_id in writer_status.active_datasets + ): + raise NotImplementedError(_SUBSCRIBERS_WITH_BACKGROUND_WRITING_ERROR) subscriber_id = uuid.uuid4().hex if not self._results_table_exists: # The results table does not exist yet - this happens with split diff --git a/src/qcodes/dataset/measurements.py b/src/qcodes/dataset/measurements.py index 53e687107b5..e996c68c722 100644 --- a/src/qcodes/dataset/measurements.py +++ b/src/qcodes/dataset/measurements.py @@ -26,7 +26,11 @@ import qcodes as qc import qcodes.validators as vals -from qcodes.dataset.data_set import DataSet, load_by_guid +from qcodes.dataset.data_set import ( + _SUBSCRIBERS_WITH_BACKGROUND_WRITING_ERROR, + DataSet, + load_by_guid, +) from qcodes.dataset.data_set_in_memory import DataSetInMem from qcodes.dataset.data_set_protocol import ( DataSetProtocol, @@ -602,6 +606,22 @@ def _calculate_write_period( return float(write_period) def __enter__(self) -> DataSaver: + # Fail before creating the run: subscribing to a DataSet that is + # written in the background is not supported, and raising only once + # the run is started would leave it incomplete. The DataSaver also + # adds the default callback and the default subscribers from the config. + if ( + self._write_in_background + and self._dataset_class is DataSetType.DataSet + and ( + self.subscribers + or qc.config.subscription.default_subscribers + or "run_tables_subscription_callback" + in (DataSaver.default_callback or {}) + ) + ): + raise NotImplementedError(_SUBSCRIBERS_WITH_BACKGROUND_WRITING_ERROR) + # multiple runners can be active at the same time. # If we just activate them in order the first one # would be the parent of the next one but that is wrong @@ -1480,6 +1500,9 @@ def add_subscriber( """ Add a subscriber to the dataset of the measurement. + Subscribers are not supported when the dataset is written in the + background, see the ``write_in_background`` argument of :meth:`run`. + Args: func: A function taking three positional arguments: a list of tuples of parameter values, an integer, a mutable variable @@ -1519,7 +1542,10 @@ def run( will be stored in background, without blocking the main thread that is executing the context manager. By default the setting for write in background will be - read from the ``qcodesrc.json`` config file. + read from the ``qcodesrc.json`` config file. Subscribers + are not supported when writing in the background for the + default ``DataSet`` class, so entering the context manager + raises a ``NotImplementedError`` if any are set up. in_memory_cache: Should measured data be keep in memory and available as part of the `dataset.cache` object. dataset_class: Enum representing the Class used to store data diff --git a/src/qcodes/dataset/subscriber.py b/src/qcodes/dataset/subscriber.py index 5c87049c3f6..506cc134632 100644 --- a/src/qcodes/dataset/subscriber.py +++ b/src/qcodes/dataset/subscriber.py @@ -4,7 +4,7 @@ import logging import time from queue import Empty, Queue -from threading import Thread +from threading import Lock, Thread from typing import TYPE_CHECKING, Any from qcodes.dataset.sqlite.connection import atomic_transaction @@ -45,12 +45,20 @@ def __init__( self.dataSet = dataSet self.table_name = dataSet.table_name + # Number of results in the dataset up to and including the ones last + # passed to the callback. Only updated while draining the queue, so it + # always matches the results that the callback has been given. self._data_set_len = len(dataSet) self.state = state self.data_queue: Queue[Any] = Queue() - self._queue_length: int = 0 + # The callback is called both from this thread and, on completion, + # from the thread that completes the dataset. Holding this lock while + # draining the queue and calling the callback keeps those calls from + # overlapping, so results are delivered in insertion order and + # done_callback only returns once every queued result is delivered. + self._callback_lock = Lock() self._stop_signal: bool = False # convert milliseconds to seconds self._loop_sleep_time = loop_sleep_time / 1000 @@ -82,8 +90,6 @@ def __init__( def _cache_data_to_queue(self, *args: Any) -> None: self.data_queue.put(args) - self._data_set_len += 1 - self._queue_length += 1 def run(self) -> None: self.log.debug("Starting subscriber") @@ -100,8 +106,10 @@ def _exhaust_queue(queue: Queue[Any]) -> list[Any]: return result_list def _call_callback_on_queue_data(self) -> None: - result_list = self._exhaust_queue(self.data_queue) - self.callback(result_list, self._data_set_len, self.state) + with self._callback_lock: + result_list = self._exhaust_queue(self.data_queue) + self._data_set_len += len(result_list) + self.callback(result_list, self._data_set_len, self.state) def _loop(self) -> None: while True: @@ -109,9 +117,11 @@ def _loop(self) -> None: self._clean_up() break - if self._queue_length >= self.min_queue_length: + # Use the queue's own size rather than a separate counter: a + # counter reset after the callback would hold back results + # inserted while the callback was running until the next insert. + if self.data_queue.qsize() >= self.min_queue_length: self._call_callback_on_queue_data() - self._queue_length = 0 time.sleep(self._loop_sleep_time) diff --git a/tests/dataset/measurement/test_measurement_context_manager.py b/tests/dataset/measurement/test_measurement_context_manager.py index d07287fb03c..9ef5b848b06 100644 --- a/tests/dataset/measurement/test_measurement_context_manager.py +++ b/tests/dataset/measurement/test_measurement_context_manager.py @@ -6,7 +6,6 @@ import re import traceback from functools import reduce -from time import sleep from typing import TYPE_CHECKING, Any, Literal import hypothesis.strategies as hst @@ -674,8 +673,6 @@ def collect_x_vals( assert collected_x_vals == given_x_vals -@pytest.mark.serial -@pytest.mark.flaky(reruns=5) def test_subscribers_called_for_all_data_points( experiment: Experiment, DAC: DummyInstrument, DMM: DummyInstrument ) -> None: @@ -717,9 +714,24 @@ def sub_get_y_vals( assert yvals == list(given_yvals) -# There is no way around it: this test is slow. We test that write_period -# works and hence we must wait for some time to elapse. Sorry. -@pytest.mark.serial +def test_subscribers_with_background_writing_not_supported( + experiment: Experiment, DAC: DummyInstrument +) -> None: + meas = Measurement(exp=experiment) + meas.register_parameter(DAC.ch1) + meas.add_subscriber(lambda results, length, state: None, state=[]) + no_of_runs = len(experiment) + + with ( + pytest.raises(NotImplementedError, match="written in the background"), + meas.run(write_in_background=True), + ): + pass + + # the error is raised before the run is created + assert len(experiment) == no_of_runs + + @settings( max_examples=5, deadline=None, @@ -735,6 +747,7 @@ def test_datasaver_scalars( experiment: Experiment, DAC: DummyInstrument, DMM: DummyInstrument, + mocker: MockerFixture, set_values: list[float], get_values: list[float], breakpoint: int, @@ -752,21 +765,24 @@ def test_datasaver_scalars( meas.register_parameter(DAC.ch1) meas.register_parameter(DMM.v1, setpoints=(DAC.ch1,)) + # Drive the clock that the DataSaver uses for write_period by hand, so + # that the outcome does not depend on how fast the machine runs the test. + now = 0.0 + + def fake_perf_counter() -> float: + return now + + mocker.patch("qcodes.dataset.measurements.perf_counter", fake_perf_counter) + with meas.run() as datasaver: for set_v, get_v in zip(set_values[:breakpoint], get_values[:breakpoint]): datasaver.add_result((DAC.ch1, set_v), (DMM.v1, get_v)) assert datasaver._dataset.number_of_results == 0 - sleep(write_period * 1.1) + now += write_period * 1.1 datasaver.add_result( (DAC.ch1, set_values[breakpoint]), (DMM.v1, get_values[breakpoint]) ) - # test that work with time intervals are often flaky - # so we add a bit more wait time here if the expected number - # of points have not been written - for _ in range(10): - if datasaver.points_written != breakpoint + 1: - sleep(write_period * 1.1) assert datasaver.points_written == breakpoint + 1 assert datasaver.run_id == no_of_runs + 1 diff --git a/tests/dataset/test_datasaver.py b/tests/dataset/test_datasaver.py index a552b1b7c23..7ae5ae3b08a 100644 --- a/tests/dataset/test_datasaver.py +++ b/tests/dataset/test_datasaver.py @@ -42,8 +42,7 @@ def reset_callback_globals() -> "None": @pytest.mark.usefixtures("experiment") -@pytest.mark.parametrize("bg_writing", [True, False]) -def test_default_callback(bg_writing: "bool") -> None: +def test_default_callback() -> None: """ The Web UI needs to know the results of an experiment with the metadata. So a default_callback class variable is set by the Web UI with a callback @@ -67,7 +66,7 @@ def test_default_callback(bg_writing: "bool") -> None: interdeps=InterDependencies_(), registered_parameters=[], ) - test_set.mark_started(start_bg_writer=bg_writing) + test_set.mark_started() test_set.mark_completed() assert CALLBACK_SNAPSHOT == "reasonable_snapshot" assert CALLBACK_RUN_ID is not None @@ -79,6 +78,39 @@ def test_default_callback(bg_writing: "bool") -> None: test_set.conn.close() +@pytest.mark.usefixtures("experiment") +def test_default_callback_with_background_writing_not_supported() -> None: + """ + Subscribers, such as the default callback, are not supported for a + dataset that is written in the background, so starting it raises. + """ + test_set = None + try: + DataSaver.default_callback = { + "run_tables_subscription_callback": callback, + "run_tables_subscription_min_wait": 1, + "run_tables_subscription_min_count": 2, + } + + test_set = new_data_set("test-dataset") + test_set.add_metadata("snapshot", "reasonable_snapshot") + DataSaver( + dataset=test_set, + write_period=0, + interdeps=InterDependencies_(), + registered_parameters=[], + ) + with pytest.raises(NotImplementedError, match="written in the background"): + test_set.mark_started(start_bg_writer=True) + assert test_set.pristine + finally: + DataSaver.default_callback = None + if test_set is not None: + # The dataset is never completed, so stop the subscriber explicitly. + test_set.unsubscribe_all() + test_set.conn.close() + + @pytest.mark.usefixtures("experiment") @pytest.mark.parametrize("bg_writing", [True, False]) def test_numpy_types(bg_writing: "bool") -> None: diff --git a/tests/dataset/test_nested_measurements.py b/tests/dataset/test_nested_measurements.py index cd151160f09..73172ebb669 100644 --- a/tests/dataset/test_nested_measurements.py +++ b/tests/dataset/test_nested_measurements.py @@ -189,8 +189,6 @@ def subscriber( return subscriber -@pytest.mark.flaky(reruns=5) -@pytest.mark.serial def test_subscription_on_dual_datasets( experiment: "Experiment", basic_subscriber: "Callable[[list[tuple[VALUE]], int, dict | None], None]", diff --git a/tests/dataset/test_subscribing.py b/tests/dataset/test_subscribing.py index ad5b84fb579..fde8edc47de 100644 --- a/tests/dataset/test_subscribing.py +++ b/tests/dataset/test_subscribing.py @@ -7,6 +7,7 @@ from numpy import ndarray import qcodes +from qcodes.dataset import new_data_set from qcodes.dataset.descriptions.dependencies import InterDependencies_ from qcodes.dataset.sqlite.connection import atomic_transaction from qcodes.parameters import ParamSpecBase @@ -134,8 +135,6 @@ def _make_broken_subscriber_config(tmp_path: "Path") -> "Generator[None, None, N yield -@pytest.mark.flaky(reruns=5) -@pytest.mark.serial def test_basic_subscription( dataset: "DataSet", basic_subscriber: "Callable[[list[tuple[VALUE]], int, dict | None], None]", @@ -179,6 +178,85 @@ def assert_expected_state() -> "None": assert len(triggers) == 0 +def test_subscribe_with_background_writing_not_supported(dataset: "DataSet") -> None: + xparam = ParamSpecBase(name="x", paramtype="numeric") + yparam = ParamSpecBase(name="y", paramtype="numeric") + dataset.set_interdependencies(InterDependencies_(dependencies={yparam: (xparam,)})) + dataset.mark_started(start_bg_writer=True) + try: + with pytest.raises(NotImplementedError, match="written in the background"): + dataset.subscribe(lambda results, length, state: None) + assert dataset.subscribers == {} + finally: + # completing the dataset shuts down the background writer + dataset.mark_completed() + + +def test_subscribe_while_another_run_writes_in_background( + dataset: "DataSet", +) -> None: + """ + Only the run that is written in the background can not be subscribed to. + Another dataset in the same database can, and can then be written in the + main thread once the background run has completed. + """ + xparam = ParamSpecBase(name="x", paramtype="numeric") + yparam = ParamSpecBase(name="y", paramtype="numeric") + idps = InterDependencies_(dependencies={yparam: (xparam,)}) + + background_run = new_data_set("background-run") + background_run.set_interdependencies(idps) + background_run.mark_started(start_bg_writer=True) + + dataset.set_interdependencies(idps) + received: list[tuple[VALUE, ...]] = [] + try: + dataset.subscribe( + lambda results, length, state: received.extend(results), + min_wait=0, + min_count=1, + ) + finally: + background_run.mark_completed() + background_run.conn.close() + + dataset.mark_started() + dataset.add_results([{"x": 1, "y": 2}]) + dataset.mark_completed() + assert received == [(1, 2)] + + +def test_subscriber_length_matches_delivered_results(dataset: "DataSet") -> None: + """ + The length passed to the callback is the number of results in the dataset + up to and including the ones passed along with it, however the inserts + and the callbacks interleave. + """ + xparam = ParamSpecBase(name="x", paramtype="numeric") + yparam = ParamSpecBase(name="y", paramtype="numeric") + dataset.set_interdependencies(InterDependencies_(dependencies={yparam: (xparam,)})) + dataset.mark_started() + + calls: list[tuple[int, int]] = [] + + def subscriber( + results: "list[tuple[VALUE, ...]]", length: int, state: None + ) -> None: + calls.append((length, len(results))) + + dataset.subscribe(subscriber, min_wait=0, min_count=1) + n_results = 1000 + for x in range(n_results): + dataset.add_results([{"x": x, "y": x}]) + dataset.mark_completed() + + delivered = 0 + for length, n_delivered in calls: + delivered += n_delivered + assert length == delivered + assert delivered == n_results + + @pytest.mark.usefixtures("working_subscriber_config") def test_subscription_from_config( dataset: "DataSet",