CORE: Communication thread exception handler. (#999)

Signed-off-by: Raul Akhmetshin <[email protected]>

Author: rakhmets

src/core/agent_data.h

@@ -99,7 +99,12 @@ class nixlAgentData {
         bool                               commThreadStop;
         bool                               useEtcd;
         std::unique_ptr<nixlTelemetry> telemetry_;
-        void commWorker(nixlAgent* myAgent);
+        std::exception_ptr commThreadException_;
+
+        void
+        commWorker(nixlAgent &myAgent) noexcept;
+        void
+        commWorkerInternal(nixlAgent *myAgent);
         void enqueueCommWork(nixl_comm_req_t request);
         void getCommWork(std::vector<nixl_comm_req_t> &req_list);
         nixl_status_t

src/core/nixl_agent.cpp

@@ -194,8 +194,7 @@ nixlAgent::nixlAgent(const std::string &name, const nixlAgentConfig &cfg) :
 
     if (data->useEtcd || cfg.useListenThread) {
         data->commThreadStop = false;
-        data->commThread =
-            std::thread(&nixlAgentData::commWorker, data.get(), this);
+        data->commThread = std::thread(&nixlAgentData::commWorker, data.get(), std::ref(*this));
     }
 }
 
@@ -204,6 +203,15 @@ nixlAgent::~nixlAgent() {
         data->commThreadStop = true;
         if(data->commThread.joinable()) data->commThread.join();
 
+        try {
+            if (data->commThreadException_) {
+                std::rethrow_exception(data->commThreadException_);
+            }
+        }
+        catch (const std::exception &e) {
+            NIXL_WARN << "Communication thread has thrown an exception: " << e.what();
+        }
+
         // Close remaining connections from comm thread
         for (auto &[remote, fd] : data->remoteSockets) {
             shutdown(fd, SHUT_RDWR);

src/core/nixl_listener.cpp

@@ -106,18 +106,22 @@ sendCommMessage(int fd, const std::string& msg) {
     };
 
     for (size_t i = 0, offset = 0, sent = 0; i < iov_size;) {
-        auto bytes = send(fd, static_cast<char *>(iov[i].iov_base) + offset, iov[i].iov_len - offset, 0);
+        auto bytes = send(fd,
+                          static_cast<char *>(iov[i].iov_base) + offset,
+                          iov[i].iov_len - offset,
+                          MSG_NOSIGNAL);
         if (bytes < 0) {
             if (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK) {
                 continue;
             }
 
             throw std::runtime_error(
-                    absl::StrFormat("sendCommMessage(fd=%d) %zu/%zu bytes failed, errno=%d",
-                                    fd,
-                                    sent,
-                                    size + sizeof(size),
-                                    errno));
+                absl::StrFormat("sendCommMessage(fd=%d, msg=%s) %zu/%zu bytes failed, errno=%d",
+                                fd,
+                                msg.c_str(),
+                                sent,
+                                size + sizeof(size),
+                                errno));
         }
 
         offset += bytes;
@@ -435,8 +439,18 @@ class nixlEtcdClient {
 
 } // unnamed namespace
 
-void nixlAgentData::commWorker(nixlAgent* myAgent){
+void
+nixlAgentData::commWorker(nixlAgent &myAgent) noexcept {
+    try {
+        commWorkerInternal(&myAgent);
+    }
+    catch (...) {
+        commThreadException_ = std::current_exception();
+    }
+}
 
+void
+nixlAgentData::commWorkerInternal(nixlAgent *myAgent) {
 #if HAVE_ETCD
     std::unique_ptr<nixlEtcdClient> etcdClient = nullptr;
     // useEtcd is set in nixlAgent constructor and is true if NIXL_ETCD_ENDPOINTS is set