Fix race condition in sqlite subscriber

.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:

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.

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``.

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

requirements.txt

@@ -243,7 +243,6 @@ packaging==26.3
     #   matplotlib
     #   nbconvert
     #   pytest
-    #   pytest-rerunfailures
     #   qcodes
     #   sphinx
     #   versioningit
@@ -306,16 +305,13 @@ pytest==9.2.0
     #   pytest-asyncio
     #   pytest-cov
     #   pytest-mock
-    #   pytest-rerunfailures
     #   pytest-xdist
 pytest-asyncio==1.4.1.post1
     # via qcodes (pyproject.toml)
 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

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

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

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,18 +106,22 @@ 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:
             if self._stop_signal:
                 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)
 

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

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:

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]",