dbCa: iocInit wait for local CA links to connect

documentation/new-notes/PR-768.md

@@ -0,0 +1,8 @@
+### iocInit wait for local CA links to connect
+
+PR [768](https://github.com/epics-base/epics-base/pull/768)
+
+During iocInit(), wait for local CA links to connect.
+With this change, database authors can be certain that
+any local CA link is connected prior to `initHookAfterIocRunning`
+and `field(PINI, RUNNING)`.

modules/database/src/ioc/db/dbCa.c

@@ -40,6 +40,7 @@
 /* We can't include dbStaticLib.h here */
 #define dbCalloc(nobj,size) callocMustSucceed(nobj,size,"dbCalloc")
 
+#include <epicsAtomic.h>
 #include "db_access_routines.h"
 #include "dbCa.h"
 #include "dbCaPvt.h"
@@ -64,6 +65,7 @@ extern int dbServiceIsolate;
 static ELLLIST workList = ELLLIST_INIT;    /* Work list for dbCaTask */
 static epicsMutexId workListLock; /*Mutual exclusions semaphores for workList*/
 static epicsEventId workListEvent; /*wakeup event for dbCaTask*/
+static size_t initOutstanding; // one count for every link with DBCA_CALLBACK_INIT_WAIT
 static int removesOutstanding = 0;
 #define removesOutstandingWarning 10000
 
@@ -351,7 +353,7 @@ static void dbCaLinkInitImpl(int isolate)
     dbCaCtl = ctlPause;
 
     dbCaWorker = epicsThreadCreateOpt("dbCaLink", dbCaTask, NULL, &opts);
-    /* wait for worker to startup and initialize dbCaClientContext */
+    /* wait for worker to startup, initialize dbCaClientContext, and connect local CA */
     epicsEventMustWait(startStopEvent);
 }
 
@@ -370,6 +372,10 @@ void dbCaRun(void)
     if (dbCaCtl == ctlPause) {
         dbCaCtl = ctlRun;
         epicsEventSignal(workListEvent);
+
+        while(epicsAtomicGetSizeT(&initOutstanding)) {
+            epicsEventMustWait(startStopEvent);
+        }
     }
 }
 
@@ -380,11 +386,12 @@ void dbCaPause(void)
         epicsEventSignal(workListEvent);
     }
 }
-
-void dbCaAddLinkCallback(struct link *plink,
-    dbCaCallback connect, dbCaCallback monitor, void *userPvt)
+void dbCaAddLinkCallbackOpt(struct dbLocker *locker, struct link *plink,
+                            dbCaCallback connect, dbCaCallback monitor, void *userPvt,
+                            unsigned flags)
 {
     caLink *pca;
+    (void)locker; /* Passed for symmetry with dbDbAddLink().  So far unused. */
 
     assert(!plink->value.pv_link.pvt);
 
@@ -396,6 +403,10 @@ void dbCaAddLinkCallback(struct link *plink,
     pca->connect = connect;
     pca->monitor = monitor;
     pca->userPvt = userPvt;
+    pca->flags = flags;
+
+    if(flags & DBCA_CALLBACK_INIT_WAIT)
+        epicsAtomicIncrSizeT(&initOutstanding);
 
     epicsMutexMustLock(pca->lock);
     plink->lset = &dbCa_lset;
@@ -405,15 +416,22 @@ void dbCaAddLinkCallback(struct link *plink,
     epicsMutexUnlock(pca->lock);
 }
 
+void dbCaAddLinkCallback(struct link *plink,
+    dbCaCallback connect, dbCaCallback monitor, void *userPvt)
+{
+    dbCaAddLinkCallbackOpt(NULL, plink, connect, monitor, userPvt, 0);
+}
+
 long dbCaAddLink(struct dbLocker *locker, struct link *plink, short dbfType)
 {
-    dbCaAddLinkCallback(plink, 0, 0, NULL);
+    dbCaAddLinkCallbackOpt(locker, plink, 0, 0, NULL, 0);
     return 0;
 }
 
 void dbCaRemoveLink(struct dbLocker *locker, struct link *plink)
 {
     caLink *pca = (caLink *)plink->value.pv_link.pvt;
+    (void)locker; /* Passed for symmetry with dbDbRemoveLink().  So far unused. */
 
     if (!pca) return;
     epicsMutexMustLock(pca->lock);
@@ -841,6 +859,8 @@ static void connectionCallback(struct connection_handler_args arg)
     }
     pca->hasReadAccess = ca_read_access(arg.chid);
     pca->hasWriteAccess = ca_write_access(arg.chid);
+    if (pca->flags & DBCA_CALLBACK_INIT_WAIT)
+        link_action |= CA_INIT_READY; // declare initialized, unless cleared below
 
     if (pca->gotFirstConnection) {
         if (pca->nelements != ca_element_count(arg.chid) ||
@@ -875,9 +895,11 @@ static void connectionCallback(struct connection_handler_args arg)
     pca->dbrType = ca_field_type(arg.chid);
     if ((plink->value.pv_link.pvlMask & pvlOptInpNative) && !pca->pgetNative) {
         link_action |= CA_MONITOR_NATIVE;
+        link_action &= ~CA_INIT_READY; // defer to eventCallback()
     }
     if ((plink->value.pv_link.pvlMask & pvlOptInpString) && !pca->pgetString) {
         link_action |= CA_MONITOR_STRING;
+        link_action &= ~CA_INIT_READY; // defer to eventCallback()
     }
     if ((plink->value.pv_link.pvlMask & pvlOptOutNative) && pca->gotOutNative) {
         link_action |= CA_WRITE_NATIVE;
@@ -889,6 +911,7 @@ static void connectionCallback(struct connection_handler_args arg)
     if (pca->dbrType != DBR_STRING) {
         /* will run connect() callback later */
         link_action |= CA_GET_ATTRIBUTES;
+        link_action &= ~CA_INIT_READY; // defer to getAttribEventCallback()
     } else {
         connect = pca->connect;
         userPvt = pca->userPvt;
@@ -909,6 +932,7 @@ static void eventCallback(struct event_handler_args arg)
     dbCaCallback monitor = 0;
     void *userPvt = 0;
     int doScan = 1;
+    short link_action = 0;
 
     assert(pca);
     epicsMutexMustLock(pca->lock);
@@ -969,10 +993,13 @@ static void eventCallback(struct event_handler_args arg)
         if ((ppv_link->pvlMask & pvlOptCP) ||
             ((ppv_link->pvlMask & pvlOptCPP) && precord->scan == 0))
         {
-            addAction(pca, CA_DBPROCESS);
+            link_action |= CA_DBPROCESS;
         }
     }
+    if (pca->flags & DBCA_CALLBACK_INIT_WAIT)
+        link_action |= CA_INIT_READY;
 done:
+    if (link_action) addAction(pca, link_action);
     epicsMutexUnlock(pca->lock);
     if (monitor) monitor(userPvt);
 }
@@ -1096,6 +1123,8 @@ static void getAttribEventCallback(struct event_handler_args arg)
     pca->alarmLimits[3] = pdbr->upper_alarm_limit;
     pca->precision = pdbr->precision;
     memcpy(pca->units, pdbr->units, MAX_UNITS_SIZE);
+    if (pca->flags & DBCA_CALLBACK_INIT_WAIT)
+        addAction(pca, CA_INIT_READY);
     epicsMutexUnlock(pca->lock);
     if (getAttributes) getAttributes(getAttributesPvt);
     if (connect) connect(userPvt);
@@ -1110,7 +1139,7 @@ static void dbCaTask(void *arg)
     dbCaClientContext = ca_current_context ();
     SEVCHK(ca_add_exception_event(exceptionCallback,NULL),
         "ca_add_exception_event");
-    epicsEventSignal(startStopEvent);
+    epicsEventSignal(startStopEvent); /* unblock dbCaLinkInitImpl() */
 
     /* channel access event loop */
     while (TRUE){
@@ -1266,6 +1295,13 @@ static void dbCaTask(void *arg)
                 db_process(prec);
                 dbScanUnlock(prec);
             }
+            if ((link_action & CA_INIT_READY) && (pca->flags & DBCA_CALLBACK_INIT_WAIT)) {
+                pca->flags &= ~DBCA_CALLBACK_INIT_WAIT;
+                if (epicsAtomicDecrSizeT(&initOutstanding)==0)
+                {
+                    epicsEventSignal(startStopEvent); /* unblock dbCaRun() */
+                }
+            }
         }
         SEVCHK(ca_flush_io(), "dbCaTask");
     }
@@ -1275,5 +1311,5 @@ shutdown:
         ca_context_destroy();
     else
         fprintf(stderr, "dbCa: chan_count = %d at shutdown\n", dbca_chan_count);
-    epicsEventSignal(startStopEvent);
+    epicsEventSignal(startStopEvent); /* unblock dbCaShutdown() */
 }

modules/database/src/ioc/db/dbCaPvt.h

@@ -21,6 +21,19 @@
 #include "epicsMutex.h"
 #include "epicsTypes.h"
 #include "link.h"
+#include "shareLib.h"
+#include "libCaAPI.h"
+
+#ifndef INC_cadef_H
+/* Copy some definitions so this header to be included from
+ * places where cadef.h and db_access.h can not.
+ */
+typedef void * chid;
+typedef void * evid;
+LIBCA_API extern const unsigned short dbr_value_size[];
+LIBCA_API short epicsShareAPI ca_field_type (chid chan);
+#define MAX_UNITS_SIZE          8
+#endif
 
 /* link_action mask */
 #define CA_CLEAR_CHANNEL        0x1
@@ -32,6 +45,7 @@
 #define CA_GET_ATTRIBUTES       0x40
 #define CA_SYNC                 0x1000
 #define CA_DBPROCESS            0x2000
+#define CA_INIT_READY           0x4000
 /* write type */
 #define CA_PUT          0x1
 #define CA_PUT_CALLBACK 0x2
@@ -63,6 +77,7 @@ typedef struct caLink
     dbCaCallback    connect;
     dbCaCallback    monitor;
     void            *userPvt;
+    unsigned        flags;
     /* The following are for write request */
     short           putType;
     dbCaCallback    putCallback;
@@ -97,4 +112,10 @@ typedef struct caLink
     unsigned long   nUpdate;
 }caLink;
 
+#define DBCA_CALLBACK_INIT_WAIT (1)
+
+void dbCaAddLinkCallbackOpt(struct dbLocker *locker, struct link *plink,
+                            dbCaCallback connect, dbCaCallback monitor,
+                            void *userPvt, unsigned flags);
+
 #endif /* INC_dbCaPvt_H */

modules/database/src/ioc/db/dbLink.c

@@ -31,7 +31,7 @@
 #include "dbAccessDefs.h"
 #include "dbAddr.h"
 #include "dbBase.h"
-#include "dbCa.h"
+#include "dbCaPvt.h"
 #include "dbCommon.h"
 #include "dbConstLink.h"
 #include "dbDbLink.h"
@@ -125,7 +125,9 @@ void dbInitLink(struct link *plink, short dbfType)
     if (dbfType == DBF_INLINK)
         plink->value.pv_link.pvlMask |= pvlOptInpNative;
 
-    dbCaAddLink(NULL, plink, dbfType);
+    int isLocal = dbChannelTest(plink->value.pv_link.pvname)==0;
+
+    dbCaAddLinkCallbackOpt(NULL, plink, NULL, NULL, NULL, isLocal ? DBCA_CALLBACK_INIT_WAIT : 0);
     if (dbfType == DBF_FWDLINK) {
         char *pperiod = strrchr(plink->value.pv_link.pvname, '.');
 

modules/database/test/ioc/db/dbCaLinkTest.c

@@ -29,13 +29,6 @@
 #include "dbEvent.h"
 #include "shareLib.h"
 
-/* Declarations from cadef.h and db_access.h which we can't include here */
-typedef void * chid;
-typedef void * evid;
-epicsShareExtern const unsigned short dbr_value_size[];
-epicsShareExtern short epicsShareAPI ca_field_type (chid chan);
-#define MAX_UNITS_SIZE          8
-
 #include "dbCaPvt.h"
 #include "errlog.h"
 #include "testMain.h"

modules/database/test/std/rec/regressTest.c

@@ -193,9 +193,8 @@ void testLinkSevr(void)
 
     startRegressTestIoc("regressLinkSevr.db");
 
-    /* wait for CA links to connect and receive an initial update */
-    testdbCaWaitForUpdateCount(dbGetDevLink(testdbRecordPtr("si2")), 1);
-    testdbCaWaitForUpdateCount(dbGetDevLink(testdbRecordPtr("li2")), 1);
+    testOk1(dbIsLinkConnected(dbGetDevLink(testdbRecordPtr("si2"))));
+    testOk1(dbIsLinkConnected(dbGetDevLink(testdbRecordPtr("li2"))));
 
     chan = dbChannelCreate("ai.SEVR");
     if(!chan)
@@ -228,7 +227,7 @@ void testLinkSevr(void)
 
 MAIN(regressTest)
 {
-    testPlan(68);
+    testPlan(70);
     testArrayLength1();
     testHexConstantLinks();
     testLinkMS();