more testing

testApp/Makefile

@@ -22,6 +22,11 @@ testtest_SRCS = testtest.cpp
 testtest_LIBS = testutils p2pcore pvAccess pvData Com
 TESTS += testtest
 
+TESTPROD_HOST += testmon
+testmon_SRCS = testmon.cpp
+testmon_LIBS = testutils p2pcore pvAccess pvData Com
+TESTS += testmon
+
 TESTSCRIPTS_HOST += $(TESTS:%=%.t)
 
 #===========================

testApp/testmon.cpp

@@ -3,64 +3,146 @@
 #include <epicsUnitTest.h>
 #include <testMain.h>
 
+#include <pv/epicsException.h>
 #include <pv/monitor.h>
 #include <pv/thread.h>
 
+#include "server.h"
+
 #include "utilities.h"
 
 namespace pvd = epics::pvData;
+namespace pva = epics::pvAccess;
 
 typedef epicsGuard<epicsMutex> Guard;
 
 namespace {
 
-struct DownStreamReq : public pvd::MonitorRequester, DumbRequester
+pvd::PVStructurePtr makeRequest(size_t bsize)
 {
-    pvd::Mutex lock;
-    bool connected;
-    bool unlistend;
-    size_t eventCnt;
-    pvd::Status connectStatus;
-    pvd::MonitorPtr mon;
-    pvd::StructureConstPtr structure;
+    pvd::StructureConstPtr dtype(pvd::getFieldCreate()->createFieldBuilder()
+                                 ->addNestedStructure("record")
+                                    ->addNestedStructure("_options")
+                                        ->add("queueSize", pvd::pvString) // yes, really.  PVA wants a string
+                                    ->endNested()
+                                 ->endNested()
+                                 ->createStructure());
 
-    DownStreamReq()
-        :DumbRequester("DownStreamReq")
-        ,connected(false)
-        ,unlistend(false)
-        ,eventCnt(0)
-    {}
+    pvd::PVStructurePtr ret(pvd::getPVDataCreate()->createPVStructure(dtype));
+    ret->getSubFieldT<pvd::PVScalar>("record._options.queueSize")->putFrom<pvd::int32>(bsize);
 
-    virtual void monitorConnect(pvd::Status const & status,
-        pvd::MonitorPtr const & monitor, pvd::StructureConstPtr const & structure)
+    return ret;
+}
+
+struct TestMonitor {
+    TestProvider::shared_pointer upstream;
+    TestPV::shared_pointer test1;
+    ScalarAccessor<pvd::int32> test1_x, test1_y;
+
+    GWServerChannelProvider::shared_pointer gateway;
+
+    TestChannelRequester::shared_pointer client_req;
+    pva::Channel::shared_pointer client;
+
+    // prepare providers and connect the client channel, don't setup monitor
+    TestMonitor()
+        :upstream(new TestProvider())
+        ,test1(upstream->addPV("test1", pvd::getFieldCreate()->createFieldBuilder()
+                               ->add("x", pvd::pvInt)
+                               ->add("y", pvd::pvInt)
+                               ->createStructure()))
+        ,test1_x(test1->value, "x")
+        ,test1_y(test1->value, "y")
+        ,gateway(new GWServerChannelProvider(upstream))
+        ,client_req(new TestChannelRequester)
+        ,client(gateway->createChannel("test1", client_req))
     {
-        Guard G(lock);
-        assert(!connected);
-        connectStatus = status;
-        mon = monitor;
-        this->structure = structure;
-        connected = true;
+        testDiag("pre-test setup");
+        if(!client)
+            testAbort("channel \"test1\" not connected");
+        test1_x = 1;
+        test1_y = 2;
     }
 
-    virtual void monitorEvent(MonitorPtr const & monitor)
+    void test_event()
     {
-        Guard G(lock);
-        mon = monitor;
-        eventCnt++;
+        testDiag("Push the initial event through from upstream to downstream");
+
+        TestChannelMonitorRequester::shared_pointer mreq(new TestChannelMonitorRequester);
+        pvd::Monitor::shared_pointer mon(client->createMonitor(mreq, makeRequest(2)));
+        if(!mon) testAbort("Failed to create monitor");
+
+        testOk1(mreq->eventCnt==0);
+        testOk1(mon->start().isSuccess());
+        upstream->dispatch(); // trigger monitorEvent() from upstream to gateway
+
+        testOk1(mreq->eventCnt==1);
+        pvd::MonitorElementPtr elem(mon->poll());
+        testOk1(!!elem.get());
+        testOk1(elem && elem->pvStructurePtr->getSubFieldT<pvd::PVInt>("x")->get()==1);
+
+        if(elem) mon->release(elem);
     }
 
-    virtual void unlisten(MonitorPtr const & monitor)
+    void test_share()
     {
-        Guard G(lock);
-        assert(!unlistend);
-        unlistend = true;
+        // here both downstream monitors are on the same Channel,
+        // which would be inefficient, and slightly unrealistic, w/ real PVA,
+        // but w/ TestProvider makes no difference
+        testDiag("Test two downstream monitors sharing the same upstream");
+
+        TestChannelMonitorRequester::shared_pointer mreq(new TestChannelMonitorRequester);
+        pvd::Monitor::shared_pointer mon(client->createMonitor(mreq, makeRequest(2)));
+        if(!mon) testAbort("Failed to create monitor");
+
+
+        TestChannelMonitorRequester::shared_pointer mreq2(new TestChannelMonitorRequester);
+        pvd::Monitor::shared_pointer mon2(client->createMonitor(mreq2, makeRequest(2)));
+        if(!mon2) testAbort("Failed to create monitor2");
+
+        testOk1(mreq->eventCnt==0);
+        testOk1(mreq2->eventCnt==0);
+        testOk1(mon->start().isSuccess());
+        testOk1(mon2->start().isSuccess());
+        upstream->dispatch(); // trigger monitorEvent() from upstream to gateway
+
+        testOk1(mreq->eventCnt==1);
+        testOk1(mreq2->eventCnt==1);
+
+        pvd::MonitorElementPtr elem(mon->poll());
+        pvd::MonitorElementPtr elem2(mon2->poll());
+        testOk1(!!elem.get());
+        testOk1(!!elem2.get());
+        testOk1(elem!=elem2);
+        testOk1(elem && elem->pvStructurePtr->getSubFieldT<pvd::PVInt>("x")->get()==1);
+        testOk1(elem2 && elem2->pvStructurePtr->getSubFieldT<pvd::PVInt>("x")->get()==1);
+
+        if(elem) mon->release(elem);
+        if(elem2) mon2->release(elem2);
     }
 };
 
+template<class C, void (C::*M)()>
+void test_method(const char *kname, const char *mname)
+{
+    try {
+        testDiag("------- %s::%s --------", kname, mname);
+        C inst;
+        (inst.*M)();
+    } catch(std::exception& e) {
+        PRINT_EXCEPTION(e);
+        testAbort("unexpected exception: %s", e.what());
+    }
+}
+
+#define TEST_METHOD(klass, method) test_method<klass, &klass::method>(#klass, #method)
+
 } // namespace
 
 MAIN(testmon)
 {
     testPlan(0);
+    TEST_METHOD(TestMonitor, test_event);
+    TEST_METHOD(TestMonitor, test_share);
     return testDone();
 }

testApp/testtest.cpp

@@ -16,21 +16,6 @@ namespace pvd = epics::pvData;
 namespace pva = epics::pvAccess;
 
 namespace {
-template<typename T>
-struct ScalarAccessor {
-    pvd::PVScalar::shared_pointer field;
-    typedef T value_type;
-    ScalarAccessor(const pvd::PVStructurePtr& s, const char *name)
-        :field(s->getSubFieldT<pvd::PVScalar>(name))
-    {}
-    operator value_type() {
-        return field->getAs<T>();
-    }
-    ScalarAccessor& operator=(T v) {
-        field->putFrom<T>(v);
-        return *this;
-    }
-};
 
 void testmonitor()
 {

testApp/utilities.cpp

@@ -8,12 +8,18 @@ typedef epicsGuardRelease<epicsMutex> UnGuard;
 namespace pvd = epics::pvData;
 namespace pva = epics::pvAccess;
 
+namespace {
+int debugdebug = getenv("TEST_EXTRA_DEBUG")!=NULL;
+}
+
 TestChannelRequester::TestChannelRequester()
     :laststate(pva::Channel::NEVER_CONNECTED)
 {}
 
 void TestChannelRequester::channelCreated(const pvd::Status& status, pva::Channel::shared_pointer const & channel)
 {
+    if(debugdebug)
+        testDiag("channelCreated %s", channel->getChannelName().c_str());
     Guard G(lock);
     laststate = pva::Channel::CONNECTED;
     this->status = status;
@@ -24,6 +30,8 @@ void TestChannelRequester::channelCreated(const pvd::Status& status, pva::Channe
 void TestChannelRequester::channelStateChange(pva::Channel::shared_pointer const & channel,
                                               pva::Channel::ConnectionState connectionState)
 {
+    if(debugdebug)
+        testDiag("channelStateChange %s %d", channel->getChannelName().c_str(), (int)connectionState);
     Guard G(lock);
     laststate = connectionState;
     wait.trigger();
@@ -60,6 +68,8 @@ void TestChannelMonitorRequester::monitorConnect(pvd::Status const & status,
                                                  pvd::MonitorPtr const & monitor,
                                                  pvd::StructureConstPtr const & structure)
 {
+    if(debugdebug)
+        testDiag("monitorConnect %p %d", monitor.get(), (int)status.isSuccess());
     Guard G(lock);
     connectStatus = status;
     dtype = structure;
@@ -69,6 +79,8 @@ void TestChannelMonitorRequester::monitorConnect(pvd::Status const & status,
 
 void TestChannelMonitorRequester::monitorEvent(pvd::MonitorPtr const & monitor)
 {
+    if(debugdebug)
+        testDiag("monitorEvent %p", monitor.get());
     mon = monitor;
     eventCnt++;
     wait.trigger();
@@ -76,6 +88,8 @@ void TestChannelMonitorRequester::monitorEvent(pvd::MonitorPtr const & monitor)
 
 void TestChannelMonitorRequester::unlisten(pvd::MonitorPtr const & monitor)
 {
+    if(debugdebug)
+        testDiag("unlisten %p", monitor.get());
     Guard G(lock);
     unlistend = true;
     wait.trigger();
@@ -210,6 +224,8 @@ TestPVChannel::createMonitor(
         monitors.insert(ret);
         static_cast<TestPVMonitor*>(ret.get())->weakself = ret; // save wrapped weak ref
     }
+    if(debugdebug)
+        testDiag("TestPVChannel::createMonitor %s %p", pv->name.c_str(), ret.get());
     requester->monitorConnect(pvd::Status(), ret, pv->dtype);
     return ret;
 }
@@ -261,14 +277,21 @@ void TestPVMonitor::destroy()
 
 pvd::Status TestPVMonitor::start()
 {
-    bool wake;
+    if(debugdebug)
+        testDiag("TestPVMonitor::start %p", this);
     {
         Guard G(channel->pv->provider->lock);
         if(finalize && buffer.empty())
             return pvd::Status();
 
-        if(!running) {
-            wake = running = needWakeup = true;
+        if(running)
+            return pvd::Status();
+        running = true;
+
+        if(this->buffer.empty()) {
+            needWakeup = true;
+            if(debugdebug)
+                testDiag(" need wakeup");
         }
 
         if(!this->free.empty()) {
@@ -287,14 +310,24 @@ pvd::Status TestPVMonitor::start()
 
             buffer.push_back(monitorElement);
             this->free.pop_front();
+            if(debugdebug)
+                testDiag(" push current");
+
+        } else {
+            inoverflow = true;
+            changedMask.clear();
+            changedMask.set(0);
+            if(debugdebug)
+                testDiag(" push overflow");
         }
     }
-    (void)wake; // todo notify?
     return pvd::Status();
 }
 
 pvd::Status TestPVMonitor::stop()
 {
+    if(debugdebug)
+        testDiag("TestPVMonitor::stop %p", this);
     Guard G(channel->pv->provider->lock);
     running = false;
     return pvd::Status();
@@ -308,14 +341,15 @@ pvd::MonitorElementPtr TestPVMonitor::poll()
         ret = buffer.front();
         buffer.pop_front();
     }
-    testDiag("pop %p", ret.get());
+    if(debugdebug)
+        testDiag("TestPVMonitor::poll %p %p", this, ret.get());
     return ret;
 }
 
 void TestPVMonitor::release(pvd::MonitorElementPtr const & monitorElement)
 {
     Guard G(channel->pv->provider->lock);
-    testDiag("release %p", monitorElement.get());
+    testDiag("TestPVMonitor::release %p %p", this, monitorElement.get());
 
     if(inoverflow) {
         assert(!buffer.empty());
@@ -329,7 +363,7 @@ void TestPVMonitor::release(pvd::MonitorElementPtr const & monitorElement)
         overflowMask.clear();
 
         buffer.push_back(monitorElement);
-        testDiag("overflow resume %p", monitorElement.get());
+        testDiag("TestPVMonitor::overflow resume %p %p", this, monitorElement.get());
         inoverflow = false;
     } else {
         this->free.push_back(monitorElement);
@@ -348,6 +382,8 @@ TestPV::TestPV(const std::string& name,
 
 void TestPV::post(const pvd::BitSet& changed, bool notify)
 {
+    if(debugdebug)
+        testDiag("post %s %d", name.c_str(), (int)notify);
     Guard G(provider->lock);
     channels_t::vector_type toupdate(channels.lock_vector());
 
@@ -364,12 +400,12 @@ void TestPV::post(const pvd::BitSet& changed, bool notify)
                 mon->inoverflow = true;
                 mon->overflowMask.or_and(mon->changedMask, changed); // oflow = prev_changed & new_changed
                 mon->changedMask |= changed;
-                testDiag("overflow");
+                if(debugdebug) testDiag("overflow");
 
             } else {
 
-                if(!mon->needWakeup)
-                    mon->needWakeup = mon->buffer.empty();
+                if(mon->buffer.empty())
+                    mon->needWakeup = true;
 
                 AUTO_REF(elem, mon->free.front());
                 elem->pvStructurePtr->copyUnchecked(*value);
@@ -378,10 +414,12 @@ void TestPV::post(const pvd::BitSet& changed, bool notify)
 
                 mon->buffer.push_back(elem);
                 mon->free.pop_front();
-                testDiag("push %p", elem.get());
+                if(debugdebug) testDiag("push %p", elem.get());
             }
 
             if(mon->needWakeup && notify) {
+                if(debugdebug) testDiag(" wakeup");
+                mon->needWakeup = false;
                 UnGuard U(G);
                 mon->requester->monitorEvent(*it2);
             }
@@ -466,6 +504,8 @@ TestProvider::createChannel(std::string const & channelName,
     } else {
         requester->channelCreated(pvd::Status(pvd::Status::STATUSTYPE_ERROR, "PV not found"), ret);
     }
+    if(debugdebug)
+        testDiag("createChannel %s %p", channelName.c_str(), ret.get());
     return ret;
 }
 
@@ -477,3 +517,42 @@ TestProvider::addPV(const std::string& name, const pvd::StructureConstPtr& tdef)
     pvs.insert(name, ret);
     return ret;
 }
+
+void TestProvider::dispatch()
+{
+    Guard G(lock);
+    if(debugdebug)
+        testDiag("TestProvider::dispatch");
+
+    pvs_t::lock_vector_type allpvs(pvs.lock_vector());
+    FOREACH(pvit, pvend, allpvs)
+    {
+        TestPV *pv = pvit->second.get();
+        TestPV::channels_t::vector_type channels(pv->channels.lock_vector());
+
+        FOREACH(chit, chend, channels)
+        {
+            TestPVChannel *chan = chit->get();
+            TestPVChannel::monitors_t::vector_type monitors(chan->monitors.lock_vector());
+
+            if(!chan->isConnected())
+                continue;
+
+            FOREACH(monit, monend, monitors)
+            {
+                TestPVMonitor *mon = monit->get();
+
+                if(mon->finalize || !mon->running)
+                    continue;
+
+                if(mon->needWakeup) {
+                    if(debugdebug)
+                        testDiag("  wakeup monitor %p", mon);
+                    mon->needWakeup = false;
+                    UnGuard U(G);
+                    mon->requester->monitorEvent(*monit);
+                }
+            }
+        }
+    }
+}

testApp/utilities.h

@@ -23,6 +23,23 @@ struct TestProvider;
         testDiag("%s : " #NAME "(%p) : %s", epics::pvData::getMessageTypeName(messageType).c_str(), this, message.c_str()); \
     }
 
+// Boilerplate reduction for accessing a scalar field
+template<typename T>
+struct ScalarAccessor {
+    epics::pvData::PVScalar::shared_pointer field;
+    typedef T value_type;
+    ScalarAccessor(const epics::pvData::PVStructurePtr& s, const char *name)
+        :field(s->getSubFieldT<epics::pvData::PVScalar>(name))
+    {}
+    operator value_type() {
+        return field->getAs<T>();
+    }
+    ScalarAccessor& operator=(T v) {
+        field->putFrom<T>(v);
+        return *this;
+    }
+};
+
 struct TestChannelRequester : public epics::pvAccess::ChannelRequester
 {
     POINTER_DEFINITIONS(TestChannelRequester);
@@ -197,6 +214,8 @@ struct TestProvider : public epics::pvAccess::ChannelProvider, std::tr1::enable_
 
     TestPV::shared_pointer addPV(const std::string& name, const epics::pvData::StructureConstPtr& tdef);
 
+    void dispatch();
+
     epicsMutex lock;
     typedef weak_value_map<std::string, TestPV> pvs_t;
     pvs_t pvs;