fixed a disconnect related problem and also improved

structure of locking

src/ca/CASG.cpp

@@ -88,15 +88,10 @@ int CASG::block ( double timeout )
     beg_time = cur_time;
     delay = 0.0;
 
-    this->client.enableCallbackPreemption ();
-
     while ( 1 ) {
-        {
-            epicsAutoMutex locker ( this->mutex );
-            if ( this->ioList.count() == 0u ) {
-                status = ECA_NORMAL;
-                break;
-            }
+        if ( this->ioList.count() == 0u ) {
+            status = ECA_NORMAL;
+            break;
         }
 
         remaining = timeout - delay;
@@ -110,7 +105,7 @@ int CASG::block ( double timeout )
             break;
         }
 
-        this->sem.wait ( remaining );
+        this->client.blockForEventAndEnableCallbacks ( this->sem, remaining );
 
         /*
          * force a time update 
@@ -120,8 +115,6 @@ int CASG::block ( double timeout )
         delay = cur_time - beg_time;
     }
 
-    this->client.disableCallbackPreemption ();
-
     return status;
 }
 

src/ca/cac.cpp

@@ -354,6 +354,12 @@ void cac::processRecvBacklog ()
 void cac::flushRequest ()
 {
     epicsAutoMutex autoMutex ( this->mutex );
+    this->flushRequestPrivate ();
+}
+
+// lock must be applied
+void cac::flushRequestPrivate ()
+{
     tsDLIterBD <tcpiiu> piiu = this->iiuList.firstIter ();
     while ( piiu.valid () ) {
         piiu->flushRequest ();
@@ -527,10 +533,6 @@ int cac::pendIO ( const double &timeout )
         return ECA_EVDISALLOW;
     }
 
-    this->enableCallbackPreemption ();
-
-    this->flushRequest ();
-   
     int status = ECA_NORMAL;
     epicsTime beg_time = epicsTime::getCurrent ();
     double remaining;
@@ -540,20 +542,29 @@ int cac::pendIO ( const double &timeout )
     else{
         remaining = timeout;
     }    
-    while ( this->pndRecvCnt > 0 ) {
-        if ( remaining < CAC_SIGNIFICANT_SELECT_DELAY ) {
-            status = ECA_TIMEOUT;
-            break;
-        }
-        this->ioDone.wait ( remaining );
-        if ( timeout != 0.0 ) {
-            double delay = epicsTime::getCurrent () - beg_time;
-            remaining = timeout - delay;
-        }    
-    }
 
     {
         epicsAutoMutex autoMutex ( this->mutex );
+
+        this->flushRequestPrivate ();
+   
+        while ( this->pndRecvCnt > 0 ) {
+            if ( remaining < CAC_SIGNIFICANT_SELECT_DELAY ) {
+                status = ECA_TIMEOUT;
+                break;
+            }
+            this->enableCallbackPreemption ();
+            {
+                epicsAutoMutexRelease autoRelease ( this->mutex );
+                this->ioDone.wait ( remaining );
+            }
+            this->disableCallbackPreemption ();
+            if ( timeout != 0.0 ) {
+                double delay = epicsTime::getCurrent () - beg_time;
+                remaining = timeout - delay;
+            }    
+        }
+
         this->readSeq++;
         this->pndRecvCnt = 0u;
         if ( this->pudpiiu ) {
@@ -561,11 +572,20 @@ int cac::pendIO ( const double &timeout )
         }
     }
 
-    this->disableCallbackPreemption ();
-
     return status;
 }
 
+void cac::blockForEventAndEnableCallbacks ( epicsEvent &event, double timeout )
+{
+    epicsAutoMutex autoMutex ( this->mutex );
+    this->enableCallbackPreemption ();
+    {
+        epicsAutoMutexRelease autoMutexRelease ( this->mutex );
+        event.wait ( timeout );
+    }
+    this->disableCallbackPreemption ();
+}
+
 int cac::pendEvent ( const double &timeout )
 {
     // prevent recursion nightmares by disabling calls to 
@@ -574,20 +594,27 @@ int cac::pendEvent ( const double &timeout )
         return ECA_EVDISALLOW;
     }
 
-    this->enableCallbackPreemption ();
+    {
+        epicsAutoMutex autoMutex ( this->mutex );
 
-    this->flushRequest ();
+        this->flushRequestPrivate ();
 
-    if ( timeout == 0.0 ) {
-        while ( true ) {
-            epicsThreadSleep ( 60.0 );
+        this->enableCallbackPreemption ();
+
+        {
+            epicsAutoMutexRelease autoMutexRelease ( this->mutex );
+            if ( timeout == 0.0 ) {
+                while ( true ) {
+                    epicsThreadSleep ( 60.0 );
+                }
+            }
+            else if ( timeout >= CAC_SIGNIFICANT_SELECT_DELAY ) {
+                epicsThreadSleep ( timeout );
+            }
         }
-    }
-    else if ( timeout >= CAC_SIGNIFICANT_SELECT_DELAY ) {
-        epicsThreadSleep ( timeout );
-    }
 
-    this->disableCallbackPreemption ();
+        this->disableCallbackPreemption ();
+    }
 
     return ECA_TIMEOUT;
 }
@@ -760,56 +787,44 @@ bool cac::setupUDP ()
     return true;
 }
 
+// lock must already be applied
 void cac::enableCallbackPreemption ()
 {
-    unsigned copy;
-    {
-        epicsAutoMutex autoMutex ( this->mutex );
-        assert ( this->recvProcessEnableRefCount < UINT_MAX );
-        copy = this->recvProcessEnableRefCount;
-        this->recvProcessEnableRefCount++;
-    }
-    if ( copy == 0u ) {
+    assert ( this->recvProcessEnableRefCount < UINT_MAX );
+    this->recvProcessEnableRefCount++;
+    if ( this->recvProcessEnableRefCount == 1u ) {
         this->recvProcessActivityEvent.signal ();
     }
 }
 
+// lock must already be applied
 void cac::disableCallbackPreemption ()
 {
-    bool wakeupNeeded;
-
-    {
-        epicsAutoMutex autoMutex ( this->mutex );
-
+    if ( ! this->recvProcessInProgress ) {
+        assert ( this->recvProcessEnableRefCount != 0u );
+        this->recvProcessEnableRefCount--;
+        return;
+    }
+    else {
+        this->recvProcessCompletionNBlockers++;
+    }
+    
+    while ( true ) {
+        {
+            epicsAutoMutexRelease autoMutexRelease ( this->mutex );
+            this->recvProcessCompletion.wait ();
+        }
+        
         if ( ! this->recvProcessInProgress ) {
-            assert ( this->recvProcessEnableRefCount != 0u );
+            assert ( this->recvProcessEnableRefCount > 0u );
             this->recvProcessEnableRefCount--;
+            assert ( this->recvProcessCompletionNBlockers > 0u );
+            this->recvProcessCompletionNBlockers--;
+            if ( this->recvProcessCompletionNBlockers > 0u ) {
+                this->recvProcessCompletion.signal ();
+            }
             return;
         }
-        else {
-            this->recvProcessCompletionNBlockers++;
-        }
-    }
-
-    while ( true ) {
-        this->recvProcessCompletion.wait ();
-
-        {
-            epicsAutoMutex autoMutex ( this->mutex );
-
-            if ( ! this->recvProcessInProgress ) {
-                assert ( this->recvProcessEnableRefCount > 0u );
-                this->recvProcessEnableRefCount--;
-                assert ( this->recvProcessCompletionNBlockers > 0u );
-                this->recvProcessCompletionNBlockers--;
-                wakeupNeeded = this->recvProcessCompletionNBlockers > 0u;
-                break;
-            }
-        }
-    }
-
-    if ( wakeupNeeded ) {
-        this->recvProcessCompletion.signal ();
     }
 }
 
@@ -957,9 +972,14 @@ void cac::flushIfRequired ( nciu &chan )
                 blockPermit = false;
             }
         }
-        this->flushRequest ();
+        this->flushRequestPrivate ();
         if ( blockPermit ) {
+            // enable / disable of call back preemption must occur here
+            // because the tcpiiu might disconnect while waiting and its
+            // pointer to this cac might become invalid
+            this->enableCallbackPreemption ();
             chan.getPIIU()->blockUntilSendBacklogIsReasonable ( this->mutex );
+            this->disableCallbackPreemption ();
         }
     }
     else {

src/ca/cac.h

@@ -111,8 +111,7 @@ public:
         const char *pFileName, unsigned lineNo );
 
     // callback preemption control
-    void enableCallbackPreemption ();
-    void disableCallbackPreemption ();
+    void blockForEventAndEnableCallbacks ( epicsEvent &event, double timeout );
 
     // diagnostics
     unsigned connectionCount () const;
@@ -189,8 +188,11 @@ private:
     bool                    recvProcessInProgress;
     bool                    recvProcessThreadExitRequest;
 
+    void flushRequestPrivate ();
     void run ();
     bool setupUDP ();
+    void enableCallbackPreemption ();
+    void disableCallbackPreemption ();
     void flushIfRequired ( nciu & ); // lock must be applied
     void recycleReadNotifyIO ( netReadNotifyIO &io );
     void recycleWriteNotifyIO ( netWriteNotifyIO &io );

src/ca/oldAccess.h

@@ -212,8 +212,7 @@ public:
     CASG * lookupCASG ( unsigned id );
     void installCASG ( CASG & );
     void uninstallCASG ( CASG & );
-    void enableCallbackPreemption ();
-    void disableCallbackPreemption ();
+    void blockForEventAndEnableCallbacks ( epicsEvent &event, double timeout );
 // perhaps these should be eliminated in deference to the exception mechanism
     int printf ( const char *pformat, ... ) const;
     int vPrintf ( const char *pformat, va_list args ) const;
@@ -492,14 +491,10 @@ inline void oldCAC::uninstallCASG ( CASG &sg )
     this->clientCtx.uninstallCASG ( sg );
 }
 
-inline void oldCAC::enableCallbackPreemption ()
+inline void oldCAC::blockForEventAndEnableCallbacks ( 
+        epicsEvent &event, double timeout )
 {
-    this->clientCtx.enableCallbackPreemption ();
-}
-
-inline void oldCAC::disableCallbackPreemption ()
-{
-    this->clientCtx.disableCallbackPreemption ();
+    this->clientCtx.blockForEventAndEnableCallbacks ( event, timeout );
 }
 
 inline void oldCAC::vSignal ( int ca_status, const char *pfilenm, 

src/ca/tcpiiu.cpp

@@ -1272,24 +1272,20 @@ bool tcpiiu::flush ()
     }
 }
 
+// ~tcpiiu() will not return while this->blockingForFlush is greater than zero
 void tcpiiu::blockUntilSendBacklogIsReasonable ( epicsMutex &mutex )
 {
-    this->pCAC()->enableCallbackPreemption ();
     assert ( this->blockingForFlush < UINT_MAX );
     this->blockingForFlush++;
     while ( this->sendQue.flushBlockThreshold(0u) && this->state == iiu_connected ) {
         epicsAutoMutexRelease autoRelease ( mutex );
         this->flushBlockEvent.wait ( 5.0 );
     }
-    assert ( this->blockingForFlush > 0u );
-    this->blockingForFlush--;
-    if ( this->blockingForFlush ) {
+    if ( this->blockingForFlush == 1 ) {
         this->flushBlockEvent.signal ();
     }
-    {
-        epicsAutoMutexRelease autoRelease ( mutex );
-        this->pCAC()->disableCallbackPreemption ();
-    }
+    assert ( this->blockingForFlush > 0u );
+    this->blockingForFlush--;
 }
 
 void tcpiiu::flushRequestIfAboveEarlyThreshold ()