Allow for clearing a plugin's processing queue on-demand

ADApp/Db/NDPluginBase.template

@@ -261,6 +261,13 @@ record(longin, "$(P)$(R)QueueFree")
     field(SCAN, "$(SCANRATE=I/O Intr)")
 }
 
+record(bo, "$(P)$(R)ClearQueue")
+{
+    field(DTYP, "asynInt32")
+    field(OUT,  "@asyn($(PORT),$(ADDR=0),$(TIMEOUT=1))CLEAR_QUEUE")
+    field(VAL,  "1")
+}
+
 # Display the fill level on the plugins input queue
 record(calc, "$(P)$(R)QueueUse") {
     field(CALC, "A-B")

ADApp/pluginSrc/NDPluginDriver.cpp

@@ -128,6 +128,7 @@ NDPluginDriver::NDPluginDriver(const char *portName, int queueSize, int blocking
     createParam(NDPluginDriverDroppedArraysString,     asynParamInt32, &NDPluginDriverDroppedArrays);
     createParam(NDPluginDriverQueueSizeString,         asynParamInt32, &NDPluginDriverQueueSize);
     createParam(NDPluginDriverQueueFreeString,         asynParamInt32, &NDPluginDriverQueueFree);
+    createParam(NDPluginDriverClearQueueString,        asynParamInt32, &NDPluginDriverClearQueue);
     createParam(NDPluginDriverMaxThreadsString,        asynParamInt32, &NDPluginDriverMaxThreads);
     createParam(NDPluginDriverNumThreadsString,        asynParamInt32, &NDPluginDriverNumThreads);
     createParam(NDPluginDriverSortModeString,          asynParamInt32, &NDPluginDriverSortMode);
@@ -155,6 +156,7 @@ NDPluginDriver::NDPluginDriver(const char *portName, int queueSize, int blocking
     setIntegerParam(NDPluginDriverDroppedOutputArrays, 0);
     setIntegerParam(NDPluginDriverQueueSize, queueSize);
     setIntegerParam(NDPluginDriverQueueFree, queueSize);
+    setIntegerParam(NDPluginDriverClearQueue, 0);
     setIntegerParam(NDPluginDriverMaxThreads, maxThreads);
     setIntegerParam(NDPluginDriverNumThreads, 1);
     setIntegerParam(NDPluginDriverBlockingCallbacks, blockingCallbacks);
@@ -732,6 +734,10 @@ asynStatus NDPluginDriver::writeInt32(asynUser *pasynUser, epicsInt32 value)
         if ((status = deleteCallbackThreads())) goto done;
         if ((status = createCallbackThreads())) goto done;
 
+    } else if ((function == NDPluginDriverClearQueue) && value) {
+        clearQueue();
+        setIntegerParam(addr, NDPluginDriverClearQueue, 0);
+
     } else if ((function == NDPluginDriverSortMode) &&
                (value == 1)) {
         status = createSortingThread();
@@ -1010,6 +1016,39 @@ asynStatus NDPluginDriver::createCallbackThreads()
     return (asynStatus) status;
 }
 
+/** Releases all NDArrays waiting in the input queue without processing them.
+  * Arrays already being processed by a plugin thread are not affected.
+  * Must be called with the lock held. */
+void NDPluginDriver::clearQueue()
+{
+    ToThreadMessage_t msg;
+    int droppedArrays, queueSize;
+    int numCleared = 0;
+    static const char *functionName = "clearQueue";
+
+    if (!pToThreadMsgQ_) return;
+
+    while (pToThreadMsgQ_->tryReceive(&msg, sizeof(msg)) == sizeof(msg)) {
+        if (msg.messageType != ToThreadMessageData) {
+            // Put back the exit message from deleteCallbackThreads so a plugin thread still receives it
+            pToThreadMsgQ_->trySend(&msg, sizeof(msg));
+            break;
+        }
+        msg.pArray->pDriver->decrementQueuedArrayCount();
+        msg.pArray->release();
+        numCleared++;
+    }
+
+    getIntegerParam(NDPluginDriverDroppedArrays, &droppedArrays);
+    setIntegerParam(NDPluginDriverDroppedArrays, droppedArrays + numCleared);
+    getIntegerParam(NDPluginDriverQueueSize, &queueSize);
+    setIntegerParam(NDPluginDriverQueueFree, queueSize - pToThreadMsgQ_->pending());
+
+    asynPrint(pasynUserSelf, ASYN_TRACE_FLOW,
+        "%s::%s dropped %d arrays from input queue\n",
+        driverName, functionName, numCleared);
+}
+
 /** Deletes the plugin threads.
   * This method is called from the destructor and whenever QueueSize or NumThreads is changed. */
 asynStatus NDPluginDriver::deleteCallbackThreads()

ADApp/pluginSrc/NDPluginDriver.h

@@ -38,6 +38,7 @@ class sortedListElement {
 #define NDPluginDriverDroppedArraysString       "DROPPED_ARRAYS"        /**< (asynInt32,    r/w) Number of dropped input arrays */
 #define NDPluginDriverQueueSizeString           "QUEUE_SIZE"            /**< (asynInt32,    r/w) Total queue elements */
 #define NDPluginDriverQueueFreeString           "QUEUE_FREE"            /**< (asynInt32,    r/w) Free queue elements */
+#define NDPluginDriverClearQueueString          "CLEAR_QUEUE"           /**< (asynInt32,    r/w) Drop all arrays in the input queue */
 #define NDPluginDriverMaxThreadsString          "MAX_THREADS"           /**< (asynInt32,    r/w) Maximum number of threads */
 #define NDPluginDriverNumThreadsString          "NUM_THREADS"           /**< (asynInt32,    r/w) Number of threads */
 #define NDPluginDriverSortModeString            "SORT_MODE"             /**< (asynInt32,    r/w) sorted callback mode */
@@ -92,6 +93,7 @@ class NDPLUGIN_API NDPluginDriver : public asynNDArrayDriver, public epicsThread
     int NDPluginDriverDroppedArrays;
     int NDPluginDriverQueueSize;
     int NDPluginDriverQueueFree;
+    int NDPluginDriverClearQueue;
     int NDPluginDriverMaxThreads;
     int NDPluginDriverNumThreads;
     int NDPluginDriverSortMode;
@@ -116,6 +118,7 @@ class NDPLUGIN_API NDPluginDriver : public asynNDArrayDriver, public epicsThread
     asynStatus startCallbackThreads();
     asynStatus deleteCallbackThreads();
     asynStatus createSortingThread();
+    void clearQueue();
 
     /* The asyn interfaces we access as a client */
     void *asynGenericPointerInterruptPvt_;

ADApp/pluginTests/Makefile

@@ -61,6 +61,7 @@ ifeq ($(WITH_BOOST),YES)
   plugin-test_SRCS += test_NDPluginROI.cpp
   plugin-test_SRCS += test_NDPluginOverlay.cpp
   plugin-test_SRCS += test_NDArrayPool.cpp
+  plugin-test_SRCS += test_NDPluginDriver.cpp
 
   # Add tests for new plugins like this:
   #plugin-test_SRCS += test_<plugin name>.cpp

ADApp/pluginTests/test_NDPluginDriver.cpp

@@ -0,0 +1,227 @@
+/*
+ * test_NDPluginDriver.cpp
+ *
+ * Tests NDPluginDriver input queue handling.
+ *  Created on: 22 Oct 2026
+ *   Author: Jakub Wlodek
+ */
+
+#include <vector>
+
+#include "boost/test/unit_test.hpp"
+
+#include <epicsEvent.h>
+#include <epicsThread.h>
+#include <asynNDArrayDriver.h>
+#include <NDPluginDriver.h>
+
+#include "testingutilities.h"
+#include "AsynPortClientContainer.h"
+
+#define QUEUE_SIZE 10
+
+// Driver that can push NDArrays to its plugins on demand
+class QueueTestDriver : public asynNDArrayDriver
+{
+public:
+    QueueTestDriver(const std::string& port)
+      : asynNDArrayDriver(port.c_str(), 1, 0, 0,
+                          asynGenericPointerMask, asynGenericPointerMask, 0, 1, 0, 0)
+    {
+    }
+
+    void sendArray(int uniqueId)
+    {
+        size_t dims[1] = {16};
+        NDArray *pArray = pNDArrayPool->alloc(1, dims, NDUInt8, 0, NULL);
+        pArray->uniqueId = uniqueId;
+        doCallbacksGenericPointer(pArray, NDArrayData, 0);
+        pArray->release();
+    }
+};
+
+// Plugin that holds its processing thread inside processCallbacks so arrays back up in the queue
+class QueueTestPlugin : public NDPluginDriver, public AsynPortClientContainer
+{
+public:
+    QueueTestPlugin(const std::string& port, const std::string& detectorPort)
+      : NDPluginDriver(port.c_str(), QUEUE_SIZE, 0, detectorPort.c_str(), 0, 1, 0, 0,
+                       asynGenericPointerMask, asynGenericPointerMask, 0, 1, 0, 0, 1),
+        AsynPortClientContainer(port),
+        holdProcessing_(true),
+        processingStarted_(epicsEventMustCreate(epicsEventEmpty)),
+        releaseProcessing_(epicsEventMustCreate(epicsEventEmpty))
+    {
+    }
+
+    ~QueueTestPlugin()
+    {
+        cleanup();
+        epicsEventDestroy(processingStarted_);
+        epicsEventDestroy(releaseProcessing_);
+    }
+
+    bool waitForProcessingStarted()
+    {
+        return epicsEventWaitWithTimeout(processingStarted_, 5.0) == epicsEventWaitOK;
+    }
+
+    void releaseProcessing()
+    {
+        lock();
+        holdProcessing_ = false;
+        unlock();
+        epicsEventSignal(releaseProcessing_);
+    }
+
+    std::vector<int> getProcessedIds()
+    {
+        lock();
+        std::vector<int> ids(processedIds_);
+        unlock();
+        return ids;
+    }
+
+protected:
+    void processCallbacks(NDArray *pArray)
+    {
+        processedIds_.push_back(pArray->uniqueId);
+        if (!holdProcessing_) return;
+        unlock();
+        epicsEventSignal(processingStarted_);
+        epicsEventMustWait(releaseProcessing_);
+        lock();
+    }
+
+private:
+    bool holdProcessing_;
+    epicsEventId processingStarted_;
+    epicsEventId releaseProcessing_;
+    std::vector<int> processedIds_;
+};
+
+struct QueueTestFixture
+{
+    QueueTestDriver *driver;
+    QueueTestPlugin *plugin;
+
+    QueueTestFixture()
+    {
+        std::string driverPort("simQueue"), pluginPort("testQueue");
+        uniqueAsynPortName(driverPort);
+        uniqueAsynPortName(pluginPort);
+
+        driver = new QueueTestDriver(driverPort);
+        plugin = new QueueTestPlugin(pluginPort, driverPort);
+        plugin->start();
+        plugin->write(NDPluginDriverEnableCallbacksString, 1);
+        plugin->write(NDPluginDriverArrayPortString, driverPort);
+    }
+
+    ~QueueTestFixture()
+    {
+        plugin->releaseProcessing();
+        delete plugin;
+        delete driver;
+    }
+
+    // Arrays are counted as queued until processCallbacks returns
+    bool waitForQueuedArrayCount(int expected)
+    {
+        for (int i = 0; i < 500; i++) {
+            if (driver->getQueuedArrayCount() == expected) return true;
+            epicsThreadSleep(0.01);
+        }
+        return false;
+    }
+
+    // Leaves array 0 held in processCallbacks and arrays 1..numQueued in the queue
+    void fillQueue(int numQueued)
+    {
+        driver->sendArray(0);
+        BOOST_REQUIRE(plugin->waitForProcessingStarted());
+        for (int i = 1; i <= numQueued; i++) {
+            driver->sendArray(i);
+        }
+        BOOST_REQUIRE_EQUAL(plugin->readInt(NDPluginDriverQueueFreeString), QUEUE_SIZE - numQueued);
+        BOOST_REQUIRE_EQUAL(driver->getQueuedArrayCount(), numQueued + 1);
+    }
+
+    void checkQueueCleared(int expectedDropped)
+    {
+        BOOST_CHECK_EQUAL(plugin->readInt(NDPluginDriverClearQueueString), 0);
+        BOOST_CHECK_EQUAL(plugin->readInt(NDPluginDriverQueueFreeString), QUEUE_SIZE);
+        BOOST_CHECK_EQUAL(plugin->readInt(NDPluginDriverDroppedArraysString), expectedDropped);
+        // Only the array held in processCallbacks remains
+        BOOST_CHECK_EQUAL(driver->getQueuedArrayCount(), 1);
+    }
+
+    void checkAllArraysReturnedToPool()
+    {
+        NDArrayPool *pool = driver->pNDArrayPool;
+        BOOST_CHECK_EQUAL(pool->getNumFree(), pool->getNumBuffers());
+    }
+};
+
+BOOST_FIXTURE_TEST_SUITE(NDPluginDriverQueueTests, QueueTestFixture)
+
+BOOST_AUTO_TEST_CASE(test_clear_queue_while_receiving_arrays)
+{
+    fillQueue(5);
+
+    plugin->write(NDPluginDriverClearQueueString, 1);
+    checkQueueCleared(5);
+
+    // Plugin still accepts new arrays after the queue is cleared
+    driver->sendArray(6);
+    driver->sendArray(7);
+    BOOST_CHECK_EQUAL(plugin->readInt(NDPluginDriverQueueFreeString), QUEUE_SIZE - 2);
+
+    plugin->releaseProcessing();
+    BOOST_REQUIRE(waitForQueuedArrayCount(0));
+
+    std::vector<int> expected;
+    expected.push_back(0);
+    expected.push_back(6);
+    expected.push_back(7);
+    std::vector<int> processed = plugin->getProcessedIds();
+    BOOST_CHECK_EQUAL_COLLECTIONS(processed.begin(), processed.end(), expected.begin(), expected.end());
+    BOOST_CHECK_EQUAL(plugin->readInt(NDPluginDriverDroppedArraysString), 5);
+    BOOST_CHECK_EQUAL(plugin->readInt(NDPluginDriverQueueFreeString), QUEUE_SIZE);
+    checkAllArraysReturnedToPool();
+}
+
+BOOST_AUTO_TEST_CASE(test_clear_queue_with_callbacks_disabled)
+{
+    fillQueue(5);
+
+    plugin->write(NDPluginDriverEnableCallbacksString, 0);
+    plugin->write(NDPluginDriverClearQueueString, 1);
+    checkQueueCleared(5);
+
+    // These never reach the plugin
+    driver->sendArray(6);
+    driver->sendArray(7);
+    BOOST_CHECK_EQUAL(plugin->readInt(NDPluginDriverQueueFreeString), QUEUE_SIZE);
+
+    plugin->releaseProcessing();
+    BOOST_REQUIRE(waitForQueuedArrayCount(0));
+
+    std::vector<int> processed = plugin->getProcessedIds();
+    BOOST_REQUIRE_EQUAL(processed.size(), (size_t)1);
+    BOOST_CHECK_EQUAL(processed[0], 0);
+    BOOST_CHECK_EQUAL(plugin->readInt(NDPluginDriverDroppedArraysString), 5);
+    BOOST_CHECK_EQUAL(plugin->readInt(NDPluginDriverQueueFreeString), QUEUE_SIZE);
+    checkAllArraysReturnedToPool();
+}
+
+BOOST_AUTO_TEST_CASE(test_clear_empty_queue)
+{
+    plugin->write(NDPluginDriverClearQueueString, 1);
+
+    BOOST_CHECK_EQUAL(plugin->readInt(NDPluginDriverClearQueueString), 0);
+    BOOST_CHECK_EQUAL(plugin->readInt(NDPluginDriverQueueFreeString), QUEUE_SIZE);
+    BOOST_CHECK_EQUAL(plugin->readInt(NDPluginDriverDroppedArraysString), 0);
+}
+
+BOOST_AUTO_TEST_SUITE_END()

docs/ADCore/NDPluginDriver.rst

@@ -83,6 +83,14 @@ following table.
     - QUEUE_FREE
     - $(P)$(R)QueueFree
     - longin
+  * - asynInt32
+    - r/w
+    - Writing 1 to this record releases all NDArrays waiting in the input queue without
+      processing them. Arrays currently being processed are not affected. The dropped
+      arrays are added to DroppedArrays.
+    - CLEAR_QUEUE
+    - $(P)$(R)ClearQueue
+    - bo
   * - N/A
     - r/o
     - The number of used queue elements.