Repository: radiocosmology/alpenhorn

Issue:
lustrehsm: node re-init results in duplicated ready tasks
By "re-init" I mean `UpdateableNode.reinit` which is triggered by changing the `StorageNode`'s I/O class or I/O config (or sometimes when juggling node names around).

The duplication happens because the reinit results in a new instance of the I/O class being created, which doesn't have any memory of what the previous instance knew.  So, if there are deferred tasks in the mix (or really any long-running task), the new I/O instance won't consider them, and instead resubmits a bunch of new jobs for things already enqueued.

Possible solutions:
* empty out the FIFO of a reinited node (meaning all jobs need resubmission).  Not a big fan of this
* have the I/O instance scan its task FIFO on start up.  Would need some collaboration with the queue itself to do this.
* store information about running tasks outside of the I/O instance and/or have some sort of "persistent" memory that gets translated across reinit.  This might work, but likely only if the I/O class hasn't changed.  Unclear if this would be implemented on a class-specific basis, or globally by alpenhorn.

Discussion:
@ketiltrout: I think, in the case of ready requests from lustrehsm, this is more of an annoyance than an actual problem: multiple ready requests for the same file are not harmful, except in their ability to gum up the queue.

Solution patch:
diff --git a/alpenhorn/daemon/auto_import.py b/alpenhorn/daemon/auto_import.py
index b3b26c44b..d86c5c4c4 100644
--- a/alpenhorn/daemon/auto_import.py
+++ b/alpenhorn/daemon/auto_import.py
@@ -422,7 +422,10 @@ def update_observer(
         # If this node isn't being auto-imported or we're force-stopping
         # it, delete a watcher if one was previously scheduled
         if node.name in _watchers:
-            _observers[node.io_class].unschedule(_watchers[node.name])
+            try:
+                _observers[node.io_class].unschedule(_watchers[node.name])
+            except KeyError:
+                pass  # wasn't being watched
             del _watchers[node.name]
     else:
         # If there's already a watcher for this node, do nothing
diff --git a/alpenhorn/daemon/update.py b/alpenhorn/daemon/update.py
index a28c2c62c..ac50c9a07 100644
--- a/alpenhorn/daemon/update.py
+++ b/alpenhorn/daemon/update.py
@@ -46,6 +46,10 @@ class updateable_base:
     def __init__(self) -> None:
         raise NotImplementedError("updateable_base cannot be instantiated directly.")
 
+    def __del__(self) -> None:
+        # Stop updating
+        self.stop()
+
     @property
     def name(self) -> str:
         """The name of this instance."""
@@ -150,6 +154,28 @@ def _get_io_class(self):
         # return the class
         return class_
 
+    def stop(self) -> None:
+        """Stop updating.
+
+        Called when the instance is deleted, and when reinit
+        occurs.  Clears the associated queue FIFO.
+        """
+
+        if self._fifo is not None:
+            pending_count, deferred_count = self._queue.clear_fifo(
+                self._fifo, keep_clear=True
+            )
+            total = pending_count + deferred_count
+            if total:
+                name = (
+                    f'Group "{self.name}"' if self.is_group else f'Node "{self.name}"'
+                )
+                tasks = "task" if total == 1 else "tasks"
+                log.info(
+                    f"Discarded {total} {tasks} ({pending_count} pending; "
+                    f"{deferred_count} deferred) while stopping Storage{name}."
+                )
+
     def reinit(self, storage: StorageNode | StorageGroup) -> bool:
         """Re-initialise the instance with a new database object.
 
@@ -167,6 +193,19 @@ def reinit(self, storage: StorageNode | StorageGroup) -> bool:
         """
         # Does I/O instance need to be re-instantiated?
         if self._check_io_reinit(storage):
+            # The check here ensures the message is only printed if self was previously
+            # fully initialised
+            if self.db and self.io:
+                log.info(
+                    "I/O config change detected for Storage"
+                    + ("Group" if self.is_group else "Node")
+                    + f' "{self.name}": attempting re-init.'
+                )
+
+            # This will do nothing, when reinit is called from __init__, because
+            # self._fifo is None
+            self.stop()
+
             self.db = storage
             self.io_class = self._get_io_class()
 
@@ -178,8 +217,13 @@ def reinit(self, storage: StorageNode | StorageGroup) -> bool:
             # Parse I/O config if present
             config = self._parse_io_config(storage.io_config)
 
+            # Generate a new fifo key and label it
+            self._fifo = (self.is_group, storage.name, time.monotonic())
+            label = f"Group {storage.name}" if self.is_group else f"Node {storage.name}"
+            self._queue.label_fifo(self._fifo, label=label)
+
             # Initialise I/O object
-            self.io = self.io_class(storage, config, self._queue)
+            self.io = self.io_class(storage, config, self._queue, self._fifo)
 
             return True
 
@@ -209,6 +253,9 @@ def __init__(self, node: StorageNode) -> None:
         config = self._parse_io_config(node.io_config)
         self.io = self.io_class.remote_class(node, config)
 
+        # This ensures updatable_base.stop() does nothing.
+        self._fifo = None
+
 
 class UpdateableNode(updateable_base):
     """Updateable storage node
@@ -229,6 +276,7 @@ class UpdateableNode(updateable_base):
 
     def __init__(self, queue: FairMultiFIFOQueue, node: StorageNode) -> None:
         self._queue = queue
+        self._fifo = None
         self._updated = False
 
         # Set to True whenever I/O tasks are started.
@@ -258,6 +306,17 @@ def __del__(self):
             #   cf. https://github.com/prometheus/client_python/pull/1077
             pass
 
+        super().__del__()
+
+    def stop(self) -> None:
+        """Stop updating.
+
+        Stops auto-import, if running, and clears all pending tasks.
+        """
+        if self._fifo is not None:
+            auto_import.update_observer(self, self._queue, force_stop=True)
+        super().stop()
+
     def reinit(self, node: StorageNode) -> bool:
         """Re-initialise the instance with a new database object.
 
@@ -699,6 +758,7 @@ def __init__(
         self._do_idle_updates = False
 
         self._queue = queue
+        self._fifo = None
 
         self.reinit(group=group, nodes=nodes, idle=idle)
 
@@ -717,6 +777,8 @@ def __del__(self):
             #   we never set the metric to anything)
             pass
 
+        super().__del__()
+
     def reinit(
         self, *, group: StorageGroup, nodes: list[UpdateableNode], idle: bool
     ) -> None:
@@ -1010,10 +1072,9 @@ def update_loop(
         vetted_nodes = {}
         for name, node in nodes.items():
             if name not in new_nodes:
-                # Stop auto-import, if running
-                auto_import.update_observer(node, queue, force_stop=True)
-
                 log.info(f'Node "{name}" no longer available.')
+                # Stop updating
+                node.stop()
                 node_avail_metric.set(0, name=name)
             else:
                 vetted_nodes[name] = node
@@ -1068,6 +1129,8 @@ def update_loop(
         for name, group in groups.items():
             if name not in new_groups:
                 log.info(f'Group "{name}" no longer available.')
+                # Stop updating
+                group.stop()
                 group_avail_metric.set(0, name=name)
             else:
                 vetted_groups[name] = group
diff --git a/alpenhorn/io/base.py b/alpenhorn/io/base.py
index c74c0ef4b..b279f5443 100644
--- a/alpenhorn/io/base.py
+++ b/alpenhorn/io/base.py
@@ -15,7 +15,7 @@
 
 if TYPE_CHECKING:
     import os
-    from collections.abc import Iterable
+    from collections.abc import Hashable, Iterable
 
     from ..daemon.update import UpdateableNode
     from ..db import (
@@ -123,11 +123,13 @@ class BaseNodeIO:
     ----------
     node : StorageNode
         the node
-    queue : FairMultiFIFOQueue
-        the task queue
     config : dict
         the parsed `node.io_config`. If `node.io_config` is None,
         this is an empty `dict`.
+    queue : FairMultiFIFOQueue
+        the task queue
+    fifo : Hashable
+        the queue fifo key to use when submitting tasks.
     """
 
     # SETUP
@@ -144,12 +146,12 @@ class BaseNodeIO:
     observer = None
 
     def __init__(
-        self, node: StorageNode, config: dict, queue: FairMultiFIFOQueue
+        self, node: StorageNode, config: dict, queue: FairMultiFIFOQueue, fifo: Hashable
     ) -> None:
         self.node = node
         self._queue = queue
         self.config = config
-        self.fifo = "n:" + node.name
+        self.fifo = fifo
 
     def set_storage(self, node: StorageNode) -> None:
         """Update the cached StorageNode instance.
@@ -459,22 +461,28 @@ class BaseGroupIO:
     ----------
     group : StorageGroup
         The group
-    queue : FairMultiFIFOQueue
-        the task queue
     config : dict
         The parsed `group.io_config`. If `group.io_config` is None,
         this is an empty `dict`.
+    queue : FairMultiFIFOQueue
+        the task queue
+    fifo : Hashable
+        the queue fifo key to use when submitting tasks.
     """
 
     # SETUP
 
     def __init__(
-        self, group: StorageGroup, config: dict, queue: FairMultiFIFOQueue
+        self,
+        group: StorageGroup,
+        config: dict,
+        queue: FairMultiFIFOQueue,
+        fifo: Hashable,
     ) -> None:
         self.group = group
         self._queue = queue
         self.config = config
-        self.fifo = "g:" + group.name
+        self.fifo = fifo
 
     def set_storage(self, group: StorageGroup) -> None:
         """Update the cached StorageGroup instance.
diff --git a/alpenhorn/io/default.py b/alpenhorn/io/default.py
index 4faa88feb..ac3d183f9 100644
--- a/alpenhorn/io/default.py
+++ b/alpenhorn/io/default.py
@@ -29,7 +29,7 @@
 from .updownlock import UpDownLock
 
 if TYPE_CHECKING:
-    from collections.abc import Iterable
+    from collections.abc import Hashable, Iterable
 
     from ..db import ArchiveFileCopy, ArchiveFileCopyRequest, StorageNode
     from ..service.queue import FairMultiFIFOQueue
@@ -79,9 +79,9 @@ class DefaultNodeIO(BaseNodeIO):
     observer = Observer
 
     def __init__(
-        self, node: StorageNode, config: dict, queue: FairMultiFIFOQueue
+        self, node: StorageNode, config: dict, queue: FairMultiFIFOQueue, fifo: Hashable
     ) -> None:
-        super().__init__(node, config, queue)
+        super().__init__(node, config, queue, fifo)
 
         # The directory tree modification lock
         self.tree_lock = UpDownLock()
diff --git a/alpenhorn/io/lustrehsm.py b/alpenhorn/io/lustrehsm.py
index b5c794b1f..4b472d28f 100644
--- a/alpenhorn/io/lustrehsm.py
+++ b/alpenhorn/io/lustrehsm.py
@@ -31,7 +31,7 @@
 
 if TYPE_CHECKING:
     import os
-    from collections.abc import Generator
+    from collections.abc import Generator, Hashable
 
     from ..db import ArchiveFile, ArchiveFileCopyRequest
     from ..scheduler import FairMultiFIFOQueue
@@ -97,9 +97,13 @@ class LustreHSMNodeIO(LustreQuotaNodeIO):
     remote_class = LustreHSMNodeRemote
 
     def __init__(
-        self, node: UpdateableNode, config: dict, queue: FairMultiFIFOQueue
+        self,
+        node: UpdateableNode,
+        config: dict,
+        queue: FairMultiFIFOQueue,
+        fifo: Hashable,
     ) -> None:
-        super().__init__(node, config, queue)
+        super().__init__(node, config, queue, fifo)
 
         self._headroom = config["headroom"] * 2**10  # convert from kiB
 
@@ -661,9 +665,13 @@ class LustreHSMGroupIO(DefaultGroupIO):
     # SETUP
 
     def __init__(
-        self, group: UpdateableGroup, config: dict, queue: FairMultiFIFOQueue
+        self,
+        group: UpdateableGroup,
+        config: dict,
+        queue: FairMultiFIFOQueue,
+        fifo: Hashable,
     ) -> None:
-        super().__init__(group, config, queue)
+        super().__init__(group, config, queue, fifo)
 
         self._threshold = self.config.get("threshold", 1000000000)  # bytes
 
diff --git a/alpenhorn/io/lustrequota.py b/alpenhorn/io/lustrequota.py
index 8ea36a2cd..0cc298088 100644
--- a/alpenhorn/io/lustrequota.py
+++ b/alpenhorn/io/lustrequota.py
@@ -20,6 +20,8 @@
 from .lfs import LFS
 
 if TYPE_CHECKING:
+    from collections.abc import Hashable
+
     from ..queue import FairMultiFIFOQueue
     from ..storage import StorageNode
 del TYPE_CHECKING
@@ -54,9 +56,9 @@ class LustreQuotaNodeIO(DefaultNodeIO):
     """
 
     def __init__(
-        self, node: StorageNode, config: dict, queue: FairMultiFIFOQueue
+        self, node: StorageNode, config: dict, queue: FairMultiFIFOQueue, fifo: Hashable
     ) -> None:
-        super().__init__(node, config, queue)
+        super().__init__(node, config, queue, fifo)
 
         quota_id = config.get("quota_id", None)
         quota_type = config.get("quota_type", None)
diff --git a/alpenhorn/scheduler/queue.py b/alpenhorn/scheduler/queue.py
index a2d347361..85e9525ba 100644
--- a/alpenhorn/scheduler/queue.py
+++ b/alpenhorn/scheduler/queue.py
@@ -28,8 +28,10 @@ class FairMultiFIFOQueue:
 
     __slots__ = [
         "_all_tasks_done",
+        "_cleared_fifos",
         "_deferrals",
         "_dlock",
+        "_fifo_labels",
         "_fifo_locks",
         "_fifos",
         "_inprogress_counts",
@@ -62,6 +64,8 @@ def __init__(self) -> None:
         self._keys_by_inprogress = [set()]
         # The set of locked FIFOs
         self._fifo_locks = set()
+        # The set of retired FIFOs
+        self._cleared_fifos = set()
 
         # The thread lock (mutex) and the conditionals (see queue.py for
         # details, which this implementation broadly follows)
@@ -77,7 +81,11 @@ def __init__(self) -> None:
         # of the primary _lock.
         self._dlock = threading.Lock()
 
-        # Metrics
+        # METRICS
+        # =======
+
+        # FIFO labels.  A dict mapping FIFO keys to labels
+        self._fifo_labels = {}
 
         # Count of tasks
         self._qcount = Metric(
@@ -85,38 +93,44 @@ def __init__(self) -> None:
         )
 
         # Marginalised over status
-        self._qcount_any = Metric(
-            "queue_count",
-            "Count of queued tasks",
-            unbound=["fifo"],
-            bound={"status": "any"},
-        )
+        self._qcount_any = self._qcount.bind(status="any")
 
         # Marignalised over fifo
-        self._qcount_all = Metric(
-            "queue_count",
-            "Count of queued tasks",
-            unbound=["status"],
-            bound={"fifo": "_ALL_"},
-        )
+        self._qcount_all = self._qcount.bind(fifo="_ALL")
 
         # Track locked fifos
         self._qlock = Metric(
             "queue_locked", "The queue fifo is locked", unbound=["fifo"]
         )
 
+    # METRIC HANDLING
+    # ===============
+
+    def _adj_metrics(self, value: int, fifo: Hashable, status: str) -> None:
+        """Adjust the queue_count metric by value.
+
+        Adds `value` (which may be negative) to the metric with fifo=fifo and
+        status=status.  Also adds one to the marginalised ..._any and ..._all
+        metrics.
+        """
+
+        try:
+            fifo_label = self._fifo_labels[fifo]
+        except KeyError:
+            fifo_label = str(fifo)
+            self._fifo_labels[fifo] = fifo_label
+
+        self._qcount.add(value, fifo=fifo_label, status=status)
+        self._qcount_any.add(value, fifo=fifo_label)
+        self._qcount_all.add(value, status=status)
+
     def _inc_metrics(self, fifo: Hashable, status: str) -> None:
         """Increment the queue_count metric.
 
         Adds one to the metric with fifo=fifo and status=status.  Also adds
         one to the marginalised ..._any and ..._all metrics.
         """
-
-        fifo = str(fifo)
-
-        self._qcount.inc(fifo=fifo, status=status)
-        self._qcount_any.inc(fifo=fifo)
-        self._qcount_all.inc(status=status)
+        self._adj_metrics(1, fifo, status)
 
     def _dec_metrics(self, fifo: Hashable, status: str) -> None:
         """Decrement the queue_count metric.
@@ -124,86 +138,33 @@ def _dec_metrics(self, fifo: Hashable, status: str) -> None:
         Subtracts one from the metric with fifo=fifo and status=status.  Also
         subtracts one from the marginalised ..._any and ..._all metrics.
         """
+        self._adj_metrics(-1, fifo, status)
 
-        fifo = str(fifo)
+    def label_fifo(self, key: Hashable, label: str) -> None:
+        """Set the metric label for `key` to `label`.
 
-        self._qcount.dec(fifo=fifo, status=status)
-        self._qcount_any.dec(fifo=fifo)
-        self._qcount_all.dec(status=status)
-
-    def task_done(self, key: Hashable) -> None:
-        """Report that a task from the FIFO named `key` is done.
-
-        After a consumer has finished processing a task provided to it
-        by a get(), the key provided by that get() must be passed back
-        into this function to report the completion of that task.
+        This sets the value of the "fifo" label used in queue metrics
+        for the FIFO named `key`.
 
         Parameters
         ----------
-        key : hashable
-            The name of the FIFO that the completed task came from.  The
-            value passed in should previously have been returned from a
-            `get()` call.
-
-        Raises
-        ------
-        ValueError
-            The FIFO named `key` has no in-progress tasks.
+        key : Hashable
+            The FIFO to label.  The FIFO does not need to exist in the queue.
+        label : string
+            The label to use for the FIFO.
+
+        Notes
+        -----
+        Labels do not need to be unique.  If two FIFOs have the same label,
+        their metrics will be combined.
+
+        For unlablled FIFOs, the result of calling `str()` on the FIFO's
+        `key` is used in place of a label.
         """
+        self._fifo_labels[key] = label
 
-        with self._all_tasks_done:
-            # The number of currently in-progress tasks (including the one
-            # that's finishing now)
-            count = self._inprogress_counts.get(key, 0)
-
-            # If the specified FIFO has no in-progress tasks, raise value error
-            if count <= 0:
-                raise ValueError(f"no unfinished tasks for FIFO {key}")
-
-            # Remove the FIFO from the ordered list
-            self._keys_by_inprogress[count].remove(key)
-
-            # Unlock this FIFO if it was locked.  If the caller has been working on
-            # an exclusive item from the FIFO, then this call must be completing it,
-            # because there can't be anything else in-progress from this FIFO.  So,
-            # it's always reasonable to try to unlock a FIFO here.
-            self._fifo_locks.discard(key)
-            self._qlock.set(0, fifo=key)
-
-            # Decrement counts
-            count -= 1
-            self._inprogress_counts[key] = count
-            self._total_inprogress -= 1
-            self._dec_metrics(fifo=key, status="in-progress")
-
-            # Put the fifo back in the ordered list.
-            #
-            # XXX Could just delete a FIFO which is now empty here.
-            #
-            # In general, there's no reason to expect an empty FIFO to be reused,
-            # by the caller, and numerous empty FIFOs will slow down get() over time,
-            # so triming could decreate get() execution time.
-            #
-            # However, in our particular case (alpenhornd), the maximum number
-            # of FIFOs is total number of nodes ever active on this host, which is
-            # generally small and static enough not to have to worry about it.
-            self._keys_by_inprogress[count].add(key)
-
-            # XXX Could trim _keys_by_inprogress here.
-            #
-            # In general, it's potentially useful: it's possible for a bunch
-            # of unnecessary empty sets to accumulate at the tail of this list,
-            # so trimming can save memory in cases where large excursions of
-            # in-progress counts happen occasionally.
-            #
-            # However, in our particular case (alpenhornd), at most, the list
-            # will have as many elements as the maximum number of worker threads
-            # that have ever existed at one time, so it's probably not worth the
-            # trouble.
-
-            # Notify waiters when there are no pending tasks
-            if self._total_queued == 0 and self._total_inprogress == 0:
-                self._all_tasks_done.notify_all()  # wakes up all waiting threads
+    # QUEUE MAINTENANCE
+    # =================
 
     def join(self) -> None:
         """Blocks until there's nothing left in the queue.
@@ -227,6 +188,81 @@ def join(self) -> None:
         # Clear the metric.  This clears the marginalised versions, too
         self._qcount.clear()
 
+    def clear_fifo(self, key: Hashable, keep_clear: bool = False) -> tuple[int, int]:
+        """Remove all items from a fifo.
+
+        Removes all pending and deferred tasks from the FIFO named `key`.
+        In-progress tasks are not removed.
+
+        If the FIFO is not present in the queue, this function successfully
+        does nothing.
+
+        Parameters
+        ----------
+        key : hashable
+            The name of the FIFO to clear
+        keep_clear : bool
+            If True, don't accept any further puts on this FIFO.  If this is
+            False (the default), the caller should not assume that the specified
+            FIFO is empty when this function returns.
+
+        Returns
+        -------
+        Returns a two-tuple of integers:
+          * pending_removed: number of pending tasks removed
+          * deferred_removed: number of deferred tasks removed
+        """
+
+        # If we're keeping it clear, add the fifo to the list first, so that
+        # no concurrent put tries to add something while we're removing stuff
+        if keep_clear:
+            self._cleared_fifos.add(key)
+
+        # Clear deferred puts
+        deferred_removed = 0
+        new_deferrals = []
+        with self._dlock:
+            # Remove everything from the heapq, creating a new one to replace it
+            for index in range(len(self._deferrals)):
+                # Pop out of old heapq and add to new if not being removed
+                item = heapq.heappop(self._deferrals)
+
+                # Here item is a 4-tuple:
+                #  0: deferral time
+                #  1: item to put
+                #  2: FIFO key
+                #  3: exclusive flag
+                if item[2] == key:
+                    deferred_removed += 1
+                else:
+                    heapq.heappush(new_deferrals, item)
+            self._deferrals = new_deferrals
+
+        # Clear queued items
+        pending_removed = 0
+        with self._not_empty:
+            # Is there anything to clear?
+            try:
+                pending_removed = len(self._fifos[key])
+                self._total_queued -= pending_removed
+
+                # Reset the FIFO to an empty deque().  We do this even if we're
+                # going to keep the FIFO clear from now on because we need to
+                # still track in-progress items
+                self._fifos[key] = deque()
+            except KeyError:
+                pass  # FIFO doesn't exist, so nothing to remove
+
+        # If the queue is now empty, notify
+        with self._all_tasks_done:
+            if self._total_queued == 0 and self._total_inprogress == 0:
+                self._all_tasks_done.notify_all()  # wakes up all waiting threads
+
+        return (pending_removed, deferred_removed)
+
+    # METADATA
+    # ========
+
     @property
     def qsize(self) -> int:
         """Total number of queued tasks.
@@ -276,6 +312,9 @@ def deferred_size(self) -> int:
         with self._dlock:
             return len(self._deferrals)
 
+    # QUEUE PRODUCERS
+    # ===============
+
     def _put(self, item: Any, key: Hashable, exclusive: bool) -> None:
         """Put `item` into the FIFO named `key` without locking.
 
@@ -350,8 +389,18 @@ def put(
         result : bool
             False if `wait` > 0 and another thread is `join()`-ing the
             queue.  True otherwise.
+
+        Raises
+        ------
+        KeyError:
+            The FIFO named `key` is not accempting items because the queue
+            has been asked to keep it clear (see `clear_fifo()`).
         """
 
+        # Check if the fifo is being kept clear
+        if key in self._cleared_fifos:
+            raise KeyError("FIFO not accepting items")
+
         if wait > 0:
             # Deferred put
             with self._dlock:
@@ -371,6 +420,83 @@ def put(
 
         return True
 
+    # QUEUE CONSUMERS
+    # ===============
+
+    def task_done(self, key: Hashable) -> None:
+        """Report that a task from the FIFO named `key` is done.
+
+        After a consumer has finished processing a task provided to it
+        by a get(), the key provided by that get() must be passed back
+        into this function to report the completion of that task.
+
+        Parameters
+        ----------
+        key : hashable
+            The name of the FIFO that the completed task came from.  The
+            value passed in should previously have been returned from a
+            `get()` call.
+
+        Raises
+        ------
+        ValueError
+            The FIFO named `key` has no in-progress tasks.
+        """
+
+        with self._all_tasks_done:
+            # The number of currently in-progress tasks (including the one
+            # that's finishing now)
+            count = self._inprogress_counts.get(key, 0)
+
+            # If the specified FIFO has no in-progress tasks, raise value error
+            if count <= 0:
+                raise ValueError(f"no unfinished tasks for FIFO {key}")
+
+            # Remove the FIFO from the ordered list
+            self._keys_by_inprogress[count].remove(key)
+
+            # Unlock this FIFO if it was locked.  If the caller has been working on
+            # an exclusive item from the FIFO, then this call must be completing it,
+            # because there can't be anything else in-progress from this FIFO.  So,
+            # it's always reasonable to try to unlock a FIFO here.
+            self._fifo_locks.discard(key)
+            self._qlock.set(0, fifo=key)
+
+            # Decrement counts
+            count -= 1
+            self._inprogress_counts[key] = count
+            self._total_inprogress -= 1
+            self._dec_metrics(fifo=key, status="in-progress")
+
+            # Put the fifo back in the ordered list.
+            #
+            # XXX Could just delete a FIFO which is now empty here.
+            #
+            # In general, there's no reason to expect an empty FIFO to be reused,
+            # by the caller, and numerous empty FIFOs will slow down get() over time,
+            # so triming could decreate get() execution time.
+            #
+            # However, in our particular case (alpenhornd), the maximum number
+            # of FIFOs is total number of nodes ever active on this host, which is
+            # generally small and static enough not to have to worry about it.
+            self._keys_by_inprogress[count].add(key)
+
+            # XXX Could trim _keys_by_inprogress here.
+            #
+            # In general, it's potentially useful: it's possible for a bunch
+            # of unnecessary empty sets to accumulate at the tail of this list,
+            # so trimming can save memory in cases where large excursions of
+            # in-progress counts happen occasionally.
+            #
+            # However, in our particular case (alpenhornd), at most, the list
+            # will have as many elements as the maximum number of worker threads
+            # that have ever existed at one time, so it's probably not worth the
+            # trouble.
+
+            # Notify waiters when there are no pending tasks
+            if self._total_queued == 0 and self._total_inprogress == 0:
+                self._all_tasks_done.notify_all()  # wakes up all waiting threads
+
     def _get(self, t: float) -> tuple[Any, Hashable] | None:
         """Iterate the `get()` loop once for at most `t` seconds.
 
@@ -394,7 +520,7 @@ def _get(self, t: float) -> tuple[Any, Hashable] | None:
             The name of the FIFO from which `item` was popped
         """
         # Set to true later if we had to skip something due to an
-        # because its queue wasn't empty
+        # exclusive flag because its queue wasn't empty
         skipped_exclusive = False
 
         timeout_at = monotonic() + t
diff --git a/alpenhorn/scheduler/task.py b/alpenhorn/scheduler/task.py
index 1e50aa4cc..0bd8687d4 100644
--- a/alpenhorn/scheduler/task.py
+++ b/alpenhorn/scheduler/task.py
@@ -100,7 +100,11 @@ def __init__(
         self._generator = None
 
         # Enqueue ourself
-        queue.put(self, key, exclusive)
+        try:
+            queue.put(self, key, exclusive)
+        except KeyError:
+            # Key no longer accepted
+            log.info(f"Ignoring task {self._name}: FIFO closed")
 
     def __call__(self) -> bool:
         """This method is invoked by the worker thread to run the task.
@@ -135,7 +139,11 @@ def __call__(self) -> bool:
                 f"Requeueing yielded task {self._name} in FIFO {self._key} "
                 f"with delay {result} seconds"
             )
-            self._queue.put(self, self._key, wait=result)
+            try:
+                self._queue.put(self, self._key, wait=result)
+            except KeyError:
+                # Key no longer accepted
+                log.info(f"Ignoring task {self._name}: FIFO closed")
             return False
         except StopIteration:
             # Function exited without yielding (i.e. we're done)


Test patch:
diff --git a/tests/daemon/test_update.py b/tests/daemon/test_update.py
index 657210b13..b0330f70b 100644
--- a/tests/daemon/test_update.py
+++ b/tests/daemon/test_update.py
@@ -170,7 +170,7 @@ def task():
     assert task_count == 3
 
 
-def test_ioload(storagegroup, storagenode, mock_lfs):
+def test_ioload(storagegroup, storagenode, queue, mock_lfs):
     """Test instantiation of the I/O classes"""
 
     for ioclass in ["Default", "Transport", "LustreHSM", None]:
@@ -193,11 +193,11 @@ def test_ioload(storagegroup, storagenode, mock_lfs):
         )
 
     for node in StorageNode.select().execute():
-        unode = update.UpdateableNode(None, node)
+        unode = update.UpdateableNode(queue, node)
         assert isinstance(unode.io, BaseNodeIO)
 
     for group in StorageGroup.select().execute():
-        ugroup = update.UpdateableGroup(queue=None, group=group, nodes=[], idle=True)
+        ugroup = update.UpdateableGroup(queue=queue, group=group, nodes=[], idle=True)
         assert isinstance(ugroup.io, BaseGroupIO)
 
 
diff --git a/tests/daemon/test_update_node.py b/tests/daemon/test_update_node.py
index 6437552a3..549fb8d8e 100644
--- a/tests/daemon/test_update_node.py
+++ b/tests/daemon/test_update_node.py
@@ -12,11 +12,11 @@
 from alpenhorn.db.storage import StorageNode
 
 
-def test_bad_ioclass(simplenode):
+def test_bad_ioclass(simplenode, queue):
     """A missing I/O class is a problem."""
 
     simplenode.io_class = "Missing"
-    unode = UpdateableNode(None, simplenode)
+    unode = UpdateableNode(queue, simplenode)
     assert unode.io_class is None
     assert unode.io is None
 
@@ -85,16 +85,41 @@ def test_reinit(storagenode, simplegroup, queue):
     assert stnode is node.db
 
 
-def test_bad_ioconfig(simplenode):
+def test_reinit_clear_fifo(storagenode, simplegroup, queue):
+    """Test that reinit clears pending tasks from the queue."""
+
+    # Create a node
+    stnode = storagenode(name="node", group=simplegroup)
+    node = UpdateableNode(queue, stnode)
+
+    # Queue something with the node's FIFO key
+    queue.put(None, node._fifo)
+
+    # Check queue
+    assert queue.qsize == 1
+
+    # Change io_config to force re-init
+    stnode = StorageNode.get(id=node.db.id)
+    stnode.io_config = "{}"
+    stnode.save()
+
+    # Check that reinit occurs
+    assert node.reinit(stnode)
+
+    # Now the queue is empty
+    assert queue.qsize == 0
+
+
+def test_bad_ioconfig(simplenode, queue):
     """io_config not resolving to a dict is an error."""
     simplenode.io_config = "true"
 
     with pytest.raises(ValueError):
-        UpdateableNode(None, simplenode)
+        UpdateableNode(queue, simplenode)
 
     # But this is fine
     simplenode.io_config = "{}"
-    UpdateableNode(None, simplenode)
+    UpdateableNode(queue, simplenode)
 
 
 def test_idle(unode, queue):
diff --git a/tests/io/test_defaultgroup.py b/tests/io/test_defaultgroup.py
index 3ee2f20e9..6070989b1 100644
--- a/tests/io/test_defaultgroup.py
+++ b/tests/io/test_defaultgroup.py
@@ -30,8 +30,8 @@ def test_too_many_nodes(storagegroup, storagenode, queue):
     """Test DefaultGroupIO rejecting more than one node."""
 
     stgroup = storagegroup(name="group")
-    node1 = UpdateableNode(None, storagenode(name="node1", group=stgroup, active=True))
-    node2 = UpdateableNode(None, storagenode(name="node2", group=stgroup, active=True))
+    node1 = UpdateableNode(queue, storagenode(name="node1", group=stgroup, active=True))
+    node2 = UpdateableNode(queue, storagenode(name="node2", group=stgroup, active=True))
 
     group = UpdateableGroup(queue=queue, group=stgroup, nodes=[node1, node2], idle=True)
     assert group._nodes is None
@@ -41,7 +41,7 @@ def test_just_enough_nodes(storagegroup, storagenode, queue):
     """Test DefaultGroupIO accepting one node."""
 
     stgroup = storagegroup(name="group")
-    node = UpdateableNode(None, storagenode(name="node1", group=stgroup, active=True))
+    node = UpdateableNode(queue, storagenode(name="node1", group=stgroup, active=True))
 
     group = UpdateableGroup(queue=queue, group=stgroup, nodes=[node], idle=True)
     assert group._nodes == [node]
diff --git a/tests/io/test_ioutil_crd.py b/tests/io/test_ioutil_crd.py
index 9df16c0ae..439a50b68 100644
--- a/tests/io/test_ioutil_crd.py
+++ b/tests/io/test_ioutil_crd.py
@@ -23,6 +23,7 @@ def mock_post_add():
 
 @pytest.fixture
 def db_setup(
+    queue,
     mock_filesize,
     mock_post_add,
     storagegroup,
@@ -40,7 +41,7 @@ def db_setup(
     )
 
     return (
-        UpdateableNode(None, node_to).io,
+        UpdateableNode(queue, node_to).io,
         archivefilecopy(file=simplefile, node=node_from, has_file="Y"),
         archivefilecopyrequest(file=simplefile, node_from=node_from, group_to=group_to),
         time.time() - 2,  # start_time
diff --git a/tests/io/test_lustrehsmgroup.py b/tests/io/test_lustrehsmgroup.py
index b60e9b995..28aa94b99 100644
--- a/tests/io/test_lustrehsmgroup.py
+++ b/tests/io/test_lustrehsmgroup.py
@@ -70,7 +70,7 @@ def test_init(storagegroup, storagenode, queue, mock_lfs):
     stgroup = storagegroup(name="group", io_class="LustreHSM")
 
     hsm = UpdateableNode(
-        None,
+        queue,
         storagenode(
             name="hsm",
             group=stgroup,
@@ -79,7 +79,7 @@ def test_init(storagegroup, storagenode, queue, mock_lfs):
         ),
     )
     smallfile = UpdateableNode(
-        None, storagenode(name="smallfile", group=stgroup, io_class="Default")
+        queue, storagenode(name="smallfile", group=stgroup, io_class="Default")
     )
 
     # These should work
diff --git a/tests/io/test_lustrehsmnode.py b/tests/io/test_lustrehsmnode.py
index d519e8218..f2e181821 100644
--- a/tests/io/test_lustrehsmnode.py
+++ b/tests/io/test_lustrehsmnode.py
@@ -49,17 +49,17 @@ def node(
     return UpdateableNode(queue, simplenode)
 
 
-def test_init_no_headroom(have_lfs, simplenode):
+def test_init_no_headroom(have_lfs, queue, simplenode):
     """No headroom is an error"""
 
     simplenode.io_class = "LustreHSM"
     simplenode.io_config = '{"quota_id": "qid", "quota_type": "group"}'
 
     with pytest.raises(KeyError):
-        UpdateableNode(None, simplenode)
+        UpdateableNode(queue, simplenode)
 
 
-def test_init_bad_release_count(simplenode, have_lfs):
+def test_init_bad_release_count(simplenode, queue, have_lfs):
     """Check for bad release_check_count."""
 
     simplenode.io_class = "LustreHSM"
@@ -69,10 +69,10 @@ def test_init_bad_release_count(simplenode, have_lfs):
     )
 
     with pytest.raises(ValueError):
-        UpdateableNode(None, simplenode)
+        UpdateableNode(queue, simplenode)
 
 
-def test_init_bad_restore_wait(simplenode, have_lfs):
+def test_init_bad_restore_wait(simplenode, queue, have_lfs):
     """Check for bad restore_wait."""
 
     simplenode.io_class = "LustreHSM"
@@ -82,7 +82,7 @@ def test_init_bad_restore_wait(simplenode, have_lfs):
     )
 
     with pytest.raises(ValueError):
-        UpdateableNode(None, simplenode)
+        UpdateableNode(queue, simplenode)
 
 
 def test_release_files_okay(queue, node):
diff --git a/tests/io/test_lustrequota.py b/tests/io/test_lustrequota.py
index 0f683f650..0f2c93a5e 100644
--- a/tests/io/test_lustrequota.py
+++ b/tests/io/test_lustrequota.py
@@ -7,31 +7,31 @@
 
 
 @pytest.fixture
-def node(simplenode, mock_lfs):
+def node(simplenode, queue, mock_lfs):
     """A LustreQuota node for testing"""
     simplenode.io_class = "LustreQuota"
     simplenode.io_config = '{"quota_id": "qid", "quota_type": "group"}'
 
-    return UpdateableNode(None, simplenode)
+    return UpdateableNode(queue, simplenode)
 
 
-def test_no_quota_id(simplenode):
+def test_no_quota_id(simplenode, queue):
     """Test instantiating I/O without specifying quota_id."""
 
     simplenode.io_class = "LustreQuota"
 
     with pytest.raises(KeyError):
-        UpdateableNode(None, simplenode)
+        UpdateableNode(queue, simplenode)
 
 
-def test_no_lfs(simplenode):
+def test_no_lfs(simplenode, queue):
     """Test crashing if lfs(1) can't be found."""
 
     simplenode.io_class = "LustreQuota"
     simplenode.io_config = '{"quota_id": "qid", "quota_type": "group"}'
 
     with pytest.raises(RuntimeError):
-        UpdateableNode(None, simplenode)
+        UpdateableNode(queue, simplenode)
 
 
 def test_quota_id(node):
@@ -48,34 +48,34 @@ def test_bytes_avail(node):
 
 
 @pytest.mark.lfs_dont_mock("quota_remaining")
-def test_quota_type(mock_lfs, simplenode, mock_run_command):
+def test_quota_type(mock_lfs, queue, simplenode, mock_run_command):
     """Test quota_type handling."""
 
     simplenode.io_class = "LustreQuota"
 
     # User quota
     simplenode.io_config = '{"quota_id": "qid", "quota_type": "user"}'
-    unode = UpdateableNode(None, simplenode)
+    unode = UpdateableNode(queue, simplenode)
     unode.io.bytes_avail()
     assert "-u qid" in " ".join(mock_run_command()["cmd"])
 
     # Group quota
     simplenode.io_config = '{"quota_id": "qid", "quota_type": "group"}'
-    unode = UpdateableNode(None, simplenode)
+    unode = UpdateableNode(queue, simplenode)
     unode.io.bytes_avail()
     assert "-g qid" in " ".join(mock_run_command()["cmd"])
 
     # Project quota
     simplenode.io_config = '{"quota_id": "qid", "quota_type": "project"}'
-    unode = UpdateableNode(None, simplenode)
+    unode = UpdateableNode(queue, simplenode)
     unode.io.bytes_avail()
     assert "-p qid" in " ".join(mock_run_command()["cmd"])
 
 
-def test_bad_quota_type(mock_lfs, simplenode):
+def test_bad_quota_type(mock_lfs, queue, simplenode):
     """Test bad quota_type handling."""
 
     simplenode.io_class = "LustreQuota"
     simplenode.io_config = '{"quota_id": "qid", "quota_type": "pool"}'
     with pytest.raises(ValueError):
-        UpdateableNode(None, simplenode)
+        UpdateableNode(queue, simplenode)
diff --git a/tests/io/test_transport.py b/tests/io/test_transport.py
index 04c3d8932..d43c0f34d 100644
--- a/tests/io/test_transport.py
+++ b/tests/io/test_transport.py
@@ -123,7 +123,7 @@ def test_group_init(transport_fleet_no_init, queue):
 
     group = UpdateableGroup(queue=queue, group=stgroup, nodes=nodes, idle=True)
     assert group._nodes == nodes
-    assert group.io.fifo == "g:group"
+    assert group.io.fifo is not None
 
 
 def test_group_init_bad(transport_fleet_no_init, queue):
diff --git a/tests/scheduler/test_queue.py b/tests/scheduler/test_queue.py
index da9dde4bb..aad1cd02c 100644
--- a/tests/scheduler/test_queue.py
+++ b/tests/scheduler/test_queue.py
@@ -2,6 +2,7 @@
 
 import threading
 from time import time
+from unittest.mock import MagicMock, patch
 
 import pytest
 
@@ -11,12 +12,18 @@ def clean_queue(queue):
     """A clean queue init and teardown"""
     yield queue
 
-    # Clean the dirty queue
+    # Check that the queue is clean by pulling everything out
+    count = 0
+    while queue.qsize > 0:
+        item, key = queue.get()
+        queue.task_done(key)
+        count += 1
+
     queue.join()
 
     # Check for clean shutdown
-    assert queue.qsize == 0
-    assert queue.inprogress_size == 0
+    assert count == 0, "Queue not clean during teardown!"
+    assert queue.inprogress_size == 0, "Task(s) left in-progress during teardown!"
 
 
 def test_emptyget(clean_queue):
@@ -193,3 +200,133 @@ def test_exclusive(clean_queue):
     # Everything's taken care of
     assert clean_queue.qsize == 0
     assert clean_queue.inprogress_size == 0
+
+
+def test_label(clean_queue):
+    """Test label_fifo()"""
+
+    clean_queue.label_fifo(key="fifo", label="label")
+
+    # Check that the label is used
+    mock = MagicMock()
+    with patch("alpenhorn.common.metrics.Metric.add", mock):
+        clean_queue.put(None, "fifo")
+        clean_queue.get()
+        clean_queue.task_done("fifo")
+
+        # Also try an unlabelled fifo
+        clean_queue.put(None, "fifo2")
+        clean_queue.get()
+        clean_queue.task_done("fifo2")
+
+    mock.assert_any_call(1, fifo="label", status="queued")
+    mock.assert_any_call(-1, fifo="label", status="queued")
+    mock.assert_any_call(1, fifo="fifo2", status="queued")
+    mock.assert_any_call(-1, fifo="fifo2", status="queued")
+
+
+def test_clear_missing(clean_queue):
+    """Clearing a non-existent FIFO should work."""
+
+    assert clean_queue.clear_fifo("MISSING") == (0, 0)
+
+
+def test_clear_nokeep(clean_queue):
+    """Test put after clearing a FIFO without keep_clear."""
+
+    assert clean_queue.clear_fifo("fifo") == (0, 0)
+
+    clean_queue.put(None, "fifo")
+    assert clean_queue.qsize == 1
+
+    # Clean up
+    item, key = clean_queue.get()
+    clean_queue.task_done(key)
+
+
+def test_clear_keep(clean_queue):
+    """Test keeping a FIFO clear."""
+
+    assert clean_queue.clear_fifo("fifo", keep_clear=True) == (0, 0)
+
+    with pytest.raises(KeyError):
+        clean_queue.put(None, "fifo")
+
+    # Queue empty
+    assert clean_queue.qsize == 0
+
+    # Try a deferred put
+    with pytest.raises(KeyError):
+        clean_queue.put(None, "fifo", wait=1)
+
+    # Queue still empty
+    assert clean_queue.qsize == 0
+
+
+def test_clear_pending(clean_queue):
+    """Test clearing pending tasks from a FIFO."""
+
+    # Queue some stuff, in several FIFOs
+    clean_queue.put(1, "fifo")
+    clean_queue.put(2, "fifo")
+    clean_queue.put(3, "fifo")
+    clean_queue.put(4, "fifo2")
+    clean_queue.put(5, "fifo2")
+    clean_queue.put(6, "fifo2")
+    assert clean_queue.qsize == 6
+    assert clean_queue.fifo_size("fifo") == 3
+
+    assert clean_queue.clear_fifo("fifo") == (3, 0)
+
+    assert clean_queue.qsize == 3
+    assert clean_queue.fifo_size("fifo") == 0
+
+    # Remove the rest
+    item, key = clean_queue.get()
+    clean_queue.task_done(key)
+    item, key = clean_queue.get()
+    clean_queue.task_done(key)
+    item, key = clean_queue.get()
+    clean_queue.task_done(key)
+
+
+def test_clear_inprogress(clean_queue):
+    """Test clearing a FIFO with something in-progress."""
+
+    # Queue some stuff, in several FIFOs
+    clean_queue.put(1, "fifo")
+    clean_queue.put(2, "fifo")
+    assert clean_queue.qsize == 2
+    assert clean_queue.inprogress_size == 0
+
+    # Retrieve the first thing
+    clean_queue.get()
+    assert clean_queue.qsize == 1
+    assert clean_queue.inprogress_size == 1
+
+    assert clean_queue.clear_fifo("fifo") == (1, 0)
+
+    assert clean_queue.qsize == 0
+    assert clean_queue.inprogress_size == 1
+
+    # Complete the task
+    clean_queue.task_done("fifo")
+
+    assert clean_queue.qsize == 0
+    assert clean_queue.inprogress_size == 0
+
+
+def test_clear_deferred(clean_queue):
+    """Test clearing deferred tasks from a FIFO."""
+
+    # Queue some stuff, in several FIFOs
+    clean_queue.put(1, "fifo", wait=1)
+    clean_queue.put(2, "fifo", wait=2)
+    clean_queue.put(3, "fifo", wait=3)
+    assert clean_queue.qsize == 0
+    assert clean_queue.deferred_size == 3
+
+    assert clean_queue.clear_fifo("fifo") == (0, 3)
+
+    assert clean_queue.qsize == 0
+    assert clean_queue.deferred_size == 0
diff --git a/tests/scheduler/test_task.py b/tests/scheduler/test_task.py
index d50411033..1059f52b6 100644
--- a/tests/scheduler/test_task.py
+++ b/tests/scheduler/test_task.py
@@ -212,3 +212,54 @@ def test_exclusive_task(queue):
 
     # Everything's taken care of
     assert queue.qsize == 0
+
+
+def test_requeue_closed(queue):
+    """Test requeueing on a closed FIFO"""
+
+    def _task(task):
+        task.requeue()
+
+    # Queue the task
+    Task(_task, queue, "fifo", requeue=True)
+    assert queue.qsize == 1
+
+    # Get the task
+    task, key = queue.get()
+
+    # Clear and close the fifo
+    assert queue.clear_fifo("fifo", keep_clear=True) == (0, 0)
+
+    # Queue is empty
+    assert queue.qsize == 0
+
+    # Run the task, to requeue it
+    task()
+    queue.task_done(key)
+
+    # Queue is empty because requeue didn't happen
+    assert queue.qsize == 0
+
+
+def test_deferred_closed(queue):
+    """Test deferring a task on a closed fifo."""
+
+    def _task(task, queue):
+        # Close the fifo here
+        queue.clear_fifo("fifo", keep_clear=True)
+
+        # Defer
+        yield 1
+
+    Task(_task, queue, "fifo", args=(queue,))
+    assert queue.qsize == 1
+    assert queue.deferred_size == 0
+
+    # Run the task until it defers
+    task, key = queue.get()
+    task()
+    queue.task_done(key)
+
+    # task is neither queued nor deferred
+    assert queue.qsize == 0
+    assert queue.deferred_size == 0