diff --git a/src/ca/tcpiiu.cpp b/src/ca/tcpiiu.cpp index 23cc218e0..61dca3ca6 100644 --- a/src/ca/tcpiiu.cpp +++ b/src/ca/tcpiiu.cpp @@ -318,7 +318,7 @@ void tcpRecvThread::run () } } else { - this->iiu.blockUntilBytesArePendingInOS(); + this->iiu.blockUntilBytesArePendingInOS (); nBytesIn = 0u; if ( this->iiu.state != tcpiiu::iiucs_connected && this->iiu.state != tcpiiu::iiucs_clean_shutdown ) { @@ -381,6 +381,7 @@ void tcpRecvThread::run () nBytesIn = pComBuf->fillFromWire ( this->iiu ); } + this->iiu.cacRef.signalRecvThreadActivity (); } if ( pComBuf ) { @@ -1329,7 +1330,9 @@ int tcpiiu::printf ( const char *pformat, ... ) // this is called virtually void tcpiiu::flushRequest () { - this->sendThreadFlushEvent.signal (); + if ( this->sendQue.occupiedBytes() > 0 ) { + this->sendThreadFlushEvent.signal (); + } } void tcpiiu::blockUntilBytesArePendingInOS () diff --git a/src/ca/udpiiu.cpp b/src/ca/udpiiu.cpp index 5a9f5987c..da9f83d88 100644 --- a/src/ca/udpiiu.cpp +++ b/src/ca/udpiiu.cpp @@ -340,6 +340,7 @@ void udpRecvThread::run () do { this->iiu.recvMsg ( this->cbMutex ); + this->iiu.cacRef.signalRecvThreadActivity (); } while ( ! this->iiu.shutdownCmd ); { diff --git a/src/ca/udpiiu.h b/src/ca/udpiiu.h index 883c6cad6..c063c75d6 100644 --- a/src/ca/udpiiu.h +++ b/src/ca/udpiiu.h @@ -98,6 +98,7 @@ public: bool pushDatagramMsg ( const caHdr &hdr, const void *pExt, ca_uint16_t extsize); void shutdown (); double roundTripDelayEstimate () const; + void fdMaskSet ( fd_set & mask ) const; // exceptions class noSocket {}; @@ -221,5 +222,10 @@ inline double udpiiu::roundTripDelayEstimate () const return this->rtteMean; } +inline void udpiiu::fdMaskSet ( fd_set & mask ) const +{ + FD_SET ( this->sock, & mask ); +} + #endif // udpiiuh diff --git a/src/ca/virtualCircuit.h b/src/ca/virtualCircuit.h index 91cf42ad4..ba3648b4f 100644 --- a/src/ca/virtualCircuit.h +++ b/src/ca/virtualCircuit.h @@ -133,6 +133,8 @@ public: bool bytesArePendingInOS () const; + void fdMaskSet ( fd_set & ) const; + private: hostNameCache hostNameCacheInstance; tcpRecvThread recvThread; @@ -247,6 +249,11 @@ inline unsigned tcpiiu::channelCount () return this->channelList.count (); } +inline void tcpiiu::fdMaskSet ( fd_set & mask ) const +{ + FD_SET ( this->sock, & mask ); +} + inline void tcpRecvThread::interruptSocketRecv () { epicsThreadId threadId = this->thread.getId ();