CA thread attach

src/ca/caChannel.cpp

@@ -27,7 +27,7 @@ using std::string;
 
 PVACCESS_REFCOUNT_MONITOR_DEFINE(caChannel);
 
-CAChannel::shared_pointer CAChannel::create(ChannelProvider::shared_pointer const & channelProvider,
+CAChannel::shared_pointer CAChannel::create(CAChannelProvider::shared_pointer const & channelProvider,
                                             std::string const & channelName,
                                             short priority,
                                             ChannelRequester::shared_pointer const & channelRequester)
@@ -185,7 +185,7 @@ void CAChannel::disconnected()
 }
 
 CAChannel::CAChannel(std::string const & _channelName,
-                     ChannelProvider::shared_pointer const & _channelProvider,
+                     CAChannelProvider::shared_pointer const & _channelProvider,
                      ChannelRequester::shared_pointer const & _channelRequester) :
     channelName(_channelName),
     channelProvider(_channelProvider),
@@ -423,6 +423,8 @@ void CAChannel::message(std::string const & message,MessageType messageType)
 
 void CAChannel::destroy()
 {
+    threadAttach();
+
     Lock lock(requestsMutex);
     {
         while (!requests.empty())
@@ -439,6 +441,11 @@ void CAChannel::destroy()
 
 /* ---------------------------------------------------------- */
 
+void CAChannel::threadAttach()
+{
+    std::tr1::static_pointer_cast<CAChannelProvider>(channelProvider)->threadAttach();
+}
+
 void CAChannel::registerRequest(ChannelRequest::shared_pointer const & request)
 {
     Lock lock(requestsMutex);
@@ -871,6 +878,8 @@ void CAChannelGet::getDone(struct event_handler_args &args)
 
 void CAChannelGet::get()
 {
+    channel->threadAttach();
+
     /*
     From R3.14.12 onwards ca_array_get_callback() replies will give a CA client application the current number
     of elements in an array field, provided they specified an element count of zero in their original request.
@@ -1176,6 +1185,8 @@ void CAChannelPut::putDone(struct event_handler_args &args)
 void CAChannelPut::put(PVStructure::shared_pointer const & pvPutStructure,
                        BitSet::shared_pointer const & /*putBitSet*/)
 {
+    channel->threadAttach();
+
     doPut putFunc = doPutFuncTable[channel->getNativeType()];
     if (putFunc)
     {
@@ -1229,6 +1240,8 @@ void CAChannelPut::getDone(struct event_handler_args &args)
 
 void CAChannelPut::get()
 {
+    channel->threadAttach();
+
     int result = ca_array_get_callback(getType, channel->getElementCount(),
                                        channel->getChannelID(), ca_put_get_handler, this);
     if (result == ECA_NORMAL)
@@ -1388,6 +1401,8 @@ void CAChannelMonitor::subscriptionEvent(struct event_handler_args &args)
 
 epics::pvData::Status CAChannelMonitor::start()
 {
+    channel->threadAttach();
+
     /*
     From R3.14.12 onwards when using the IOC server and the C++ client libraries monitor callbacks
     replies will give a CA client application the current number of elements in an array field,
@@ -1419,6 +1434,8 @@ epics::pvData::Status CAChannelMonitor::start()
 
 epics::pvData::Status CAChannelMonitor::stop()
 {
+    channel->threadAttach();
+
     int result = ca_clear_subscription(eventID);
 
     if (result == ECA_NORMAL)
@@ -1466,6 +1483,8 @@ void CAChannelMonitor::cancel()
 
 void CAChannelMonitor::destroy()
 {
+    channel->threadAttach();
+
     ca_clear_subscription(eventID);
 
     // TODO

src/ca/caChannel.h

@@ -12,6 +12,8 @@
 /* for CA */
 #include <cadef.h>
 
+#include <caProvider.h>
+
 namespace epics {
 namespace pvAccess {
 namespace ca {
@@ -24,7 +26,7 @@ class CAChannel :
 public:
     POINTER_DEFINITIONS(CAChannel);
 
-    static shared_pointer create(ChannelProvider::shared_pointer const & channelProvider,
+    static shared_pointer create(CAChannelProvider::shared_pointer const & channelProvider,
                                             std::string const & channelName,
                                             short priority,
                                             ChannelRequester::shared_pointer const & channelRequester);
@@ -95,19 +97,21 @@ public:
 
     /* ---------------------------------------------------------------- */
 
+    void threadAttach();
+
     void registerRequest(ChannelRequest::shared_pointer const & request);
     void unregisterRequest(ChannelRequest::shared_pointer const & request);
 
 private:
 
     CAChannel(std::string const & channelName,
-              ChannelProvider::shared_pointer const & channelProvider,
+              CAChannelProvider::shared_pointer const & channelProvider,
               ChannelRequester::shared_pointer const & channelRequester);
     void activate(short priority);
 
     std::string channelName;
     
-    ChannelProvider::shared_pointer channelProvider;
+    CAChannelProvider::shared_pointer channelProvider;
     ChannelRequester::shared_pointer channelRequester;
 
     chid channelID;

src/ca/caProvider.cpp

@@ -27,7 +27,7 @@ using namespace epics::pvAccess::ca;
 
 std::string CAChannelProvider::PROVIDER_NAME = "ca";
 
-CAChannelProvider::CAChannelProvider()
+CAChannelProvider::CAChannelProvider() : current_context(0)
 {
     initialize();
 }
@@ -75,6 +75,8 @@ Channel::shared_pointer CAChannelProvider::createChannel(
         ChannelRequester::shared_pointer const & channelRequester,
         short priority)
 {
+    threadAttach();
+
     static std::string emptyString;
     return createChannel(channelName, channelRequester, priority, emptyString);
 }
@@ -119,6 +121,11 @@ void CAChannelProvider::destroy()
     ca_context_destroy();
 }
 
+void CAChannelProvider::threadAttach()
+{
+    ca_attach_context(current_context);
+}
+
 void CAChannelProvider::registerChannel(Channel::shared_pointer const & channel)
 {
     Lock lock(channelsMutex);
@@ -140,6 +147,8 @@ void CAChannelProvider::initialize()
                 "to start channel access:") + ca_message(result));
     }
 
+    current_context = ca_current_context();
+
     // TODO create a ca_poll thread, if ca_disable_preemptive_callback
 }
 

src/ca/caProvider.h

@@ -7,6 +7,8 @@
 #ifndef CAPROVIDER_H
 #define CAPROVIDER_H
 
+#include <cadef.h>
+
 #include <pv/pvAccess.h>
 #include <map>
 
@@ -57,6 +59,8 @@ public:
 
     /* ---------------------------------------------------------------- */
 
+    void threadAttach();
+
     void registerChannel(Channel::shared_pointer const & channel);
     void unregisterChannel(Channel::shared_pointer const & channel);
 
@@ -64,6 +68,8 @@ private:
 
     void initialize();
 
+    ca_client_context* current_context;
+
     epics::pvData::Mutex channelsMutex;
     // TODO std::unordered_map
     // void* is not the nicest thing, but there is no fast weak_ptr==