Merge commit '731767b086e8146f31c1f9038568e3c852c609f7'

* commit '731767b086e8146f31c1f9038568e3c852c609f7': SharedPV fix onFirstConnect() on monitor creation SharedPV allow use of Channel::destroy() to trigger onLastDisconnect() SharedPV defer onFirstConnect() to first getField/Put/Monitor client.h add some exception guards more SharedPV rpc state tracking more UDP debugging fix global string constants fix SharedPV allow rpc() while close()'d RPC troubleshooting minor SharedPV too many notifiedConn client.h minor
epics-base · Oct 8, 2018 · 8d2edc0 · 8d2edc0
2 parents 30f07f2 + 731767b
commit 8d2edc0
Show file tree

Hide file tree

Showing 14 changed files with 208 additions and 81 deletions.
diff --git a/src/client/clientGet.cpp b/src/client/clientGet.cpp
@@ -50,7 +50,11 @@ struct Getter : public pvac::detail::CallbackStorage,
         pvac::ClientChannel::GetCallback *C=cb;
         cb = 0;
         CallbackUse U(G);
-        C->getDone(event);
+        try {
+            C->getDone(event);
+        } catch(std::exception& e) {
+            LOG(pva::logLevelInfo, "Lost exception during getDone(): %s", e.what());
+        }
     }
 
     virtual std::string name() const OVERRIDE FINAL

diff --git a/src/client/clientPut.cpp b/src/client/clientPut.cpp
@@ -54,7 +54,11 @@ struct Putter : public pvac::detail::CallbackStorage,
         pvac::ClientChannel::PutCallback *C=cb;
         cb = 0;
         CallbackUse U(G);
-        C->putDone(event);
+        try {
+            C->putDone(event);
+        } catch(std::exception& e) {
+            LOG(pva::logLevelInfo, "Lost exception during putDone(): %s", e.what());
+        }
     }
 
     virtual std::string name() const OVERRIDE FINAL
@@ -205,13 +209,13 @@ namespace pvac {
 Operation
 ClientChannel::put(PutCallback* cb,
                    epics::pvData::PVStructure::const_shared_pointer pvRequest,
-                   bool getcurrent)
+                   bool getprevious)
 {
     if(!impl) throw std::logic_error("Dead Channel");
     if(!pvRequest)
         pvRequest = pvd::createRequest("field()");
 
-    std::tr1::shared_ptr<Putter> ret(Putter::build(cb, getcurrent));
+    std::tr1::shared_ptr<Putter> ret(Putter::build(cb, getprevious));
 
     {
         Guard G(ret->mutex);

diff --git a/src/client/clientRPC.cpp b/src/client/clientRPC.cpp
@@ -29,6 +29,7 @@ struct RPCer : public pvac::detail::CallbackStorage,
     operation_type::shared_pointer op;
 
     pvac::ClientChannel::GetCallback *cb;
+    // 'event' may be modified as long as cb!=NULL
     pvac::GetEvent event;
 
     pvd::PVStructure::const_shared_pointer args;
@@ -70,7 +71,7 @@ struct RPCer : public pvac::detail::CallbackStorage,
     }
 
     // called automatically via wrapped_shared_from_this
-    virtual void cancel()
+    virtual void cancel() OVERRIDE FINAL
     {
         std::tr1::shared_ptr<RPCer> keepalive(internal_shared_from_this());
         CallbackGuard G(*this);
@@ -86,7 +87,7 @@ struct RPCer : public pvac::detail::CallbackStorage,
 
     virtual void channelRPCConnect(
         const epics::pvData::Status& status,
-        pva::ChannelRPC::shared_pointer const & operation)
+        pva::ChannelRPC::shared_pointer const & operation) OVERRIDE FINAL
     {
         std::tr1::shared_ptr<RPCer> keepalive(internal_shared_from_this());
         CallbackGuard G(*this);
@@ -110,6 +111,7 @@ struct RPCer : public pvac::detail::CallbackStorage,
     {
         std::tr1::shared_ptr<RPCer> keepalive(internal_shared_from_this());
         CallbackGuard G(*this);
+        if(!cb) return;
         event.message = "Disconnect";
 
         callEvent(G);
@@ -118,7 +120,7 @@ struct RPCer : public pvac::detail::CallbackStorage,
     virtual void requestDone(
         const epics::pvData::Status& status,
         pva::ChannelRPC::shared_pointer const & operation,
-        epics::pvData::PVStructure::shared_pointer const & pvResponse)
+        epics::pvData::PVStructure::shared_pointer const & pvResponse) OVERRIDE FINAL
     {
         std::tr1::shared_ptr<RPCer> keepalive(internal_shared_from_this());
         CallbackGuard G(*this);

diff --git a/src/client/pva/client.h b/src/client/pva/client.h
@@ -385,7 +385,7 @@ class epicsShareClass ClientChannel
     //!                    If false, then previous=NULL
     Operation put(PutCallback* cb,
                   epics::pvData::PVStructure::const_shared_pointer pvRequest = epics::pvData::PVStructure::const_shared_pointer(),
-                  bool getcurrent = false);
+                  bool getprevious = false);
 
     //! Synchronious put operation
     inline

diff --git a/src/pva/pv/pvaConstants.h b/src/pva/pv/pvaConstants.h
@@ -18,6 +18,7 @@
 #   define epicsExportSharedSymbols
 #	undef pvaConstantsepicsExportSharedSymbols
 #endif
+#include <shareLib.h>
 
 namespace epics {
 namespace pvAccess {
@@ -73,13 +74,13 @@ const epics::pvData::int16 INVALID_DATA_TYPE = 0xFFFF;
 const epics::pvData::int32 INVALID_IOID = 0;
 
 /** Default PVA provider name. */
-const std::string PVACCESS_DEFAULT_PROVIDER;
+epicsShareExtern const std::string PVACCESS_DEFAULT_PROVIDER;
 
 /** "All-providers registered" PVA provider name. */
-const std::string PVACCESS_ALL_PROVIDERS;
+epicsShareExtern const std::string PVACCESS_ALL_PROVIDERS;
 
 /** Name of the system env. variable to turn on debugging. */
-const std::string PVACCESS_DEBUG;
+epicsShareExtern const std::string PVACCESS_DEBUG;
 }
 }
 

diff --git a/src/pva/pvaVersion.cpp b/src/pva/pvaVersion.cpp
@@ -8,16 +8,17 @@
 
 #define epicsExportSharedSymbols
 #include <pv/pvaVersion.h>
+#include <pv/pvaConstants.h>
 
 using std::stringstream;
 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";
+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,

diff --git a/src/remote/blockingUDPTransport.cpp b/src/remote/blockingUDPTransport.cpp
@@ -225,19 +225,30 @@ void BlockingUDPTransport::run() {
                                      0, (sockaddr*)&fromAddress,
                                      &addrStructSize);
 
-            if(likely(bytesRead>0)) {
+            if(likely(bytesRead>=0)) {
                 // successfully got datagram
                 bool ignore = false;
                 for(size_t i = 0; i <_ignoredAddresses.size(); i++)
                 {
                     if(_ignoredAddresses[i].ia.sin_addr.s_addr==fromAddress.ia.sin_addr.s_addr)
                     {
                         ignore = true;
+                        if(pvAccessIsLoggable(logLevelDebug)) {
+                            char strBuffer[64];
+                            sockAddrToDottedIP(&fromAddress.sa, strBuffer, sizeof(strBuffer));
+                            LOG(logLevelDebug, "UDP Ignore (%d) %s x- %s", bytesRead, _remoteName.c_str(), strBuffer);
+                        }
                         break;
                     }
                 }
 
                 if(likely(!ignore)) {
+                    if(pvAccessIsLoggable(logLevelDebug)) {
+                        char strBuffer[64];
+                        sockAddrToDottedIP(&fromAddress.sa, strBuffer, sizeof(strBuffer));
+                        LOG(logLevelDebug, "UDP Rx (%d) %s <- %s", bytesRead, _remoteName.c_str(), strBuffer);
+                    }
+
                     _receiveBuffer.setPosition(RECEIVE_BUFFER_PRE_RESERVE);
                     _receiveBuffer.setLimit(RECEIVE_BUFFER_PRE_RESERVE+bytesRead);
 
@@ -253,8 +264,7 @@ void BlockingUDPTransport::run() {
                             __FILE__, __LINE__);
                     }
                 }
-            }
-            else if (unlikely(bytesRead == -1)) {
+            } else {
 
                 int socketError = SOCKERRNO;
 
@@ -400,7 +410,7 @@ bool BlockingUDPTransport::send(const char* buffer, size_t length, const osiSock
 {
     if (IS_LOGGABLE(logLevelDebug))
     {
-        LOG(logLevelDebug, "Sending %zu bytes %s -> %s.",
+        LOG(logLevelDebug, "UDP Tx (%zu) %s -> %s.",
             length, _remoteName.c_str(), inetAddressToString(address).c_str());
     }
 

diff --git a/src/server/pva/sharedstate.h b/src/server/pva/sharedstate.h
@@ -206,7 +206,10 @@ class epicsShareClass SharedPV
     //! Used for initial Monitor update and Get operations.
     epics::pvData::BitSet valid;
 
-    bool notifiedConn; // whether onFirstConnect() has been, or is being, called
+    // whether onFirstConnect() has been, or is being, called.
+    // Set when the first getField, Put, or Monitor (but not RPC) is created.
+    // Cleared when the last Channel is destroyed.
+    bool notifiedConn;
 
     int debugLvl;
 

diff --git a/src/server/responseHandlers.cpp b/src/server/responseHandlers.cpp
@@ -2988,6 +2988,7 @@ ServerChannelRPCRequesterImpl::ServerChannelRPCRequesterImpl(
     const pvAccessID ioid, Transport::shared_pointer const & transport):
     BaseChannelRequester(context, channel, ioid, transport),
     _channelRPC(), _pvResponse()
+  ,_status(Status::fatal("Invalid State"))
 
 {
 }
@@ -3090,6 +3091,7 @@ void ServerChannelRPCRequesterImpl::send(ByteBuffer* buffer, TransportSendContro
                 SerializationHelper::serializeStructureFull(buffer, control, _pvResponse);
             }
         }
+        _status = Status::fatal("Stale state");
     }
 
     stopRequest();