Add configurable burst filter time

caPutLogApp/caPutJsonLogShellCommands.cpp

@@ -31,40 +31,44 @@ extern "C"
     extern int caPutLogJsonMsgQueueSize;
 
     /* Initalisation */
-    int caPutJsonLogInit(const char * address, caPutJsonLogConfig config){
+    int caPutJsonLogInit(const char * address, caPutJsonLogConfig config, double timeout){
         CaPutJsonLogTask *logger =  CaPutJsonLogTask::getInstance();
-        if (logger != NULL) return logger->initialize(address, config);
+        if (logger != NULL) return logger->initialize(address, config, timeout);
         else return -1;
     }
 
     static const iocshArg caPutJsonLogInitArg0 = {"address", iocshArgString};
     static const iocshArg caPutJsonLogInitArg1 = {"config", iocshArgInt};
+    static const iocshArg caPutJsonLogInitArg2 = {"burst timeout", iocshArgDouble};
     static const iocshArg *const caPutJsonLogInitArgs[] = {
         &caPutJsonLogInitArg0,
-        &caPutJsonLogInitArg1
+        &caPutJsonLogInitArg1,
+        &caPutJsonLogInitArg2
     };
-    static const iocshFuncDef caPutJsonLogInitDef = {"caPutJsonLogInit", 2, caPutJsonLogInitArgs};
+    static const iocshFuncDef caPutJsonLogInitDef = {"caPutJsonLogInit", 3, caPutJsonLogInitArgs};
     static void caPutJsonLogInitCall(const iocshArgBuf *args)
     {
-        caPutJsonLogInit(args[0].sval, static_cast<caPutJsonLogConfig>(args[1].ival));
+        caPutJsonLogInit(args[0].sval, static_cast<caPutJsonLogConfig>(args[1].ival), args[2].dval);
     }
 
 
     /* Reconfigure */
-    int caPutJsonLogReconf(caPutJsonLogConfig config){
+    int caPutJsonLogReconf(caPutJsonLogConfig config, double timeout){
         CaPutJsonLogTask *logger =  CaPutJsonLogTask::getInstance();
-        if (logger != NULL)  return logger->reconfigure(config);
+        if (logger != NULL)  return logger->reconfigure(config, timeout);
         else return -1;
     }
 
     static const iocshArg caPutJsonLogReconfArg0 = {"config", iocshArgInt};
+    static const iocshArg caPutJsonLogReconfArg1 = {"burst timeout", iocshArgDouble};
     static const iocshArg *const caPutJsonLogReconfArgs[] = {
-        &caPutJsonLogReconfArg0
+        &caPutJsonLogReconfArg0,
+        &caPutJsonLogReconfArg1
     };
     static const iocshFuncDef caPutJsonLogReconfDef = {"caPutJsonLogReconf", 1, caPutJsonLogReconfArgs};
     static void caPutJsonLogReconfCall(const iocshArgBuf *args)
     {
-        caPutJsonLogReconf(static_cast<caPutJsonLogConfig>(args[0].ival));
+        caPutJsonLogReconf(static_cast<caPutJsonLogConfig>(args[0].ival), args[1].dval);
     }
 
     /* Report */
@@ -118,6 +122,23 @@ extern "C"
         caPutJsonLogAddMetadata(args[0].sval, args[1].sval);
     }
 
+    /* Change burst filter timeout */
+    int caPutJsonLogSetBurstTimeout(double timeout){
+        CaPutJsonLogTask *logger =  CaPutJsonLogTask::getInstance();
+        if (logger != NULL)  return logger->setBurstTimeout(timeout);
+        else return -1;
+    }
+
+    static const iocshArg caPutJsonLogSetBurstTimeoutArg0 = {"burst timeout", iocshArgDouble};
+    static const iocshArg *const caPutJsonLogSetBurstTimeoutArgs[] = {
+        &caPutJsonLogSetBurstTimeoutArg0
+    };
+    static const iocshFuncDef caPutJsonLogSetBurstTimeoutDef = {"caPutJsonLogSetBurstTimeout", 1, caPutJsonLogSetBurstTimeoutArgs};
+    static void caPutJsonLogSetBurstTimeoutCall(const iocshArgBuf *args)
+    {
+        caPutJsonLogSetBurstTimeout(args[0].dval);
+    }
+
     /* Register JSON IOCsh commands */
     static void caPutJsonLogRegister(void)
     {
@@ -130,6 +151,7 @@ extern "C"
             iocshRegister(&caPutJsonLogShowDef,caPutJsonLogShowCall);
             iocshRegister(&caPutLogInitDef,caPutLogInitCall);
             iocshRegister(&caPutJsonLogAddMetadataDef,caPutJsonLogAddMetadataCall);
+            iocshRegister(&caPutJsonLogSetBurstTimeoutDef,caPutJsonLogSetBurstTimeoutCall);
             caPutLogRegisterDone = 2;
             break;
 

caPutLogApp/caPutJsonLogTask.cpp

@@ -88,7 +88,7 @@ CaPutJsonLogTask::~CaPutJsonLogTask()
     }
 }
 
-caPutJsonLogStatus CaPutJsonLogTask::reconfigure(caPutJsonLogConfig config)
+caPutJsonLogStatus CaPutJsonLogTask::reconfigure(caPutJsonLogConfig config, double timeout)
 {
     if ((config < caPutJsonLogNone)
         || (config > caPutJsonLogAllNoFilter)) {
@@ -97,6 +97,9 @@ caPutJsonLogStatus CaPutJsonLogTask::reconfigure(caPutJsonLogConfig config)
     } else {
         epics::atomic::set(this->config, config);
     }
+
+    this->setBurstTimeout(timeout);
+
     return caPutJsonLogSuccess;
 }
 
@@ -151,12 +154,12 @@ const std::map<std::string, std::string>& CaPutJsonLogTask::getMetadata()
     return metadata;
 }
 
-caPutJsonLogStatus CaPutJsonLogTask::initialize(const char* addresslist, caPutJsonLogConfig config)
+caPutJsonLogStatus CaPutJsonLogTask::initialize(const char* addresslist, caPutJsonLogConfig config, double timeout)
 {
     caPutJsonLogStatus status;
 
     // Store passed configuration parameters
-    this->reconfigure(config);
+    this->reconfigure(config, timeout);
 
     // Check if user enabled the logger
     if (config == caPutJsonLogNone) {
@@ -278,6 +281,16 @@ caPutJsonLogStatus CaPutJsonLogTask::configureServerLogging(const char* address)
     return caPutJsonLogSuccess;
 }
 
+caPutJsonLogStatus CaPutJsonLogTask::setBurstTimeout( double timeout )
+{
+    if (timeout > 0.0) {
+        this->burstTimeout = timeout;
+    } else {
+        this->burstTimeout = DEFAULT_BURST_TIMEOUT;
+    }
+    return caPutJsonLogSuccess;
+}
+
 caPutJsonLogStatus CaPutJsonLogTask::configurePvLogging()
 {
     char *caPutJsonLogPVEnv;
@@ -331,7 +344,7 @@ void CaPutJsonLogTask::caPutJsonLogTask(void *arg)
         int msgSize;
 
         // Receive new put with timeout
-        msgSize = this->caPutJsonLogQ.receive(&pnext, sizeof(LOGDATA *), 5.0);
+        msgSize = this->caPutJsonLogQ.receive(&pnext, sizeof(LOGDATA *), this->burstTimeout);
 
         /* Timeout */
         if (msgSize == -1) {

caPutLogApp/caPutJsonLogTask.h

@@ -91,10 +91,11 @@ class epicsShareClass CaPutJsonLogTask {
      *
      * @param address IP address or hostname of the log server. Can include a port number after a colon,
      *           if port number is not specified, default value will be used.
-     * @param config Configuration paramteter. Valid value are -1 <= config <= 2.
+     * @param config Configuration parameter. Valid value are -1 <= config <= 2.
+     * @param timeout Burst filter timeout parameter.
      * @return caPutJsonLogStatus Status code.
      */
-    caPutJsonLogStatus initialize(const char* address, caPutJsonLogConfig config);
+    caPutJsonLogStatus initialize(const char* address, caPutJsonLogConfig config, double timeout);
 
     /**
      * @brief Add a put details packed as a ::LOGDATA structure to a queue for processing.
@@ -131,9 +132,10 @@ class epicsShareClass CaPutJsonLogTask {
      *
      * @param config New configuration. Valid value are -1 <= config <= 2. Invalid
      * value will default to 1 "caPutJsonLogAll".
+     * @param timeout New burst filter timeout value
      * @return int Status code.
      */
-    caPutJsonLogStatus reconfigure(caPutJsonLogConfig config);
+    caPutJsonLogStatus reconfigure(caPutJsonLogConfig config, double timeout);
 
     /**
      * @brief Print report client logger information to the console.
@@ -178,6 +180,14 @@ class epicsShareClass CaPutJsonLogTask {
      */
     const std::map<std::string, std::string>& getMetadata();
 
+    /**
+     * @brief Change the burst filter timeout
+     *
+     * @param timeout Time between puts to the same PV that will be considered a burst
+     * @return int Status code.
+     */
+    caPutJsonLogStatus setBurstTimeout(double timeout);
+
 private:
 
     // Singleton instance of this class.
@@ -186,6 +196,8 @@ class epicsShareClass CaPutJsonLogTask {
     // Logger configuration
     int config; // To modify or read this value only epicsAtomic methods should be used
 
+    double burstTimeout;
+
     // Interthread communication
     epicsMessageQueue caPutJsonLogQ;
 

caPutLogApp/caPutLog.c

@@ -51,12 +51,12 @@ void caPutLogShow (int level)
 /*
  *  caPutLogReconf()
  */
-int caPutLogReconf (int config)
+int caPutLogReconf (int config, double timeout)
 {
     if (config < 0)
         caPutLogTaskStop();
     else
-        caPutLogTaskStart(config);
+        caPutLogTaskStart(config, timeout);
     caPutLogClientFlush();
     return caPutLogSuccess;
 }
@@ -69,7 +69,7 @@ static void caPutLogExitProc(void *arg)
 /*
  *  caPutLogInit()
  */
-int caPutLogInit (const char *addr_str, int config)
+int caPutLogInit (const char *addr_str, int config, double timeout)
 {
     int status;
 
@@ -96,7 +96,7 @@ int caPutLogInit (const char *addr_str, int config)
         return caPutLogError;
     }
 
-    status = caPutLogTaskStart(config);
+    status = caPutLogTaskStart(config, timeout);
     if (status) {
         return caPutLogError;
     }

caPutLogApp/caPutLog.h

@@ -17,10 +17,11 @@ extern "C" {
 #define caPutLogAll         1   /* log all puts */
 #define caPutLogAllNoFilter 2   /* log all puts no filtering on same PV*/
 
-epicsShareFunc int caPutLogInit (const char *addr_str, int config);
-epicsShareFunc int caPutLogReconf (int config);
+epicsShareFunc int caPutLogInit (const char *addr_str, int config, double timeout);
+epicsShareFunc int caPutLogReconf (int config, double timeout);
 epicsShareFunc void caPutLogShow (int level);
 epicsShareFunc void caPutLogSetTimeFmt (const char *format);
+epicsShareFunc void caPutLogSetBurstTimeout (double timeout);
 
 #ifdef __cplusplus
 }

caPutLogApp/caPutLogShellCommands.c

@@ -15,24 +15,28 @@
 
 static const iocshArg caPutLogInitArg0 = {"address", iocshArgString};
 static const iocshArg caPutLogInitArg1 = {"config", iocshArgInt};
+static const iocshArg caPutLogInitArg2 = {"burst timeout", iocshArgDouble};
 static const iocshArg *const caPutLogInitArgs[] = {
     &caPutLogInitArg0,
-    &caPutLogInitArg1
+    &caPutLogInitArg1,
+    &caPutLogInitArg2
 };
-static const iocshFuncDef caPutLogInitDef = {"caPutLogInit", 2, caPutLogInitArgs};
+static const iocshFuncDef caPutLogInitDef = {"caPutLogInit", 3, caPutLogInitArgs};
 static void caPutLogInitCall(const iocshArgBuf *args)
 {
-    caPutLogInit(args[0].sval, args[1].ival);
+    caPutLogInit(args[0].sval, args[1].ival, args[2].dval);
 }
 
 static const iocshArg caPutLogReconfArg0 = {"config", iocshArgInt};
+static const iocshArg caPutLogReconfArg1 = {"burst timeout", iocshArgDouble};
 static const iocshArg *const caPutLogReconfArgs[] = {
-    &caPutLogReconfArg0
+    &caPutLogReconfArg0,
+    &caPutLogReconfArg1
 };
-static const iocshFuncDef caPutLogReconfDef = {"caPutLogReconf", 1, caPutLogReconfArgs};
+static const iocshFuncDef caPutLogReconfDef = {"caPutLogReconf", 2, caPutLogReconfArgs};
 static void caPutLogReconfCall(const iocshArgBuf *args)
 {
-    caPutLogReconf(args[0].ival);
+    caPutLogReconf(args[0].ival, args[1].dval);
 }
 
 static const iocshArg caPutLogShowArg0 = {"level", iocshArgInt};
@@ -69,6 +73,16 @@ static void caPutJsonLogInitCall(const iocshArgBuf *args)
     #endif
 }
 
+static const iocshArg caPutLogSetBurstTimeoutArg0 = {"burst timeout", iocshArgDouble};
+static const iocshArg *const caPutLogSetBurstTimeoutArgs[] = {
+    &caPutLogSetBurstTimeoutArg0
+};
+static const iocshFuncDef caPutLogSetBurstTimeoutDef = {"caPutLogSetBurstTimeout", 1, caPutLogSetBurstTimeoutArgs};
+static void caPutLogSetBurstTimeoutCall(const iocshArgBuf *args)
+{
+    caPutLogSetBurstTimeout(args[0].dval);
+}
+
 static void caPutLogRegister(void)
 {
     extern int caPutLogRegisterDone;
@@ -80,6 +94,7 @@ static void caPutLogRegister(void)
         iocshRegister(&caPutLogShowDef,caPutLogShowCall);
         iocshRegister(&caPutLogSetTimeFmtDef,caPutLogSetTimeFmtCall);
         iocshRegister(&caPutJsonLogInitDef,caPutJsonLogInitCall);
+        iocshRegister(&caPutLogSetBurstTimeoutDef,caPutLogSetBurstTimeoutCall);
         caPutLogRegisterDone = 1;
         break;
 

caPutLogApp/caPutLogTask.c

@@ -109,6 +109,7 @@ static DBADDR *pcaPutLogPV;             /* Pointer to PV address structure,
 static epicsMessageQueueId caPutLogQ;   /* Mailbox for caPutLogTask */
 
 static volatile int caPutLogConfig;
+static volatile double burstTimeout;
 
 int caPutLogDebug = 0;
 epicsExportAddress(int, caPutLogDebug);
@@ -121,7 +122,7 @@ int caPutLogTotalCount = 0;
 #define isDbrNumeric(type) ((type) > DBR_STRING && (type) <= DBR_ENUM)
 
 /* Start Rng Log Task */
-int caPutLogTaskStart(int config)
+int caPutLogTaskStart(int config, double timeout)
 {
     epicsThreadId threadId;
     char *caPutLogPVEnv;
@@ -160,6 +161,7 @@ int caPutLogTaskStart(int config)
     }
 
     caPutLogConfig = config;
+    burstTimeout = (timeout > 0.0) ? timeout : DEFAULT_BURST_TIMEOUT;
 
     if (epicsThreadGetId("caPutLog")) {
         if (caPutLogDebug)
@@ -222,13 +224,23 @@ void caPutLogSetTimeFmt (const char *format)
         timeFormat = format;
 }
 
+void caPutLogSetBurstTimeout(double timeout)
+{
+    if (timeout > 0.0) {
+        burstTimeout = timeout;
+    } else {
+        burstTimeout = DEFAULT_BURST_TIMEOUT;
+        errlogSevPrintf(errlogMinor, "caPutLog: invalid timeout value = %f. Setting default value = %f\n", timeout, burstTimeout);
+    }
+}
+
 static void caPutLogTask(void *arg)
 {
     int sent = FALSE;
     int burst = 0;
     int config;
     int msg_size;
-
+    double timeout;
     LOGDATA *pcurrent, *pnext;
     VALUE old_value, max_value, min_value;
     VALUE *pold=&old_value, *pmax=&max_value, *pmin=&min_value;
@@ -255,7 +267,7 @@ static void caPutLogTask(void *arg)
     while (caPutLogConfig != caPutLogNone) {                 /* Main Server Loop */
 
         /* Receive next message */
-        msg_size = epicsMessageQueueReceiveWithTimeout(caPutLogQ, &pnext, MSG_SIZE, 5.0);
+        msg_size = epicsMessageQueueReceiveWithTimeout(caPutLogQ, &pnext, MSG_SIZE, burstTimeout);
         config = caPutLogConfig;
 
         if (msg_size == -1) {   /* timeout */
@@ -312,8 +324,6 @@ static void caPutLogTask(void *arg)
             caPutLogDataFree(pcurrent);
             pcurrent = pnext;
 
-            epicsAtomicIncrIntT(&caPutLogTotalCount);
-
             /* Set new old_value */
             val_assign(pold, &pcurrent->old_value, pcurrent->type);
 
@@ -322,7 +332,7 @@ static void caPutLogTask(void *arg)
             val_assign(pmin, &pcurrent->new_value.value, pcurrent->type);
 
             sent = FALSE;
-            burst = 0;
+            burst = FALSE;
         }
     }
     errlogSevPrintf(errlogInfo, "caPutLog: log task exiting\n");
@@ -406,10 +416,6 @@ static void log_msg(const VALUE *pold_value, const LOGDATA *pLogData,
         if (len >= space) { do_log(msg, space-1, YES); return; }
         len += val_to_string(msg+len, space-len, pmax, pLogData->type);
         if (len >= space) { do_log(msg, space-1, YES); return; }
-
-        /* burst count */
-        len += epicsSnprintf(msg+len, space-len, " burst=%d", burst);
-        if (len >= space) { do_log(msg, space-1, YES); return; }
     }
     do_log(msg, len, NO);
 }