Support for TCP Searches

src/remote/blockingUDPTransport.cpp

@@ -435,8 +435,8 @@ bool BlockingUDPTransport::send(const char* buffer, size_t length, const osiSock
 {
     if (IS_LOGGABLE(logLevelDebug))
     {
-        LOG(logLevelDebug, "UDP Tx (%zu) %s -> %s.",
-            length, _remoteName.c_str(), inetAddressToString(address).c_str());
+        LOG(logLevelDebug, "UDP Tx (%lu) %s -> %s.",
+            static_cast<unsigned long>(length), _remoteName.c_str(), inetAddressToString(address).c_str());
     }
 
     int retval = sendto(_channel, buffer,
@@ -460,8 +460,8 @@ bool BlockingUDPTransport::send(ByteBuffer* buffer, const osiSockAddr& address)
 
     if (IS_LOGGABLE(logLevelDebug))
     {
-        LOG(logLevelDebug, "Sending %zu bytes %s -> %s.",
-            buffer->getRemaining(), _remoteName.c_str(), inetAddressToString(address).c_str());
+        LOG(logLevelDebug, "Sending %lu bytes %s -> %s.",
+            static_cast<unsigned long>(buffer->getRemaining()), _remoteName.c_str(), inetAddressToString(address).c_str());
     }
 
     int retval = sendto(_channel, buffer->getBuffer(),
@@ -498,8 +498,8 @@ bool BlockingUDPTransport::send(ByteBuffer* buffer, InetAddressType target) {
 
         if (IS_LOGGABLE(logLevelDebug))
         {
-            LOG(logLevelDebug, "Sending %zu bytes %s -> %s.",
-                buffer->getRemaining(), _remoteName.c_str(), inetAddressToString(_sendAddresses[i]).c_str());
+            LOG(logLevelDebug, "Sending %lu bytes %s -> %s.",
+                static_cast<unsigned long>(buffer->getRemaining()), _remoteName.c_str(), inetAddressToString(_sendAddresses[i]).c_str());
         }
 
         int retval = sendto(_channel, buffer->getBuffer(),
@@ -684,7 +684,7 @@ void initializeUDPTransports(bool serverFlag,
                 }
             }
             LOG(logLevelDebug,
-                "Broadcast address #%zu: %s. (%sunicast)", i, inetAddressToString(list[i]).c_str(),
+                "Broadcast address #%lu: %s. (%sunicast)", static_cast<unsigned long>(i), inetAddressToString(list[i]).c_str(),
                 isunicast[i]?"":"not ");
         }
 
@@ -714,7 +714,7 @@ void initializeUDPTransports(bool serverFlag,
     getSocketAddressList(ignoreAddressVector, ignoreAddressList, 0, 0);
 
     //
-    // Setup UDP trasport(s) (per interface)
+    // Setup UDP transport(s) (per interface)
     //
 
     InetAddrVector tappedNIF;

src/remote/channelSearchManager.cpp

@@ -71,12 +71,14 @@ static const int MAX_FALLBACK_COUNT_VALUE = (1 << 7) + 1;
 static const int MAX_FRAMES_AT_ONCE = 10;
 static const int DELAY_BETWEEN_FRAMES_MS = 50;
 
+static const int MAX_NAME_SERVER_SEARCH_COUNT = 3;
 
 ChannelSearchManager::ChannelSearchManager(Context::shared_pointer const & context) :
     m_context(context),
     m_responseAddress(), // initialized in activate()
     m_canceled(),
     m_sequenceNumber(0),
+    m_nsSearchCounter(0),
     m_sendBuffer(MAX_UDP_UNFRAGMENTED_SEND),
     m_channels(),
     m_lastTimeSent(),
@@ -135,6 +137,7 @@ void ChannelSearchManager::registerSearchInstance(SearchInstance::shared_pointer
     if (m_canceled.get())
         return;
 
+    LOG(logLevelDebug, "Registering search instance: %s", channel->getSearchInstanceName().c_str());
     bool immediateTrigger;
     {
         Lock guard(m_channelMutex);
@@ -154,6 +157,7 @@ void ChannelSearchManager::registerSearchInstance(SearchInstance::shared_pointer
 
 void ChannelSearchManager::unregisterSearchInstance(SearchInstance::shared_pointer const & channel)
 {
+    LOG(logLevelDebug, "Unregistering search instance: %s", channel->getSearchInstanceName().c_str());
     Lock guard(m_channelMutex);
     pvAccessID id = channel->getSearchInstanceID();
     m_channels.erase(id);
@@ -180,6 +184,7 @@ void ChannelSearchManager::searchResponse(const ServerGUID & guid, pvAccessID ci
         SearchInstance::shared_pointer si(channelsIter->second.lock());
 
         // remove from search list
+        LOG(logLevelDebug, "Removing cid %d from the channel map", cid);
         m_channels.erase(cid);
 
         guard.unlock();
@@ -188,6 +193,30 @@ void ChannelSearchManager::searchResponse(const ServerGUID & guid, pvAccessID ci
         if(si)
             si->searchResponse(guid, minorRevision, serverAddress);
     }
+    // Release all NS connections if there are
+    // no more channels to search for
+    releaseNameServerTransport();
+}
+
+void ChannelSearchManager::releaseNameServerTransport(bool forceRelease)
+{
+    bool releaseAllConnections = true;
+    if(m_channels.size() == 0)
+    {
+        // No more channels to search for, release all connections
+        m_nsSearchCounter = 0;
+        m_context.lock()->releaseNameServerSearchTransport(m_nsTransport, releaseAllConnections);
+        m_nsTransport.reset();
+    }
+    else if(forceRelease)
+    {
+        // There are channels to search for, release only connection
+        // that is currently used
+        releaseAllConnections = false;
+        m_nsSearchCounter = 0;
+        m_context.lock()->releaseNameServerSearchTransport(m_nsTransport, releaseAllConnections);
+        m_nsTransport.reset();
+    }
 }
 
 void ChannelSearchManager::newServerDetected()
@@ -196,6 +225,46 @@ void ChannelSearchManager::newServerDetected()
     callback();
 }
 
+void ChannelSearchManager::send(epics::pvData::ByteBuffer* buffer, TransportSendControl* control)
+{
+    control->startMessage(CMD_SEARCH, 4+1+3+16+2+1+4+2);
+    buffer->putInt(m_sequenceNumber);
+    buffer->putByte((int8_t)QOS_REPLY_REQUIRED); // CAST_POSITION
+    buffer->putByte((int8_t)0); // reserved
+    buffer->putShort((int16_t)0); // reserved
+
+    osiSockAddr anyAddress;
+    memset(&anyAddress, 0, sizeof(anyAddress));
+    anyAddress.ia.sin_family = AF_INET;
+    anyAddress.ia.sin_port = htons(0);
+    anyAddress.ia.sin_addr.s_addr = htonl(INADDR_ANY);
+    encodeAsIPv6Address(buffer, &anyAddress);
+    buffer->putShort((int16_t)ntohs(anyAddress.ia.sin_port));
+    buffer->putByte((int8_t)1);
+    SerializeHelper::serializeString("tcp", buffer, control);
+    buffer->putShort((int16_t)0);  // initial channel count
+
+    vector<SearchInstance::shared_pointer> toSend;
+    {
+        Lock guard(m_channelMutex);
+        toSend.reserve(m_channels.size());
+
+        for(m_channels_t::iterator channelsIter = m_channels.begin();
+            channelsIter != m_channels.end(); channelsIter++)
+        {
+            SearchInstance::shared_pointer inst(channelsIter->second.lock());
+            if(!inst) continue;
+            toSend.push_back(inst);
+        }
+    }
+
+    vector<SearchInstance::shared_pointer>::iterator siter = toSend.begin();
+    for (; siter != toSend.end(); siter++)
+    {
+        generateSearchRequestMessage(*siter, buffer, control);
+    }
+}
+
 void ChannelSearchManager::initializeSendBuffer()
 {
     // for now OK, since it is only set here
@@ -236,15 +305,41 @@ void ChannelSearchManager::flushSendBuffer()
 {
     Lock guard(m_mutex);
 
+    // UDP transport
     Transport::shared_pointer tt = m_context.lock()->getSearchTransport();
     BlockingUDPTransport::shared_pointer ut = std::tr1::static_pointer_cast<BlockingUDPTransport>(tt);
 
+    // UDP search
     m_sendBuffer.putByte(CAST_POSITION, (int8_t)0x80);  // unicast, no reply required
     ut->send(&m_sendBuffer, inetAddressType_unicast);
 
     m_sendBuffer.putByte(CAST_POSITION, (int8_t)0x00);  // b/m-cast, no reply required
     ut->send(&m_sendBuffer, inetAddressType_broadcast_multicast);
 
+    // Name server transport
+    if(m_nsTransport)
+    {
+        // Reset transport (current connection only)
+        // after max. number of attempts is reached.
+        if (m_nsSearchCounter >= MAX_NAME_SERVER_SEARCH_COUNT)
+        {
+            LOG(logLevelDebug, "Resetting name server transport after %d search attempts", m_nsSearchCounter);
+            releaseNameServerTransport(true);
+        }
+    }
+
+    if(!m_nsTransport)
+    {
+        m_nsTransport = m_context.lock()->getNameServerSearchTransport();
+    }
+
+    // Name server search
+    if(m_nsTransport)
+    {
+        m_nsSearchCounter++;
+        LOG(logLevelDebug, "Initiating name server search for %d channels, search attempt %d", int(m_channels.size()), m_nsSearchCounter);
+        m_nsTransport->enqueueSendRequest(shared_from_this());
+    }
     initializeSendBuffer();
 }
 
@@ -253,7 +348,6 @@ bool ChannelSearchManager::generateSearchRequestMessage(SearchInstance::shared_p
         ByteBuffer* requestMessage, TransportSendControl* control)
 {
     epics::pvData::int16 dataCount = requestMessage->getShort(DATA_COUNT_POSITION);
-
     dataCount++;
 
     /*
@@ -262,6 +356,8 @@ bool ChannelSearchManager::generateSearchRequestMessage(SearchInstance::shared_p
     */
 
     const std::string& name(channel->getSearchInstanceName());
+    LOG(logLevelDebug, "Searching for channel: %s", name.c_str());
+
     // not nice...
     const int addedPayloadSize = sizeof(int32)/sizeof(int8) + (1 + sizeof(int32)/sizeof(int8) + name.length());
     if(((int)requestMessage->getRemaining()) < addedPayloadSize)

src/remote/codec.cpp

@@ -1608,8 +1608,8 @@ void BlockingServerTCPTransportCodec::destroyAllChannels() {
     {
         LOG(
             logLevelDebug,
-            "Transport to %s still has %zu channel(s) active and closing...",
-            _socketName.c_str(), _channels.size());
+            "Transport to %s still has %lu channel(s) active and closing...",
+            _socketName.c_str(), static_cast<unsigned long>(_channels.size()));
     }
 
     _channels_t temp;
@@ -1761,7 +1761,7 @@ bool BlockingClientTCPTransportCodec::acquire(ClientChannelImpl::shared_pointer
 
     if (IS_LOGGABLE(logLevelDebug))
     {
-        LOG(logLevelDebug, "Acquiring transport to %s.", _socketName.c_str());
+        LOG(logLevelDebug, "Acquiring transport to %s for channel cid %d.", _socketName.c_str(), client->getID());
     }
 
     _owners[client->getID()] = ClientChannelImpl::weak_pointer(client);
@@ -1789,8 +1789,8 @@ void BlockingClientTCPTransportCodec::internalClose() {
         {
             LOG(
                 logLevelDebug,
-                "Transport to %s still has %zu client(s) active and closing...",
-                _socketName.c_str(), refs);
+                "Transport to %s still has %lu client(s) active and closing...",
+                _socketName.c_str(), static_cast<unsigned long>(refs));
         }
 
         TransportClientMap_t::iterator it = _owners.begin();
@@ -1828,6 +1828,11 @@ void BlockingClientTCPTransportCodec::release(pvAccessID clientID) {
     }
 }
 
+bool BlockingClientTCPTransportCodec::isUsed()
+{
+    return (_owners.size() > 0);
+}
+
 void BlockingClientTCPTransportCodec::send(ByteBuffer* buffer,
                                            TransportSendControl* control)
 {

src/remote/pv/channelSearchManager.h

@@ -53,6 +53,7 @@ class SearchInstance {
 
 class ChannelSearchManager :
         public epics::pvData::TimerCallback,
+        public TransportSender,
         public std::tr1::enable_shared_from_this<ChannelSearchManager>
 {
 public:
@@ -99,13 +100,19 @@ class ChannelSearchManager :
     /// Timer stooped callback.
     virtual void timerStopped() OVERRIDE FINAL;
 
+    // Transport sender interface.
+    virtual void send(epics::pvData::ByteBuffer* buffer, TransportSendControl* control) OVERRIDE FINAL;
+
     /**
      * Private constructor.
      * @param context
      */
     ChannelSearchManager(Context::shared_pointer const & context);
     void activate();
 
+    // Releases name server transport.
+    void releaseNameServerTransport(bool forceRelease=false);
+
 private:
 
     bool generateSearchRequestMessage(SearchInstance::shared_pointer const & channel, bool allowNewFrame, bool flush);
@@ -140,6 +147,16 @@ class ChannelSearchManager :
      */
     int32_t m_sequenceNumber;
 
+    /**
+     * Name server search attempt counter.
+     */
+    int m_nsSearchCounter;
+
+    /**
+     * Name server transport
+     */
+    Transport::shared_pointer m_nsTransport;
+
     /**
      * Send byte buffer (frame)
      */

src/remote/pv/codec.h

@@ -612,6 +612,8 @@ class BlockingClientTCPTransportCodec :
 
     virtual void release(pvAccessID clientId) OVERRIDE FINAL;
 
+    virtual bool isUsed() OVERRIDE FINAL;
+
     virtual void send(epics::pvData::ByteBuffer* buffer,
                       TransportSendControl* control) OVERRIDE FINAL;
 

src/remote/pv/remote.h

@@ -18,7 +18,6 @@
 #include <osiSock.h>
 
 #include <pv/serialize.h>
-#include <pv/pvType.h>
 #include <pv/byteBuffer.h>
 #include <pv/timer.h>
 #include <pv/pvData.h>
@@ -185,6 +184,11 @@ class epicsShareClass Transport : public epics::pvData::DeserializableControl {
      */
     virtual void release(pvAccessID clientId) = 0;
 
+    /**
+     * Is transport used
+     */
+    virtual bool isUsed() {return false;}
+
     /**
      * Get protocol type (tcp, udp, ssl, etc.).
      * @return protocol type.
@@ -305,6 +309,8 @@ class Context {
 
     virtual std::tr1::shared_ptr<Channel> getChannel(pvAccessID id) = 0;
     virtual Transport::shared_pointer getSearchTransport() = 0;
+    virtual Transport::shared_pointer getNameServerSearchTransport() {return Transport::shared_pointer();}
+    virtual void releaseNameServerSearchTransport(const Transport::shared_pointer& nsTransport=Transport::shared_pointer(), bool releaseAllConnections=true) {}
 };
 
 /**

src/remote/transportRegistry.cpp

@@ -127,7 +127,7 @@ void TransportRegistry::clear()
     if(temp.empty())
         return;
 
-    LOG(logLevelDebug, "Context still has %zu transport(s) active and closing...", temp.size());
+    LOG(logLevelDebug, "Context still has %lu transport(s) active and closing...", static_cast<unsigned long>(temp.size()));
 
     for(transports_t::iterator it(temp.begin()), end(temp.end());
         it != end; ++it)