epics/pvAccess
0
0
Fork 0
Code Issues Pull Requests Projects Wiki Activity

volatile cleanup

Browse Source
This commit is contained in:
Matej Sekoranja
2011-02-07 15:35:52 +01:00
parent de8b632381
commit f268ec5f8d
9 changed files with 83 additions and 64 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
pvAccessApp/remote/blockingClientTCPTransport.cpp
View File

@@ -46,7 +46,7 @@ namespace epics {
setSendQueueFlushStrategy(IMMEDIATE);
// setup connection timeout timer (watchdog)
epicsTimeGetCurrent(const_cast<epicsTimeStamp*> (&_aliveTimestamp));
epicsTimeGetCurrent(&_aliveTimestamp);
context->getTimer()->schedulePeriodic(_timerNode, beaconInterval,
beaconInterval);
@@ -64,8 +64,11 @@ namespace epics {
epicsTimeStamp currentTime;
epicsTimeGetCurrent(&currentTime);
double diff = epicsTimeDiffInSeconds(&currentTime,
const_cast<epicsTimeStamp*> (&_aliveTimestamp));
_ownersMutex.lock();
// no exception expected here
double diff = epicsTimeDiffInSeconds(&currentTime, &_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<TransportClient*>::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<epicsTimeStamp*> (&_aliveTimestamp));
Lock guard(&_ownersMutex);
epicsTimeGetCurrent(&_aliveTimestamp);
if(_unresponsiveTransport) responsiveTransport();
}
void BlockingClientTCPTransport::responsiveTransport() {
Lock lock(&_ownersMutex);
if(_unresponsiveTransport) {
_unresponsiveTransport = false;
Lock lock(&_ownersMutex);
set<TransportClient*>::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<TransportClient*>::iterator it = _owners.begin();
for(; it!=_owners.end(); it++)
(*it)->transportChanged();

23
pvAccessApp/remote/blockingTCP.h
View File

@@ -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<epics::pvData::Mutex*>(&_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, <code>null</code> 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;

41
pvAccessApp/remote/blockingTCPAcceptor.cpp
View File

@@ -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);

16
pvAccessApp/remote/blockingTCPTransport.cpp
View File

@@ -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()<size&&!_closed)
flush(false);
if(!_closed) THROW_BASE_EXCEPTION("transport closed");
if(_closed) THROW_BASE_EXCEPTION("transport closed");
}
void BlockingTCPTransport::endMessage(bool hasMoreSegments) {
@@ -424,9 +426,11 @@ namespace epics {
_socketBuffer->setLimit(min(_storedPosition+_storedPayloadSize,
_storedLimit));
// TODO sync _closed
// add if missing...
if(!_closed&&_socketBuffer->getRemaining()<size) ensureData(
size);
if(!_closed&&_socketBuffer->getRemaining()<size)
ensureData(size);
}
if(_closed) THROW_BASE_EXCEPTION("transport closed");
@@ -435,6 +439,7 @@ namespace epics {
void BlockingTCPTransport::processReadCached(bool nestedCall,
ReceiveStage inStage, int requiredBytes, bool addToBuffer) {
try {
// TODO sync _closed
while(!_closed) {
if(_stage==READ_FROM_SOCKET||inStage!=NONE) {
int currentStartPosition;
@@ -759,6 +764,7 @@ namespace epics {
}
void BlockingTCPTransport::processSendQueue() {
// TODO sync _closed
while(!_closed) {
TransportSender* sender;

7
pvAccessApp/remote/blockingUDP.h
View File

@@ -37,7 +37,8 @@ namespace epics {
virtual ~BlockingUDPTransport();
virtual bool isClosed() const {
virtual bool isClosed() {
Lock guard(&_mutex);
return _closed;
}
@@ -81,7 +82,7 @@ namespace epics {
// noop
}
virtual bool isVerified() const {
virtual bool isVerified() {
return false;
}
@@ -184,7 +185,7 @@ namespace epics {
}
protected:
bool volatile _closed;
bool _closed;
/**
* Response handler.

18
pvAccessApp/remote/channelSearchManager.h
View File

@@ -174,11 +174,11 @@ private:
/**
* Number of search attempts in one frame.
*/
volatile int32 _searchAttempts;
int32 _searchAttempts;
/**
* Number of search responses in one frame.
*/
volatile int32 _searchRespones;
int32 _searchRespones;
/**
* Number of frames per search try.
*/
@@ -190,11 +190,11 @@ private:
/**
* Start sequence number (first frame number within a search try).
*/
volatile int32 _startSequenceNumber;
int32 _startSequenceNumber;
/**
* End sequence number (last frame number within a search try).
*/
volatile int32 _endSequenceNumber;
int32 _endSequenceNumber;
/**
* This timer index.
*/
@@ -223,7 +223,7 @@ private:
/**
* Cancel this instance.
*/
volatile bool _canceled;
bool _canceled;
/**
* Time of last response check.
*/
@@ -327,11 +327,11 @@ private:
/**
* Canceled flag.
*/
volatile bool _canceled;
bool _canceled;
/**
* Round-trip time (RTT) mean.
*/
volatile double _rttmean;
double _rttmean;
/**
* Search timers array.
* Each timer with a greater index has longer (doubled) search period.
@@ -348,7 +348,7 @@ private:
/**
* Search (datagram) sequence number.
*/
volatile int32 _sequenceNumber;
int32 _sequenceNumber;
/**
* Max search period (in ms).
*/
@@ -372,7 +372,7 @@ private:
/**
* Time of last frame send.
*/
volatile int64 _timeAtLastSend;
int64 _timeAtLastSend;
/**
* Set of registered channels.
*/

4
pvAccessApp/remote/remote.h
View File

@@ -235,13 +235,13 @@ namespace epics {
* Check connection status.
* @return <code>true</code> 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.

8
testApp/client/MockClientImpl.cpp
View File

@@ -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()

4
testApp/utils/transportRegistryTest.cpp
View File

@@ -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) {};