Merge pull request #202 from maciekpac/release/ccl_2021.15.9-arc

Maria1Petrova · web-flow · commit 6cee0313a02f · 2026-04-23T12:10:29.000+02:00
Intel(R) oneAPI Collective Communications Library (oneCCL) 2021.15.9
diff --git a/CMakeLists.txt b/CMakeLists.txt
@@ -343,7 +343,7 @@ endif()
 
 set(CCL_MAJOR_VERSION     "2021")
 set(CCL_MINOR_VERSION     "15")
-set(CCL_UPDATE_VERSION    "8")
+set(CCL_UPDATE_VERSION    "9")
 set(CCL_PRODUCT_STATUS    "Gold")
 string(TIMESTAMP CCL_PRODUCT_BUILD_DATE "%Y-%m-%dT %H:%M:%SZ")
 get_vcs_properties("git")
diff --git a/man/doxconfig b/man/doxconfig
@@ -1,5 +1,5 @@
 PROJECT_NAME           = "Intel® oneAPI Collective Communications Library"
-PROJECT_NUMBER         = "2021.15.8"
+PROJECT_NUMBER         = "2021.15.9"
 
 INPUT = ../src/common/env/vars.hpp ../src/common/env/vars_experimental.hpp
 
diff --git a/src/atl/ofi/atl_ofi_comm.cpp b/src/atl/ofi/atl_ofi_comm.cpp
@@ -59,14 +59,26 @@ atl_status_t atl_ofi_comm::barrier(size_t ep_idx, atl_req_t& req) {
     int tag_comm_id = (comm_id != atl_comm_id_storage::invalid_comm_id)
                           ? comm_id
                           : atl_comm_id_storage::max_comm_id;
-    int tagc = tag_counter++;
+    int tagc = tag_counter_barrier++;
+
+    LOG_DEBUG("ofi_barrier: comm_rank: ",
+              rank,
+              ", comm_size: ",
+              size,
+              ", comm_id: ",
+              comm_id,
+              ", tag_comm_id: ",
+              tag_comm_id,
+              ", tag_counter: ",
+              tagc);
+
     int src, dst;
     const int len = 1;
     char sendbuf[len], recvbuf[len];
     int mask = 0x1;
     while (mask < size) {
         dst = (rank + mask) % size;
-        src = (rank - mask + size) % size;
+        src = (rank + size - mask) % size;
         atl_req send_req, recv_req;
         uint64_t op_tag = tag_creator->create(rank, tag_comm_id, tagc, 1);
         do {
@@ -98,6 +110,17 @@ atl_status_t atl_ofi_comm::barrier(size_t ep_idx, atl_req_t& req) {
         mask <<= 1;
     }
 
+    LOG_DEBUG("ofi_barrier done: comm_rank: ",
+              rank,
+              ", comm_size: ",
+              size,
+              ", comm_id: ",
+              comm_id,
+              ", tag_comm_id: ",
+              tag_comm_id,
+              ", tag_counter: ",
+              tagc);
+
     ofi_req->comp_state = ATL_OFI_COMP_COMPLETED;
     return ATL_STATUS_SUCCESS;
 }
@@ -205,6 +228,19 @@ atl_status_t atl_ofi_comm::allgatherv(size_t ep_idx,
     atl_ofi_req_t* ofi_req = ((atl_ofi_req_t*)req.internal);
     ofi_req->comp_state = ATL_OFI_COMP_COMPLETED;
 
+    LOG_DEBUG("ofi_allgatherv done: comm_rank: ",
+              rank,
+              ", comm_size: ",
+              size,
+              ", send_len: ",
+              send_len,
+              ", comm_id: ",
+              comm_id,
+              ", tag_comm_id: ",
+              tag_comm_id,
+              ", tag_counter: ",
+              tag_counter);
+
     tag_counter++;
 
     return ATL_STATUS_SUCCESS;
diff --git a/src/atl/ofi/atl_ofi_comm.hpp b/src/atl/ofi/atl_ofi_comm.hpp
@@ -185,4 +185,5 @@ class atl_ofi_comm : public atl_base_comm {
     atl_status_t init_transport(bool is_new);
 
     uint64_t tag_counter = 0;
+    uint64_t tag_counter_barrier = 0;
 };
diff --git a/src/coll/algorithms/allgatherv/sycl/allgatherv_large_sycl.cpp b/src/coll/algorithms/allgatherv/sycl/allgatherv_large_sycl.cpp
@@ -33,12 +33,16 @@ ccl::event allgatherv_large(sycl::queue& q,
     std::shared_ptr<ccl_comm> pair_comm = comm->get_pair_comm();
     std::shared_ptr<ccl_comm> even_comm = comm->get_even_comm();
 
+    bool is_arc = is_arc_card(ccl::ze::get_device_family(global_stream->get_ze_device()));
+
     const size_t dsize = ccl::global_data::get().dtypes->get(dtype).size();
     // use full vector (>= 8 bytes) if buffers and data size are 4 byte aligned
     bool use_full_vector = can_use_full_vector(send_buf, recv_buf, send_count * dsize);
     // TODO : generalize constraints for different hardware.
     // kernels with remote access is best performant at 64 bytes alignment (sycl_kernels_line_size/2) on PVC
-    const size_t align_size = ccl::global_data::env().sycl_kernels_line_size / 2;
+    // for BMG, relax the alignment check so that we don't change the path
+    // tmp_buf = 1 path is not implemented in this branch
+    const size_t align_size = is_arc ? dsize : ccl::global_data::env().sycl_kernels_line_size / 2;
     const bool is_aligned = (send_count * dsize) % align_size == 0;
     // use tmp buf for types < 4 byte size with odd count or non 4 byte aligned data
     // use tmp buf when data count bytes is not 64 byte aligned
@@ -47,6 +51,7 @@ ccl::event allgatherv_large(sycl::queue& q,
                              ((!use_full_vector || !is_aligned) && ccl::global_data::env().sycl_auto_use_tmp_buf);
 
     if (is_tmp_used) {
+        CCL_THROW_IF_NOT(!is_arc);
         CCL_THROW_IF_NOT(!ccl::global_data::env().sycl_copy_engine,
                          "allgatherv using copy engines not supported with tmp buffer");
 
@@ -72,7 +77,7 @@ ccl::event allgatherv_large(sycl::queue& q,
             }
         }
 
-        if (is_arc_card(ccl::ze::get_device_family(global_stream->get_ze_device()))) {
+        if (is_arc) {
             // only need output buffer
             std::vector<void*> ptrs{ recv_buf }; // index 0
             auto [sched, exchange_entry] = do_ipc_exchange(comm, global_stream, ptrs);
diff --git a/src/coll/algorithms/allreduce/sycl/allreduce_large_sycl_ring.hpp b/src/coll/algorithms/allreduce/sycl/allreduce_large_sycl_ring.hpp
@@ -332,16 +332,18 @@ void allreduce_large_su_ring_write_multi_kernel(const void *send_buf,
     std::vector<sycl::event> dep_events = get_sycl_events(deps);
     sycl::event work_event;
 
-    const int pipeline_size = N;
+    const int pipeline_size = std::max(N, ccl_large_tmp_bufs::buf_count);
 
     const size_t my_offset_count = count_per_rank * rank;
     const size_t recv_count = count_per_rank;
     const size_t recv_bytes = recv_count * dsize;
     void *recv_buf_rank_offset = (T *)recv_buf + my_offset_count;
     void *peer_recv_buf_rank_offset = (T *)peer_recv + my_offset_count;
-    const size_t chunk_size = ccl::global_data::env().sycl_tmp_buf_size / pipeline_size;
-    const size_t rem_chunk_size = recv_bytes % chunk_size;
-    const size_t num_chunks = recv_bytes / chunk_size + (rem_chunk_size != 0);
+    const size_t chunk_size = ccl::global_data::env().sycl_tmp_buf_size / pipeline_size /
+                              sizeof(message_t) * sizeof(message_t);
+    const size_t chunk_count = chunk_size / dsize;
+    const size_t rem_chunk_count = recv_count % chunk_count;
+    const size_t num_chunks = recv_count / chunk_count + (rem_chunk_count != 0);
 
     std::array<void *, ARC_MAX_NUM> work_bufs;
     std::array<void *, ARC_MAX_NUM> remote_work_bufs;
@@ -356,13 +358,12 @@ void allreduce_large_su_ring_write_multi_kernel(const void *send_buf,
 
     int slot = 0;
     for (size_t nc = 0; nc < num_chunks; nc++) {
-        const size_t chunk_offset = nc * chunk_size;
-        const size_t data_count =
-            ((nc < recv_bytes / chunk_size) ? chunk_size : rem_chunk_size) / dsize;
+        const size_t chunk_offset = nc * chunk_count * dsize;
+        const size_t data_count = (nc < recv_count / chunk_count) ? chunk_count : rem_chunk_count;
 
         // starting indexes
-        int s = (rank - 1 + N) % N;
-        int r = (s - 1 + N) % N;
+        int s = (rank + N - 1) % N;
+        int r = (s + N - 1) % N;
 
         for (int i = 0; i < N - 1; i++) {
             int next_slot = (slot + 1) % pipeline_size;
@@ -413,7 +414,7 @@ void allreduce_large_su_ring_write_multi_kernel(const void *send_buf,
             }
 
             s = r;
-            r = (s - 1 + N) % N;
+            r = (s + N - 1) % N;
             slot = next_slot;
         }
     }
diff --git a/src/coll/algorithms/recv/sycl/recv_pcie.cpp b/src/coll/algorithms/recv/sycl/recv_pcie.cpp
@@ -34,26 +34,33 @@ ccl::event recv_ll(const void *recv_buf,
     const int comm_size = node_comm->size();
     const int comm_rank = node_comm->rank();
 
-    //    coll_init(comm, global_stream);
-
     auto ccl_dtype = ccl::global_data::get().dtypes->get(dtype);
     size_t dt_sz = ccl_dtype.size();
     size_t recv_size = recv_count * ccl_dtype.size();
 
     bool p2p = node_comm->get_topo_manager().has_p2p_access();
     uint32_t pattern = node_comm->get_rt_pattern(pattern_type::recv, peer_rank);
 
+    LOG_DEBUG("recv_ll recv_count: ", recv_count, " peer rank: ", peer_rank);
+
     std::vector<sycl::event> dep_events = get_sycl_events(deps);
 
     auto lambda = [&]<typename T, template <typename, int> class Proto>(int NRanks) {
         T *peerbuf0[NRanks];
         T *peerbuf1[NRanks];
+        T *ipcbuf0;
+        T *ipcbuf1;
+        // use small tmp buffer. Can not use large tmp buffer because
+        // there is only single copy of the buffer.
+        // Otherwise a barrier is needed her
+        auto [local_tmp_buf, remote_ptrs] = node_comm->get_all_tmp_bufs(true);
         for (int i = 0; i < NRanks; i++) {
-            peerbuf0[i] = (T *)get_remote_node_tmp_buf(0, comm)[i];
-            peerbuf1[i] = (T *)get_remote_node_tmp_buf(1, comm)[i];
+            peerbuf0[i] = (T *)remote_ptrs[i];
+            peerbuf1[i] = (T *)((char *)remote_ptrs[i] + ccl_tmp_bufs::buf_size / 2);
         }
-        T *ipcbuf0 = (T *)get_tmp_buf(0, comm);
-        T *ipcbuf1 = (T *)get_tmp_buf(1, comm);
+        ipcbuf0 = (T *)local_tmp_buf;
+        ipcbuf1 = (T *)((char *)local_tmp_buf + ccl_tmp_bufs::buf_size / 2);
+
         sycl::event e = Recv<T, Proto, RingTransmit>::launch(NRanks,
                                                              (T *)recv_buf,
                                                              ipcbuf0,
diff --git a/src/coll/algorithms/send/sycl/send_pcie.cpp b/src/coll/algorithms/send/sycl/send_pcie.cpp
@@ -34,26 +34,33 @@ ccl::event send_ll(const void *send_buf,
     const int comm_size = node_comm->size();
     const int comm_rank = node_comm->rank();
 
-    //    coll_init(comm, global_stream);
-
     auto ccl_dtype = ccl::global_data::get().dtypes->get(dtype);
     size_t dt_sz = ccl_dtype.size();
     size_t send_size = send_count * ccl_dtype.size();
 
     bool p2p = node_comm->get_topo_manager().has_p2p_access();
     uint32_t pattern = node_comm->get_rt_pattern(pattern_type::send, peer_rank);
 
+    LOG_DEBUG("send_ll send_count: ", send_count, " peer rank: ", peer_rank);
+
     std::vector<sycl::event> dep_events = get_sycl_events(deps);
 
     auto lambda = [&]<typename T, template <typename, int> class Proto>(int NRanks) {
         T *peerbuf0[NRanks];
         T *peerbuf1[NRanks];
+        T *ipcbuf0;
+        T *ipcbuf1;
+        // use small tmp buffer. Can not use large tmp buffer because
+        // there is only single copy of the buffer.
+        // Otherwise a barrier is needed here
+        auto [local_tmp_buf, remote_ptrs] = node_comm->get_all_tmp_bufs(true);
         for (int i = 0; i < NRanks; i++) {
-            peerbuf0[i] = (T *)get_remote_node_tmp_buf(0, comm)[i];
-            peerbuf1[i] = (T *)get_remote_node_tmp_buf(1, comm)[i];
+            peerbuf0[i] = (T *)remote_ptrs[i];
+            peerbuf1[i] = (T *)((char *)remote_ptrs[i] + ccl_tmp_bufs::buf_size / 2);
         }
-        T *ipcbuf0 = (T *)get_tmp_buf(0, comm);
-        T *ipcbuf1 = (T *)get_tmp_buf(1, comm);
+        ipcbuf0 = (T *)local_tmp_buf;
+        ipcbuf1 = (T *)((char *)local_tmp_buf + ccl_tmp_bufs::buf_size / 2);
+
         sycl::event e = Send<T, Proto, RingTransmit>::launch(NRanks,
                                                              (T *)send_buf,
                                                              ipcbuf0,
diff --git a/src/coll/algorithms/utils/sycl_coll_base.cpp b/src/coll/algorithms/utils/sycl_coll_base.cpp
@@ -331,7 +331,7 @@ void coll_init(ccl_comm *comm, ccl_stream *global_stream) {
 
             //set up temp buf to be used for large collectives
             // WA : use smaller tmp buffer for client GPUs
-            if (is_arc &&
+            if (is_arc && ccl::global_data::env().sycl_ll_buffer_global &&
                 ccl::global_data::env().sycl_tmp_buf_size == tmp_bufs_count * 128 * 1024 * 1024) {
                 ccl::global_data::env().sycl_tmp_buf_size =
                     tmp_bufs_count * calculate_ll_buf_size(q);
@@ -607,7 +607,8 @@ void coll_initExt(ccl_comm *comm,
             //set up temp buf to be used for large collectives
             // WA : use smaller tmp buffer for client GPUs
             if (is_arc_card(ccl::ze::get_device_family(global_stream->get_ze_device())) &&
-                ccl::global_data::env().sycl_tmp_buf_size == 3 * 128 * 1024 * 1024) {
+                ccl::global_data::env().sycl_ll_buffer_global &&
+                ccl::global_data::env().sycl_tmp_buf_size == tmp_bufs_count * 128 * 1024 * 1024) {
                 ccl::global_data::env().sycl_tmp_buf_size =
                     tmp_bufs_count * calculate_ll_buf_size(q);
                 LOG_DEBUG("MT: Allocate LL ring buffer of size: ",
diff --git a/src/coll/algorithms/utils/transmit/ring_transmit.hpp b/src/coll/algorithms/utils/transmit/ring_transmit.hpp
@@ -181,8 +181,20 @@ class RingTransmit : public Proto<T, SubGroupSize> {
             retry |= recvMessages(messages, localScatterSink[peer][slot][wireId], flag);
         } while (sycl::any_of_group(sycl::ext::oneapi::this_work_item::get_sub_group(), retry));
 
-        shuffleData(v);
-        accumMessages(v, messages);
+        if constexpr (sizeof(T) > sizeof(flag)) {
+            // restore data, accumulate, shuffle
+            restoreData(messages);
+            accumMessages(v, messages);
+            shuffleData(v);
+        }
+        else {
+            // datasize is smaller than flag,
+            // no overlap between datatype and flag
+            // can accumulate shuffled data:
+            // less ops to perform, faster
+            shuffleData(v);
+            accumMessages(v, messages);
+        }
         insertFlags(v, flag);
 
         sendMessages(scatterSink[peer][slot][wireId], v);
@@ -208,8 +220,20 @@ class RingTransmit : public Proto<T, SubGroupSize> {
             retry |= recvMessages(messages, localScatterSink[peer][slot][wireId], flag);
         } while (sycl::any_of_group(sycl::ext::oneapi::this_work_item::get_sub_group(), retry));
 
-        shuffleData(v);
-        accumMessages(v, messages);
+        if constexpr (sizeof(T) > sizeof(flag)) {
+            // restore data, accumulate, shuffle
+            restoreData(messages);
+            accumMessages(v, messages);
+            shuffleData(v);
+        }
+        else {
+            // datasize is smaller than flag,
+            // no overlap between datatype and flag
+            // can accumulate shuffled data:
+            // less ops to perform, faster
+            shuffleData(v);
+            accumMessages(v, messages);
+        }
 
         insertFlags(v, flag);
         sendMessages(gatherSink[peer][slot][wireId], v);
@@ -294,8 +318,20 @@ class RingTransmit : public Proto<T, SubGroupSize> {
             retry |= recvMessages(messages, localScatterSink[peer][slot][wireId], flag);
         } while (sycl::any_of_group(sycl::ext::oneapi::this_work_item::get_sub_group(), retry));
 
-        shuffleData(v);
-        accumMessages(v, messages);
+        if constexpr (sizeof(T) > sizeof(flag)) {
+            // restore data, accumulate, shuffle
+            restoreData(messages);
+            accumMessages(v, messages);
+            shuffleData(v);
+        }
+        else {
+            // datasize is smaller than flag,
+            // no overlap between datatype and flag
+            // can accumulate shuffled data:
+            // less ops to perform, faster
+            shuffleData(v);
+            accumMessages(v, messages);
+        }
 
         insertFlags(v, flag);
         restoreData(v);
diff --git a/src/comm/comm.cpp b/src/comm/comm.cpp
@@ -382,16 +382,16 @@ int ccl_comm::detect_numa_nodes_from_gpu_topology(std::shared_ptr<ccl_comm> node
 
     // Step 3: Determine number of NUMA nodes and assign ranks to NUMA groups
     std::map<int, std::vector<int>> numa_to_ranks; // NUMA node -> list of ranks
-    int numa_nodes_per_host = 0;
-
     for (int rank = 0; rank < node_size; rank++) {
         int numa = all_gpu_numa_info[rank].numa_node;
         if (numa >= 0) {
             numa_to_ranks[numa].push_back(rank);
-            numa_nodes_per_host = std::max(numa_nodes_per_host, numa + 1);
         }
     }
 
+    // Get number of numa nodes in our system
+    int numa_nodes_per_host = static_cast<int>(numa_to_ranks.size());
+
     // Log the detected topology
     if (numa_nodes_per_host > 0) {
         LOG_DEBUG("Detected ", numa_nodes_per_host, " NUMA nodes from GPU topology");
diff --git a/src/topology/topo_manager.cpp b/src/topology/topo_manager.cpp
@@ -1008,13 +1008,34 @@ bool topo_manager::check_p2p_access() const {
 }
 
 bool topo_manager::check_p2p_atomics() const {
+    static int atomics_enable = -1;
+
+    /* with ZE_AFFINITY_MASK, if there is only one device visible,
+     * i.e. this rank's own device, the self checking result
+     * must be ignored */
+    if (atomics_matrix.size() == 1) {
+        if (atomics_enable != -1) {
+            // old results are valid because it is same GPU
+            return atomics_enable;
+        }
+        if (ze::is_arc_card(ccl::global_data::get().ze_data->devices[0].family)) {
+            return false;
+        }
+        // FIXME: be conservative, assume no atomics
+        // however, the no atomics version may not barrier
+        // may not work on PVC
+        return false;
+    }
+
     for (size_t i = 0; i < atomics_matrix.size(); i++) {
         for (size_t j = 0; j < atomics_matrix[i].size(); j++) {
             if (!atomics_matrix[i][j]) {
+                atomics_enable = 0;
                 return false;
             }
         }
     }
+    atomics_enable = 1;
     return true;
 }
 
diff --git a/third-party-programs.txt b/third-party-programs.txt
