Initializers

.ci/jenkins/lib/build-matrix.yaml

@@ -59,7 +59,7 @@ steps:
     parallel: false
     timeout: "${TEST_TIMEOUT}"
     run: |
-      .gitlab/test_cpp.sh ${NIXL_INSTALL_DIR}
+      for i in $(seq 1 30); do .gitlab/test_cpp.sh ${NIXL_INSTALL_DIR} ; done
 
   - name: Test Python
     parallel: false

.ci/jenkins/lib/test-matrix.yaml

@@ -25,7 +25,7 @@ timeout_minutes: 240
 # label is defined at jenkins slave configuration, we want to run the job on a gpu agent and be able to esaly replace it without having to change this file
 runs_on_agents:
   - {nodeLabel: 'H100'}
-  # - {nodeLabel: 'DGX'}
+  - {nodeLabel: 'DGX'}
 
 matrix:
   axes:

src/core/nixl_listener.cpp

@@ -35,10 +35,12 @@ namespace {
 
 static const std::string invalid_label = "invalid";
 
-int connectToIP(std::string ip_addr, int port) {
+int
+connectToIP(std::string ip_addr, int port) {
 
-    struct sockaddr_in listenerAddr;
-    listenerAddr.sin_port   = htons(port);
+    struct sockaddr_in listenerAddr {};
+
+    listenerAddr.sin_port = htons(port);
     listenerAddr.sin_family = AF_INET;
 
     if (inet_pton(AF_INET, ip_addr.c_str(), &listenerAddr.sin_addr) <= 0) {

src/infra/nixl_descriptors.cpp

@@ -112,19 +112,23 @@ nixlBlobDesc::nixlBlobDesc(const nixlBasicDesc &desc,
 }
 
 nixlBlobDesc::nixlBlobDesc(const nixl_blob_t &blob) {
+    if (blob.size() < sizeof(nixlBasicDesc)) {
+        NIXL_ERROR << "Blob size is less than the size of nixlBasicDesc";
+        addr = 0;
+        len = 0;
+        devId = 0;
+        metaInfo.resize(0);
+        return;
+    }
     size_t meta_size = blob.size() - sizeof(nixlBasicDesc);
     if (meta_size > 0) {
         metaInfo.resize(meta_size);
         blob.copy(reinterpret_cast<char*>(this), sizeof(nixlBasicDesc));
-        blob.copy(reinterpret_cast<char*>(&metaInfo[0]),
-                 meta_size, sizeof(nixlBasicDesc));
+        blob.copy(reinterpret_cast<char *>(&metaInfo[0]), meta_size, sizeof(nixlBasicDesc));
     } else if (meta_size == 0) {
-        blob.copy(reinterpret_cast<char*>(this), sizeof(nixlBasicDesc));
-    } else { // Error
-        addr  = 0;
-        len   = 0;
-        devId = 0;
-        metaInfo.resize(0);
+        blob.copy(reinterpret_cast<char *>(this), sizeof(nixlBasicDesc));
+    } else {
+        NIXL_ASSERT(false) << "Negative meta size";
     }
 }
 

src/plugins/ucx/ucx_backend.cpp

@@ -1841,6 +1841,7 @@ nixl_status_t nixlUcxEngine::genNotif(const std::string &remote_agent, const std
     switch(ret) {
     case NIXL_IN_PROG:
         /* do not track the request */
+        // TODO: why are we releasing a request that is still in progress?
         getWorker(getWorkerId())->reqRelease(req);
     case NIXL_SUCCESS:
         break;

src/utils/stream/metadata_stream.cpp

@@ -132,21 +132,21 @@ std::string nixlMDStreamListener::recvFromClient() {
         return recvData;
 }
 
-void nixlMDStreamListener::recvFromClients(int clientSocket) {
-        char    buffer[RECV_BUFFER_SIZE];
-        int     bytes_read;
-
-        while ((bytes_read = recv(clientSocket, buffer,
-                                  sizeof(buffer), 0)) > 0) {
-              buffer[bytes_read] = '\0';
-              // Return ack
-              std::string ack = "Message received";
-              send(clientSocket, ack.c_str(), ack.size(), 0);
-              std::string recv_message(buffer);
-              NIXL_DEBUG << "Message Received" << recv_message;
-        }
-        close(clientSocket);
-        NIXL_DEBUG << "Client Disconnected";
+void
+nixlMDStreamListener::recvFromClients(int clientSocket) {
+    char buffer[RECV_BUFFER_SIZE + 1];
+    int bytes_read;
+
+    while ((bytes_read = recv(clientSocket, buffer, sizeof(buffer), 0)) > 0) {
+        buffer[bytes_read] = '\0';
+        // Return ack
+        std::string ack = "Message received";
+        send(clientSocket, ack.c_str(), ack.size(), 0);
+        std::string recv_message(buffer);
+        NIXL_DEBUG << "Message Received" << recv_message;
+    }
+    close(clientSocket);
+    NIXL_DEBUG << "Client Disconnected";
 }
 
 void nixlMDStreamListener::startListenerForClient() {
@@ -169,12 +169,14 @@ nixlMDStreamClient::~nixlMDStreamClient() {
     closeStream();
 }
 
-bool nixlMDStreamClient::setupClient() {
+bool
+nixlMDStreamClient::setupClient() {
     setupStream();
 
-    struct sockaddr_in listenerAddr;
+    struct sockaddr_in listenerAddr {};
+
     listenerAddr.sin_family = AF_INET;
-    listenerAddr.sin_port   = htons(port);
+    listenerAddr.sin_port = htons(port);
 
     if (inet_pton(AF_INET, listenerAddress.c_str(),
                   &listenerAddr.sin_addr) <= 0) {

src/utils/ucx/gpu_xfer_req_h.cpp

@@ -54,31 +54,33 @@ createGpuXferReq(const nixlUcxEp &ep,
     ucp_elements.reserve(local_mems.size());
 
     for (size_t i = 0; i < local_mems.size(); i++) {
-        ucp_device_mem_list_elem_t ucp_elem;
-        ucp_elem.field_mask = UCP_DEVICE_MEM_LIST_ELEM_FIELD_MEMH |
-            UCP_DEVICE_MEM_LIST_ELEM_FIELD_RKEY | UCP_DEVICE_MEM_LIST_ELEM_FIELD_LOCAL_ADDR |
-            UCP_DEVICE_MEM_LIST_ELEM_FIELD_REMOTE_ADDR | UCP_DEVICE_MEM_LIST_ELEM_FIELD_LENGTH;
-        ucp_elem.memh = local_mems[i].getMemh();
-        ucp_elem.rkey = remote_rkeys[i]->get();
-        ucp_elem.local_addr = local_mems[i].getBase();
-        ucp_elem.remote_addr = remote_addrs[i];
-        ucp_elem.length = local_mems[i].getSize();
+        ucp_device_mem_list_elem_t ucp_elem = {
+            .field_mask = UCP_DEVICE_MEM_LIST_ELEM_FIELD_MEMH |
+                UCP_DEVICE_MEM_LIST_ELEM_FIELD_RKEY | UCP_DEVICE_MEM_LIST_ELEM_FIELD_LOCAL_ADDR |
+                UCP_DEVICE_MEM_LIST_ELEM_FIELD_REMOTE_ADDR | UCP_DEVICE_MEM_LIST_ELEM_FIELD_LENGTH,
+            .memh = local_mems[i].getMemh(),
+            .rkey = remote_rkeys[i]->get(),
+            .local_addr = local_mems[i].getBase(),
+            .remote_addr = remote_addrs[i],
+            .length = local_mems[i].getSize(),
+        };
         ucp_elements.push_back(ucp_elem);
     }
 
-    ucp_device_mem_list_params_t params;
-    params.field_mask = UCP_DEVICE_MEM_LIST_PARAMS_FIELD_ELEMENTS |
+ucp_device_mem_list_params_t params = {
+    .field_mask = UCP_DEVICE_MEM_LIST_PARAMS_FIELD_ELEMENTS |
         UCP_DEVICE_MEM_LIST_PARAMS_FIELD_ELEMENT_SIZE |
-        UCP_DEVICE_MEM_LIST_PARAMS_FIELD_NUM_ELEMENTS;
-    params.elements = ucp_elements.data();
-    params.element_size = sizeof(ucp_device_mem_list_elem_t);
-    params.num_elements = ucp_elements.size();
-
-    ucp_device_mem_list_handle_h ucx_handle;
-    ucs_status_t ucs_status = ucp_device_mem_list_create(ep.getEp(), &params, &ucx_handle);
-    if (ucs_status != UCS_OK) {
-        throw std::runtime_error(std::string("Failed to create device memory list: ") +
-                                 ucs_status_string(ucs_status));
+        UCP_DEVICE_MEM_LIST_PARAMS_FIELD_NUM_ELEMENTS,
+    .elements = ucp_elements.data(),
+    .element_size = sizeof(ucp_device_mem_list_elem_t),
+    .num_elements = ucp_elements.size(),
+};
+
+ucp_device_mem_list_handle_h ucx_handle;
+ucs_status_t ucs_status = ucp_device_mem_list_create(ep.getEp(), &params, &ucx_handle);
+if (ucs_status != UCS_OK) {
+    throw std::runtime_error(std::string("Failed to create device memory list: ") +
+                             ucs_status_string(ucs_status));
     }
 
     NIXL_DEBUG << "Created device memory list: ep=" << ep.getEp() << " handle=" << ucx_handle