elevated priority of CAC timer event thread

src/ca/cac.cpp

@@ -45,6 +45,7 @@ cac::cac () :
 {
 	long status;
     static threadOnceId once = OSITHREAD_ONCE_INIT;
+    unsigned abovePriority;
 
     threadOnce ( &once, cacInitRecursionLock, 0);
 
@@ -56,6 +57,21 @@ cac::cac () :
         throwWithLocation ( caErrorCode (ECA_INTERNAL) );
 	}
 
+    {
+        threadBoolStatus tbs;
+        unsigned selfPriority = threadGetPrioritySelf ();
+
+        tbs = threadLowestPriorityLevelAbove ( selfPriority, &abovePriority);
+        if ( tbs != tbsSuccess ) {
+            abovePriority = selfPriority;
+        }
+    }
+
+    this->pTimerQueue = new osiTimerQueue ( abovePriority );
+    if ( ! this->pTimerQueue ) {
+        throwWithLocation ( caErrorCode (ECA_ALLOCMEM) );
+    }
+
 	ellInit (&this->ca_taskVarList);
 	ellInit (&this->putCvrtBuf);
     ellInit (&this->fdInfoFreeList);
@@ -230,6 +246,8 @@ cac::~cac ()
     semMutexDestroy (this->ca_client_lock);
 
     osiSockRelease ();
+
+    delete this->pTimerQueue;
 }
 
 void cac::safeDestroyNMIU (unsigned id)

src/ca/iocinf.h

@@ -725,7 +725,7 @@ public:
     void lock () const;
     void unlock () const;
 
-    osiTimerQueue           timerQueue;
+    osiTimerQueue           *pTimerQueue;
     ELLLIST                 activeCASGOP;
     ELLLIST                 putCvrtBuf;
     ELLLIST                 fdInfoFreeList;

src/ca/nciu.cpp

@@ -57,7 +57,7 @@ nciu::nciu (cac *pcac, cacChannel &chan, const char *pNameIn) :
     /*
      * reset broadcasted search counters
      */
-    pcac->pudpiiu->searchTmr.reset (0.0);
+    pcac->pudpiiu->searchTmr.reset ( CA_RECAST_DELAY );
 
     this->f_fullyConstructed = true;
 

src/ca/searchTimer.cpp

@@ -39,14 +39,18 @@ searchTimer::searchTimer (udpiiu &iiuIn, osiTimerQueue &queueIn) :
 //
 // searchTimer::reset ()
 //
-void searchTimer::reset (double delayToNextTry)
+void searchTimer::reset ( double delayToNextTry )
 {
     LOCK (this->iiu.pcas);
     this->retry = 0;
     this->period = CA_RECAST_DELAY;
     UNLOCK (this->iiu.pcas);
 
-    if (this->timeRemaining()>delayToNextTry) {
+    if ( delayToNextTry < CA_RECAST_DELAY ) {
+        delayToNextTry = CA_RECAST_DELAY;
+    }
+
+    if ( this->timeRemaining () > delayToNextTry ) {
         this->reschedule (delayToNextTry);
         debugPrintf ( ("reschedualed search timer for completion in %f sec\n", delayToNextTry) );
     }

src/ca/tcpiiu.cpp

@@ -321,7 +321,7 @@ extern "C" void cacRecvThreadTCP (void *pParam)
         }
         tid = threadCreate ("CAC-TCP-send", priorityOfSend,
                 threadGetStackSize (threadStackMedium), cacSendThreadTCP, piiu);
-        if (tid) {
+        if ( tid ) {
             while (1) {
                 piiu->recvMsg ();
                 if ( piiu->state != iiu_connected ) {
@@ -331,9 +331,13 @@ extern "C" void cacRecvThreadTCP (void *pParam)
             }
         }
         else {
+            semBinaryGive (piiu->sendThreadExitSignal);
             piiu->shutdown ();
         }
     }
+    else {
+        semBinaryGive (piiu->sendThreadExitSignal);
+    }
     semBinaryGive (piiu->recvThreadExitSignal);
 }
 
@@ -341,8 +345,8 @@ extern "C" void cacRecvThreadTCP (void *pParam)
 // tcpiiu::tcpiiu ()
 //
 tcpiiu::tcpiiu (cac *pcac, const struct sockaddr_in &ina, unsigned minorVersion, class bhe &bheIn) :
-    tcpRecvWatchdog (pcac->ca_connectTMO, pcac->timerQueue, CA_V43 (CA_PROTOCOL_VERSION, minorVersion) ),
-    tcpSendWatchdog (pcac->ca_connectTMO, pcac->timerQueue),
+    tcpRecvWatchdog (pcac->ca_connectTMO, *pcac->pTimerQueue, CA_V43 (CA_PROTOCOL_VERSION, minorVersion) ),
+    tcpSendWatchdog (pcac->ca_connectTMO, *pcac->pTimerQueue),
     netiiu (pcac),
     bhe (bheIn)
 {
@@ -1460,8 +1464,8 @@ void tcpiiu::disconnect (nciu *chan)
      * try to reconnect
      */
     assert (this->pcas->pudpiiu);
-    this->pcas->pudpiiu->addToChanList (chan);
-    this->pcas->pudpiiu->searchTmr.reset (0.0);
+    this->pcas->pudpiiu->addToChanList ( chan );
+    this->pcas->pudpiiu->searchTmr.reset ( CA_RECAST_DELAY );
     UNLOCK (this->pcas);
 }
 

src/ca/udpiiu.cpp

@@ -358,11 +358,11 @@ int repeater_installed (udpiiu *piiu)
 //
 // udpiiu::udpiiu ()
 //
-udpiiu::udpiiu (cac *pcac) :
-    netiiu (pcac),
-    searchTmr (*this, pcac->timerQueue), 
-    repeaterSubscribeTmr (*this, pcac->timerQueue),
-    shutdownCmd (false)
+udpiiu::udpiiu ( cac *pcac ) :
+    netiiu ( pcac ),
+    searchTmr ( *this, *pcac->pTimerQueue ), 
+    repeaterSubscribeTmr ( *this, *pcac->pTimerQueue ),
+    shutdownCmd ( false )
 {
     static const unsigned short PORT_ANY = 0u;
     osiSockAddr addr;
@@ -597,19 +597,22 @@ udpiiu::~udpiiu ()
  */
 void udpiiu::shutdown ()
 {
-    int status;
+    LOCK (this->pcas);
+    if ( ! this->shutdownCmd ) {
+        int status;
 
-    //
-    // use of shutdown () for this purpose on UDP
-    // sockets does not work on certain OS (i.e. solaris).
-    //
-    status = socket_close ( this->sock );
-    if ( status ) {
-        errlogPrintf ( "CAC UDP socket close error was %s\n", 
-            SOCKERRSTR (SOCKERRNO) );
+        this->shutdownCmd = true;
+        //
+        // use of shutdown () for this purpose on UDP
+        // sockets does not work on certain OS (i.e. solaris).
+        //
+        status = socket_close ( this->sock );
+        if ( status ) {
+            errlogPrintf ( "CAC UDP socket close error was %s\n", 
+                SOCKERRSTR (SOCKERRNO) );
+        }
     }
-
-    this->shutdownCmd = true;
+    UNLOCK (this->pcas);
     semBinaryGive (this->xmitSignal);
 }