From f268ec5f8d751b664f4cdab4d4c70f1d4ee0ec06 Mon Sep 17 00:00:00 2001 From: Matej Sekoranja Date: Mon, 7 Feb 2011 15:35:52 +0100 Subject: [PATCH] volatile cleanup --- .../remote/blockingClientTCPTransport.cpp | 26 +++++++----- pvAccessApp/remote/blockingTCP.h | 23 ++++++----- pvAccessApp/remote/blockingTCPAcceptor.cpp | 41 +++++++++++-------- pvAccessApp/remote/blockingTCPTransport.cpp | 16 +++++--- pvAccessApp/remote/blockingUDP.h | 7 ++-- pvAccessApp/remote/channelSearchManager.h | 18 ++++---- pvAccessApp/remote/remote.h | 4 +- testApp/client/MockClientImpl.cpp | 8 ++-- testApp/utils/transportRegistryTest.cpp | 4 +- 9 files changed, 83 insertions(+), 64 deletions(-) diff --git a/pvAccessApp/remote/blockingClientTCPTransport.cpp b/pvAccessApp/remote/blockingClientTCPTransport.cpp index 42adb22..b3f274c 100644 --- a/pvAccessApp/remote/blockingClientTCPTransport.cpp +++ b/pvAccessApp/remote/blockingClientTCPTransport.cpp @@ -46,7 +46,7 @@ namespace epics { setSendQueueFlushStrategy(IMMEDIATE); // setup connection timeout timer (watchdog) - epicsTimeGetCurrent(const_cast (&_aliveTimestamp)); + epicsTimeGetCurrent(&_aliveTimestamp); context->getTimer()->schedulePeriodic(_timerNode, beaconInterval, beaconInterval); @@ -64,8 +64,11 @@ namespace epics { epicsTimeStamp currentTime; epicsTimeGetCurrent(¤tTime); - double diff = epicsTimeDiffInSeconds(¤tTime, - const_cast (&_aliveTimestamp)); + _ownersMutex.lock(); + // no exception expected here + double diff = epicsTimeDiffInSeconds(¤tTime, &_aliveTimestamp); + _ownersMutex.unlock(); + if(diff>2*_connectionTimeout) { unresponsiveTransport(); } @@ -76,10 +79,10 @@ namespace epics { } void BlockingClientTCPTransport::unresponsiveTransport() { + Lock lock(&_ownersMutex); if(!_unresponsiveTransport) { _unresponsiveTransport = true; - Lock lock(&_ownersMutex); set::iterator it = _owners.begin(); for(; it!=_owners.end(); it++) (*it)->transportUnresponsive(); @@ -88,9 +91,8 @@ namespace epics { bool BlockingClientTCPTransport::acquire(TransportClient* client) { Lock lock(&_mutex); - if(_closed) return false; - + char ipAddrStr[48]; ipAddrToDottedIP(&_socketAddress.ia, ipAddrStr, sizeof(ipAddrStr)); errlogSevPrintf(errlogInfo, "Acquiring transport to %s.", ipAddrStr); @@ -135,14 +137,15 @@ namespace epics { } void BlockingClientTCPTransport::release(TransportClient* client) { + Lock lock(&_mutex); if(_closed) return; - + char ipAddrStr[48]; ipAddrToDottedIP(&_socketAddress.ia, ipAddrStr, sizeof(ipAddrStr)); errlogSevPrintf(errlogInfo, "Releasing transport to %s.", ipAddrStr); - Lock lock(&_ownersMutex); + Lock lock2(&_ownersMutex); _owners.erase(client); // not used anymore @@ -151,14 +154,15 @@ namespace epics { } void BlockingClientTCPTransport::aliveNotification() { - epicsTimeGetCurrent(const_cast (&_aliveTimestamp)); + Lock guard(&_ownersMutex); + epicsTimeGetCurrent(&_aliveTimestamp); if(_unresponsiveTransport) responsiveTransport(); } void BlockingClientTCPTransport::responsiveTransport() { + Lock lock(&_ownersMutex); if(_unresponsiveTransport) { _unresponsiveTransport = false; - Lock lock(&_ownersMutex); set::iterator it = _owners.begin(); for(; it!=_owners.end(); it++) @@ -168,8 +172,8 @@ namespace epics { void BlockingClientTCPTransport::changedTransport() { _introspectionRegistry->reset(); - Lock lock(&_ownersMutex); + Lock lock(&_ownersMutex); set::iterator it = _owners.begin(); for(; it!=_owners.end(); it++) (*it)->transportChanged(); diff --git a/pvAccessApp/remote/blockingTCP.h b/pvAccessApp/remote/blockingTCP.h index 99cb1a9..5e07f51 100644 --- a/pvAccessApp/remote/blockingTCP.h +++ b/pvAccessApp/remote/blockingTCP.h @@ -54,7 +54,8 @@ namespace epics { ResponseHandler* responseHandler, int receiveBufferSize, int16 priority); - virtual bool isClosed() const { + virtual bool isClosed() { + Lock guard(&_mutex); return _closed; } @@ -108,8 +109,8 @@ namespace epics { virtual int getSocketReceiveBufferSize() const; - virtual bool isVerified() const { - Lock lock(const_cast(&_verifiedMutex)); + virtual bool isVerified() { + Lock lock(&_verifiedMutex); return _verified; } @@ -354,7 +355,7 @@ namespace epics { * Connection status * NOTE: synced by _mutex */ - bool volatile _closed; + bool _closed; // NOTE: synced by _mutex bool _sendThreadExited; @@ -513,7 +514,7 @@ namespace epics { /** * Unresponsive transport flag. */ - volatile bool _unresponsiveTransport; + bool _unresponsiveTransport; /** * Timer task node. @@ -523,7 +524,7 @@ namespace epics { /** * Timestamp of last "live" event on this transport. */ - volatile epicsTimeStamp _aliveTimestamp; + epicsTimeStamp _aliveTimestamp; epics::pvData::Mutex _mutex; epics::pvData::Mutex _ownersMutex; @@ -712,7 +713,7 @@ namespace epics { /** * Last SID cache. */ - volatile pvAccessID _lastChannelSID; + pvAccessID _lastChannelSID; /** * Channel table (SID -> channel mapping). @@ -753,7 +754,7 @@ namespace epics { * @return bind socket address, null if not binded. */ osiSockAddr* getBindAddress() { - return _bindAddress; + return &_bindAddress; } /** @@ -770,7 +771,7 @@ namespace epics { /** * Bind server socket address. */ - osiSockAddr* _bindAddress; + osiSockAddr _bindAddress; /** * Server socket channel. @@ -785,7 +786,9 @@ namespace epics { /** * Destroyed flag. */ - volatile bool _destroyed; + bool _destroyed; + + Mutex _mutex; epicsThreadId _threadId; diff --git a/pvAccessApp/remote/blockingTCPAcceptor.cpp b/pvAccessApp/remote/blockingTCPAcceptor.cpp index 4cf3895..1d58fda 100644 --- a/pvAccessApp/remote/blockingTCPAcceptor.cpp +++ b/pvAccessApp/remote/blockingTCPAcceptor.cpp @@ -29,7 +29,7 @@ namespace epics { BlockingTCPAcceptor::BlockingTCPAcceptor(Context* context, int port, int receiveBufferSize) : - _context(context), _bindAddress(NULL), _serverSocketChannel( + _context(context), _bindAddress(), _serverSocketChannel( INVALID_SOCKET), _receiveBufferSize(receiveBufferSize), _destroyed(false), _threadId(NULL) { initialize(port); @@ -37,20 +37,17 @@ namespace epics { BlockingTCPAcceptor::~BlockingTCPAcceptor() { destroy(); - - if(_bindAddress!=NULL) delete _bindAddress; } int BlockingTCPAcceptor::initialize(in_port_t port) { // specified bind address - _bindAddress = new osiSockAddr; - _bindAddress->ia.sin_family = AF_INET; - _bindAddress->ia.sin_port = htons(port); - _bindAddress->ia.sin_addr.s_addr = htonl(INADDR_ANY); + _bindAddress.ia.sin_family = AF_INET; + _bindAddress.ia.sin_port = htons(port); + _bindAddress.ia.sin_addr.s_addr = htonl(INADDR_ANY); char strBuffer[64]; char ipAddrStr[48]; - ipAddrToDottedIP(&_bindAddress->ia, ipAddrStr, sizeof(ipAddrStr)); + ipAddrToDottedIP(&_bindAddress.ia, ipAddrStr, sizeof(ipAddrStr)); int tryCount = 0; while(tryCount<2) { @@ -71,20 +68,20 @@ namespace epics { else { // try to bind int retval = ::bind(_serverSocketChannel, - &_bindAddress->sa, sizeof(sockaddr)); + &_bindAddress.sa, sizeof(sockaddr)); if(retval<0) { epicsSocketConvertErrnoToString(strBuffer, sizeof(strBuffer)); errlogSevPrintf(errlogMinor, "Socket bind error: %s", strBuffer); - if(_bindAddress->ia.sin_port!=0) { + if(_bindAddress.ia.sin_port!=0) { // failed to bind to specified bind address, // try to get port dynamically, but only once errlogSevPrintf( errlogMinor, "Configured TCP port %d is unavailable, trying to assign it dynamically.", port); - _bindAddress->ia.sin_port = htons(0); + _bindAddress.ia.sin_port = htons(0); } else { ::close(_serverSocketChannel); @@ -95,11 +92,11 @@ namespace epics { // bind succeeded // update bind address, if dynamically port selection was used - if(ntohs(_bindAddress->ia.sin_port)==0) { + if(ntohs(_bindAddress.ia.sin_port)==0) { socklen_t sockLen = sizeof(sockaddr); // read the actual socket info retval = ::getsockname(_serverSocketChannel, - &_bindAddress->sa, &sockLen); + &_bindAddress.sa, &sockLen); if(retval<0) { // error obtaining port number epicsSocketConvertErrnoToString(strBuffer, @@ -111,7 +108,7 @@ namespace epics { errlogSevPrintf( errlogInfo, "Using dynamically assigned TCP port %d.", - ntohs(_bindAddress->ia.sin_port)); + ntohs(_bindAddress.ia.sin_port)); } } @@ -135,7 +132,7 @@ namespace epics { this); // all OK, return - return ntohs(_bindAddress->ia.sin_port); + return ntohs(_bindAddress.ia.sin_port); } // successful bind } // successfully obtained socket tryCount++; @@ -149,14 +146,21 @@ namespace epics { void BlockingTCPAcceptor::handleEvents() { // rise level if port is assigned dynamically char ipAddrStr[48]; - ipAddrToDottedIP(&_bindAddress->ia, ipAddrStr, sizeof(ipAddrStr)); + ipAddrToDottedIP(&_bindAddress.ia, ipAddrStr, sizeof(ipAddrStr)); errlogSevPrintf(errlogInfo, "Accepting connections at %s.", ipAddrStr); bool socketOpen = true; char strBuffer[64]; - while(!_destroyed&&socketOpen) { + while(socketOpen) { + + { + Lock guard(&_mutex); + if (_destroyed) + break; + } + osiSockAddr address; osiSocklen_t len = sizeof(sockaddr); @@ -241,12 +245,13 @@ namespace epics { } void BlockingTCPAcceptor::destroy() { + Lock guard(&_mutex); if(_destroyed) return; _destroyed = true; if(_serverSocketChannel!=INVALID_SOCKET) { char ipAddrStr[48]; - ipAddrToDottedIP(&_bindAddress->ia, ipAddrStr, + ipAddrToDottedIP(&_bindAddress.ia, ipAddrStr, sizeof(ipAddrStr)); errlogSevPrintf(errlogInfo, "Stopped accepting connections at %s.", ipAddrStr); diff --git a/pvAccessApp/remote/blockingTCPTransport.cpp b/pvAccessApp/remote/blockingTCPTransport.cpp index ee4204c..4452616 100644 --- a/pvAccessApp/remote/blockingTCPTransport.cpp +++ b/pvAccessApp/remote/blockingTCPTransport.cpp @@ -136,11 +136,11 @@ namespace epics { close(true); TransportSender* sender; - while (sender = _monitorSendQueue->extract()) + while ((sender = _monitorSendQueue->extract())) sender->release(); delete _monitorSendQueue; - while (sender = _sendQueue->extract()) + while ((sender = _sendQueue->extract())) sender->release(); delete _sendQueue; @@ -314,11 +314,13 @@ namespace epics { temp<<_maxPayloadSize<<" available."; THROW_BASE_EXCEPTION(temp.str().c_str()); } + + // TODO sync _closed while(_sendBuffer->getRemaining()setLimit(min(_storedPosition+_storedPayloadSize, _storedLimit)); + // TODO sync _closed + // add if missing... - if(!_closed&&_socketBuffer->getRemaining()getRemaining()true if connected. */ - virtual bool isClosed() const =0; + virtual bool isClosed() =0; /** * Get transport verification status. * @return verification flag. */ - virtual bool isVerified() const =0; + virtual bool isVerified() =0; /** * Notify transport that it is has been verified. diff --git a/testApp/client/MockClientImpl.cpp b/testApp/client/MockClientImpl.cpp index 6fe2408..8f4a5ca 100644 --- a/testApp/client/MockClientImpl.cpp +++ b/testApp/client/MockClientImpl.cpp @@ -163,7 +163,7 @@ class MockChannelGet : public ChannelGet ChannelGetRequester* m_channelGetRequester; PVStructure* m_pvStructure; BitSet* m_bitSet; - volatile bool m_first; + bool m_first; private: ~MockChannelGet() @@ -218,7 +218,7 @@ class MockChannelPut : public ChannelPut ChannelPutRequester* m_channelPutRequester; PVStructure* m_pvStructure; BitSet* m_bitSet; - volatile bool m_first; + bool m_first; private: ~MockChannelPut() @@ -272,9 +272,9 @@ class MockMonitor : public Monitor, public MonitorElement PVStructure* m_pvStructure; BitSet* m_changedBitSet; BitSet* m_overrunBitSet; - volatile bool m_first; + bool m_first; Mutex* m_lock; - volatile int m_count; + int m_count; private: ~MockMonitor() diff --git a/testApp/utils/transportRegistryTest.cpp b/testApp/utils/transportRegistryTest.cpp index 5311851..406b63f 100644 --- a/testApp/utils/transportRegistryTest.cpp +++ b/testApp/utils/transportRegistryTest.cpp @@ -36,8 +36,8 @@ namespace epics { virtual void aliveNotification(){}; virtual void changedTransport(){}; virtual void close(bool force){}; - virtual bool isClosed() const{return false;}; - virtual bool isVerified() const{return false;}; + virtual bool isClosed() {return false;}; + virtual bool isVerified() {return false;}; virtual void verified(){}; virtual void enqueueSendRequest(TransportSender* sender){}; virtual void ensureData(int) {};