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

general indent beautification

Browse Source
This commit is contained in:
Matej Sekoranja
2016-03-10 23:36:16 +01:00
parent d139c42d17
commit 5279d247ca
110 changed files with 21778 additions and 21525 deletions
Download Patch File Download Diff File Expand all files Collapse all files

245
src/remote/simpleChannelSearchManagerImpl.cpp
View File

@@ -43,15 +43,15 @@ SimpleChannelSearchManagerImpl::create(Context::shared_pointer const & context)
{
SimpleChannelSearchManagerImpl::shared_pointer thisPtr(new SimpleChannelSearchManagerImpl(context));
thisPtr->activate();
return thisPtr;
return thisPtr;
}
SimpleChannelSearchManagerImpl::SimpleChannelSearchManagerImpl(Context::shared_pointer const & context) :
m_context(context),
m_responseAddress(*context->getSearchTransport()->getRemoteAddress()),
m_canceled(),
m_sequenceNumber(0),
m_sendBuffer(MAX_UDP_UNFRAGMENTED_SEND),
m_canceled(),
m_sequenceNumber(0),
m_sendBuffer(MAX_UDP_UNFRAGMENTED_SEND),
m_channels(),
m_lastTimeSent(),
m_mockTransportSendControl(),
@@ -60,56 +60,56 @@ SimpleChannelSearchManagerImpl::SimpleChannelSearchManagerImpl(Context::shared_p
m_mutex()
{
// initialize send buffer
initializeSendBuffer();
// initialize send buffer
initializeSendBuffer();
// initialize random seed with some random value
// initialize random seed with some random value
srand ( time(NULL) );
}
void SimpleChannelSearchManagerImpl::activate()
{
// add some jitter so that all the clients do not send at the same time
double period = ATOMIC_PERIOD + (rand() % (2*PERIOD_JITTER_MS+1) - PERIOD_JITTER_MS)/(double)1000;
// add some jitter so that all the clients do not send at the same time
double period = ATOMIC_PERIOD + (rand() % (2*PERIOD_JITTER_MS+1) - PERIOD_JITTER_MS)/(double)1000;
Context::shared_pointer context = m_context.lock();
if (context.get())
Context::shared_pointer context = m_context.lock();
if (context.get())
context->getTimer()->schedulePeriodic(shared_from_this(), period, period);
//new Thread(this, "pvAccess immediate-search").start();
//new Thread(this, "pvAccess immediate-search").start();
}
SimpleChannelSearchManagerImpl::~SimpleChannelSearchManagerImpl()
{
// shared_from_this() is not allowed from destructor
// be sure to call cancel() first
// shared_from_this() is not allowed from destructor
// be sure to call cancel() first
// cancel();
}
void SimpleChannelSearchManagerImpl::cancel()
{
Lock guard(m_mutex);
if (m_canceled.get())
return;
m_canceled.set();
Context::shared_pointer context = m_context.lock();
if (context.get())
context->getTimer()->cancel(shared_from_this());
if (m_canceled.get())
return;
m_canceled.set();
Context::shared_pointer context = m_context.lock();
if (context.get())
context->getTimer()->cancel(shared_from_this());
}
int32_t SimpleChannelSearchManagerImpl::registeredCount()
{
Lock guard(m_channelMutex);
return static_cast<int32_t>(m_channels.size());
Lock guard(m_channelMutex);
return static_cast<int32_t>(m_channels.size());
}
void SimpleChannelSearchManagerImpl::registerSearchInstance(SearchInstance::shared_pointer const & channel, bool penalize)
{
if (m_canceled.get())
return;
if (m_canceled.get())
return;
bool immediateTrigger;
{
@@ -125,43 +125,43 @@ void SimpleChannelSearchManagerImpl::registerSearchInstance(SearchInstance::shar
}
if (immediateTrigger)
callback();
callback();
}
void SimpleChannelSearchManagerImpl::unregisterSearchInstance(SearchInstance::shared_pointer const & channel)
{
Lock guard(m_channelMutex);
pvAccessID id = channel->getSearchInstanceID();
std::map<pvAccessID,SearchInstance::shared_pointer>::iterator channelsIter = m_channels.find(id);
if(channelsIter != m_channels.end())
m_channels.erase(id);
Lock guard(m_channelMutex);
pvAccessID id = channel->getSearchInstanceID();
std::map<pvAccessID,SearchInstance::shared_pointer>::iterator channelsIter = m_channels.find(id);
if(channelsIter != m_channels.end())
m_channels.erase(id);
}
void SimpleChannelSearchManagerImpl::searchResponse(const GUID & guid, pvAccessID cid, int32_t /*seqNo*/, int8_t minorRevision, osiSockAddr* serverAddress)
{
Lock guard(m_channelMutex);
std::map<pvAccessID,SearchInstance::shared_pointer>::iterator channelsIter = m_channels.find(cid);
if(channelsIter == m_channels.end())
{
guard.unlock();
Lock guard(m_channelMutex);
std::map<pvAccessID,SearchInstance::shared_pointer>::iterator channelsIter = m_channels.find(cid);
if(channelsIter == m_channels.end())
{
guard.unlock();
// enable duplicate reports
SearchInstance::shared_pointer si = std::tr1::dynamic_pointer_cast<SearchInstance>(m_context.lock()->getChannel(cid));
if (si)
SearchInstance::shared_pointer si = std::tr1::dynamic_pointer_cast<SearchInstance>(m_context.lock()->getChannel(cid));
if (si)
si->searchResponse(guid, minorRevision, serverAddress);
}
else
{
SearchInstance::shared_pointer si = channelsIter->second;
}
else
{
SearchInstance::shared_pointer si = channelsIter->second;
// remove from search list
m_channels.erase(cid);
guard.unlock();
m_channels.erase(cid);
// then notify SearchInstance
guard.unlock();
// then notify SearchInstance
si->searchResponse(guid, minorRevision, serverAddress);
}
}
}
void SimpleChannelSearchManagerImpl::newServerDetected()
@@ -173,17 +173,17 @@ void SimpleChannelSearchManagerImpl::newServerDetected()
void SimpleChannelSearchManagerImpl::initializeSendBuffer()
{
// for now OK, since it is only set here
m_sequenceNumber++;
m_sequenceNumber++;
// new buffer
m_sendBuffer.clear();
// new buffer
m_sendBuffer.clear();
m_sendBuffer.putByte(PVA_MAGIC);
m_sendBuffer.putByte(PVA_VERSION);
m_sendBuffer.putByte((EPICS_BYTE_ORDER == EPICS_ENDIAN_BIG) ? 0x80 : 0x00); // data + 7-bit endianess
m_sendBuffer.putByte(CMD_SEARCH);
m_sendBuffer.putInt(4+1+3+16+2+1); // "zero" payload
m_sendBuffer.putInt(m_sequenceNumber);
m_sendBuffer.putInt(m_sequenceNumber);
// multicast vs unicast mask
m_sendBuffer.putByte((int8_t)0);
@@ -201,15 +201,15 @@ void SimpleChannelSearchManagerImpl::initializeSendBuffer()
m_sendBuffer.putByte((int8_t)1);
// TODO "tcp" constant
SerializeHelper::serializeString("tcp", &m_sendBuffer, &m_mockTransportSendControl);
m_sendBuffer.putShort((int16_t)0); // count
m_sendBuffer.putShort((int16_t)0); // count
}
void SimpleChannelSearchManagerImpl::flushSendBuffer()
{
Lock guard(m_mutex);
Lock guard(m_mutex);
Transport::shared_pointer tt = m_context.lock()->getSearchTransport();
BlockingUDPTransport::shared_pointer ut = std::tr1::static_pointer_cast<BlockingUDPTransport>(tt);
BlockingUDPTransport::shared_pointer ut = std::tr1::static_pointer_cast<BlockingUDPTransport>(tt);
m_sendBuffer.putByte(CAST_POSITION, (int8_t)0x80); // unicast, no reply required
ut->send(&m_sendBuffer, inetAddressType_unicast);
@@ -222,10 +222,10 @@ void SimpleChannelSearchManagerImpl::flushSendBuffer()
bool SimpleChannelSearchManagerImpl::generateSearchRequestMessage(SearchInstance::shared_pointer const & channel,
ByteBuffer* requestMessage, TransportSendControl* control)
ByteBuffer* requestMessage, TransportSendControl* control)
{
epics::pvData::int16 dataCount = requestMessage->getShort(DATA_COUNT_POSITION);
dataCount++;
/*
@@ -245,36 +245,36 @@ bool SimpleChannelSearchManagerImpl::generateSearchRequestMessage(SearchInstance
requestMessage->putInt(PAYLOAD_POSITION, requestMessage->getPosition() - PVA_MESSAGE_HEADER_SIZE);
requestMessage->putShort(DATA_COUNT_POSITION, dataCount);
return true;
}
}
bool SimpleChannelSearchManagerImpl::generateSearchRequestMessage(SearchInstance::shared_pointer const & channel,
bool allowNewFrame, bool flush)
bool allowNewFrame, bool flush)
{
Lock guard(m_mutex);
bool success = generateSearchRequestMessage(channel, &m_sendBuffer, &m_mockTransportSendControl);
// buffer full, flush
if(!success)
{
flushSendBuffer();
if(allowNewFrame)
generateSearchRequestMessage(channel, &m_sendBuffer, &m_mockTransportSendControl);
if (flush)
flushSendBuffer();
return true;
}
Lock guard(m_mutex);
bool success = generateSearchRequestMessage(channel, &m_sendBuffer, &m_mockTransportSendControl);
// buffer full, flush
if(!success)
{
flushSendBuffer();
if(allowNewFrame)
generateSearchRequestMessage(channel, &m_sendBuffer, &m_mockTransportSendControl);
if (flush)
flushSendBuffer();
return true;
}
if (flush)
flushSendBuffer();
if (flush)
flushSendBuffer();
return flush;
return flush;
}
void SimpleChannelSearchManagerImpl::boost()
{
Lock guard(m_channelMutex);
Lock guard2(m_userValueMutex);
std::map<pvAccessID,SearchInstance::shared_pointer>::iterator channelsIter = m_channels.begin();
for(; channelsIter != m_channels.end(); channelsIter++)
Lock guard(m_channelMutex);
Lock guard2(m_userValueMutex);
std::map<pvAccessID,SearchInstance::shared_pointer>::iterator channelsIter = m_channels.begin();
for(; channelsIter != m_channels.end(); channelsIter++)
{
int32_t& userValue = channelsIter->second->getUserValue();
userValue = BOOST_VALUE;
@@ -283,62 +283,62 @@ void SimpleChannelSearchManagerImpl::boost()
void SimpleChannelSearchManagerImpl::callback()
{
// high-frequency beacon anomaly trigger guard
// high-frequency beacon anomaly trigger guard
{
Lock guard(m_mutex);
Lock guard(m_mutex);
epics::pvData::TimeStamp now;
now.getCurrent();
now.getCurrent();
int64_t nowMS = now.getMilliseconds();
if (nowMS - m_lastTimeSent < 100)
return;
m_lastTimeSent = nowMS;
}
if (nowMS - m_lastTimeSent < 100)
return;
m_lastTimeSent = nowMS;
}
int count = 0;
int frameSent = 0;
int frameSent = 0;
vector<SearchInstance::shared_pointer> toSend;
{
Lock guard(m_channelMutex);
toSend.reserve(m_channels.size());
std::map<pvAccessID,SearchInstance::shared_pointer>::iterator channelsIter = m_channels.begin();
for(; channelsIter != m_channels.end(); channelsIter++)
toSend.push_back(channelsIter->second);
}
Lock guard(m_channelMutex);
toSend.reserve(m_channels.size());
std::map<pvAccessID,SearchInstance::shared_pointer>::iterator channelsIter = m_channels.begin();
for(; channelsIter != m_channels.end(); channelsIter++)
toSend.push_back(channelsIter->second);
}
vector<SearchInstance::shared_pointer>::iterator siter = toSend.begin();
for (; siter != toSend.end(); siter++)
{
m_userValueMutex.lock();
m_userValueMutex.lock();
int32_t& countValue = (*siter)->getUserValue();
bool skip = !isPowerOfTwo(countValue);
if (countValue >= MAX_COUNT_VALUE)
countValue = MAX_FALLBACK_COUNT_VALUE;
else
countValue++;
m_userValueMutex.unlock();
// back-off
if (skip)
continue;
count++;
bool skip = !isPowerOfTwo(countValue);
if (generateSearchRequestMessage(*siter, true, false))
frameSent++;
if (frameSent == MAX_FRAMES_AT_ONCE)
{
epicsThreadSleep(DELAY_BETWEEN_FRAMES_MS/(double)1000.0);
frameSent = 0;
}
if (countValue >= MAX_COUNT_VALUE)
countValue = MAX_FALLBACK_COUNT_VALUE;
else
countValue++;
m_userValueMutex.unlock();
// back-off
if (skip)
continue;
count++;
if (generateSearchRequestMessage(*siter, true, false))
frameSent++;
if (frameSent == MAX_FRAMES_AT_ONCE)
{
epicsThreadSleep(DELAY_BETWEEN_FRAMES_MS/(double)1000.0);
frameSent = 0;
}
}
if (count > 0)
flushSendBuffer();
flushSendBuffer();
}
bool SimpleChannelSearchManagerImpl::isPowerOfTwo(int32_t x)
@@ -350,5 +350,6 @@ void SimpleChannelSearchManagerImpl::timerStopped()
{
}
}}
}
}