Merge 046c1f2 into f1268ad

Author: sveseli

src/pva/Makefile

@@ -8,5 +8,6 @@ INC += pv/pvaVersion.h
 INC += pv/pvaVersionNum.h
 INC += pv/clientFactory.h
 
+pvAccess_SRCS += pvaConstants.cpp
 pvAccess_SRCS += pvaVersion.cpp
 pvAccess_SRCS += clientFactory.cpp

src/pva/pv/pvaConstants.h

@@ -22,8 +22,7 @@
 #endif
 #include <shareLib.h>
 
-namespace epics {
-namespace pvAccess {
+namespace epics { namespace pvAccess {
 
 /** PVA protocol magic number */
 const epics::pvData::int8 PVA_MAGIC = static_cast<epics::pvData::int8>(0xCA);
@@ -43,6 +42,9 @@ const epics::pvData::int32 PVA_SERVER_PORT = 5075;
 /** Default PVA beacon port. */
 const epics::pvData::int32 PVA_BROADCAST_PORT = 5076;
 
+/** Default UDP sender port. */
+const epics::pvData::int32 PVA_UDP_SENDER_PORT = 0;
+
 /** PVA protocol message header size. */
 const epics::pvData::int16 PVA_MESSAGE_HEADER_SIZE = 8;
 
@@ -86,7 +88,7 @@ epicsShareExtern const std::string PVACCESS_ALL_PROVIDERS;
 
 /** Name of the system env. variable to turn on debugging. */
 epicsShareExtern const std::string PVACCESS_DEBUG;
-}
-}
+
+}}
 
 #endif /* PVACONSTANTS_H_ */

src/pva/pvaConstants.cpp

@@ -0,0 +1,16 @@
+/**
+* Copyright - See the COPYRIGHT that is included with this distribution.
+* pvAccessCPP is distributed subject to a Software License Agreement found
+* in file LICENSE that is included with this distribution.
+*/
+
+#define epicsExportSharedSymbols
+#include <pv/pvaConstants.h>
+
+namespace epics { namespace pvAccess {
+
+const std::string PVACCESS_DEFAULT_PROVIDER("local");
+const std::string PVACCESS_ALL_PROVIDERS("<all>");
+const std::string PVACCESS_DEBUG("EPICS_PVA_DEBUG");
+
+}}

src/pva/pvaVersion.cpp

@@ -16,10 +16,6 @@ using std::string;
 namespace epics {
 namespace pvAccess {
 
-const std::string PVACCESS_DEFAULT_PROVIDER("local");
-const std::string PVACCESS_ALL_PROVIDERS("<all>");
-const std::string PVACCESS_DEBUG("EPICS_PVA_DEBUG");
-
 Version::Version(std::string const & productName,
                  std::string const & implementationLangugage,
                  int majorVersion, int minorVersion,

src/remote/abstractResponseHandler.cpp

@@ -10,6 +10,7 @@
 #include <pv/reftrack.h>
 
 #define epicsExportSharedSymbols
+#include <pv/pvaConstants.h>
 #include <pv/remote.h>
 #include <pv/hexDump.h>
 

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(),
@@ -580,6 +580,7 @@ void initializeUDPTransports(bool serverFlag,
                              const ResponseHandler::shared_pointer& responseHandler,
                              BlockingUDPTransport::shared_pointer& sendTransport,
                              int32& listenPort,
+                             int32& senderPort,
                              bool autoAddressList,
                              const std::string& addressList,
                              const std::string& ignoreAddressList)
@@ -595,7 +596,7 @@ void initializeUDPTransports(bool serverFlag,
     osiSockAddr anyAddress;
     memset(&anyAddress, 0, sizeof(anyAddress));
     anyAddress.ia.sin_family = AF_INET;
-    anyAddress.ia.sin_port = htons(0);
+    anyAddress.ia.sin_port = htons(senderPort);
     anyAddress.ia.sin_addr.s_addr = htonl(INADDR_ANY);
 
     sendTransport = connector.connect(responseHandler, anyAddress, protoVer);
@@ -684,7 +685,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 +715,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

@@ -135,6 +135,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 +155,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 +182,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 +191,15 @@ void ChannelSearchManager::searchResponse(const ServerGUID & guid, pvAccessID ci
         if(si)
             si->searchResponse(guid, minorRevision, serverAddress);
     }
+    releaseNameServerTransport();
+}
+
+void ChannelSearchManager::releaseNameServerTransport()
+{
+    if(m_channels.size() == 0)
+    {
+        m_context.lock()->releaseNameServerSearchTransport();
+    }
 }
 
 void ChannelSearchManager::newServerDetected()
@@ -196,6 +208,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
@@ -245,6 +297,13 @@ void ChannelSearchManager::flushSendBuffer()
     m_sendBuffer.putByte(CAST_POSITION, (int8_t)0x00);  // b/m-cast, no reply required
     ut->send(&m_sendBuffer, inetAddressType_broadcast_multicast);
 
+    // Name server search
+    Transport::shared_pointer nsTransport = m_context.lock()->getNameServerSearchTransport();
+    if(nsTransport)
+    {
+        LOG(logLevelDebug, "Initiating name server search for %d channels", int(m_channels.size()));
+        nsTransport->enqueueSendRequest(shared_from_this());
+    }
     initializeSendBuffer();
 }
 
@@ -253,7 +312,6 @@ bool ChannelSearchManager::generateSearchRequestMessage(SearchInstance::shared_p
         ByteBuffer* requestMessage, TransportSendControl* control)
 {
     epics::pvData::int16 dataCount = requestMessage->getShort(DATA_COUNT_POSITION);
-
     dataCount++;
 
     /*
@@ -262,6 +320,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/blockingUDP.h

@@ -76,7 +76,7 @@ class BlockingUDPTransport :
     }
 
     virtual std::string getType() const OVERRIDE FINAL {
-        return std::string("udp");
+        return "udp";
     }
 
     virtual std::size_t getReceiveBufferSize() const OVERRIDE FINAL {
@@ -401,6 +401,7 @@ void initializeUDPTransports(
     const ResponseHandler::shared_pointer& responseHandler,
     BlockingUDPTransport::shared_pointer& sendTransport,
     epics::pvData::int32& listenPort,
+    epics::pvData::int32& senderPort,
     bool autoAddressList,
     const std::string& addressList,
     const std::string& ignoreAddressList);

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();
+
 private:
 
     bool generateSearchRequestMessage(SearchInstance::shared_pointer const & channel, bool allowNewFrame, bool flush);