Repository: PennyLaneAI/pennylane
Pull request #7075 (2025-03-12 15:47:43)

Add utilities module to qml.ftqc for workload support
### Before submitting

Please complete the following checklist when submitting a PR:

- [x] All new features must include a unit test.
      If you've fixed a bug or added code that should be tested, add a test to the
      test directory!

- [x] All new functions and code must be clearly commented and documented.
      If you do make documentation changes, make sure that the docs build and
      render correctly by running `make docs`.

- [x] Ensure that the test suite passes, by running `make test`.

- [x] Add a new entry to the `doc/releases/changelog-dev.md` file, summarizing the
      change, and including a link back to the PR.

- [x] The PennyLane source code conforms to
      [PEP8 standards](https://www.python.org/dev/peps/pep-0008/).
      We check all of our code against [Pylint](https://www.pylint.org/).
      To lint modified files, simply `pip install pylint`, and then
      run `pylint pennylane/path/to/file.py`.

When all the above are checked, delete everything above the dashed
line and fill in the pull request template.

------------------------------------------------------------------------------------------------------------

**Context:** This PR adds utility methods for the FTQC module. Included is a `QubitMgr` class, to assist with dynamically recycling qubit wire labels during execution.

**Description of the Change:** As above.

**Benefits:** Required to support the `qml.ftqc` module workloads.

**Possible Drawbacks:** Potential rewrites may be required with future dynamic allocation supports.

**Related GitHub Issues:**


github-actions[bot]: Hello. You may have forgotten to update the changelog!
Please edit [`doc/releases/changelog-dev.md`](/PennyLaneAI/pennylane/blob/master/doc/releases/changelog-dev.md) with:
* A one-to-two sentence description of the change. You may include a small working example for new features.
* A link back to this PR.
* Your name (or GitHub username) in the contributors section.
mlxd: [sc-86310]
mlxd: Thanks @multiphaseCFD @lillian542  and @joeycarter 

Patch:
diff --git a/doc/releases/changelog-dev.md b/doc/releases/changelog-dev.md
index af30237e1..22382f923 100644
--- a/doc/releases/changelog-dev.md
+++ b/doc/releases/changelog-dev.md
@@ -189,6 +189,9 @@
   [(#7064)](https://github.com/PennyLaneAI/pennylane/pull/7064)
 
 <h3>Improvements 🛠</h3>
+
+* A new utility module `qml.ftqc.utils` is provided, with support for functionality such as dynamic qubit recycling.
+  [(#7075)](https://github.com/PennyLaneAI/pennylane/pull/7075/)
   
 * The decompositions of `qml.SX`, `qml.X` and `qml.Y` use `qml.GlobalPhase` instead of `qml.PhaseShift`.
   [(#7073)](https://github.com/PennyLaneAI/pennylane/pull/7073)  
diff --git a/pennylane/ftqc/__init__.py b/pennylane/ftqc/__init__.py
index 46bc3013a..fd9323c3d 100644
--- a/pennylane/ftqc/__init__.py
+++ b/pennylane/ftqc/__init__.py
@@ -40,11 +40,13 @@ from .conditional_measure import cond_measure
 from .lattice import Lattice, generate_lattice
 from .graph_state_preparation import GraphStatePrep
 from .qubit_graph import QubitGraph
+from .utils import QubitMgr
 
 __all__ = [
     "GraphStatePrep",
     "Lattice",
     "QubitGraph",
+    "QubitMgr",
     "ParametricMidMeasureMP",
     "XMidMeasureMP",
     "YMidMeasureMP",
diff --git a/pennylane/ftqc/utils.py b/pennylane/ftqc/utils.py
new file mode 100644
index 000000000..5ea797b54
--- /dev/null
+++ b/pennylane/ftqc/utils.py
@@ -0,0 +1,229 @@
+# Copyright 2025 Xanadu Quantum Technologies Inc.
+
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+
+#     http://www.apache.org/licenses/LICENSE-2.0
+
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+r"""
+This module contains utility data-structures and algorithms supporting functionality in the
+ftqc module.
+"""
+from threading import RLock
+
+
+class QubitMgr:
+    r"""
+    The ``QubitMgr`` object maintains a list of active and inactive qubit wire indices used and for use
+    during execution of a workload. Its purpose is to allow tracking of free qubit indices that
+    are in the :math:`\vert 0 \rangle` state to participate in MCM-based workloads, under the assumption of reset
+    upon measurement. Qubit wires indices will be tracked with a monotonically increasing set
+    of values, starting from the initial input ``start_idx``.
+
+    Args:
+        num_qubits (int): Total number of wire indices to track.
+        start_idx (int): Starting index of wires to track. Defaults to 0.
+
+    **Example:**
+        The following MBQC example workload uses the ``QubitMgr`` to assist with recycling of indices
+        between iterations
+
+        .. code-block:: python3
+
+            import pennylane as qml
+            from pennylane.ftqc import QubitGraph, diagonalize_mcms, generate_lattice, measure_x, measure_y
+            dev = qml.device('null.qubit')
+
+            @qml.qnode(dev, mcm_method="one-shot")
+            def circuit_mbqc(start_state, angles):
+                q_mgr = QubitMgr(num_qubits=5, start_idx=0)
+                input_idx = q_mgr.acquire_qubit()
+
+                # prep input node
+                qml.StatePrep(start_state, wires=[input_idx])
+
+                # prep and consume graph state iteratively
+                for i in range(num_iter):
+                    # Acquire 4 free qubit indices
+                    graph_wires = q_mgr.acquire_qubits(4)
+
+                    # Denote the index for the final output state
+                    output_idx = graph_wires[-1]
+
+                    # Prepare the state
+                    qml.ftqc.GraphStatePrep(lattice.graph, wires=graph_wires)
+
+                    # entangle input and graph using first qubit
+                    qml.CZ([input_idx, graph_wires[0]])
+
+                    # MBQC Z rotation: X, X, +/- angle, X
+                    # Reset operations allow qubits to be returned to the pool
+                    m0 = measure_x(input_idx, reset=True)
+                    m1 = measure_x(graph_wires[0], reset=True)
+                    m2 = cond_measure(m1, partial(measure, angle=angle, reset=True), partial(measure, angle=-angle, reset=True))(plane="XY", wires=graph_wires[1])
+                    m3 = measure_x(graph_wires[2], reset=True)
+
+                    # corrections based on measurement outcomes
+                    qml.cond((m0+m2)%2, qml.Z)(graph_wires[3])
+                    qml.cond((m1+m3)%2, qml.X)(graph_wires[3])
+
+                    # The input qubit can be freed and the output qubit becomes the next iteration's input
+                    q_mgr.release_qubit(input_idx)
+                    input_idx = output_idx
+
+                    # We can now free all but the last qubit, which has become the new input_idx
+                    q_mgr.release_qubits(graph_wires[0:-1])
+
+                # Perform the measurements on the output qubit from the last iteration
+                return qml.expval(X(output_idx)), qml.expval(Y(output_idx)), qml.expval(Z(output_idx))
+
+        For each loop iteration, the measured and reset wire labels are returned to the ``QubitMgr`` instance, which are then reallocated
+        on the next step, which when combined with the MCM resets allows for qubit index recycling.
+
+    """
+
+    def __init__(self, num_qubits: int = 0, start_idx: int = 0):
+        # All resources are protected via a re-entrant lock, to ensure exclusive access when
+        # acquiring/accessing/releasing qubits.
+        self._lock = RLock()
+        self._num_qubits = num_qubits
+        self._active = set()
+        is_valid = lambda x: (isinstance(x, int) and x >= 0)
+        if is_valid(num_qubits) and is_valid(start_idx):
+            self._inactive = set(range(start_idx, start_idx + num_qubits, 1))
+        else:
+            raise TypeError(
+                f"Index counts and starting values must be positive integers. Received {num_qubits} and {start_idx}"
+            )
+
+    def __repr__(self):
+        return f"QubitMgr(num_qubits={self.num_qubits}, active={self.active}, inactive={self.inactive})"
+
+    @property
+    def num_qubits(self) -> int:
+        """Defines the total number of wire indices tracked by the manager.
+
+        Returns:
+            int: total number of qubit wire indices
+        """
+
+        return self._num_qubits
+
+    @property
+    def active(self) -> set:
+        """
+        Defines the active wire indices. Any wire index in this set is unavailable for use, as it may
+        be participating in existing algorithms and/or not be in a reset state.
+
+        Returns:
+            set[int]: active wire indices
+        """
+        with self._lock:
+            return self._active
+
+    @property
+    def inactive(self) -> set:
+        r"""
+        Defines the inactive wire indices. Any wire index in this set is available for use, and is
+        assumed to be in a reset (:math:`\vert 0 \rangle`) state.
+
+        Returns:
+            set[int]: inactive wire indices
+        """
+
+        with self._lock:
+            return self._inactive
+
+    @property
+    def all_qubits(self) -> set:
+        """
+        Defines all active and inactive wire indices.
+
+        Returns:
+            set[int]: union of active and inactive wire indices
+        """
+        with self._lock:
+            return self.inactive | self.active
+
+    def acquire_qubit(self) -> int:
+        """
+        Acquires an available qubit wire index from the inactive pool, and makes it active.
+        If there are no inactive qubits available a RuntimeError will be raised.
+
+        Returns:
+            int: newly activated qubit wire index
+        """
+        with self._lock:
+            try:
+                idx = self._inactive.pop()
+                self._active.add(idx)
+                return idx
+            except Exception as exc:
+                raise RuntimeError(
+                    "Cannot allocate any additional wire indices. Execution aborted."
+                ) from exc
+
+    def acquire_qubits(self, num_qubits: int) -> list[int]:
+        """
+        Acquires num_qubits qubit wire indices from the inactive pool, and makes them active.
+        If there are no inactive qubits available a RuntimeError will be raised.
+
+        .. seealso:: :meth:`~.QubitMgr.acquire_qubit`.
+
+        Returns:
+            list[int]: newly activated qubit wire indices
+        """
+        indices = []
+        if num_qubits > 0:
+            with self._lock:
+                while True:
+                    indices.append(self.acquire_qubit())
+                    if len(indices) == num_qubits:
+                        break
+        return indices
+
+    def release_qubit(self, idx: int) -> None:
+        """
+        Release an active qubit wire index, idx, from the active pool, and makes it inactive.
+        If idx is not in the active pool a RuntimeError will be raised.
+        """
+        with self._lock:
+            try:
+                self._active.remove(idx)
+            except Exception as exc:
+                raise RuntimeError(
+                    f"Wire index {idx} not found in active set. Execution aborted."
+                ) from exc
+            self._inactive.add(idx)
+
+    def release_qubits(self, indices: list[int]) -> None:
+        """
+        Release the list of active qubit wire indices, indices, from the active pool, and makes them inactive.
+        If any of the given indices are not in the active pool a RuntimeError will be raised.
+
+        .. seealso:: :meth:`~.QubitMgr.release_qubit`.
+        """
+        with self._lock:
+            for idx in indices:
+                self.release_qubit(idx)
+
+    def reserve_qubit(self, idx: int) -> None:
+        """
+        Explicitly reserve the qubit wire index, idx, to be active.
+        If given index is not in the active pool a RuntimeError will be raised.
+        """
+        with self._lock:
+            if idx in self._inactive:
+                self._inactive.remove(idx)
+                self._active.add(idx)
+            else:
+                raise RuntimeError(
+                    f"Qubit index {idx} not found in inactive set. Execution aborted."
+                )


Test patch:
diff --git a/tests/ftqc/test_utils.py b/tests/ftqc/test_utils.py
new file mode 100644
index 000000000..88d16bd56
--- /dev/null
+++ b/tests/ftqc/test_utils.py
@@ -0,0 +1,215 @@
+# Copyright 2018-2025 Xanadu Quantum Technologies Inc.
+
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+
+#     http://www.apache.org/licenses/LICENSE-2.0
+
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Unit tests for the ftqc.utils module"""
+
+import itertools as it
+import random
+
+import pytest
+
+from pennylane.ftqc import QubitMgr
+
+# pylint: disable=too-few-public-methods, too-many-public-methods
+
+
+num_qubits_vals = [1, 3, 7]
+acquire_num_vals = [0, 1, 5]
+offsets_vals = [0, 8, 9]
+
+
+def _is_valid(num_q, acquire_q):
+    "Utility function to select for valid/invalid parameters in tests"
+    if num_q < acquire_q:
+        return False
+    return True
+
+
+class TestQubitMgr:
+    """Tests for the dynamic qubit wire manager instance"""
+
+    def test_default_init(self):
+        "Default initialized manager holds 0 qubit wire indices"
+        mgr = QubitMgr()
+        assert mgr.num_qubits == 0
+        assert len(mgr.active) == 0
+        assert len(mgr.inactive) == 0
+        assert len(mgr.all_qubits) == 0
+
+    @pytest.mark.parametrize("num_qubits, offset", list(it.product(num_qubits_vals, offsets_vals)))
+    def test_explicit_init(self, num_qubits, offset):
+        "Test for valid initialization of QubitMgr"
+
+        mgr = QubitMgr(num_qubits, offset)
+        assert mgr.num_qubits == num_qubits
+        assert len(mgr.active) == 0
+        assert len(mgr.inactive) == num_qubits
+        assert len(mgr.all_qubits) == num_qubits
+        assert mgr.inactive == set(range(offset, offset + num_qubits, 1))
+        assert mgr.inactive == mgr.all_qubits
+        assert (
+            str(mgr)
+            == f"QubitMgr(num_qubits={num_qubits}, active={mgr.active}, inactive={mgr.inactive})"
+        )
+
+    @pytest.mark.parametrize(
+        "num_qubits, offset",
+        [
+            ("a", 0),
+            (0, "a"),
+            (1, range(3)),
+            (range(5), 0),
+        ],
+    )
+    def test_invalid_init(self, num_qubits, offset):
+        "Test for invalid initialization of QubitMgr"
+
+        with pytest.raises(
+            TypeError, match="Index counts and starting values must be positive integers"
+        ):
+            _ = QubitMgr(num_qubits, offset)
+
+    @pytest.mark.parametrize("num_qubits, offset", list(it.product(num_qubits_vals, offsets_vals)))
+    def test_acquire_qubit_valid(self, num_qubits, offset):
+        "Test that we can acquire a single qubits and make it active"
+
+        mgr = QubitMgr(num_qubits, offset)
+
+        assert len(mgr.inactive) == num_qubits
+        assert len(mgr.active) == 0
+        assert len(mgr.active.intersection(mgr.inactive)) == 0
+
+        idx_list = mgr.acquire_qubit()
+
+        assert idx_list == offset
+        assert set([idx_list]) == mgr.active
+        assert (
+            str(mgr)
+            == f"QubitMgr(num_qubits={num_qubits}, active={mgr.active}, inactive={mgr.inactive})"
+        )
+
+    @pytest.mark.parametrize(
+        "num_qubits, acquire_num, offset",
+        list(it.product(num_qubits_vals, acquire_num_vals, offsets_vals)),
+    )
+    def test_acquire_qubits(self, num_qubits, acquire_num, offset):
+        "Test that we can acquire multiple qubits and make them active"
+
+        if not _is_valid(num_qubits, acquire_num):
+            pytest.skip()
+
+        mgr = QubitMgr(num_qubits, offset)
+        idx_list = mgr.acquire_qubits(acquire_num)
+
+        assert set(idx_list) == mgr.active
+        assert len(mgr.active.intersection(mgr.inactive)) == 0
+
+    @pytest.mark.parametrize(
+        "num_qubits, acquire_num, offset",
+        list(it.product(range(5), acquire_num_vals, offsets_vals)),
+    )
+    def test_acquire_qubits_invalid(self, num_qubits, acquire_num, offset):
+        "Test that we have checks for invalid qubit acquisition requests"
+        if _is_valid(num_qubits, acquire_num):
+            pytest.skip()
+
+        mgr = QubitMgr(num_qubits, offset)
+        if num_qubits > 0:  # Pre-burn qubits for valid acquisition
+            mgr.acquire_qubits(num_qubits)
+        with pytest.raises(RuntimeError, match="Cannot allocate any additional wire indices"):
+            mgr.acquire_qubit()
+
+        with pytest.raises(RuntimeError, match="Cannot allocate any additional wire indices"):
+            mgr.acquire_qubits(acquire_num)
+
+    @pytest.mark.parametrize(
+        "num_qubits, acquire_num, offset",
+        list(it.product(num_qubits_vals, acquire_num_vals, offsets_vals)),
+    )
+    def test_release_qubit(self, num_qubits, acquire_num, offset):
+        "Test that we can acquire and release qubits individually"
+
+        if not _is_valid(num_qubits, acquire_num):
+            pytest.skip()
+        mgr = QubitMgr(num_qubits, offset)
+        idx_list = mgr.acquire_qubits(acquire_num)
+
+        for e in idx_list:
+            assert e in mgr.active
+            assert e not in mgr.inactive
+            mgr.release_qubit(e)
+            assert e not in mgr.active
+            assert e in mgr.inactive
+
+    @pytest.mark.parametrize(
+        "num_qubits, acquire_num, offset",
+        list(it.product(num_qubits_vals, acquire_num_vals, offsets_vals)),
+    )
+    def test_release_qubits(self, num_qubits, acquire_num, offset):
+        "Test that we can acquire and release a fixed number of qubits"
+
+        if not _is_valid(num_qubits, acquire_num):
+            pytest.skip()
+        mgr = QubitMgr(num_qubits, offset)
+        idx_list = mgr.acquire_qubits(acquire_num)
+
+        assert set(idx_list) == mgr.active
+        assert set(idx_list).intersection(mgr.inactive) == set()
+        mgr.release_qubits(idx_list)
+        if acquire_num > 0:
+            assert mgr.inactive.intersection(idx_list) == set(idx_list)
+            assert set(idx_list).intersection(mgr.active) == set()
+
+    @pytest.mark.parametrize(
+        "num_qubits, acquire_num, offset",
+        list(it.product(num_qubits_vals, acquire_num_vals, offsets_vals)),
+    )
+    def test_release_qubits_invalid(self, num_qubits, acquire_num, offset):
+        "Test that we have checks for invalid qubit release requests"
+        if _is_valid(num_qubits, acquire_num):
+            pytest.skip()
+
+        mgr = QubitMgr(num_qubits, offset)
+        with pytest.raises(RuntimeError, match="not found in active set"):
+            mgr.release_qubit(101)
+
+        with pytest.raises(RuntimeError, match="not found in active set"):
+            mgr.release_qubits([99, 88, 77])
+
+    @pytest.mark.parametrize("num_qubits, offset", list(it.product(num_qubits_vals, offsets_vals)))
+    def test_reserve_qubit(self, num_qubits, offset):
+        "Test that we can selectively reserve and make active a user-specified qubit wire index"
+        mgr = QubitMgr(num_qubits, offset)
+
+        # Ensure randomly ordered indices for reservation
+        q_inact = list(mgr.inactive)
+        random.shuffle(q_inact)
+
+        for q_idx in q_inact:
+            assert q_idx not in mgr.active
+            assert q_idx in mgr.inactive
+            mgr.reserve_qubit(q_idx)
+            assert q_idx in mgr.active
+            assert q_idx not in mgr.inactive
+
+    @pytest.mark.parametrize("num_qubits, offset", list(it.product(num_qubits_vals, offsets_vals)))
+    def test_reserve_qubit_invalid(self, num_qubits, offset):
+        "Test that we can selectively reserve and make active a user-specified qubit wire index"
+        mgr = QubitMgr(num_qubits, offset)
+
+        # Ensure randomly ordered indices for reservation
+        q_invalid = list(range(99, 100, 1))
+
+        for q_idx in q_invalid:
+            with pytest.raises(RuntimeError, match="not found in inactive set"):
+                mgr.reserve_qubit(q_idx)