Mirgate kmeans daal to DPC++

examples/kmeans/run-gpu-standalone.sh

@@ -9,7 +9,6 @@ DATA_FILE=$HDFS_ROOT/data/sample_kmeans_data.txt
 APP_JAR=target/oap-mllib-examples-$OAP_MLLIB_VERSION.jar
 APP_CLASS=org.apache.spark.examples.ml.KMeansExample
 
-USE_DPU=true
 DEVICE=GPU
 RESOURCE_FILE=$PWD/IntelGpuResourceFile.json
 WORKER_GPU_AMOUNT=4
@@ -28,7 +27,6 @@ time $SPARK_HOME/bin/spark-submit --master $SPARK_MASTER \
     --conf "spark.sql.shuffle.partitions=$SPARK_DEFAULT_PARALLELISM" \
     --conf "spark.driver.extraClassPath=$SPARK_DRIVER_CLASSPATH" \
     --conf "spark.executor.extraClassPath=$SPARK_EXECUTOR_CLASSPATH" \
-    --conf "spark.oap.mllib.isDPU=$USE_DPU" \
     --conf "spark.oap.mllib.device=$DEVICE" \
     --conf "spark.worker.resourcesFile=$RESOURCE_FILE" \
     --conf "spark.worker.resource.gpu.amount=$WORKER_GPU_AMOUNT" \

examples/kmeans/run.sh

@@ -6,6 +6,7 @@ source ../../conf/env.sh
 # The data file should be copied to $HDFS_ROOT before running examples
 DATA_FILE=$HDFS_ROOT/data/sample_kmeans_data.txt
 
+DEVICE=CPU
 APP_JAR=target/oap-mllib-examples-$OAP_MLLIB_VERSION.jar
 APP_CLASS=org.apache.spark.examples.ml.KMeansExample
 
@@ -20,6 +21,7 @@ time $SPARK_HOME/bin/spark-submit --master $SPARK_MASTER \
     --conf "spark.sql.shuffle.partitions=$SPARK_DEFAULT_PARALLELISM" \
     --conf "spark.driver.extraClassPath=$SPARK_DRIVER_CLASSPATH" \
     --conf "spark.executor.extraClassPath=$SPARK_EXECUTOR_CLASSPATH" \
+    --conf "spark.oap.mllib.device=$DEVICE" \
     --conf "spark.shuffle.reduceLocality.enabled=false" \
     --conf "spark.network.timeout=1200s" \
     --conf "spark.task.maxFailures=1" \

mllib-dal/src/main/java/com/intel/oneapi/dal/table/ColumnAccessor.java

@@ -39,5 +39,4 @@ private native int[] cPullInt(long cObject, long cColumnIndex, long cRowStartInd
                                   long cRowEndIndex, int computeDeviceIndex);
     private native float[] cPullFloat(long cObject, long cColumnIndex, long cRowStartIndex,
                                       long cRowEndIndex, int computeDeviceIndex);
-
 }

mllib-dal/src/main/native/Communicator.hpp

@@ -0,0 +1,73 @@
+/*******************************************************************************
+* Copyright 2021 Intel Corporation
+*
+* Licensed under the Apache License, Version 2.0 (the "License");
+* you may not use this file except in compliance with the License.
+* You may obtain a copy of the License at
+*
+*     http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*******************************************************************************/
+
+#pragma once
+
+//cpp ccl host communicator
+#ifdef ONEDAL_DATA_PARALLEL
+
+#include "oneapi/ccl.hpp"
+#include "oneapi/dal/detail/ccl/communicator.hpp"
+#include "Singleton.hpp"
+
+namespace de = oneapi::dal::detail;
+namespace oneapi::dal::preview::spmd {
+
+namespace backend {
+struct ccl {};
+} // namespace backend
+class ccl_info {
+    friend class de::singleton<ccl_info>;
+
+private:
+    ccl_info(int size, int rank_id, const ccl::string &ip_port) {
+        rank = rank_id;
+        rank_count = size;
+        ccl::string ccl_ip_port(ip_port);
+        auto kvs_attr = ccl::create_kvs_attr();
+        kvs_attr.set<ccl::kvs_attr_id::ip_port>(ccl_ip_port);
+        kvs = ccl::create_main_kvs(kvs_attr);
+    }
+
+public:
+    ccl::shared_ptr_class<ccl::kvs> kvs;
+    int rank;
+    int rank_count;
+};
+
+template <typename Backend>
+communicator<device_memory_access::none> make_communicator(int size, int rank, const ccl::string &ip_port) {
+    auto& info = de::singleton<ccl_info>::get(size, rank, ip_port);
+    // integral cast
+    return oneapi::dal::detail::ccl_communicator<device_memory_access::none>{ info.kvs,
+                                                                      info.rank,
+                                                                      info.rank_count };
+}
+
+template <typename Backend>
+communicator<device_memory_access::usm> make_communicator(sycl::queue& queue, int size, int rank, const ccl::string &ip_port) {
+    auto& info = de::singleton<ccl_info>::get(size, rank, ip_port);
+    return oneapi::dal::detail::ccl_communicator<device_memory_access::usm>{
+        queue,
+        info.kvs,
+        oneapi::dal::detail::integral_cast<std::int64_t>(info.rank),
+        oneapi::dal::detail::integral_cast<std::int64_t>(info.rank_count)
+    };
+}
+
+} // namespace oneapi::dal::preview::spmd
+
+#endif

mllib-dal/src/main/native/GPU.cpp

@@ -3,26 +3,12 @@
 #include <unistd.h>
 
 #include "GPU.h"
+
 typedef std::shared_ptr<sycl::queue> queuePtr;
 
 static std::mutex mtx;
 static std::vector<sycl::queue> cVector;
 
-static sycl::queue getSyclQueue(const sycl::device device) {
-    mtx.lock();
-    if (!cVector.empty()) {
-        auto device = cVector[0];
-        mtx.unlock();
-        return device;
-    } else {
-        sycl::queue queue{device};
-        cVector.push_back(queue);
-        auto device = cVector[0];
-        mtx.unlock();
-        return device;
-    }
-}
-
 static std::vector<sycl::device> get_gpus() {
     auto platforms = sycl::platform::get_platforms();
     for (auto p : platforms) {
@@ -82,6 +68,21 @@ sycl::device getAssignedGPU(ccl::communicator &comm, int size, int rankId,
     return rank_gpu;
 }
 
+static sycl::queue getSyclQueue(const sycl::device device) {
+    mtx.lock();
+    if (!cVector.empty()) {
+        const auto device = cVector[0];
+        mtx.unlock();
+        return device;
+    } else {
+        sycl::queue queue{device};
+        cVector.push_back(queue);
+        const auto device = cVector[0];
+        mtx.unlock();
+        return device;
+    }
+}
+
 sycl::queue getQueue(const compute_device device) {
     std::cout << "Get Queue" << std::endl;
 

mllib-dal/src/main/native/GPU.h

@@ -1,5 +1,9 @@
 #pragma once
 
+#ifndef ONEDAL_DATA_PARALLEL
+#define ONEDAL_DATA_PARALLEL
+#endif
+
 #include "service.h"
 #include <CL/cl.h>
 #include <CL/sycl.hpp>