Merge ~info-martin-konrad/epics-base:callbackQueueStatus into ~epics-core/epics-base/+git/epics-base:3.16

[Martin Konrad]

Allow scanOnce and callback queue utilization to be monitored from the IOC shell and iocStats. Fixes lp:1786540

[Ralph Lange (ralph-lange)]

I like the idea of adding queue health status.

Number of overruns is good and useful data.

Given the common usage pattern of those queues (a large number of entries is put on the queue by a driver or at IOC init for PINI processing, then a number of threads are working on taking items off) I think that an iocStats or manual check of the current usage does not tell much as it most probably misses the moment of high usage.
A simple high water mark mechanism might generate data that is a lot more useful.

I would also consider resetting the stats at every read of their values for easy one-record reporting by iocStats. (Cumulative values are not nice in the archiver.)

[Andrew Johnson (anj)]

Core Group review at ESS: Please implement Ralph's high water mark (which needs implementing inside the queueing code) and use epicsAtomic types to protect access to the usage data.

review: Needs Fixing

[Martin Konrad (info-martin-konrad)]

I have implemented the high-water mark Ralph suggested. This is in fact very useful on small IOCs which are booting fast. The one I'm mainly testing with is seeing very heavy load after IOC start and I can watch the queue usage drop slowly in the first minutes after IOC boot (so in some cases the PVs showing the current queue usage also make sense).

I'm now also leveraging epicsAtomic to make my code thread safe.

I understand Ralph's concerns about cumulative values. However, I'm a little hesitant to reset the stats at every read since we offer read out from both the IOC shell and via iocStats. If a read by iocStats resets the number of queue overruns once a second chances are I'll never see anything on the IOC shell...

[mdavidsaver (mdavidsaver)]

Looking at this again, I find myself wondering about adding so many new public functions. 4 query functions and 1 printer, repeated 3 times. This in addition to the ring buffer feature addition. I'd rather see the 4x query functions combined into one which accepts a struct. Something like:

typedef struct QueueStats {
    size_t size;
    size_t numUsed[NUM_CALLBACK_PRIORITIES];
    size_t maxUsed[NUM_CALLBACK_PRIORITIES];
    size_t numOverflow[NUM_CALLBACK_PRIORITIES];
} QueueStats;

void queryQueue(int reset, QueueStates* result);

I think this would be a better match for the (only) prospective outside user, devIocStats.

The printer functions are accessible through iocsh, so it doesn't seem like they need to be public.

Also, I like the idea of combining fetch with (conditional) reset to zero. I've seen and used this idea in the past, and found it a good fit for how I ended up access these stats.

I'm not going to push for this too much if there is a desire to see this in 3.16.2.

At minimum though, the idea of const-ness should apply to the C API as well as C++. eg. epicsRingPointerGetHighWaterMark() and epicsRingBytesHighWaterMark() should take a pointer to const.

review: Needs Fixing

[Martin Konrad (info-martin-konrad)]

Thanks for the feedback. I reworked this to include Michael's comments. I have also updated my IOCStats PR. You might want to look at my code again - this ended up to be quite a few changes.

[Andrew Johnson (anj)]

Hi Martin, if you committed any changes I think you forgot to push them to launchpad, this merge request still only shows your original code. There may still be a chance to get it into the final 3.16.2 release.

[Martin Konrad (info-martin-konrad)]

Andrew, I double checked, this is my latest code. It has the const-ness change, the reset feature and it has been modified to pull all values at the same time. Note that I have squashed my commits together to avoid cluttering the history.

[Andrew Johnson (anj)]

@mdavidsaver Please re-review this merge, should it go into 3.16.2?

[Andrew Johnson (anj)]

Some older C compilers (MSVC, older GCC) will fail while building this code, there are several places where it declares variables after the first statement in a block. It's fine to do that in C++ code, but still not in our C code yet.

I don't understand the use of epicsAtomic functions for accessing callbackQueueSize. The queue size must be set before callbackIsInit and can't be changed after that (unless we go through callbackCleanup() but that clears callbackIsInit anyway and if callbackQueueStatus() is called then it returns an error). Did I miss something?

review: Needs Fixing

[Martin Konrad (info-martin-konrad)]

I tweaked the code to make pre-C99 compilers happy.

I have also removed the unneeded epicsAtomic from the callbackQueueSize variable.

Let me know if you want me to squash the commits.

[Andrew Johnson (anj)]

Thanks for those changes. Once you have created a Merge Request I would ask you to *never* squash commits, as doing that prevents us from seeing what additional changes you have made since we last looked at the proposal (hence my confusion on 2018-10-30). If you really want to hide the evidence of your having taken a U-turn from the final history, resubmitting it as a new Merge Request at the end of the review (or after a major rewrite) is probably the best way to handle that.

However I'm still unsure about accepting this; see my code-specific comments below. Further review from the other core developers is still welcome.

I also think these changes deserve some test code; we have test programs in src/libCom/test for both the bytes and pointer versions of the ring-buffer that could be extended, although the bytes version is single-threaded and just tests basic functionality — no inter-thread issues or tests for a spin-locked ring buffer.

[Martin Konrad (info-martin-konrad)]

I'm glad you asked me to write tests - that helped to I catch an issue in the code...

Do we need multi-threaded tests for this feature?

[Martin Konrad (info-martin-konrad)]

After our discussion I switched from atomics to using the existing spin lock. This should make the code a little bit easier to understand.

[Andrew Johnson (anj)]

Merging this, but one change I've made is to rename your QueuePrintStatus() routines and the QueueStatus iocsh commands (and their associated routines and data structures) to QueueShow for consistency with other subsystems. We normally use a Show suffix for routines that print stuff to stdout, and Status for querying whether a subsystem is happy, so your QueueStatus() API routines are fine. This is also necessary to maintain command-line compatibility with the VxWorks shell.

review: Approve

diff --git a/src/ioc/db/callback.c b/src/ioc/db/callback.c
index ae07414..1b6c249 100644
--- a/src/ioc/db/callback.c
+++ b/src/ioc/db/callback.c
@@ -54,6 +54,7 @@ typedef struct cbQueueSet {
     epicsEventId semWakeUp;
     epicsRingPointerId queue;
     int queueOverflow;
+    int queueOverflows;
     int shutdown;
     int threadsConfigured;
     int threadsRunning;
@@ -103,6 +104,51 @@ int callbackSetQueueSize(int size)
     return 0;
 }
 
+int callbackQueueStatus(const int reset, callbackQueueStats *result)
+{
+    int ret;
+    if (!callbackIsInit) return -1;
+    if (result) {
+        int prio;
+        result->size = callbackQueueSize;
+        for(prio = 0; prio < NUM_CALLBACK_PRIORITIES; prio++) {
+            epicsRingPointerId qId = callbackQueue[prio].queue;
+            result->numUsed[prio] = epicsRingPointerGetUsed(qId);
+            result->maxUsed[prio] = epicsRingPointerGetHighWaterMark(qId);
+            result->numOverflow[prio] = epicsAtomicGetIntT(&callbackQueue[prio].queueOverflows);
+        }
+        ret = 0;
+    } else {
+        ret = -2;
+    }
+    if (reset) {
+        int prio;
+        for(prio = 0; prio < NUM_CALLBACK_PRIORITIES; prio++) {
+            epicsRingPointerResetHighWaterMark(callbackQueue[prio].queue);
+        }
+    }
+    return ret;
+}
+
+void callbackQueuePrintStatus(const int reset)
+{
+    callbackQueueStats stats;
+    if (callbackQueueStatus(reset, &stats) == -1) {
+        fprintf(stderr, "Callback system not initialized, yet. Please run "
+            "iocInit before using this command.\n");
+    } else {
+        int prio;
+        printf("PRIORITY  HIGH-WATER MARK  ITEMS IN Q  Q SIZE  %% USED  Q OVERFLOWS\n");
+        for (prio = 0; prio < NUM_CALLBACK_PRIORITIES; prio++) {
+            double qusage = 100.0 * stats.numUsed[prio] / stats.size;
+            printf("%8s  %15d  %10d  %6d  %6.1f  %11d\n",
+                   threadNamePrefix[prio], stats.maxUsed[prio],
+                   stats.numUsed[prio], stats.size, qusage,
+                   stats.numOverflow[prio]);
+        }
+    }
+}
+
 int callbackParallelThreads(int count, const char *prio)
 {
     if (callbackIsInit) {
@@ -290,6 +336,7 @@ int callbackRequest(CALLBACK *pcallback)
     if (!pushOK) {
         epicsInterruptContextMessage(fullMessage[priority]);
         mySet->queueOverflow = TRUE;
+        epicsAtomicIncrIntT(&mySet->queueOverflows);
         return S_db_bufFull;
     }
     epicsEventSignal(mySet->semWakeUp);
diff --git a/src/ioc/db/callback.h b/src/ioc/db/callback.h
index fa626d1..dd13cdb 100644
--- a/src/ioc/db/callback.h
+++ b/src/ioc/db/callback.h
@@ -48,6 +48,13 @@ typedef epicsCallback CALLBACK;
 
 typedef void    (*CALLBACKFUNC)(struct callbackPvt*);
 
+typedef struct callbackQueueStats {
+    int size;
+    int numUsed[NUM_CALLBACK_PRIORITIES];
+    int maxUsed[NUM_CALLBACK_PRIORITIES];
+    int numOverflow[NUM_CALLBACK_PRIORITIES];
+} callbackQueueStats;
+
 #define callbackSetCallback(PFUN, PCALLBACK) \
     ( (PCALLBACK)->callback = (PFUN) )
 #define callbackSetPriority(PRIORITY, PCALLBACK) \
@@ -73,6 +80,8 @@ epicsShareFunc void callbackCancelDelayed(CALLBACK *pcallback);
 epicsShareFunc void callbackRequestProcessCallbackDelayed(
     CALLBACK *pCallback, int Priority, void *pRec, double seconds);
 epicsShareFunc int callbackSetQueueSize(int size);
+epicsShareFunc int callbackQueueStatus(const int reset, callbackQueueStats *result);
+void callbackQueuePrintStatus(const int reset);
 epicsShareFunc int callbackParallelThreads(int count, const char *prio);
 
 #ifdef __cplusplus
diff --git a/src/ioc/db/dbIocRegister.c b/src/ioc/db/dbIocRegister.c
index 4d0b88c..cecf756 100644
--- a/src/ioc/db/dbIocRegister.c
+++ b/src/ioc/db/dbIocRegister.c
@@ -296,6 +296,17 @@ static void scanOnceSetQueueSizeCallFunc(const iocshArgBuf *args)
     scanOnceSetQueueSize(args[0].ival);
 }
 
+/* scanOnceQueueStatus */
+static const iocshArg scanOnceQueueStatusArg0 = { "reset",iocshArgInt};
+static const iocshArg * const scanOnceQueueStatusArgs[1] =
+    {&scanOnceQueueStatusArg0};
+static const iocshFuncDef scanOnceQueueStatusFuncDef =
+    {"scanOnceQueueStatus",1,scanOnceQueueStatusArgs};
+static void scanOnceQueueStatusCallFunc(const iocshArgBuf *args)
+{
+    scanOnceQueuePrintStatus(args[0].ival);
+}
+
 /* scanppl */
 static const iocshArg scanpplArg0 = { "rate",iocshArgDouble};
 static const iocshArg * const scanpplArgs[1] = {&scanpplArg0};
@@ -335,6 +346,17 @@ static void callbackSetQueueSizeCallFunc(const iocshArgBuf *args)
     callbackSetQueueSize(args[0].ival);
 }
 
+/* callbackQueueStatus */
+static const iocshArg callbackQueueStatusArg0 = { "reset", iocshArgInt};
+static const iocshArg * const callbackQueueStatusArgs[1] =
+    {&callbackQueueStatusArg0};
+static const iocshFuncDef callbackQueueStatusFuncDef =
+    {"callbackQueueStatus",1,callbackQueueStatusArgs};
+static void callbackQueueStatusCallFunc(const iocshArgBuf *args)
+{
+    callbackQueuePrintStatus(args[0].ival);
+}
+
 /* callbackParallelThreads */
 static const iocshArg callbackParallelThreadsArg0 = { "no of threads", iocshArgInt};
 static const iocshArg callbackParallelThreadsArg1 = { "priority", iocshArgString};
@@ -441,12 +463,14 @@ void dbIocRegister(void)
     iocshRegister(&dbLockShowLockedFuncDef,dbLockShowLockedCallFunc);
 
     iocshRegister(&scanOnceSetQueueSizeFuncDef,scanOnceSetQueueSizeCallFunc);
+    iocshRegister(&scanOnceQueueStatusFuncDef,scanOnceQueueStatusCallFunc);
     iocshRegister(&scanpplFuncDef,scanpplCallFunc);
     iocshRegister(&scanpelFuncDef,scanpelCallFunc);
     iocshRegister(&postEventFuncDef,postEventCallFunc);
     iocshRegister(&scanpiolFuncDef,scanpiolCallFunc);
 
     iocshRegister(&callbackSetQueueSizeFuncDef,callbackSetQueueSizeCallFunc);
+    iocshRegister(&callbackQueueStatusFuncDef,callbackQueueStatusCallFunc);
     iocshRegister(&callbackParallelThreadsFuncDef,callbackParallelThreadsCallFunc);
 
     /* Needed before callback system is initialized */
diff --git a/src/ioc/db/dbScan.c b/src/ioc/db/dbScan.c
index e5c78fe..e0e14e0 100644
--- a/src/ioc/db/dbScan.c
+++ b/src/ioc/db/dbScan.c
@@ -24,6 +24,7 @@
 #include "cantProceed.h"
 #include "dbDefs.h"
 #include "ellLib.h"
+#include "epicsAtomic.h"
 #include "epicsEvent.h"
 #include "epicsMutex.h"
 #include "epicsPrint.h"
@@ -63,6 +64,7 @@ static volatile enum ctl scanCtl;
 static int onceQueueSize = 1000;
 static epicsEventId onceSem;
 static epicsRingBytesId onceQ;
+static int onceQOverruns = 0;
 static epicsThreadId onceTaskId;
 static void *exitOnce;
 
@@ -676,6 +678,7 @@ int scanOnceCallback(struct dbCommon *precord, once_complete cb, void *usr)
     if (!pushOK) {
         if (newOverflow) errlogPrintf("scanOnce: Ring buffer overflow\n");
         newOverflow = FALSE;
+        epicsAtomicIncrIntT(&onceQOverruns);
     } else {
         newOverflow = TRUE;
     }
@@ -722,6 +725,40 @@ int scanOnceSetQueueSize(int size)
     return 0;
 }
 
+int scanOnceQueueStatus(const int reset, scanOnceQueueStats *result)
+{
+    int ret;
+    if (!onceQ) return -1;
+    if (result) {
+        result->size = epicsRingBytesSize(onceQ) / sizeof(onceEntry);
+        result->numUsed = epicsRingBytesUsedBytes(onceQ) / sizeof(onceEntry);
+        result->maxUsed = epicsRingBytesHighWaterMark(onceQ) / sizeof(onceEntry);
+        result->numOverflow = epicsAtomicGetIntT(&onceQOverruns);
+        ret = 0;
+    } else {
+        ret = -2;
+    }
+    if (reset) {
+        epicsRingBytesResetHighWaterMark(onceQ);
+    }
+    return ret;
+}
+
+void scanOnceQueuePrintStatus(const int reset)
+{
+    scanOnceQueueStats stats;
+    if (scanOnceQueueStatus(reset, &stats) == -1) {
+        fprintf(stderr, "scanOnce system not initialized, yet. Please run "
+            "iocInit before using this command.\n");
+    } else {
+        double qusage = 100.0 * stats.numUsed / stats.size;
+        printf("PRIORITY  HIGH-WATER MARK  ITEMS IN Q  Q SIZE  %% USED  Q OVERFLOWS\n");
+        printf("%8s  %15d  %10d  %6d  %6.1f  %11d\n", "scanOnce", stats.maxUsed,
+               stats.numUsed, stats.size, qusage,
+               epicsAtomicGetIntT(&onceQOverruns));
+    }
+}
+
 static void initOnce(void)
 {
     if ((onceQ = epicsRingBytesLockedCreate(sizeof(onceEntry)*onceQueueSize)) == NULL) {
diff --git a/src/ioc/db/dbScan.h b/src/ioc/db/dbScan.h
index d483a0c..830d3a8 100644
--- a/src/ioc/db/dbScan.h
+++ b/src/ioc/db/dbScan.h
@@ -42,6 +42,13 @@ struct dbCommon;
 typedef void (*io_scan_complete)(void *usr, IOSCANPVT, int prio);
 typedef void (*once_complete)(void *usr, struct dbCommon*);
 
+typedef struct scanOnceQueueStats {
+    int size;
+    int numUsed;
+    int maxUsed;
+    int numOverflow;
+} scanOnceQueueStats;
+
 epicsShareFunc long scanInit(void);
 epicsShareFunc void scanRun(void);
 epicsShareFunc void scanPause(void);
@@ -57,6 +64,8 @@ epicsShareFunc double scanPeriod(int scan);
 epicsShareFunc int scanOnce(struct dbCommon *);
 epicsShareFunc int scanOnceCallback(struct dbCommon *, once_complete cb, void *usr);
 epicsShareFunc int scanOnceSetQueueSize(int size);
+epicsShareFunc int scanOnceQueueStatus(const int reset, scanOnceQueueStats *result);
+void scanOnceQueuePrintStatus(const int reset);
 
 /*print periodic lists*/
 epicsShareFunc int scanppl(double rate);
diff --git a/src/libCom/ring/epicsRingBytes.c b/src/libCom/ring/epicsRingBytes.c
index cb7e52e..ab048e4 100644
--- a/src/libCom/ring/epicsRingBytes.c
+++ b/src/libCom/ring/epicsRingBytes.c
@@ -38,6 +38,7 @@ typedef struct ringPvt {
     volatile int   nextPut;
     volatile int   nextGet;
     int            size;
+    int            highWaterMark;
     volatile char buffer[1]; /* actually larger */
 }ringPvt;
 
@@ -47,6 +48,7 @@ epicsShareFunc epicsRingBytesId  epicsShareAPI epicsRingBytesCreate(int size)
     if(!pring)
         return NULL;
     pring->size = size + SLOP;
+    pring->highWaterMark = 0;
     pring->nextGet = 0;
     pring->nextPut = 0;
     pring->lock    = 0;
@@ -118,7 +120,7 @@ epicsShareFunc int epicsShareAPI epicsRingBytesPut(
 {
     ringPvt *pring = (ringPvt *)id;
     int nextGet, nextPut, size;
-    int freeCount, copyCount, topCount;
+    int freeCount, copyCount, topCount, used;
 
     if (pring->lock) epicsSpinLock(pring->lock);
     nextGet = pring->nextGet;
@@ -131,8 +133,9 @@ epicsShareFunc int epicsShareAPI epicsRingBytesPut(
             if (pring->lock) epicsSpinUnlock(pring->lock);
             return 0;
         }
-        if (nbytes)
+        if (nbytes) {
             memcpy ((void *)&pring->buffer[nextPut], value, nbytes);
+        }
         nextPut += nbytes;
     }
     else {
@@ -143,8 +146,9 @@ epicsShareFunc int epicsShareAPI epicsRingBytesPut(
         }
         topCount = size - nextPut;
         copyCount = (nbytes > topCount) ?  topCount : nbytes;
-        if (copyCount)
+        if (copyCount) {
             memcpy ((void *)&pring->buffer[nextPut], value, copyCount);
+        }
         nextPut += copyCount;
         if (nextPut == size) {
             int nLeft = nbytes - copyCount;
@@ -155,6 +159,10 @@ epicsShareFunc int epicsShareAPI epicsRingBytesPut(
     }
     pring->nextPut = nextPut;
 
+    used = nextPut - nextGet;
+    if (used < 0) used += pring->size;
+    if (used > pring->highWaterMark) pring->highWaterMark = used;
+
     if (pring->lock) epicsSpinUnlock(pring->lock);
     return nbytes;
 }
@@ -224,3 +232,20 @@ epicsShareFunc int epicsShareAPI epicsRingBytesIsFull(epicsRingBytesId id)
 {
     return (epicsRingBytesFreeBytes(id) <= 0);
 }
+
+epicsShareFunc int epicsShareAPI epicsRingBytesHighWaterMark(epicsRingBytesIdConst id)
+{
+    ringPvt *pring = (ringPvt *)id;
+    return pring->highWaterMark;
+}
+
+epicsShareFunc void epicsShareAPI epicsRingBytesResetHighWaterMark(epicsRingBytesId id)
+{
+    ringPvt *pring = (ringPvt *)id;
+    int used;
+    if (pring->lock) epicsSpinLock(pring->lock);
+    used = pring->nextGet - pring->nextPut;
+    if (used < 0) used += pring->size;
+    pring->highWaterMark = used;
+    if (pring->lock) epicsSpinUnlock(pring->lock);
+}
diff --git a/src/libCom/ring/epicsRingBytes.h b/src/libCom/ring/epicsRingBytes.h
index 011829b..3dc0081 100644
--- a/src/libCom/ring/epicsRingBytes.h
+++ b/src/libCom/ring/epicsRingBytes.h
@@ -24,6 +24,7 @@ extern "C" {
 #include "shareLib.h"
 
 typedef void *epicsRingBytesId;
+typedef void const *epicsRingBytesIdConst;
 
 epicsShareFunc epicsRingBytesId  epicsShareAPI epicsRingBytesCreate(int nbytes);
 /* Same, but secured by a spinlock */
@@ -39,6 +40,8 @@ epicsShareFunc int  epicsShareAPI epicsRingBytesUsedBytes(epicsRingBytesId id);
 epicsShareFunc int  epicsShareAPI epicsRingBytesSize(epicsRingBytesId id);
 epicsShareFunc int  epicsShareAPI epicsRingBytesIsEmpty(epicsRingBytesId id);
 epicsShareFunc int  epicsShareAPI epicsRingBytesIsFull(epicsRingBytesId id);
+epicsShareFunc int  epicsShareAPI epicsRingBytesHighWaterMark(epicsRingBytesIdConst id);
+epicsShareFunc void epicsShareAPI epicsRingBytesResetHighWaterMark(epicsRingBytesId id);
 
 #ifdef __cplusplus
 }
diff --git a/src/libCom/ring/epicsRingPointer.cpp b/src/libCom/ring/epicsRingPointer.cpp
index 9c144ce..709ab65 100644
--- a/src/libCom/ring/epicsRingPointer.cpp
+++ b/src/libCom/ring/epicsRingPointer.cpp
@@ -90,3 +90,15 @@ epicsShareFunc int epicsShareAPI epicsRingPointerIsFull(epicsRingPointerId id)
     voidPointer *pvoidPointer = reinterpret_cast<voidPointer*>(id);
     return((pvoidPointer->isFull()) ? 1 : 0);
 }
+
+epicsShareFunc int epicsShareAPI epicsRingPointerGetHighWaterMark(epicsRingPointerIdConst id)
+{
+    voidPointer const *pvoidPointer = reinterpret_cast<voidPointer const*>(id);
+    return(pvoidPointer->getHighWaterMark());
+}
+
+epicsShareFunc void epicsShareAPI epicsRingPointerResetHighWaterMark(epicsRingPointerId id)
+{
+    voidPointer *pvoidPointer = reinterpret_cast<voidPointer*>(id);
+    pvoidPointer->resetHighWaterMark();
+}
diff --git a/src/libCom/ring/epicsRingPointer.h b/src/libCom/ring/epicsRingPointer.h
index 48d6203..68bf8f5 100644
--- a/src/libCom/ring/epicsRingPointer.h
+++ b/src/libCom/ring/epicsRingPointer.h
@@ -40,18 +40,22 @@ public: /* Functions */
     int getSize() const;
     bool isEmpty() const;
     bool isFull() const;
+    int getHighWaterMark() const;
+    void resetHighWaterMark();
 
 private: /* Prevent compiler-generated member functions */
     /* default constructor, copy constructor, assignment operator */
     epicsRingPointer();
     epicsRingPointer(const epicsRingPointer &);
     epicsRingPointer& operator=(const epicsRingPointer &);
+    int getUsedNoLock() const;
 
 private: /* Data */
     epicsSpinId lock;
     volatile int nextPush;
     volatile int nextPop;
     int size;
+    int highWaterMark;
     T  * volatile * buffer;
 };
 
@@ -59,6 +63,7 @@ extern "C" {
 #endif /*__cplusplus */
 
 typedef void *epicsRingPointerId;
+typedef void const *epicsRingPointerIdConst;
 
 epicsShareFunc epicsRingPointerId  epicsShareAPI epicsRingPointerCreate(int size);
 /* Same, but secured by a spinlock */
@@ -74,6 +79,8 @@ epicsShareFunc int  epicsShareAPI epicsRingPointerGetUsed(epicsRingPointerId id)
 epicsShareFunc int  epicsShareAPI epicsRingPointerGetSize(epicsRingPointerId id);
 epicsShareFunc int  epicsShareAPI epicsRingPointerIsEmpty(epicsRingPointerId id);
 epicsShareFunc int  epicsShareAPI epicsRingPointerIsFull(epicsRingPointerId id);
+epicsShareFunc int  epicsShareAPI epicsRingPointerGetHighWaterMark(epicsRingPointerIdConst id);
+epicsShareFunc void epicsShareAPI epicsRingPointerResetHighWaterMark(epicsRingPointerId id);
 
 /* This routine was incorrectly named in previous releases */
 #define epicsRingPointerSize epicsRingPointerGetSize
@@ -95,7 +102,8 @@ epicsShareFunc int  epicsShareAPI epicsRingPointerIsFull(epicsRingPointerId id);
 
 template <class T>
 inline epicsRingPointer<T>::epicsRingPointer(int sz, bool locked) :
-    lock(0), nextPush(0), nextPop(0), size(sz+1), buffer(new T* [sz+1])
+    lock(0), nextPush(0), nextPop(0), size(sz+1), highWaterMark(0),
+    buffer(new T* [sz+1])
 {
     if (locked)
         lock = epicsSpinCreate();
@@ -121,6 +129,8 @@ inline bool epicsRingPointer<T>::push(T *p)
     }
     buffer[next] = p;
     nextPush = newNext;
+    int used = getUsedNoLock();
+    if (used > highWaterMark) highWaterMark = used;
     if (lock) epicsSpinUnlock(lock);
     return(true);
 }
@@ -162,11 +172,18 @@ inline int epicsRingPointer<T>::getFree() const
 }
 
 template <class T>
-inline int epicsRingPointer<T>::getUsed() const
+inline int epicsRingPointer<T>::getUsedNoLock() const
 {
-    if (lock) epicsSpinLock(lock);
     int n = nextPush - nextPop;
     if (n < 0) n += size;
+    return n;
+}
+
+template <class T>
+inline int epicsRingPointer<T>::getUsed() const
+{
+    if (lock) epicsSpinLock(lock);
+    int n = getUsedNoLock();
     if (lock) epicsSpinUnlock(lock);
     return n;
 }
@@ -196,6 +213,20 @@ inline bool epicsRingPointer<T>::isFull() const
     return((count == 0) || (count == size));
 }
 
+template <class T>
+inline int epicsRingPointer<T>::getHighWaterMark() const
+{
+    return highWaterMark;
+}
+
+template <class T>
+inline void epicsRingPointer<T>::resetHighWaterMark()
+{
+    if (lock) epicsSpinLock(lock);
+    highWaterMark = getUsedNoLock();
+    if (lock) epicsSpinUnlock(lock);
+}
+
 #endif /* __cplusplus */
 
 #endif /* INCepicsRingPointerh */
diff --git a/src/libCom/test/ringBytesTest.c b/src/libCom/test/ringBytesTest.c
index 6cef933..bb91d02 100644
--- a/src/libCom/test/ringBytesTest.c
+++ b/src/libCom/test/ringBytesTest.c
@@ -30,7 +30,8 @@ typedef struct info {
     epicsRingBytesId	ring;
 }info;
 
-static void check(epicsRingBytesId ring, int expectedFree)
+static void check(epicsRingBytesId ring, int expectedFree,
+                  int expectedHighWaterMark)
 {
     int expectedUsed = RINGSIZE - expectedFree;
     int expectedEmpty = (expectedUsed == 0);
@@ -39,11 +40,14 @@ static void check(epicsRingBytesId ring, int expectedFree)
     int nUsed = epicsRingBytesUsedBytes(ring);
     int isEmpty = epicsRingBytesIsEmpty(ring);
     int isFull = epicsRingBytesIsFull(ring);
+    int highWaterMark = epicsRingBytesHighWaterMark(ring);
     
     testOk(nFree == expectedFree, "Free: %d == %d", nFree, expectedFree);
     testOk(nUsed == expectedUsed, "Used: %d == %d", nUsed, expectedUsed);
     testOk(isEmpty == expectedEmpty, "Empty: %d == %d", isEmpty, expectedEmpty);
     testOk(isFull == expectedFull, "Full: %d == %d", isFull, expectedFull);
+    testOk(highWaterMark == expectedHighWaterMark, "HighWaterMark: %d == %d",
+           highWaterMark, expectedHighWaterMark);
 }
     
 MAIN(ringBytesTest)
@@ -55,7 +59,7 @@ MAIN(ringBytesTest)
     char get[RINGSIZE+1];
     epicsRingBytesId ring;
 
-    testPlan(245);
+    testPlan(292);
 
     pinfo = calloc(1,sizeof(info));
     if (!pinfo) {
@@ -70,50 +74,54 @@ MAIN(ringBytesTest)
     if (!ring) {
         testAbort("epicsRingBytesCreate failed");
     }
-    check(ring, RINGSIZE);
+    check(ring, RINGSIZE, 0);
 
     for (i = 0 ; i < sizeof(put) ; i++)
         put[i] = i;
     for(i = 0 ; i < RINGSIZE ; i++) {
         n = epicsRingBytesPut(ring, put, i);
         testOk(n==i, "ring put %d", i);
-        check(ring, RINGSIZE-i);
+        check(ring, RINGSIZE-i, i);
         n = epicsRingBytesGet(ring, get, i);
         testOk(n==i, "ring get %d", i);
-        check(ring, RINGSIZE);
+        check(ring, RINGSIZE, i);
         testOk(memcmp(put,get,i)==0, "get matches write");
     }
 
+    epicsRingBytesResetHighWaterMark(ring);
+
     for(i = 0 ; i < RINGSIZE ; i++) {
         n = epicsRingBytesPut(ring, put+i, 1);
         testOk(n==1, "ring put 1, %d", i);
-        check(ring, RINGSIZE-1-i);
+        check(ring, RINGSIZE-1-i, i + 1);
     }
     n = epicsRingBytesPut(ring, put+RINGSIZE, 1);
     testOk(n==0, "put to full ring");
-    check(ring, 0);
+    check(ring, 0, RINGSIZE);
     for(i = 0 ; i < RINGSIZE ; i++) {
         n = epicsRingBytesGet(ring, get+i, 1);
         testOk(n==1, "ring get 1, %d", i);
-        check(ring, 1+i);
+        check(ring, 1+i, RINGSIZE);
     }
     testOk(memcmp(put,get,RINGSIZE)==0, "get matches write");
     n = epicsRingBytesGet(ring, get+RINGSIZE, 1);
     testOk(n==0, "get from empty ring");
-    check(ring, RINGSIZE);
+    check(ring, RINGSIZE, RINGSIZE);
+
+    epicsRingBytesResetHighWaterMark(ring);
 
     n = epicsRingBytesPut(ring, put, RINGSIZE+1);
     testOk(n==0, "ring put beyond ring capacity (%d, expected 0)",n);
-    check(ring, RINGSIZE);
+    check(ring, RINGSIZE, 0);
     n = epicsRingBytesPut(ring, put, 1);
     testOk(n==1, "ring put %d", 1);
-    check(ring, RINGSIZE-1);
+    check(ring, RINGSIZE-1, 1);
     n = epicsRingBytesPut(ring, put, RINGSIZE);
     testOk(n==0, "ring put beyond ring capacity (%d, expected 0)",n);
-    check(ring, RINGSIZE-1);
+    check(ring, RINGSIZE-1, 1);
     n = epicsRingBytesGet(ring, get, 1);
     testOk(n==1, "ring get %d", 1);
-    check(ring, RINGSIZE);
+    check(ring, RINGSIZE, 1);
 
     epicsRingBytesDelete(ring);
     epicsEventDestroy(consumerEvent);
diff --git a/src/libCom/test/ringPointerTest.c b/src/libCom/test/ringPointerTest.c
index 65a3494..d351708 100644
--- a/src/libCom/test/ringPointerTest.c
+++ b/src/libCom/test/ringPointerTest.c
@@ -64,6 +64,7 @@ static void testSingle(void)
     testOk1(epicsRingPointerGetFree(ring)==rsize);
     testOk1(epicsRingPointerGetSize(ring)==rsize);
     testOk1(epicsRingPointerGetUsed(ring)==0);
+    testOk1(epicsRingPointerGetHighWaterMark(ring)==0);
 
     testOk1(epicsRingPointerPop(ring)==NULL);
 
@@ -75,6 +76,10 @@ static void testSingle(void)
     testOk1(epicsRingPointerGetFree(ring)==rsize-1);
     testOk1(epicsRingPointerGetSize(ring)==rsize);
     testOk1(epicsRingPointerGetUsed(ring)==1);
+    testOk1(epicsRingPointerGetHighWaterMark(ring)==1);
+
+    epicsRingPointerResetHighWaterMark(ring);
+    testOk1(epicsRingPointerGetHighWaterMark(ring)==1);
 
     testDiag("Fill it up");
     for(i=2; i<2*rsize; i++) {
@@ -92,6 +97,7 @@ static void testSingle(void)
     testOk1(epicsRingPointerGetFree(ring)==0);
     testOk1(epicsRingPointerGetSize(ring)==rsize);
     testOk1(epicsRingPointerGetUsed(ring)==rsize);
+    testOk1(epicsRingPointerGetHighWaterMark(ring)==rsize);
 
     testDiag("Drain it out");
     for(i=1; i<2*rsize; i++) {
@@ -108,6 +114,7 @@ static void testSingle(void)
     testOk1(epicsRingPointerGetFree(ring)==rsize);
     testOk1(epicsRingPointerGetSize(ring)==rsize);
     testOk1(epicsRingPointerGetUsed(ring)==0);
+    testOk1(epicsRingPointerGetHighWaterMark(ring)==rsize);
 
     testDiag("Fill it up again");
     for(i=2; i<2*rsize; i++) {
@@ -236,7 +243,7 @@ MAIN(ringPointerTest)
 {
     int prio = epicsThreadGetPrioritySelf();
 
-    testPlan(37);
+    testPlan(42);
     testSingle();
     if (prio)
         epicsThreadSetPriority(epicsThreadGetIdSelf(), epicsThreadPriorityScanLow);