testpvalink avoid false positive races

pdbApp/pv/qsrv.h

@@ -19,12 +19,16 @@
 extern "C" {
 #endif
 
+struct link; /* aka. DBLINK from link.h */
+
 /** returns QSRV_VERSION_INT captured at compilation time */
 epicsShareExtern unsigned qsrvVersion(void);
 
 /** returns QSRV_ABI_VERSION_INT captured at compilation time */
 epicsShareExtern unsigned qsrvABIVersion(void);
 
+epicsShareFunc void testqsrvWaitForLinkEvent(struct link *plink);
+
 /** Call before testIocShutdownOk()
  @code
    testdbPrepare();

pdbApp/pvalink.cpp

@@ -138,6 +138,23 @@ void testqsrvCleanup(void)
     }
 }
 
+void testqsrvWaitForLinkEvent(struct link *plink)
+{
+    std::tr1::shared_ptr<pvaLinkChannel> lchan;
+    {
+        DBScanLocker lock(plink->precord);
+
+        if(plink->type!=JSON_LINK || !plink->value.json.jlink || plink->value.json.jlink->pif!=&lsetPVA) {
+            testAbort("Not a PVA link");
+        }
+        pvaLink *pval = static_cast<pvaLink*>(plink->value.json.jlink);
+        lchan = pval->lchan;
+    }
+    if(lchan) {
+        lchan->run_done.wait();
+    }
+}
+
 static
 void installPVAAddLinkHook()
 {

pdbApp/pvalink.h

@@ -121,6 +121,7 @@ struct pvaLinkChannel : public pvac::ClientChannel::MonitorCallback,
     static size_t num_instances;
 
     pvd::Mutex lock;
+    epicsEvent run_done; // used by testing code
 
     pvac::ClientChannel chan;
     pvac::Monitor op_mon;

pdbApp/pvalink_channel.cpp

@@ -270,8 +270,13 @@ void pvaLinkChannel::run()
 
         // pop next update from monitor queue.
         // still under lock to safeguard concurrent calls to lset functions
-        if(connected && !op_mon.poll())
+        if(connected && !op_mon.poll()) {
+            TRACE(<<"empty");
+            run_done.signal();
             return; // monitor queue is empty, nothing more to do here
+        }
+
+        TRACE(<<(connected_latched?"connected":"disconnected"));
 
         assert(!connected || !!op_mon.root);
 
@@ -305,7 +310,7 @@ void pvaLinkChannel::run()
 
         // at this point we know we will re-queue, but not immediately
         // so an expected error won't get us stuck in a tight loop.
-        requeue = queued = true;
+        requeue = queued = connected_latched;
 
         if(links_changed) {
             // a link has been added or removed since the last update.
@@ -361,6 +366,8 @@ void pvaLinkChannel::run()
     if(requeue) {
         // re-queue until monitor queue is empty
         pvaGlobal->queue.add(shared_from_this());
+    } else {
+        run_done.signal();
     }
 }
 

pdbApp/pvalink_lset.cpp

@@ -99,7 +99,6 @@ int pvaIsConnected(const DBLINK *plink)
     TRY {
         TRACE(<<plink->precord->name<<" "<<self->channelName);
         Guard G(self->lchan->lock);
-        if(!self->valid()) return -1;
 
         return self->valid();
 
@@ -161,9 +160,10 @@ long pvaGetValue(DBLINK *plink, short dbrType, void *pbuffer,
         long *pnRequest)
 {
     TRY {
-        TRACE(<<plink->precord->name<<" "<<self->channelName);
         Guard G(self->lchan->lock);
 
+        TRACE(<<plink->precord->name<<" "<<self->channelName<<" conn="<<(self->valid()?"T":"F"));
+
         if(!self->valid()) {
             // disconnected
             if(self->ms != pvaLink::NMS) {
@@ -371,10 +371,11 @@ long pvaPutValue(DBLINK *plink, short dbrType,
         const void *pbuffer, long nRequest)
 {
     TRY {
-        TRACE(<<plink->precord->name<<" "<<self->channelName);
         (void)self;
         Guard G(self->lchan->lock);
 
+        TRACE(<<plink->precord->name<<" "<<self->channelName<<" nReq="<<nRequest<<" conn="<<(self->valid()?"T":"F"));
+
         if(nRequest < 0) return -1;
 
         if(!self->valid()) {

testApp/testpvalink.cpp

@@ -1,10 +1,13 @@
 
 #include <dbUnitTest.h>
 #include <testMain.h>
+#include <longinRecord.h>
+#include <longoutRecord.h>
 
 #include <pv/qsrv.h>
 #include "utilities.h"
 #include "pvalink.h"
+#include "pv/qsrv.h"
 
 namespace {
 
@@ -12,40 +15,50 @@ void testGet()
 {
     testDiag("==== testGet ====");
 
+    longinRecord *li1 = (longinRecord*)testdbRecordPtr("src:li1");
+
+    while(!dbIsLinkConnected(&li1->inp))
+        testqsrvWaitForLinkEvent(&li1->inp);
+
     testdbGetFieldEqual("target:li.VAL", DBF_LONG, 42);
-    testdbGetFieldEqual("src:li1.VAL", DBF_LONG, 0);
+
+    testdbGetFieldEqual("src:li1.VAL", DBF_LONG, 0); // value before first process
+
     testdbGetFieldEqual("src:li1.INP", DBF_STRING, "{\"pva\":\"target:li\"}");
 
     testdbPutFieldOk("src:li1.PROC", DBF_LONG, 1);
-    //TODO: wait for dbEvent queue update
-    epicsThreadSleep(0.1);
 
     testdbGetFieldEqual("src:li1.VAL", DBF_LONG, 42);
 
     testdbPutFieldOk("src:li1.INP", DBF_STRING, "{\"pva\":\"target:ai\"}");
 
-    testdbGetFieldEqual("src:li1.VAL", DBF_LONG, 42);
+    while(!dbIsLinkConnected(&li1->inp))
+        testqsrvWaitForLinkEvent(&li1->inp);
+
+    testdbGetFieldEqual("src:li1.VAL", DBF_LONG, 42); // changing link doesn't automatically process
 
-    //TODO: wait for pvalink worker update
-    epicsThreadSleep(0.1);
     testdbPutFieldOk("src:li1.PROC", DBF_LONG, 1);
-    //TODO: wait for dbEvent queue update
-    epicsThreadSleep(0.1);
 
-    testdbGetFieldEqual("src:li1.VAL", DBF_LONG, 4);
+    testdbGetFieldEqual("src:li1.VAL", DBF_LONG, 4); // now it's changed
 }
 
 void testPut()
 {
     testDiag("==== testPut ====");
-    testdbGetFieldEqual("target:li2.VAL", DBF_LONG, 43);
-    testdbGetFieldEqual("src:li2.VAL", DBF_LONG, 0);
-    testdbGetFieldEqual("src:li2.OUT", DBF_STRING, "{\"pva\":\"target:li2\"}");
 
-    testdbPutFieldOk("src:li2.VAL", DBF_LONG, 14);
+    longoutRecord *lo2 = (longoutRecord*)testdbRecordPtr("src:lo2");
+
+    while(!dbIsLinkConnected(&lo2->out))
+        testqsrvWaitForLinkEvent(&lo2->out);
+
+    testdbGetFieldEqual("target:li2.VAL", DBF_LONG, 43);
+    testdbGetFieldEqual("src:lo2.VAL", DBF_LONG, 0);
+    testdbGetFieldEqual("src:lo2.OUT", DBF_STRING, "{\"pva\":\"target:li2\"}");
+
+    testdbPutFieldOk("src:lo2.VAL", DBF_LONG, 14);
 
     testdbGetFieldEqual("target:li2.VAL", DBF_LONG, 14);
-    testdbGetFieldEqual("src:li2.VAL", DBF_LONG, 14);
+    testdbGetFieldEqual("src:lo2.VAL", DBF_LONG, 14);
 }
 
 } // namespace

testApp/testpvalink.db

@@ -16,6 +16,6 @@ record(longin, "target:li2") {
     field(VAL, "43")
 }
 
-record(longout, "src:li2") {
+record(longout, "src:lo2") {
     field(OUT, {pva:"target:li2"})
 }