MonitorCacheEntry Struct Reference

#include "chancache.h"

Inheritance diagram for MonitorCacheEntry:

Collaboration diagram for MonitorCacheEntry:

Public Types
typedef std::vector< epicsUInt8 >	pvrequest_t
typedef weak_set< MonitorUser >	interested_t
Public Types inherited from epics::pvAccess::MonitorRequester
typedef Monitor	operation_type

Public Member Functions
	POINTER_DEFINITIONS (MonitorCacheEntry)
epicsMutex &	mutex () const
	MonitorCacheEntry (ChannelCacheEntry *ent, const epics::pvData::PVStructure::shared_pointer &pvr)
virtual	~MonitorCacheEntry ()
virtual void	monitorConnect (epics::pvData::Status const &status, epics::pvData::MonitorPtr const &monitor, epics::pvData::StructureConstPtr const &structure)
virtual void	monitorEvent (epics::pvData::MonitorPtr const &monitor)
virtual void	unlisten (epics::pvData::MonitorPtr const &monitor)
virtual std::string	getRequesterName ()
Public Member Functions inherited from epics::pvAccess::MonitorRequester
	POINTER_DEFINITIONS (MonitorRequester)
virtual	~MonitorRequester ()
virtual void	monitorConnect (epics::pvData::Status const &status, MonitorPtr const &monitor, epics::pvData::StructureConstPtr const &structure)=0
virtual void	monitorEvent (MonitorPtr const &monitor)=0
virtual void	unlisten (MonitorPtr const &monitor)=0
Public Member Functions inherited from epics::pvAccess::ChannelBaseRequester
	POINTER_DEFINITIONS (ChannelBaseRequester)
	ChannelBaseRequester ()
virtual	~ChannelBaseRequester ()
virtual void	channelDisconnect (bool destroy)

Public Attributes
weak_pointer	weakref
ChannelCacheEntry *const	chan
const size_t	bufferSize
bool	havedata
bool	done
size_t	nwakeups
size_t	nevents
epics::pvData::StructureConstPtr	typedesc
epics::pvData::MonitorElement::shared_pointer	lastelem
epics::pvData::MonitorPtr	mon
epics::pvData::Status	startresult
interested_t	interested

Static Public Attributes
static size_t	num_instances
Static Public Attributes inherited from epics::pvAccess::ChannelBaseRequester
static size_t	num_instances

Detailed Description

Definition at line 24 of file chancache.h.

Member Typedef Documentation

typedef weak_set<MonitorUser> MonitorCacheEntry::interested_t

Definition at line 52 of file chancache.h.

typedef std::vector<epicsUInt8> MonitorCacheEntry::pvrequest_t

Definition at line 37 of file chancache.h.

Constructor & Destructor Documentation

MonitorCacheEntry::MonitorCacheEntry	(	ChannelCacheEntry *	ent,
		const epics::pvData::PVStructure::shared_pointer &	pvr
	)

Definition at line 34 of file moncache.cpp.

    :chan(ent)
    ,bufferSize(getS<pvd::uint32>(pvr, "record._options.queueSize", 2)) // should be same default as pvAccess, but not required
    ,havedata(false)
    ,done(false)
    ,nwakeups(0)
    ,nevents(0)
{
    epicsAtomicIncrSizeT(&num_instances);
}

MonitorCacheEntry::~MonitorCacheEntry ( )

virtual

Definition at line 45 of file moncache.cpp.

{
    pvd::Monitor::shared_pointer M;
    M.swap(mon);
    if(M) {
        M->destroy();
    }
    epicsAtomicDecrSizeT(&num_instances);
    const_cast<ChannelCacheEntry*&>(chan) = NULL; // spoil to fault use after free
}

Member Function Documentation

std::string MonitorCacheEntry::getRequesterName ( )

virtual

Definition at line 239 of file moncache.cpp.

{
    return "MonitorCacheEntry";
}

void MonitorCacheEntry::monitorConnect ( epics::pvData::Status const &  status, epics::pvData::MonitorPtr const &  monitor, epics::pvData::StructureConstPtr const &  structure  )

virtual

Definition at line 57 of file moncache.cpp.

{
    interested_t::vector_type tonotify;
    {
        Guard G(mutex());
        if(typedesc) {
            // we shouldn't have to deal with monitor type change since we
            // destroy() Monitors on Channel disconnect.
            std::cerr<<"monitorConnect() w/ new type.  Monitor has outlived it's connection.\n";
            monitor->stop();
            //TODO: unlisten()
            return;
        }
        typedesc = structure;

        if(status.isSuccess()) {
            startresult = monitor->start();
        } else {
            startresult = status;
        }

        if(startresult.isSuccess()) {
            lastelem.reset(new pvd::MonitorElement(pvd::getPVDataCreate()->createPVStructure(structure)));
        }

        // set typedesc and startresult for futured MonitorUsers
        // and copy snapshot of already interested MonitorUsers
        tonotify = interested.lock_vector();
    }

    if(!startresult.isSuccess())
        std::cout<<"upstream monitor start() fails\n";

    shared_pointer self(weakref); // keeps us alive all MonitorUsers are destroy()ed

    for(interested_t::vector_type::const_iterator it = tonotify.begin(),
        end = tonotify.end(); it!=end; ++it)
    {
        pvd::MonitorRequester::shared_pointer req((*it)->req);
        if(req) {
            req->monitorConnect(startresult, *it, structure);
        }
    }
}

void MonitorCacheEntry::monitorEvent ( epics::pvData::MonitorPtr const &  monitor )

virtual

Definition at line 110 of file moncache.cpp.

{
    /* PVA is being tricky, the Monitor* passed to monitorConnect()
     * isn't the same one we see here!
     * The original was a ChannelMonitorImpl, we now see a MonitorStrategyQueue
     * owned by the original, which delegates deserialization and accumulation
     * of deltas into complete events for us.
     * However, we don't want to keep the MonitorStrategyQueue as it's
     * destroy() method is a no-op!
     */

    epicsAtomicIncrSizeT(&nwakeups);

    shared_pointer self(weakref); // keeps us alive in case all MonitorUsers are destroy()ed

    pva::MonitorElementPtr update;

    typedef std::vector<MonitorUser::shared_pointer> dsnotify_t;
    dsnotify_t dsnotify;

    {
        Guard G(mutex()); // MCE and MU guarded by the same mutex
        if(!havedata)
            havedata = true;

        //TODO: flow control, if all MU buffers are full, break before poll()==NULL
        while((update=monitor->poll()))
        {
            epicsAtomicIncrSizeT(&nevents);

            lastelem->pvStructurePtr->copyUnchecked(*update->pvStructurePtr,
                                                    *update->changedBitSet);
            *lastelem->changedBitSet = *update->changedBitSet;
            *lastelem->overrunBitSet = *update->overrunBitSet;
            monitor->release(update);
            update.reset();

            interested_t::iterator IIT(interested); // recursively locks interested.mutex() (assumes this->mutex() is interestd.mutex())
            for(interested_t::value_pointer pusr = IIT.next(); pusr; pusr = IIT.next())
            {
                MonitorUser *usr = pusr.get();

                {
                    Guard G(usr->mutex());
                    if(usr->initial)
                        continue; // no start() yet
                    // TODO: track overflow when !running (after stop())?
                    if(!usr->running || usr->empty.empty()) {
                        usr->inoverflow = true;

                        /* overrun |= lastelem->overrun           // upstream overflows
                         * overrun |= changed & lastelem->changed // downstream overflows
                         * changed |= lastelem->changed           // accumulate changes
                         */

                        *usr->overflowElement->overrunBitSet |= *lastelem->overrunBitSet;
                        usr->overflowElement->overrunBitSet->or_and(*usr->overflowElement->changedBitSet,
                                                                    *lastelem->changedBitSet);
                        *usr->overflowElement->changedBitSet |= *lastelem->changedBitSet;

                        usr->overflowElement->pvStructurePtr->copyUnchecked(*lastelem->pvStructurePtr,
                                                                            *lastelem->changedBitSet);

                        epicsAtomicIncrSizeT(&usr->ndropped);
                        continue;
                    }
                    // we only come out of overflow when downstream release()s an element to us
                    // empty.empty() does not imply inoverflow,
                    // however inoverflow does imply empty.empty()
                    assert(!usr->inoverflow);

                    if(usr->filled.empty())
                        dsnotify.push_back(pusr);

                    pvd::MonitorElementPtr elem(usr->empty.front());

                    *elem->overrunBitSet = *lastelem->overrunBitSet;
                    *elem->changedBitSet = *lastelem->changedBitSet;
                    // Note: can't use changed mask to optimize this copy since we don't know
                    //       the state of the free element
                    elem->pvStructurePtr->copyUnchecked(*lastelem->pvStructurePtr);

                    usr->filled.push_back(elem);
                    usr->empty.pop_front();

                    epicsAtomicIncrSizeT(&usr->nevents);
                }
            }
        }
    }

    // unlock here, race w/ stop(), unlisten()?
    //TODO: notify from worker thread

    FOREACH(dsnotify_t::iterator, it,end,dsnotify) {
        MonitorUser *usr = (*it).get();
        pvd::MonitorRequester::shared_pointer req(usr->req);
        epicsAtomicIncrSizeT(&usr->nwakeups);
        req->monitorEvent(*it); // notify when first item added to empty queue, may call poll(), release(), and others
    }
}

epicsMutex& MonitorCacheEntry::mutex ( ) const

inline

Definition at line 35 of file chancache.h.

{ return interested.mutex(); }

MonitorCacheEntry::POINTER_DEFINITIONS

(

MonitorCacheEntry

)

void MonitorCacheEntry::unlisten ( epics::pvData::MonitorPtr const &  monitor )

virtual

Definition at line 214 of file moncache.cpp.

{
    pvd::Monitor::shared_pointer M;
    interested_t::vector_type tonotify;
    {
        Guard G(mutex());
        M.swap(mon);
        tonotify = interested.lock_vector();
        // assume that upstream won't call monitorEvent() again

        // cause future downstream start() to error
        startresult = pvd::Status(pvd::Status::STATUSTYPE_ERROR, "upstream unlisten()");
    }
    if(M) {
        M->destroy();
    }
    FOREACH(interested_t::vector_type::iterator, it, end, tonotify) {
        MonitorUser *usr = it->get();
        pvd::MonitorRequester::shared_pointer req(usr->req);
        if(usr->inuse.empty()) // TODO: what about stopped?
            req->unlisten(*it);
    }
}

Member Data Documentation

const size_t MonitorCacheEntry::bufferSize

Definition at line 32 of file chancache.h.

ChannelCacheEntry* const MonitorCacheEntry::chan

Definition at line 30 of file chancache.h.

bool MonitorCacheEntry::done

Definition at line 40 of file chancache.h.

bool MonitorCacheEntry::havedata

Definition at line 39 of file chancache.h.

interested_t MonitorCacheEntry::interested

Definition at line 53 of file chancache.h.

epics::pvData::MonitorElement::shared_pointer MonitorCacheEntry::lastelem

value of upstream monitor (accumulation of all deltas) changed/overflow bit masks of last delta

Definition at line 48 of file chancache.h.

epics::pvData::MonitorPtr MonitorCacheEntry::mon

Definition at line 49 of file chancache.h.

size_t MonitorCacheEntry::nevents

Definition at line 42 of file chancache.h.

size_t MonitorCacheEntry::num_instances

static

Definition at line 27 of file chancache.h.

size_t MonitorCacheEntry::nwakeups

Definition at line 41 of file chancache.h.

epics::pvData::Status MonitorCacheEntry::startresult

Definition at line 50 of file chancache.h.

epics::pvData::StructureConstPtr MonitorCacheEntry::typedesc

Definition at line 44 of file chancache.h.

weak_pointer MonitorCacheEntry::weakref

Definition at line 28 of file chancache.h.

---

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

• modules/pva2pva/p2pApp/chancache.h
• modules/pva2pva/p2pApp/moncache.cpp