expand dbEvent synchronization

Changes method of waiting in db_flush_extra_labor_event(),
which also blocks if labor is pending.

Adds testMonitorSync()

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

@@ -557,6 +557,40 @@ static void event_remove ( struct event_que *ev_que,
     pevent->npend--;
 }
 
+/* synchronize with worker thread.
+ *
+ * On return, any previously pending events or extra labor have been handled.
+ *
+ * caller must lock evUser->lock
+ */
+static
+void db_sync_event (struct event_user * const evUser)
+{
+    /* grab current cycle counter, then wait for it to change */
+    epicsUInt32 curSeq = evUser->pflush_seq;
+    event_waiter wait;
+    wait.wake = epicsEventCreate(epicsEventEmpty); /* failure allowed */
+
+    ellAdd(&evUser->waiters, &wait.node);
+    do {
+        epicsMutexUnlock( evUser->lock );
+        /* ensure worker will cycle at least once */
+        epicsEventMustTrigger(evUser->ppendsem);
+
+        if(wait.wake) {
+            epicsEventMustWait(wait.wake);
+        } else {
+            epicsThreadSleep(0.01); /* ick. but better than cantProceed() */
+        }
+
+        epicsMutexMustLock ( evUser->lock );
+    } while(curSeq == evUser->pflush_seq);
+    ellDelete(&evUser->waiters, &wait.node);
+    /* destroy under lock to ensure epicsEventMustTrigger() has returned */
+    if(wait.wake)
+        epicsEventDestroy(wait.wake);
+}
+
 /*
  * DB_CANCEL_EVENT()
  *
@@ -594,34 +628,9 @@ void db_cancel_event (dbEventSubscription event)
     UNLOCKEVQUE (que);
 
     if(sync) {
-        /* cycle through worker */
-        struct event_user *evUser = que->evUser;
-        epicsUInt32 curSeq;
-        event_waiter wait;
-        wait.wake = epicsEventCreate(epicsEventEmpty); /* may fail */
-
-        epicsMutexMustLock ( evUser->lock );
-        ellAdd(&evUser->waiters, &wait.node);
-        /* grab current cycle counter, then wait for it to change */
-        curSeq = evUser->pflush_seq;
-        do {
-            epicsMutexUnlock( evUser->lock );
-            /* ensure worker will cycle at least once */
-            epicsEventMustTrigger(evUser->ppendsem);
-
-            if(wait.wake) {
-                epicsEventMustWait(wait.wake);
-            } else {
-                epicsThreadSleep(0.01); /* ick. but better than cantProceed() */
-            }
-
-            epicsMutexMustLock ( evUser->lock );
-        } while(curSeq == evUser->pflush_seq);
-        ellDelete(&evUser->waiters, &wait.node);
-        /* destroy under lock to ensure epicsEventMustTrigger() has returned */
-        if(wait.wake)
-            epicsEventDestroy(wait.wake);
-        epicsMutexUnlock( evUser->lock );
+        epicsMutexMustLock ( que->evUser->lock );
+        db_sync_event(que->evUser);
+        epicsMutexUnlock( que->evUser->lock );
     }
 }
 
@@ -635,10 +644,10 @@ void db_flush_extra_labor_event (dbEventCtx ctx)
     struct event_user * const evUser = (struct event_user *) ctx;
 
     epicsMutexMustLock ( evUser->lock );
-    while ( evUser->extraLaborBusy ) {
-        epicsMutexUnlock ( evUser->lock );
-        epicsThreadSleep(0.1);
-        epicsMutexMustLock ( evUser->lock );
+    if ( evUser->extraLaborBusy || (evUser->extra_labor && evUser->extralabor_sub) ) {
+        db_sync_event(evUser);
+        // At this point, original labor completed.
+        // Do not wait for any additional labor queued afterwards.
     }
     epicsMutexUnlock ( evUser->lock );
 }
@@ -1027,9 +1036,7 @@ static void event_task (void *pParm)
          * labor to this task
          */
         epicsMutexMustLock ( evUser->lock );
-        evUser->extraLaborBusy = TRUE;
         if ( evUser->extra_labor && evUser->extralabor_sub ) {
-            evUser->extra_labor = FALSE;
             pExtraLaborSub = evUser->extralabor_sub;
             pExtraLaborArg = evUser->extralabor_arg;
         }
@@ -1037,12 +1044,14 @@ static void event_task (void *pParm)
             pExtraLaborSub = NULL;
             pExtraLaborArg = NULL;
         }
+        evUser->extra_labor = FALSE;
         if ( pExtraLaborSub ) {
+            evUser->extraLaborBusy = TRUE;
             epicsMutexUnlock ( evUser->lock );
             (*pExtraLaborSub)(pExtraLaborArg);
             epicsMutexMustLock ( evUser->lock );
+            evUser->extraLaborBusy = FALSE;
         }
-        evUser->extraLaborBusy = FALSE;
 
         for ( ev_que = &evUser->firstque; ev_que; ev_que = ev_que->nextque ) {
             /* unlock during iteration is safe as event_que will not be free'd */

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

@@ -434,6 +434,17 @@ void testMonitorWait(testMonitor *mon)
     }
 }
 
+static void dummylabor(void* unused) {(void)unused;}
+
+void testMonitorSync(testMonitor *mon)
+{
+    // db_flush_extra_labor_event() only blocks if there is actual labor pending
+    (void)db_add_extra_labor_event(testEvtCtx, dummylabor, NULL);
+    (void)db_post_extra_labor(testEvtCtx);
+    db_flush_extra_labor_event(testEvtCtx);
+    (void)db_add_extra_labor_event(testEvtCtx, NULL, NULL);
+}
+
 unsigned testMonitorCount(testMonitor *mon, unsigned reset)
 {
     unsigned count;

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

@@ -139,21 +139,45 @@ DBCORE_API dbCommon* testdbRecordPtr(const char* pv);
 
 typedef struct testMonitor testMonitor;
 
-/** Setup monitoring the named PV for changes */
+/** Setup monitoring the named PV for changes
+ *
+ * @param[in] pvname Requested PV name.  Must be valid for dbChannelCreate().
+ * @param[in] dbe_mask A bitwise or of DBE_VALUE and friends.
+ * @param[in] opt Currently unused.  Set to zero.
+ * @returns Newly allocated testMonitor object, which caller must testMonitorDestroy()
+ *
+ * Calls testAbort() on failure.  Will never return NULL.
+ *
+ * @since 3.16.0.1
+ */
 DBCORE_API testMonitor* testMonitorCreate(const char* pvname, unsigned dbe_mask, unsigned opt);
-/** Stop monitoring */
+/** Stop monitoring
+ *
+ * @since 3.16.0.1
+ */
 DBCORE_API void testMonitorDestroy(testMonitor*);
 /** Return immediately if it has been updated since create, last wait,
  * or reset (count w/ reset=1).
  * Otherwise, block until the value of the target PV is updated.
+ *
+ * @since 3.16.0.1
  */
 DBCORE_API void testMonitorWait(testMonitor*);
+/** Synchronize with dbEvent working for subscription.
+ *
+ * On return, any updates previously posted for this subscriptions have been delivered.
+ *
+ * @since UNRELEASED
+ */
+DBCORE_API void testMonitorSync(testMonitor*);
 /** Return the number of monitor events which have occured since create,
  * or a previous reset (called reset=1).
  * Calling w/ reset=0 only returns the count.
  * Calling w/ reset=1 resets the count to zero and ensures that the next
  * wait will block unless subsequent events occur.  Returns the previous
  * count.
+ *
+ * @since 3.16.0.1
  */
 DBCORE_API unsigned testMonitorCount(testMonitor*, unsigned reset);