epics::pvAccess::MonitorFIFO Class Reference

#include "monitor.h"

Inheritance diagram for epics::pvAccess::MonitorFIFO:

Collaboration diagram for epics::pvAccess::MonitorFIFO:

Classes
struct	Config
struct	Source

Public Member Functions
	POINTER_DEFINITIONS (MonitorFIFO)
	MonitorFIFO (const std::tr1::shared_ptr< MonitorRequester > &requester, const pvData::PVStructure::const_shared_pointer &pvRequest, const Source::shared_pointer &source=Source::shared_pointer(), Config *conf=0)
virtual	~MonitorFIFO ()
const std::tr1::shared_ptr< MonitorRequester >	getRequester () const
void	show (std::ostream &strm) const
virtual void	destroy () OVERRIDE FINAL
void	setFreeHighMark (double level)
void	open (const epics::pvData::StructureConstPtr &type)
	Mark subscription as "open" with the associated structure type. More...
void	close ()
	Abnormal closure (eg. due to upstream dis-connection) More...
void	finish ()
	Successful closure (eg. RDB query done) More...
bool	tryPost (const pvData::PVStructure &value, const epics::pvData::BitSet &changed, const epics::pvData::BitSet &overrun=epics::pvData::BitSet(), bool force=false)
void	post (const pvData::PVStructure &value, const epics::pvData::BitSet &changed, const epics::pvData::BitSet &overrun=epics::pvData::BitSet())
	Consume a free slot if available, otherwise squash with most recent. More...
void	notify ()
virtual epics::pvData::Status	start () OVERRIDE FINAL
virtual epics::pvData::Status	stop () OVERRIDE FINAL
virtual MonitorElementPtr	poll () OVERRIDE FINAL
virtual void	release (MonitorElementPtr const &monitorElement) OVERRIDE FINAL
virtual void	getStats (Stats &s) const OVERRIDE FINAL
virtual void	reportRemoteQueueStatus (epics::pvData::int32 freeElements) OVERRIDE FINAL
size_t	freeCount () const
	Number of unused FIFO slots at this moment, which may changed in the next. More...
Public Member Functions inherited from epics::pvAccess::Monitor
	POINTER_DEFINITIONS (Monitor)
virtual	~Monitor ()
Public Member Functions inherited from epics::pvAccess::Destroyable
	POINTER_DEFINITIONS (Destroyable)

Friends
void	providerRegInit (void *)

Additional Inherited Members
Public Types inherited from epics::pvAccess::Monitor
typedef MonitorRequester	requester_type
Protected Member Functions inherited from epics::pvAccess::Destroyable
virtual	~Destroyable ()

Detailed Description

Utility implementation of Monitor.

The Monitor interface defines the downstream (consumer facing) side of a FIFO. This class is a concrete implementation of this FIFO, including the upstream (producer facing) side.

In addition to MonitorRequester, which provides callbacks to the downstream side, The MonitorFIFO::Source class provides callbacks to the upstream side.

The simplest usage is to create (as shown below), then put update into the FIFO using post() and tryPost(). These methods behave the same when the queue is not full, but differ when it is. Additionally, tryPost() has an argument 'force'. Together there are three actions

post(value, changed) - combines the new update with the last (most recent) in the FIFO.

tryPost(value, changed, ..., false) - Makes no change to the FIFO and returns false.

tryPost(value, changed, ..., true) - Over-fills the FIFO with the new element, then returns false.

Note

Calls to post() or tryPost() must be followed with a call to notify(). Callers of notify() must not hold any locks, or a deadlock is possible.

The intent of tryPost() with force=true is to aid code which is transferring values from some upstream buffer and this FIFO. Such code can be complicated if an item is removed from the upstream buffer, but can't be put into this downstream FIFO. Rather than being forced to effectivly maintain a third FIFO, code can use force=true.

In either case, tryPost()==false indicates the the FIFO is full.

eg. simple usage in a sub-class for Channel named MyChannel.

pva::Monitor::shared_pointer
MyChannel::createMonitor(const pva::MonitorRequester::shared_pointer &requester,
                         const pvd::PVStructure::shared_pointer &pvRequest)
{
    std::tr1::shared_ptr<pva::MonitorFIFO> ret(new pva::MonitorFIFO(requester, pvRequest));
    ret->open(spamtype);
    ret->notify();
    // ret->post(...); // maybe initial update
}

Definition at line 258 of file monitor.h.

Constructor & Destructor Documentation

epics::pvAccess::MonitorFIFO::MonitorFIFO	(	const std::tr1::shared_ptr< MonitorRequester > &	requester,
		const pvData::PVStructure::const_shared_pointer &	pvRequest,
		const Source::shared_pointer &	source = Source::shared_pointer(),
		Config *	conf = 0
	)

Parameters

requester	Downstream/consumer callbacks
pvRequest	Downstream provided options
source	Upstream/producer callbacks
conf	Upstream provided options. Updated with actual values used. May be NULL to use defaults.

Definition at line 37 of file monitor.cpp.

    :conf(inconf ? *inconf : Config())
    ,requester(requester)
    ,pvRequest(pvRequest)
    ,upstream(source)
    ,state(Closed)
    ,pipeline(false)
    ,running(false)
    ,finished(false)
    ,needConnected(false)
    ,needEvent(false)
    ,needUnlisten(false)
    ,needClosed(false)
    ,freeHighLevel(0u)
    ,flowCount(0)
{
    REFTRACE_INCREMENT(num_instances);

    if(conf.maxCount==0)
        conf.maxCount = 1;

    if(conf.defCount==0)
        conf.defCount = 1;

    pvd::PVScalar::const_shared_pointer O(pvRequest->getSubField<pvd::PVScalar>("record._options.queueSize"));
    if(O && conf.actualCount==0) {
        try {
            conf.actualCount = O->getAs<pvd::uint32>();
        } catch(std::exception& e) {
            std::ostringstream strm;
            strm<<"invalid queueSize : "<<e.what();
            requester->message(strm.str());
        }
    }

    if(conf.actualCount==0)
        conf.actualCount = conf.defCount;

    if(conf.actualCount > conf.maxCount)
        conf.actualCount = conf.maxCount;

    O = pvRequest->getSubField<pvd::PVScalar>("record._options.pipeline");
    if(O) {
        try {
            pipeline = O->getAs<pvd::boolean>();
        } catch(std::exception& e) {
            std::ostringstream strm;
            strm<<"invalid pipeline : "<<e.what();
            requester->message(strm.str());
        }
    }

    setFreeHighMark(0.00);

    if(inconf)
        *inconf = conf;
}

epics::pvAccess::MonitorFIFO::~MonitorFIFO ( )

virtual

Definition at line 97 of file monitor.cpp.

                          {
    REFTRACE_DECREMENT(num_instances);
}

Member Function Documentation

void epics::pvAccess::MonitorFIFO::close

(

)

Abnormal closure (eg. due to upstream dis-connection)

Definition at line 190 of file monitor.cpp.

{
    Guard G(mutex);
    needClosed = state==Opened;
    state = Closed;
}

void epics::pvAccess::MonitorFIFO::destroy ( )

virtual

Destroy this instance.

Implements epics::pvAccess::Destroyable.

Definition at line 101 of file monitor.cpp.

{}

void epics::pvAccess::MonitorFIFO::finish

(

)

Successful closure (eg. RDB query done)

Definition at line 197 of file monitor.cpp.

{
    Guard G(mutex);
    if(state==Closed)
        throw std::logic_error("Can not finish() a closed Monitor");
    else if(finished)
        return; // no-op

    finished = true;
    if(inuse.empty() && running && state==Opened)
        needUnlisten = true;
}

size_t epics::pvAccess::MonitorFIFO::freeCount

(

)

const

Number of unused FIFO slots at this moment, which may changed in the next.

Definition at line 501 of file monitor.cpp.

{
    Guard G(mutex);
    return _freeCount();
}

const std::tr1::shared_ptr<MonitorRequester> epics::pvAccess::MonitorFIFO::getRequester ( ) const

inline

Access to MonitorRequester passed to ctor, or NULL if it has already been destroyed.

Since

>6.1.0

Definition at line 297 of file monitor.h.

{ return requester.lock(); }

void epics::pvAccess::MonitorFIFO::getStats ( Stats &  s ) const

virtual

Reimplemented from epics::pvAccess::Monitor.

Definition at line 461 of file monitor.cpp.

{
    Guard G(mutex);
    s.nempty = empty.size() + returned.size();
    s.nfilled = inuse.size();
    s.noutstanding = conf.actualCount - s.nempty - s.nfilled;
}

void epics::pvAccess::MonitorFIFO::notify

(

)

Call after calling any other upstream interface methods (open()/close()/finish()/post()/...) when no upstream mutexes are locked. Do not call from Source::freeHighMark(). This is done automatically. Call any MonitorRequester methods.

Definition at line 318 of file monitor.cpp.

{
    Monitor::shared_pointer self;
    MonitorRequester::shared_pointer req;
    pvd::StructureConstPtr type;
    bool conn = false,
         evt = false,
         unl = false,
         clo = false;
    pvd::Status err;

    {
        Guard G(mutex);

        std::swap(conn, needConnected);
        std::swap(evt, needEvent);
        std::swap(unl, needUnlisten);
        std::swap(clo, needClosed);
        std::swap(err, error);

        if(conn | evt | unl | clo) {
            req = requester.lock();
            self = shared_from_this();
        }
        if(conn && err.isSuccess())
            type = mapper.requested();
    }

    if(!req)
        return;
    if(conn && err.isSuccess())
        req->monitorConnect(pvd::Status(), self, type);
    else if(conn)
        req->monitorConnect(err, self, type);
    if(evt)
        req->monitorEvent(self);
    if(unl)
        req->unlisten(self);
    if(clo)
        req->channelDisconnect(false);
}

void epics::pvAccess::MonitorFIFO::open

(

const epics::pvData::StructureConstPtr &

type

)

Mark subscription as "open" with the associated structure type.

Definition at line 137 of file monitor.cpp.

{
    std::string message;
    {
        Guard G(mutex);

        if(state!=Closed)
            throw std::logic_error("Monitor already open.  Must close() before re-openning");
        else if(needClosed)
            throw std::logic_error("Monitor needs notify() between close() and open().");
        else if(finished)
            throw std::logic_error("Monitor finished.  re-open() not possible");

        // keep the code simpler.
        // never try to re-use elements, even on re-open w/o type change.
        empty.clear();
        inuse.clear();
        returned.clear();

        // fill up empty.
        pvd::PVDataCreatePtr create(pvd::getPVDataCreate());

        try {
            mapper.compute(*create->createPVStructure(type), *pvRequest, conf.mapperMode);
            message = mapper.warnings();

            while(empty.size() < conf.actualCount+1) {
                MonitorElementPtr elem(new MonitorElement(mapper.buildRequested()));
                empty.push_back(elem);
            }

            state = Opened;
            error = pvd::Status(); // ok

            assert(inuse.empty());
            assert(empty.size()>=2);
            assert(returned.empty());
            assert(conf.actualCount>=1);

        }catch(std::runtime_error& e){
            // error from compute()
            error = pvd::Status::error(e.what());
            state = Error;
        }
        needConnected = true;
    }
    if(message.empty()) return;
    requester_type::shared_pointer req(requester.lock());
    if(req) {
        req->message(message, warningMessage);
    }
}

epics::pvAccess::MonitorFIFO::POINTER_DEFINITIONS

(

MonitorFIFO

)

MonitorElementPtr epics::pvAccess::MonitorFIFO::poll ( )

virtual

If monitor has occurred return data.

Returns

monitorElement for modified data. Must call get to determine if data is available.

May recursively call MonitorRequester::unlisten()

Implements epics::pvAccess::Monitor.

Definition at line 397 of file monitor.cpp.

{
    MonitorElementPtr ret;
    Monitor::shared_pointer self;
    MonitorRequester::shared_pointer req;

    {
        Guard G(mutex);

        if(!inuse.empty() && inuse.size() + empty.size() > 1) {
            ret = inuse.front();
            inuse.pop_front();
            if(inuse.empty() && finished) {
                self = shared_from_this();
                req = requester.lock();
            }
        }

        assert(!inuse.empty() || !empty.empty());
    }

    if(req) {
        req->unlisten(self);
    }

    return ret;
}

void epics::pvAccess::MonitorFIFO::post	(	const pvData::PVStructure &	value,
		const epics::pvData::BitSet &	changed,
		const epics::pvData::BitSet &	overrun = epics::pvData::BitSet()
	)

Consume a free slot if available, otherwise squash with most recent.

Definition at line 259 of file monitor.cpp.

{
    Guard G(mutex);

    if(state!=Opened || finished) return;
    assert(!empty.empty() || !inuse.empty());

    const bool use_empty = !empty.empty();

    MonitorElementPtr elem;

    if(use_empty) {
        // space in window, or entering overflow, fill an empty element

        assert(!empty.empty());

        elem = empty.front();

    } else {
        // window full and already in overflow
        // squash with last element
        assert(!inuse.empty());
        elem = inuse.back();
    }

    if(conf.dropEmptyUpdates && !changed.logical_and(mapper.requestedMask()))
        return; // drop empty update

    scratch.clear();
    mapper.copyBaseToRequested(value, changed, *elem->pvStructurePtr, scratch);

    if(use_empty) {
        *elem->changedBitSet = scratch;
        elem->overrunBitSet->clear();
        mapper.maskBaseToRequested(overrun, *elem->overrunBitSet);

        if(inuse.empty() && running)
            needEvent = true;

        inuse.push_back(elem);
        empty.pop_front();
        if(pipeline)
            flowCount--;

    } else {
        // in overflow
        // squash
        elem->overrunBitSet->or_and(*elem->changedBitSet, scratch);
        *elem->changedBitSet |= scratch;
        oscratch.clear();
        mapper.maskBaseToRequested(overrun, oscratch);
        elem->overrunBitSet->or_and(oscratch, scratch);

        // leave as inuse.back()
    }
}

void epics::pvAccess::MonitorFIFO::release ( MonitorElementPtr const &  monitorElement )

virtual

Release a MonitorElement that was returned by poll. A poll() must be called after the release() to check the presence of any modified data.

Parameters

monitorElement

Implements epics::pvAccess::Monitor.

Definition at line 425 of file monitor.cpp.

{
    size_t nempty;
    {
        Guard G(mutex);

        assert(!inuse.empty() || !empty.empty());

        const pvd::StructureConstPtr& type((!inuse.empty() ? inuse.front() : empty.back())->pvStructurePtr->getStructure());

        if(elem->pvStructurePtr->getStructure() != type // return of old type
                || empty.size()+returned.size()>=conf.actualCount+1) // return of force'd
            return; // ignore it

        if(pipeline) {
            // work done during reportRemoteQueueStatus()
            returned.push_back(elem);
            return;
        }

        bool below = _freeCount() <= freeHighLevel;

        empty.push_front(elem);

        bool above = _freeCount() > freeHighLevel;

        if(!below || !above || !upstream)
            return;

        nempty = _freeCount();
    }

    upstream->freeHighMark(this, nempty);
    notify();
}

void epics::pvAccess::MonitorFIFO::reportRemoteQueueStatus ( epics::pvData::int32  freeElements )

virtual

Report remote queue status.

Parameters

freeElements

number of free elements.

Reimplemented from epics::pvAccess::Monitor.

Definition at line 469 of file monitor.cpp.

{
    if(nfree<=0 || !pipeline)
        return; // paranoia

    size_t nempty;
    {
        Guard G(mutex);

        bool below = _freeCount() <= freeHighLevel;

        size_t nack = std::min(size_t(nfree), returned.size());
        flowCount += nfree;

        buffer_t::iterator end(returned.begin());
        std::advance(end, nack);

        // remove[0, nack) from returned and append to empty
        empty.splice(empty.end(), returned, returned.begin(), end);

        bool above = _freeCount() > freeHighLevel;

        if(!below || !above || empty.size()<=1 || !upstream)
            return;

        nempty = _freeCount();
    }

    upstream->freeHighMark(this, nempty);
    notify();
}

void epics::pvAccess::MonitorFIFO::setFreeHighMark

(

double

level

)

Level, as a percentage of empty buffer slots, at which to call Source::freeHighMark(). Trigger condition is when number of free buffer slots goes above this level. In range [0.0, 1.0)

Definition at line 127 of file monitor.cpp.

{
    level = std::max(0.0, std::min(level, 1.0));
    pvd::uint32 lvl = std::max(size_t(0), std::min(size_t(conf.actualCount * level), conf.actualCount-1));

    Guard G(mutex);

    freeHighLevel = lvl;
}

void epics::pvAccess::MonitorFIFO::show

(

std::ostream &

strm

)

const

Definition at line 104 of file monitor.cpp.

{
    // const (after ctor) bits
    strm<<"MonitorFIFO"
          " pipeline="<<pipeline
        <<" size="<<conf.actualCount
        <<" freeHighLevel="<<freeHighLevel
        <<"\n";

    Guard G(mutex);

    switch(state) {
    case Closed: strm<<"  Closed"; break;
    case Opened: strm<<"  Opened"; break;
    case Error:  strm<<"  Error:"<<error; break;
    }

    strm<<" running="<<running<<" finished="<<finished<<"\n";
    strm<<"  #empty="<<empty.size()<<" #returned="<<returned.size()<<" #inuse="<<inuse.size()<<" flowCount="<<flowCount<<"\n";
    strm<<"  events "<<(needConnected?'C':'_')<<(needEvent?'E':'_')<<(needUnlisten?'U':'_')<<(needClosed?'X':'_')
        <<"\n";
}

pvd::Status epics::pvAccess::MonitorFIFO::start ( )

virtual

Start monitoring.

Returns

completion status.

Implements epics::pvAccess::Monitor.

Definition at line 360 of file monitor.cpp.

{
    Monitor::shared_pointer self;
    MonitorRequester::shared_pointer req;

    {
        Guard G(mutex);

        if(state==Closed)
            throw std::logic_error("Monitor can't start() before open()");

        if(running || state!=Opened)
            return pvd::Status();

        if(!inuse.empty()) {
            self = shared_from_this();
            req = requester.lock();
        }

        running = true;
    }

    if(req)
        req->monitorEvent(self);

    return pvd::Status();
}

pvd::Status epics::pvAccess::MonitorFIFO::stop ( )

virtual

Stop Monitoring.

Returns

completion status.

Implements epics::pvAccess::Monitor.

Definition at line 388 of file monitor.cpp.

{
    Guard G(mutex);

    running = false;

    return pvd::Status();
}

bool epics::pvAccess::MonitorFIFO::tryPost	(	const pvData::PVStructure &	value,
		const epics::pvData::BitSet &	changed,
		const epics::pvData::BitSet &	overrun = epics::pvData::BitSet(),
		bool	force = false
	)

Consume a free slot if available. otherwise ... if !force take no action and return false. if force then attempt to allocate and fill a new slot, then return false. The extra slot will be free'd after it is consumed.

Definition at line 210 of file monitor.cpp.

{
    Guard G(mutex);

    if(state!=Opened || finished) return false; // when Error, act as always "full"
    assert(!empty.empty() || !inuse.empty());

    const bool havefree = _freeCount()>0u;

    MonitorElementPtr elem;
    if(conf.dropEmptyUpdates && !changed.logical_and(mapper.requestedMask())) {
        // drop empty update
    } else if(havefree) {
        // take an empty element
        elem = empty.front();
        empty.pop_front();
    } else if(force) {
        // allocate an extra element
        elem.reset(new MonitorElement(mapper.buildRequested()));
    }

    if(elem) {
        try {
            elem->changedBitSet->clear();
            mapper.copyBaseToRequested(value, changed,
                                       *elem->pvStructurePtr, *elem->changedBitSet);
            elem->overrunBitSet->clear();
            mapper.maskBaseToRequested(overrun, *elem->overrunBitSet);

            if(inuse.empty() && running)
                needEvent = true;
            inuse.push_back(elem);
        }catch(...){
            if(havefree) {
                empty.push_front(elem);
            }
            throw;
        }
        if(pipeline)
            flowCount--;
    }

    return _freeCount()>0u;
}

Friends And Related Function Documentation

void providerRegInit ( void *  )

friend

Definition at line 208 of file ChannelAccessFactory.cpp.

{
    epicsSignalInstallSigAlarmIgnore();
    epicsSignalInstallSigPipeIgnore();

    providerRegGbl = new providerRegGbl_t;
    providerRegGbl->clients->add("pva", createClientProvider);

    registerRefCounter("ServerContextImpl", &ServerContextImpl::num_instances);
    registerRefCounter("ServerChannel", &ServerChannel::num_instances);
    registerRefCounter("Transport (ABC)", &Transport::num_instances);
    registerRefCounter("BlockingTCPTransportCodec", &detail::BlockingTCPTransportCodec::num_instances);
    registerRefCounter("BlockingUDPTransport", &BlockingUDPTransport::num_instances);
    registerRefCounter("ChannelProvider (ABC)", &ChannelProvider::num_instances);
    registerRefCounter("Channel (ABC)", &Channel::num_instances);
    registerRefCounter("ChannelRequester (ABC)", &ChannelRequester::num_instances);
    registerRefCounter("ChannelBaseRequester (ABC)", &ChannelBaseRequester::num_instances);
    registerRefCounter("ChannelRequest (ABC)", &ChannelRequest::num_instances);
    registerRefCounter("ResponseHandler (ABC)", &ResponseHandler::num_instances);
    registerRefCounter("MonitorFIFO", &MonitorFIFO::num_instances);
    pvas::registerRefTrackServer();
    registerRefCounter("pvas::SharedChannel", &pvas::detail::SharedChannel::num_instances);
    registerRefCounter("pvas::SharedPut", &pvas::detail::SharedPut::num_instances);
    registerRefCounter("pvas::SharedRPC", &pvas::detail::SharedRPC::num_instances);
    registerRefCounter("pvas::SharedPV", &pvas::SharedPV::num_instances);
}

---

The documentation for this class was generated from the following files:

• modules/pvAccess/src/client/pv/monitor.h
• modules/pvAccess/src/client/monitor.cpp