Re: Camonitor with client dictated update rate

["Johnson, Andrew N. via Tech-talk" <tech-talk@aps.anl.gov>]

Hi Emanuele,

On 6/2/19 4:25 PM, Emanuele Laface via Tech-talk wrote:

I have this scenario: N PVs of waveforms (arrays) of 10^5 data in each PV.

The IOc runs at 14 Hz, it means that if I subscribe in monitor mode I will be updated 14 times per second with N x 10^5 numbers.

Now I have an interface that I want to update just once per second with the content of these waveforms.

If I am in monitor, I can easily get the value once per second but my python script will anyway use the bandwidth and the cpu to process at 14 Hz because the threads of the monitoring will be updated at that frequency.

If I use caget once per second I need to deal with all the timeouts that can occur if some pv is down, disconnected, unreachable etc.

I haven’t find a reliable way to get the data once per second in a full asynchronous mode but without rejecting the data that anyway occupy bandwidth and cpu.

So from your description you really want/need the filtering to happen at the server. The existing solution for this is to add an additional record to the IOC for each waveform that you want to see at 1 Hz. The "slow data" record would have SCAN="1 second" and its INP link would be a DB link pointing to the equivalent "fast data" waveform record. Then once every second the "slow" record reads the current array from the "fast" record and publishes it from its own VAL field.

However I just converted the existing "sync" server-side filter into a fairly simplistic "dec" (decimate) filter which allows a client to request only one in every N monitor updates from a channel. Here I'm monitoring a 1Hz counter anj:count but also asking for a decimated view of that counter that gives me only one in three of the updates:

tux% camonitor anj:count 'anj:count.{"dec":{"n":3}}'
anj:count                      2019-06-03 12:26:26.719940 16  
anj:count.{"dec":{"n":3}}      2019-06-03 12:26:26.719940 16  
anj:count                      2019-06-03 12:26:27.719992 17  
anj:count                      2019-06-03 12:26:28.719933 18  
anj:count                      2019-06-03 12:26:29.719946 19  
anj:count.{"dec":{"n":3}}      2019-06-03 12:26:29.719946 19  
anj:count                      2019-06-03 12:26:30.719933 20  
anj:count                      2019-06-03 12:26:31.719952 21  

My demonstration shows a scalar value but it will work for any array field too. You can add this filter directly to any 3.15 or later IOC: Just build the attached source file into the IOC and add the registrar entry below to a DBD file that it also loads:

registrar(decInitialize)

If this filter works for you I will add it to the next 3.15 release of Base.

- Andrew

-- 
Complexity comes for free, Simplicity you have to work for.

/*************************************************************************\
* Copyright (c) 2010 Brookhaven National Laboratory.
* Copyright (c) 2010 Helmholtz-Zentrum Berlin
*     fuer Materialien und Energie GmbH.
* EPICS BASE is distributed subject to a Software License Agreement found
* in file LICENSE that is included with this distribution.
\*************************************************************************/

/*
 *  Author: Ralph Lange <Ralph.Lange@bessy.de>
 */

#include <stdio.h>

#include "freeList.h"
#include "db_field_log.h"
#include "chfPlugin.h"
#include "dbState.h"
#include "epicsAssert.h"
#include "epicsExport.h"

typedef struct myStruct {
    epicsInt32 n, i;
} myStruct;

static void *myStructFreeList;

static const
chfPluginArgDef opts[] = {
    chfInt32(myStruct, n, "n", 1, 0),
    chfPluginArgEnd
};

static void * allocPvt(void)
{
    myStruct *my = (myStruct*) freeListCalloc(myStructFreeList);
    return (void *) my;
}

static void freePvt(void *pvt)
{
    freeListFree(myStructFreeList, pvt);
}

static int parse_ok(void *pvt)
{
    myStruct *my = (myStruct*) pvt;

    if (my->n < 1)
        return -1;

    return 0;
}

static db_field_log* filter(void* pvt, dbChannel *chan, db_field_log *pfl) {
    db_field_log *passfl = NULL;
    myStruct *my = (myStruct*) pvt;
    epicsInt32 i = my->i;

    if (pfl->ctx == dbfl_context_read)
        return pfl;

    if (i++ == 0)
        passfl = pfl;

    if (i == my->n)
        i = 0;
    my->i = i;

    return passfl;
}

static void channelRegisterPre(dbChannel *chan, void *pvt,
                               chPostEventFunc **cb_out, void **arg_out, db_field_log *probe)
{
    *cb_out = filter;
    *arg_out = pvt;
}

static void channel_report(dbChannel *chan, void *pvt, int level, const unsigned short indent)
{
    myStruct *my = (myStruct*) pvt;
    printf("%*sDecimate (dec): n=%d, i=%d\n", indent, "",
           my->n, my->i);
}

static chfPluginIf pif = {
    allocPvt,
    freePvt,

    NULL, /* parse_error, */
    parse_ok,

    NULL, /* channel_open, */
    channelRegisterPre,
    NULL, /* channelRegisterPost, */
    channel_report,
    NULL /* channel_close */
};

static void decInitialize(void)
{
    static int firstTime = 1;

    if (!firstTime) return;
    firstTime = 0;

    if (!myStructFreeList)
        freeListInitPvt(&myStructFreeList, sizeof(myStruct), 64);

    chfPluginRegister("dec", &pif, opts);
}

epicsExportRegistrar(decInitialize);