Workflow: queue/cache memory estimates and high-memory confirmation

src/dashpva/gui/theme.qss

@@ -357,6 +357,11 @@ QTreeWidget#treeWidgetConfig[sectionActive="false"] {
     color: palette(mid);
 }
 QLabel#labelProfile, QLabel#labelProfileDescription { color: $INFO; font-weight: bold; }
+QLabel[memoryEstimate="true"] {
+    color: $TEXT_MUTED;
+    font-style: italic;
+    font-size: $FONT_CAPTION;
+}
 QComboBox#comboBoxProfile { border: 1px solid $INFO; border-radius: 3px; padding: 2px 6px; }
 QLineEdit#lineEditProfileDescription { color: $TEXT_SECONDARY; background: transparent; border: none; }
 QPushButton#buttonCollapseAll, QPushButton#buttonExpandAll { font-size: $FONT_BODY; padding: 4px 8px; min-height: 28px; }

src/dashpva/workflow/workflow.py

@@ -72,6 +72,11 @@
 _TREE_ORIGINAL_ROLE = Qt.UserRole        # column 1: the untouched typed value
 _TREE_RENDERED_ROLE = Qt.UserRole + 1    # column 1: the text we rendered
 
+_REFERENCE_PIXELS = 1_000_000
+_REFERENCE_IMAGE_BYTES = _REFERENCE_PIXELS * 2
+_REFERENCE_Q_BYTES = _REFERENCE_PIXELS * 3 * 8
+_MEMORY_CONFIRM_BYTES = 1024 ** 3
+
 
 class _OpaqueValue:
     """Carries a Python value through Qt item data untouched.
@@ -456,6 +461,8 @@ def __init__(self, parent=None):
         self.buttonRunAnalysisConsumer.clicked.connect(self.run_analysis_consumer)
         self.buttonStopAnalysisConsumer.clicked.connect(self.stop_analysis_consumer)
 
+        self._setup_memory_feedback()
+
         # DB init / refresh buttons
         self.buttonInitDb.clicked.connect(self._init_db_and_recheck)
         self.buttonRefreshDb.clicked.connect(self._refresh_db)
@@ -626,6 +633,85 @@ def _on_analysis_processor_file_changed(self, processor_file):
                 )
             self.spinBoxNConsumersAnalysis.setToolTip("")
 
+    @staticmethod
+    def _format_memory(byte_count: int) -> str:
+        value = float(byte_count)
+        for unit in ('B', 'KiB', 'MiB', 'GiB', 'TiB'):
+            if value < 1024 or unit == 'TiB':
+                return f'{value:.1f} {unit}'
+            value /= 1024
+
+    @classmethod
+    def _frame_memory_text(cls, frame_count: int, rich_text: bool = False) -> str:
+        image = cls._format_memory(frame_count * _REFERENCE_IMAGE_BYTES)
+        image_q = cls._format_memory(
+            frame_count * (_REFERENCE_IMAGE_BYTES + _REFERENCE_Q_BYTES)
+        )
+        if rich_text:
+            image = f'<b>{image}</b>'
+            image_q = f'<b>{image_q}</b>'
+        return f'Estimated for 1 MP: images {image}; with HKL {image_q}'
+
+    def _setup_memory_feedback(self) -> None:
+        queue_tip = (
+            'Number of full input updates retained by the upstream PVA server for this '
+            'subscription. Larger values absorb short bursts but do not improve sustained '
+            'processing speed; they increase RAM use and latency.'
+        )
+        cache_tip = (
+            'Number of full incoming frames retained by the collector while ordering producer '
+            'updates. This is separate from the server queue and profile frame cache.'
+        )
+        controls = (
+            (self.spinBoxServerQueueSizeAssociator, self.gridLayout, 7, queue_tip),
+            (self.spinBoxServerQueueSizeCollector, self.gridLayout_3, 9, queue_tip),
+            (self.spinBoxCollectorCacheSize, self.gridLayout_3, 10, cache_tip),
+            (self.spinBoxServerQueueSizeAnalysis, self.gridLayout_4, 7, queue_tip),
+        )
+        self._memory_estimate_labels = []
+        for spin_box, layout, row, tooltip in controls:
+            label = QtWidgets.QLabel()
+            label.setProperty('memoryEstimate', True)
+            label.setToolTip(tooltip)
+            spin_box.setToolTip(tooltip)
+            field = QtWidgets.QWidget()
+            field_layout = QtWidgets.QVBoxLayout(field)
+            field_layout.setContentsMargins(0, 0, 0, 0)
+            field_layout.setSpacing(2)
+            layout.removeWidget(spin_box)
+            field_layout.addWidget(spin_box)
+            field_layout.addWidget(label)
+            layout.addWidget(field, row, 1, 1, 2)
+            spin_box.valueChanged.connect(
+                lambda value, estimate_label=label: estimate_label.setText(
+                    self._frame_memory_text(value, rich_text=True)
+                )
+            )
+            label.setText(self._frame_memory_text(spin_box.value(), rich_text=True))
+            self._memory_estimate_labels.append(label)
+
+    def _confirm_memory_risk(self, title: str, entries: list[tuple[str, int]]) -> bool:
+        risky = [(name, frames) for name, frames in entries
+                 if frames * (_REFERENCE_IMAGE_BYTES + _REFERENCE_Q_BYTES)
+                 >= _MEMORY_CONFIRM_BYTES]
+        if not risky:
+            return True
+        details = '\n'.join(
+            f'• {name}: {frames} frame(s) — {Workflow._frame_memory_text(frames)}'
+            for name, frames in risky
+        )
+        reply = QMessageBox.warning(
+            self,
+            'High Memory Setting',
+            f'{title} may retain a large amount of frame data.\n\n{details}\n\n'
+            'Actual use depends on detector size, data type, compression, metadata, and the '
+            'number of subscriptions. Larger buffers do not improve sustained throughput.\n\n'
+            'Start anyway?',
+            QMessageBox.Yes | QMessageBox.No,
+            QMessageBox.No,
+        )
+        return reply == QMessageBox.Yes
+
     # ------------------------------------------------------------------ #
     # Database controls
     # ------------------------------------------------------------------ #
@@ -1658,6 +1744,7 @@ def _drop(idx):
             w = table.cellWidget(idx, 3)
             if w is not None:
                 w.setEnabled(False)
+            _update_cache_notice()
 
         for r, (kind, key, old, new) in enumerate(rows):
             key_item = QTableWidgetItem(
@@ -1694,6 +1781,44 @@ def _drop(idx):
             table.setCellWidget(r, 3, btn)
 
         layout.addWidget(table)
+
+        cache_paths = {
+            'CACHE_OPTIONS.ALIGNMENT.MAX_CACHE_SIZE',
+            'CACHE_OPTIONS.SCAN.MAX_CACHE_SIZE',
+        }
+        cache_rows = [
+            index for index, (_kind, key, _old, _new) in enumerate(rows)
+            if key in cache_paths
+        ]
+        cache_notice = QtWidgets.QLabel()
+        cache_notice.setWordWrap(True)
+
+        def _update_cache_notice():
+            estimates = []
+            for index in cache_rows:
+                if dropped_flags[index]:
+                    continue
+                cell = table.item(index, 2)
+                try:
+                    frame_count = int(cell.text())
+                except (AttributeError, TypeError, ValueError):
+                    continue
+                estimates.append(
+                    f'{rows[index][1]} = {frame_count}: '
+                    f'{self._frame_memory_text(frame_count)}'
+                )
+            cache_notice.setText(
+                'Frame-cache memory estimate (Save HDF5/Post-scan; not needed for ordinary '
+                'live viewing):\n' + '\n'.join(estimates)
+                if estimates else ''
+            )
+            cache_notice.setVisible(bool(estimates))
+
+        if cache_rows:
+            table.itemChanged.connect(lambda _item: _update_cache_notice())
+            layout.addWidget(cache_notice)
+            _update_cache_notice()
+
         btns = QDialogButtonBox(QDialogButtonBox.Ok | QDialogButtonBox.Cancel)
         btns.accepted.connect(dlg.accept)
         btns.rejected.connect(dlg.reject)
@@ -2528,6 +2653,16 @@ def run_associator_consumers(self):
             self.logger.warning('Start Metadata Associator ignored — already running')
             QtWidgets.QMessageBox.warning(self, 'Warning', 'Associator Consumers are already running.')
             return
+        queue_frames = (
+            self.spinBoxServerQueueSizeAssociator.value()
+            * self.spinBoxNConsumersAssociator.value()
+        )
+        if not Workflow._confirm_memory_risk(
+            self,
+            'Starting the Metadata Associator',
+            [('Server queues across consumers', queue_frames)],
+        ):
+            return
 
         # Reload before building anything below, so a profile switched (or
         # edited via the separate HKL Setup process) since the last reload is
@@ -2896,6 +3031,17 @@ def run_collector(self):
         if 'collector' in self.processes:
             QtWidgets.QMessageBox.warning(self, 'Warning', 'Collector is already running.')
             return
+        producer_count = max(1, int(self.lineEditProducerIdList.text()))
+        if not Workflow._confirm_memory_risk(
+            self,
+            'Starting the Collector',
+            [
+                ('Server queues across producers',
+                 self.spinBoxServerQueueSizeCollector.value() * producer_count),
+                ('Collector cache', self.spinBoxCollectorCacheSize.value()),
+            ],
+        ):
+            return
         self._save_collector_last()
 
         producer_id_list = [str(i) for i in range(1, int(self.lineEditProducerIdList.text()) + 1)]
@@ -2962,6 +3108,16 @@ def run_analysis_consumer(self):
         if 'analysis_consumer' in self.processes:
             QtWidgets.QMessageBox.warning(self, 'Warning', 'Analysis Consumer is already running.')
             return
+        queue_frames = (
+            self.spinBoxServerQueueSizeAnalysis.value()
+            * self.spinBoxNConsumersAnalysis.value()
+        )
+        if not Workflow._confirm_memory_risk(
+            self,
+            'Starting the Analysis Consumer',
+            [('Server queues across consumers', queue_frames)],
+        ):
+            return
         self._on_analysis_processor_file_changed(
             self.comboBoxProcessorFileAnalysis.currentText()
         )

tests/unit/test_workflow_memory_estimates.py

@@ -0,0 +1,61 @@
+# Copyright © 2026, UChicago Argonne, LLC
+# All Rights Reserved
+# Software Name: DashPVA
+# By: Argonne National Laboratory
+#
+# BSD OPEN SOURCE LICENSE
+#
+# Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met:
+#
+# 1. Redistributions of source code must retain the above copyright notice, this list of conditions and the following disclaimer.
+# 2. Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the following disclaimer in the documentation and/or other materials provided with the distribution.
+# 3. Neither the name of the copyright holder nor the names of its contributors may be used to endorse or promote products derived from this software without specific prior written permission.
+#
+# ******************************************************************************************************
+# DISCLAIMER
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+# ******************************************************************************************************
+
+
+"""Workflow queue/cache memory estimates and the high-memory confirmation."""
+
+from dashpva.workflow import workflow
+from dashpva.workflow.workflow import Workflow
+
+
+def test_frame_memory_text_scales_with_frame_count():
+    one = Workflow._frame_memory_text(1)
+    assert one.startswith("Estimated for 1 MP: images")
+    assert "with HKL" in one
+    assert "1.9 MiB" in one
+    assert "24.8 MiB" in Workflow._frame_memory_text(1)
+    assert "GiB" in Workflow._frame_memory_text(1000)
+
+
+def test_frame_memory_rich_text_bolds_sizes():
+    text = Workflow._frame_memory_text(1, rich_text=True)
+    assert "images <b>1.9 MiB</b>" in text
+    assert "with HKL <b>24.8 MiB</b>" in text
+
+
+def test_small_buffers_start_without_asking(monkeypatch):
+    asked = []
+    monkeypatch.setattr(workflow.QMessageBox, "warning", lambda *a, **k: asked.append(a))
+    assert Workflow._confirm_memory_risk(None, "Start", [("Server queue", 10)])
+    assert asked == []
+
+
+def test_large_buffers_ask_and_respect_no(monkeypatch):
+    asked = []
+
+    def _warning(*args, **_kwargs):
+        asked.append((args[1], args[2]))
+        return workflow.QMessageBox.No
+
+    monkeypatch.setattr(workflow.QMessageBox, "warning", _warning)
+    assert not Workflow._confirm_memory_risk(None, "Starting the Analysis Consumer",
+                                             [("Server queue", 1000)])
+    title, text = asked[0]
+    assert title == "High Memory Setting"
+    assert "1000 frame(s)" in text