Merge branch 'openucx:master' into patch-1

Author: Laur1nMartins

contrib/test_jenkins.sh

@@ -657,10 +657,18 @@ run_ucx_perftest_cuda_device() {
 	ucx_perftest="$ucx_inst/bin/ucx_perftest"
 	ucp_test_args="-b $ucx_inst_ptest/test_types_ucp_device_cuda"
 
-	# TODO: Run on all GPUs
+	# TODO: Run on all GPUs & NICs combinations
 	ucp_client_args="-a cuda:0 $(hostname)"
+	gda_tls="cuda_copy,rc,rc_gda"
+	cuda_ipc_tls="cuda_copy,rc,cuda_ipc"
 
-	run_client_server_app "$ucx_perftest" "$ucp_test_args" "$ucp_client_args" 0 0
+	# TODO: Run with cuda_ipc_tls
+	for tls in "$gda_tls"
+	do
+		export UCX_TLS=${tls}
+		run_client_server_app "$ucx_perftest" "$ucp_test_args" "$ucp_client_args" 0 0
+	done
+	unset UCX_TLS
 }
 
 #

contrib/ucx_perftest_config/test_types_ucp_device_cuda

@@ -2,6 +2,8 @@
 # UCP basic device cuda tests
 #
 ucp_device_cuda_bw_1k_1thread         -t ucp_put_multi_bw -m cuda -s 1024 -n 10000
-ucp_device_cuda_bw_1k_128threads      -t ucp_put_multi_bw -m cuda -s 1024 -n 10000 -T 128
+# TODO - Increase number of threads after adjusting perftest.
+ucp_device_cuda_bw_1k_128threads      -t ucp_put_multi_bw -m cuda -s 1024 -n 10000 -T 32
 ucp_device_cuda_lat_1k_1thread        -t ucp_put_multi_lat -m cuda -s 1024 -n 10000
-ucp_device_cuda_lat_1k_128threads     -t ucp_put_multi_lat -m cuda -s 1024 -n 10000 -T 128
+# TODO - Increase number of threads after adjusting perftest.
+ucp_device_cuda_lat_1k_128threads     -t ucp_put_multi_lat -m cuda -s 1024 -n 10000 -T 32

src/tools/perf/cuda/cuda_kernel.cuh

@@ -79,9 +79,11 @@ UCS_F_DEVICE size_t ucx_bitset_popcount(const uint8_t *set, size_t bits) {
     return count;
 }
 
-UCS_F_DEVICE size_t ucx_bitset_ffs(const uint8_t *set, size_t bits, size_t from) {
+UCS_F_DEVICE size_t ucx_bitset_ffns(const uint8_t *set, size_t bits,
+                                    size_t from)
+{
     for (size_t i = from; i < bits; i++) {
-        if (UCX_BIT_GET(set, i)) {
+        if (!UCX_BIT_GET(set, i)) {
             return i;
         }
     }

src/tools/perf/cuda/ucp_cuda_kernel.cu

@@ -18,7 +18,9 @@
 
 class ucp_perf_cuda_request_manager {
 public:
-    __device__ ucp_perf_cuda_request_manager(size_t size) : m_size(size)
+    __device__
+    ucp_perf_cuda_request_manager(size_t size, ucp_device_request *requests) :
+        m_size(size), m_requests(requests)
     {
         assert(m_size <= CAPACITY);
         for (size_t i = 0; i < m_size; ++i) {
@@ -54,7 +56,7 @@ public:
     __device__ ucp_device_request_t &get_request()
     {
         assert(get_pending_count() < m_size);
-        size_t index = ucx_bitset_ffs(m_pending, m_size, 0);
+        size_t index = ucx_bitset_ffns(m_pending, m_size, 0);
         UCX_BIT_SET(m_pending, index);
         return m_requests[index];
     }
@@ -69,7 +71,7 @@ private:
     static const size_t CAPACITY = 128;
 
     size_t               m_size;
-    ucp_device_request_t m_requests[CAPACITY];
+    ucp_device_request_t *m_requests;
     uint8_t              m_pending[UCX_BITSET_SIZE(CAPACITY)];
 };
 
@@ -81,24 +83,29 @@ ucp_perf_cuda_put_multi_bw_kernel(ucx_perf_cuda_context &ctx,
                                   const void *address, uint64_t remote_address,
                                   size_t length)
 {
+    extern __shared__ ucp_device_request requests[];
     ucx_perf_cuda_time_t last_report_time = ucx_perf_cuda_get_time_ns();
     ucx_perf_counter_t max_iters          = ctx.max_iters;
     uint64_t *sn                          = ucx_perf_cuda_get_sn(address, length);
-    ucp_perf_cuda_request_manager request_mgr(ctx.max_outstanding);
+    ucp_device_request *thread_requests =
+            &requests[ctx.max_outstanding * threadIdx.x];
+    ucp_perf_cuda_request_manager request_mgr(ctx.max_outstanding,
+                                              thread_requests);
     ucs_status_t status;
 
     for (ucx_perf_counter_t idx = 0; idx < max_iters; idx++) {
         while (request_mgr.get_pending_count() >= ctx.max_outstanding) {
             status = request_mgr.progress<level>(1);
-            if (status != UCS_OK) {
+            if (UCS_STATUS_IS_ERR(status)) {
                 break;
             }
         }
 
         *sn = idx + 1;
         ucp_device_request_t &req = request_mgr.get_request();
         status = ucp_device_put_single<level>(mem_list, mem_list_index, address,
-                                              remote_address, length, 0, &req);
+                                              remote_address, length,
+                                              UCP_DEVICE_FLAG_NODELAY, &req);
         if (status != UCS_OK) {
             break;
         }
@@ -109,7 +116,7 @@ ucp_perf_cuda_put_multi_bw_kernel(ucx_perf_cuda_context &ctx,
 
     while (request_mgr.get_pending_count() > 0) {
         status = request_mgr.progress<level>(max_iters);
-        if (status != UCS_OK) {
+        if (UCS_STATUS_IS_ERR(status)) {
             break;
         }
     }
@@ -135,17 +142,19 @@ ucp_perf_cuda_put_single(ucp_device_mem_list_handle_h mem_list,
                          unsigned mem_list_index, const void *address,
                          uint64_t remote_address, size_t length)
 {
-    ucp_device_request_t req;
+    extern __shared__ ucp_device_request requests[];
+    ucp_device_request *req = &requests[threadIdx.x];
     ucs_status_t status;
 
     status = ucp_device_put_single<level>(mem_list, mem_list_index, address,
-                                          remote_address, length, 0, &req);
+                                          remote_address, length,
+                                          UCP_DEVICE_FLAG_NODELAY, req);
     if (status != UCS_OK) {
         return status;
     }
 
     do {
-        status = ucp_device_progress_req<level>(&req);
+        status = ucp_device_progress_req<level>(req);
     } while (status == UCS_INPROGRESS);
 
     return status;
@@ -220,8 +229,9 @@ public:
         ucp_perf_barrier(&m_perf);
         ucx_perf_test_start_clock(&m_perf);
 
-        ucp_perf_cuda_put_multi_latency_kernel
-            <UCS_DEVICE_LEVEL_THREAD><<<1, thread_count>>>(
+        ucp_perf_cuda_put_multi_latency_kernel<UCS_DEVICE_LEVEL_THREAD>
+            <<<1, thread_count,
+               thread_count * sizeof(ucp_device_request)>>>(
                 gpu_ctx(), handle.get(), 0, m_perf.send_buffer,
                 m_perf.ucp.remote_addr, length, m_perf.recv_buffer, my_index);
         CUDA_CALL_RET(UCS_ERR_NO_DEVICE, cudaGetLastError);
@@ -250,10 +260,12 @@ public:
             }
 
             unsigned thread_count = m_perf.params.device_thread_count;
-            ucp_perf_cuda_put_multi_bw_kernel
-                <UCS_DEVICE_LEVEL_THREAD><<<1, thread_count>>>(
-                    gpu_ctx(), handle.get(), 0, m_perf.send_buffer,
-                    m_perf.ucp.remote_addr, length);
+            ucp_perf_cuda_put_multi_bw_kernel<UCS_DEVICE_LEVEL_THREAD>
+                    <<<1, thread_count,
+                       thread_count * m_perf.params.max_outstanding *
+                               sizeof(ucp_device_request)>>>(
+                            gpu_ctx(), handle.get(), 0, m_perf.send_buffer,
+                            m_perf.ucp.remote_addr, length);
             CUDA_CALL_RET(UCS_ERR_NO_DEVICE, cudaGetLastError);
             wait_for_kernel(length);
         } else if (my_index == 0) {

src/ucp/api/device/ucp_device_impl.h

@@ -60,6 +60,23 @@ UCS_F_DEVICE void ucp_device_request_init(uct_device_ep_t *device_ep,
 }
 
 
+/**
+ * Macro for device put operations with retry logic
+ */
+#define UCP_DEVICE_PUT_BLOCKING(_level, _uct_device_ep_put, _device_ep, ...) \
+    ({ \
+        ucs_status_t _status; \
+        do { \
+            _status = _uct_device_ep_put<_level>(_device_ep, __VA_ARGS__); \
+            if (_status != UCS_ERR_NO_RESOURCE) { \
+                break; \
+            } \
+            _status = uct_device_ep_progress<_level>(_device_ep); \
+        } while (!UCS_STATUS_IS_ERR(_status)); \
+        _status; \
+    })
+
+
 UCS_F_DEVICE ucs_status_t ucp_device_prepare_single(
         ucp_device_mem_list_handle_h mem_list_h, unsigned mem_list_index,
         ucp_device_request_t *req, uct_device_ep_t *&device_ep,
@@ -115,6 +132,7 @@ ucp_device_prepare_multi(ucp_device_mem_list_handle_h mem_list_h,
  *
  * The routine returns a request that can be progressed and checked for
  * completion with @ref ucp_device_progress_req.
+ * The routine returns only after the message has been posted or an error has occurred.
  *
  * This routine can be called repeatedly with the same handle and different
  * addresses and length. The flags parameter can be used to modify the behavior
@@ -149,8 +167,9 @@ UCS_F_DEVICE ucs_status_t ucp_device_put_single(
         return status;
     }
 
-    return uct_device_ep_put_single<level>(device_ep, uct_elem, address,
-                                           remote_address, length, flags, comp);
+    return UCP_DEVICE_PUT_BLOCKING(level, uct_device_ep_put_single, device_ep,
+                                   uct_elem, address, remote_address, length,
+                                   flags, comp);
 }
 
 
@@ -225,6 +244,7 @@ UCS_F_DEVICE ucs_status_t ucp_device_counter_inc(
  *
  * The routine returns a request that can be progressed and checked for
  * completion with @ref ucp_device_progress_req.
+ * The routine returns only after all the messages have been posted or an error has occurred.
  *
  * This routine can be called repeatedly with the same handle and different
  * @a addresses, @a lengths and counter related parameters. The @a flags
@@ -261,11 +281,11 @@ UCS_F_DEVICE ucs_status_t ucp_device_put_multi(
         return status;
     }
 
-    return uct_device_ep_put_multi<level>(device_ep, uct_mem_list,
-                                          mem_list_h->mem_list_length,
-                                          addresses, remote_addresses, lengths,
-                                          counter_inc_value,
-                                          counter_remote_address, flags, comp);
+    return UCP_DEVICE_PUT_BLOCKING(level, uct_device_ep_put_multi, device_ep,
+                                   uct_mem_list, mem_list_h->mem_list_length,
+                                   addresses, remote_addresses, lengths,
+                                   counter_inc_value, counter_remote_address,
+                                   flags, comp);
 }
 
 
@@ -292,6 +312,7 @@ UCS_F_DEVICE ucs_status_t ucp_device_put_multi(
  *
  * The routine returns a request that can be progressed and checked for
  * completion with @ref ucp_device_progress_req.
+ * The routine returns only after all the messages have been posted or an error has occurred.
  *
  * This routine can be called repeatedly with the same handle and different
  * mem_list_indices, addresses, lengths and increment related parameters. The
@@ -334,10 +355,11 @@ UCS_F_DEVICE ucs_status_t ucp_device_put_multi_partial(
         return status;
     }
 
-    return uct_device_ep_put_multi_partial<level>(
-            device_ep, uct_mem_list, mem_list_indices, mem_list_count,
-            addresses, remote_addresses, lengths, counter_index,
-            counter_inc_value, counter_remote_address, flags, comp);
+    return UCP_DEVICE_PUT_BLOCKING(level, uct_device_ep_put_multi_partial,
+                                   device_ep, uct_mem_list, mem_list_indices,
+                                   mem_list_count, addresses, remote_addresses,
+                                   lengths, counter_index, counter_inc_value,
+                                   counter_remote_address, flags, comp);
 }
 
 
@@ -364,6 +386,28 @@ UCS_F_DEVICE uint64_t ucp_device_counter_read(const void *counter_ptr)
 }
 
 
+/**
+ * @ingroup UCP_DEVICE
+ * @brief Write value to the counter memory area.
+ *
+ * This function can be used to set counter to a specific value.
+ *
+ * The counter memory area must be initialized with the host function
+ * @ref ucp_device_counter_init.
+ *
+ * @tparam      level       Level of cooperation of the transfer.
+ * @param [in]  counter_ptr Counter memory area.
+ * @param [in]  value       Value to write.
+ *
+ */
+template<ucs_device_level_t level = UCS_DEVICE_LEVEL_THREAD>
+UCS_F_DEVICE void ucp_device_counter_write(void *counter_ptr, uint64_t value)
+{
+    return ucs_device_atomic64_write(
+            reinterpret_cast<uint64_t*>(counter_ptr), value);
+}
+
+
 /**
  * @ingroup UCP_DEVICE
  * @brief Progress a device request containing a batch of operations.
@@ -390,7 +434,12 @@ UCS_F_DEVICE ucs_status_t ucp_device_progress_req(ucp_device_request_t *req)
     }
 
     status = uct_device_ep_progress<level>(req->device_ep);
-    return (status != UCS_OK ? status : UCS_INPROGRESS);
+    if (status != UCS_OK) {
+        return status;
+    }
+
+    return (ucs_likely(req->comp.count == 0)) ? req->comp.status :
+                                                UCS_INPROGRESS;
 }
 
 #endif /* UCP_DEVICE_IMPL_H */

src/ucp/core/ucp_device.c

@@ -271,7 +271,8 @@ static ucs_status_t ucp_device_mem_list_create_handle(
     }
 
     if (i == 0) {
-        ucs_error("failed to select lane");
+        ucs_error("failed to select lane for local device %s",
+                  ucs_topo_sys_device_get_name(local_sys_dev));
         return UCS_ERR_NO_RESOURCE;
     }
 

src/ucp/core/ucp_ep.c

@@ -321,7 +321,7 @@ ucp_ep_peer_mem_get(ucp_context_h context, ucp_ep_h ep, uint64_t address,
 
     data->size = size;
     ucp_ep_rkey_unpack_internal(ep, rkey_buf, 0, UCS_BIT(rkey_ptr_md_index), 0,
-                                UCS_SYS_DEVICE_ID_UNKNOWN, 1, &data->rkey);
+                                UCS_SYS_DEVICE_ID_UNKNOWN, &data->rkey);
     rkey_index = ucs_bitmap2idx(data->rkey->md_map, rkey_ptr_md_index);
     status     = uct_rkey_ptr(data->rkey->tl_rkey[rkey_index].cmpt,
                               &data->rkey->tl_rkey[rkey_index].rkey, address,