monitor.h

Go to the documentation of this file.

/* monitor.h */
/*
 * Copyright information and license terms for this software can be
 * found in the file LICENSE that is included with the distribution
 */
#ifndef MONITOR_H
#define MONITOR_H

#include <list>
#include <ostream>

#ifdef epicsExportSharedSymbols
#   define monitorEpicsExportSharedSymbols
#   undef epicsExportSharedSymbols
#endif

#include <epicsMutex.h>
#include <pv/status.h>
#include <pv/pvData.h>
#include <pv/sharedPtr.h>
#include <pv/bitSet.h>
#include <pv/createRequest.h>

#ifdef monitorEpicsExportSharedSymbols
#   define epicsExportSharedSymbols
#       undef monitorEpicsExportSharedSymbols
#endif

#include <pv/requester.h>
#include <pv/destroyable.h>

#include <shareLib.h>

namespace epics { namespace pvAccess {

class MonitorRequester;
class MonitorElement;
typedef std::tr1::shared_ptr<MonitorElement> MonitorElementPtr;
typedef std::vector<MonitorElementPtr> MonitorElementPtrArray;

class Monitor;
typedef std::tr1::shared_ptr<Monitor> MonitorPtr;


class epicsShareClass MonitorElement {
public:
    POINTER_DEFINITIONS(MonitorElement);
    MonitorElement(epics::pvData::PVStructurePtr const & pvStructurePtr);
    const epics::pvData::PVStructurePtr pvStructurePtr;
    const epics::pvData::BitSet::shared_pointer changedBitSet;
    const epics::pvData::BitSet::shared_pointer overrunBitSet;

    class Ref;
};

class epicsShareClass Monitor : public virtual Destroyable{
    public:
    POINTER_DEFINITIONS(Monitor);
    typedef MonitorRequester requester_type;

    virtual ~Monitor(){}
    virtual epics::pvData::Status start() = 0;
    virtual epics::pvData::Status stop() = 0;
    virtual MonitorElementPtr poll() = 0;
    virtual void release(MonitorElementPtr const & monitorElement) = 0;

    struct Stats {
        size_t nfilled; 
        size_t noutstanding; 
        size_t nempty; 
    };

    virtual void getStats(Stats& s) const {
        s.nfilled = s.noutstanding = s.nempty = 0;
    }

    virtual void reportRemoteQueueStatus(epics::pvData::int32 freeElements) {}
};

class MonitorElement::Ref
{
    Monitor* mon;
    MonitorElementPtr elem;
public:
    Ref() :mon(0), elem() {}
    Ref(Monitor& M) :mon(&M), elem(mon->poll()) {}
    Ref(const Monitor::shared_pointer& M) :mon(M.get()), elem(mon->poll()) {}
    ~Ref() { reset(); }
#if __cplusplus>=201103L
    Ref(Ref&& o) :mon(o.mon), elem(o.elem) {
        o.mon = 0;
        o.elem.reset();
    }
#endif
    void swap(Ref& o) {
        std::swap(mon , o.mon);
        std::swap(elem, o.elem);
    }
    MonitorElementPtr letGo() {
        MonitorElementPtr ret;
        elem.swap(ret);
        return ret;
    }
    void attach(Monitor& M) {
        reset();
        mon = &M;
    }
    void attach(const Monitor::shared_pointer& M) {
        reset();
        mon = M.get();
    }
    bool next() {
        if(elem) mon->release(elem);
        elem = mon->poll();
        return !!elem;
    }
    void reset() {
        if(elem && mon) mon->release(elem);
        elem.reset();
    }
    Ref& operator++() {// prefix increment.  aka "++(*this)"
        next();
        return *this;
    }
#if __cplusplus>=201103L
    inline explicit operator bool() const { return elem.get(); }
#else
private:
    typedef const Monitor* const * hidden_bool_type;
public:
    operator hidden_bool_type() const { return elem.get() ? &mon : 0; }
#endif
    inline MonitorElement* operator->() { return elem.get(); }
    inline MonitorElement& operator*() { return *elem; }
    inline MonitorElement* get() { return elem.get(); }

    inline bool operator==(const Ref& o) const { return elem==o.elem; }
    inline bool operator!=(const Ref& o) const { return !(*this==o); }

    EPICS_NOT_COPYABLE(Ref)
};

#if __cplusplus>=201103L
// used by c++11 for-range
inline MonitorElement::Ref begin(Monitor& mon) { return MonitorElement::Ref(mon); }
inline MonitorElement::Ref end(Monitor& mon) { return MonitorElement::Ref(); }
#endif // __cplusplus<201103L

class epicsShareClass MonitorFIFO : public Monitor,
                                    public std::tr1::enable_shared_from_this<MonitorFIFO>
{
public:
    POINTER_DEFINITIONS(MonitorFIFO);
    struct epicsShareClass Source {
        POINTER_DEFINITIONS(Source);
        virtual ~Source();
        virtual void freeHighMark(MonitorFIFO *mon, size_t numEmpty) {}
    };
    struct epicsShareClass Config {
        size_t maxCount,    
               defCount,    
               actualCount; 
        bool dropEmptyUpdates; 
        epics::pvData::PVRequestMapper::mode_t mapperMode; 
        Config();
    };

    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();

    inline const std::tr1::shared_ptr<MonitorRequester> getRequester() const { return requester.lock(); }

    void show(std::ostream& strm) const;

    virtual void destroy() OVERRIDE FINAL;

    // configuration

    void setFreeHighMark(double level);

    // up-stream interface (putting data into FIFO)
    void open(const epics::pvData::StructureConstPtr& type);
    void close();
    void finish();
    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());
    void notify();

    // down-stream interface (taking data from FIFO)
    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; // may call Source::freeHighMark()
    virtual void getStats(Stats& s) const OVERRIDE FINAL;
    virtual void reportRemoteQueueStatus(epics::pvData::int32 freeElements) OVERRIDE FINAL;

    size_t freeCount() const;
private:
    size_t _freeCount() const;

    friend void providerRegInit(void*);
    static size_t num_instances;

    // const after ctor
    Config conf;

    // locking here is complicated...
    // our entry points which make callbacks are:
    //   notify() -> MonitorRequester::monitorConnect()
    //            -> MonitorRequester::monitorEvent()
    //            -> MonitorRequester::unlisten()
    //            -> ChannelBaseRequester::channelDisconnect()
    //   start()  -> MonitorRequester::monitorEvent()
    //   release()                 -> Source::freeHighMark()
    //                             -> notify() -> ...
    //   reportRemoteQueueStatus() -> Source::freeHighMark()
    //                             -> notify() -> ...
    mutable epicsMutex mutex;

    // ownership is archored at the downstream (consumer) end.
    // strong refs are:
    //   downstream -> MonitorFIFO -> Source
    // weak refs are:
    //   MonitorRequester <- MonitorFIFO <- upstream

    // so we expect that downstream will hold a strong ref to us,
    // and we keep a weak ref to downstream's MonitorRequester
    const std::tr1::weak_ptr<MonitorRequester> requester;

    const epics::pvData::PVStructure::const_shared_pointer pvRequest;

    // then we expect to keep a strong ref to upstream (Source)
    // and expect that upstream will have only a weak ref to us.
    const Source::shared_pointer upstream;

    enum state_t {
        Closed, // not open()'d
        Opened, // successful open()
        Error,  // unsuccessful open()
    } state;
    bool pipeline; // const after ctor
    bool running; // start() vs. stop()
    bool finished; // finish() called
    epics::pvData::BitSet scratch, oscratch; // using during post to avoid re-alloc

    bool needConnected;
    bool needEvent;
    bool needUnlisten;
    bool needClosed;

    epics::pvData::Status error; // Set when entering Error state

    size_t freeHighLevel;
    epicsInt32 flowCount;

    epics::pvData::PVRequestMapper mapper;

    typedef std::list<MonitorElementPtr> buffer_t;
    // we allocate one extra buffer element to hold data when post()
    // while all elements poll()'d.  So there will always be one
    // element on either the empty or inuse lists
    buffer_t inuse, empty, returned;
    /* our elements are in one of 4 states
     * Empty - on empty list
     * In Use - on inuse list
     * Polled - Returnedd from poll().  Not tracked
     * Returned - only if pipeline==true, release()'d but not ack'd
     */

    EPICS_NOT_COPYABLE(MonitorFIFO)
};

static inline
std::ostream& operator<<(std::ostream& strm, const MonitorFIFO& fifo) {
    fifo.show(strm);
    return strm;
}

}}

namespace epics { namespace pvData {

using epics::pvAccess::MonitorElement;
using epics::pvAccess::MonitorElementPtr;
using epics::pvAccess::MonitorElementPtrArray;
using epics::pvAccess::Monitor;
using epics::pvAccess::MonitorPtr;
}}

#endif  /* MONITOR_H */