oap-project · minmingzhu · Sep 18, 2023 · Sep 18, 2023 · Sep 19, 2023 · Sep 19, 2023
diff --git a/mllib-dal/src/main/native/CorrelationImpl.cpp b/mllib-dal/src/main/native/CorrelationImpl.cpp
@@ -172,12 +172,12 @@ static void doCorrelationOneAPICompute(
         printHomegenTable(result_train.get_cor_matrix());
         auto t2 = std::chrono::high_resolution_clock::now();
         auto duration =
-            std::chrono::duration_cast<std::chrono::milliseconds>(t2 - t1)
+            (float)std::chrono::duration_cast<std::chrono::milliseconds>(t2 -
+                                                                         t1)
                 .count();
-        logger::println(
-            logger::INFO,
-            "Correlation batch(native): computing step took %d secs.",
-            duration / 1000);
+        logger::println(logger::INFO,
+                        "Correlation (native): training step took %f secs.",
+                        duration / 1000);
         // Return all covariance & mean
         jclass clazz = env->GetObjectClass(resultObj);
 

diff --git a/mllib-dal/src/main/native/DecisionForestClassifierImpl.cpp b/mllib-dal/src/main/native/DecisionForestClassifierImpl.cpp
@@ -247,6 +247,7 @@ static jobject doRFClassifierOneAPICompute(
             .set_max_tree_depth(maxTreeDepth)
             .set_max_bins(maxBins);
 
+    auto t1 = std::chrono::high_resolution_clock::now();
     const auto result_train =
         preview::train(comm, df_desc, hFeaturetable, hLabeltable);
     const auto result_infer =
@@ -261,7 +262,14 @@ static jobject doRFClassifierOneAPICompute(
         printHomegenTable(result_infer.get_responses());
         logger::println(logger::INFO, "Probabilities results:\n");
         printHomegenTable(result_infer.get_probabilities());
-
+        auto t2 = std::chrono::high_resolution_clock::now();
+        auto duration =
+            (float)std::chrono::duration_cast<std::chrono::milliseconds>(t2 -
+                                                                         t1)
+                .count();
+        logger::println(logger::INFO,
+                        "RF Classifier (native): training step took %f secs.",
+                        duration / 1000);
         // convert to java hashmap
         trees = collect_model(env, result_train.get_model(), classCount);
 

diff --git a/mllib-dal/src/main/native/KMeansImpl.cpp b/mllib-dal/src/main/native/KMeansImpl.cpp
@@ -270,10 +270,11 @@ static jlong doKMeansOneAPICompute(
         printHomegenTable(result_train.get_model().get_centroids());
         auto t2 = std::chrono::high_resolution_clock::now();
         auto duration =
-            std::chrono::duration_cast<std::chrono::milliseconds>(t2 - t1)
+            (float)std::chrono::duration_cast<std::chrono::milliseconds>(t2 -
+                                                                         t1)
                 .count();
         logger::println(logger::INFO,
-                        "KMeans (native): training step took %d secs",
+                        "KMeans (native): training step took %f secs",
                         duration / 1000);
         // Get the class of the input object
         jclass clazz = env->GetObjectClass(resultObj);

diff --git a/mllib-dal/src/main/native/LinearRegressionImpl.cpp b/mllib-dal/src/main/native/LinearRegressionImpl.cpp
@@ -237,12 +237,21 @@ static jlong doLROneAPICompute(
     linear_regression_gpu::train_input local_input{xtrain, ytrain};
     const auto linear_regression_desc =
         linear_regression_gpu::descriptor<GpuAlgorithmFPType>(fitIntercept);
-
+    auto t1 = std::chrono::high_resolution_clock::now();
     linear_regression_gpu::train_result result_train =
         preview::train(comm, linear_regression_desc, xtrain, ytrain);
     if (isRoot) {
         HomogenTablePtr result_matrix = std::make_shared<homogen_table>(
             result_train.get_model().get_betas());
+        auto t2 = std::chrono::high_resolution_clock::now();
+        auto duration =
+            (float)std::chrono::duration_cast<std::chrono::milliseconds>(t2 -
+                                                                         t1)
+                .count();
+        logger::println(
+            logger::INFO,
+            "Linear regression (native): training step took %f secs.",
+            duration / 1000);
         saveHomogenTablePtrToVector(result_matrix);
         return (jlong)result_matrix.get();
     } else {

diff --git a/mllib-dal/src/main/native/OneCCL.cpp b/mllib-dal/src/main/native/OneCCL.cpp
@@ -60,7 +60,8 @@ JNIEXPORT jint JNICALL Java_com_intel_oap_mllib_OneCCL_00024_c_1init(
     JNIEnv *env, jobject obj, jint size, jint rank, jstring ip_port,
     jint computeDeviceOrdinal, jobject param) {
 
-    logger::println(logger::INFO, "OneCCL (native): init");
+    logger::println(logger::INFO, "OneCCL (native): init rank %d size %d", rank,
+                    size);
     const char *str = env->GetStringUTFChars(ip_port, 0);
     ccl::string ccl_ip_port(str);
     auto &singletonCCLInit = CCLInitSingleton::get(size, rank, ccl_ip_port);
@@ -87,7 +88,12 @@ JNIEXPORT jint JNICALL Java_com_intel_oap_mllib_OneCCL_00024_c_1init(
 #ifdef CPU_GPU_PROFILE
     case ComputeDevice::gpu: {
         auto gpus = get_gpus();
-        sycl::queue queue{gpus[0]};
+        auto gpus_count = gpus.size();
+        logger::println(logger::INFO, "OneCCL (native): gpus_count %d",
+                        gpus_count);
+        sycl::device selected_device;
+        selected_device = gpus[rank % gpus_count];
+        sycl::queue queue{selected_device};
         auto t1 = std::chrono::high_resolution_clock::now();
         auto comm = oneapi::dal::preview::spmd::make_communicator<
             oneapi::dal::preview::spmd::backend::ccl>(queue, size, rank,

diff --git a/mllib-dal/src/main/native/PCAImpl.cpp b/mllib-dal/src/main/native/PCAImpl.cpp
@@ -198,26 +198,26 @@ static void doPCAOneAPICompute(
 
     auto t1 = std::chrono::high_resolution_clock::now();
     const auto result = preview::compute(comm, cov_desc, htable);
-    auto t2 = std::chrono::high_resolution_clock::now();
-    auto duration =
-        std::chrono::duration_cast<std::chrono::milliseconds>(t2 - t1).count();
-    logger::println(logger::INFO, "PCA (native): Covariance step took %d secs",
-                    duration / 1000);
     if (isRoot) {
         using float_t = GpuAlgorithmFPType;
         using method_t = pca_gpu::method::precomputed;
         using task_t = pca_gpu::task::dim_reduction;
         using descriptor_t = pca_gpu::descriptor<float_t, method_t, task_t>;
         const auto pca_desc = descriptor_t().set_deterministic(true);
 
-        t1 = std::chrono::high_resolution_clock::now();
         const auto result_train =
             preview::train(comm, pca_desc, result.get_cov_matrix());
-        t2 = std::chrono::high_resolution_clock::now();
-        duration =
-            std::chrono::duration_cast<std::chrono::milliseconds>(t2 - t1)
+        logger::println(logger::INFO, "Eigenvectors:");
+        printHomegenTable(result_train.get_eigenvectors());
+        logger::println(logger::INFO, "Eigenvalues:");
+        printHomegenTable(result_train.get_eigenvalues());
+
+        auto t2 = std::chrono::high_resolution_clock::now();
+        auto duration =
+            (float)std::chrono::duration_cast<std::chrono::milliseconds>(t2 - t1)
                 .count();
-        logger::println(logger::INFO, "PCA (native): Eigen step took %d secs",
+        logger::println(logger::INFO,
+                        "PCA (native): training step took %f secs",
                         duration / 1000);
         // Return all eigenvalues & eigenvectors
         // Get the class of the input object
@@ -227,10 +227,6 @@ static void doPCAOneAPICompute(
             env->GetFieldID(clazz, "pcNumericTable", "J");
         jfieldID explainedVarianceNumericTableField =
             env->GetFieldID(clazz, "explainedVarianceNumericTable", "J");
-        logger::println(logger::INFO, "Eigenvectors:");
-        printHomegenTable(result_train.get_eigenvectors());
-        logger::println(logger::INFO, "Eigenvalues:");
-        printHomegenTable(result_train.get_eigenvalues());
 
         HomogenTablePtr eigenvectors =
             std::make_shared<homogen_table>(result_train.get_eigenvectors());

diff --git a/mllib-dal/src/main/native/SummarizerImpl.cpp b/mllib-dal/src/main/native/SummarizerImpl.cpp
@@ -230,7 +230,7 @@ static void doSummarizerOneAPICompute(
                                                                          t1)
                 .count();
         logger::println(logger::INFO,
-                        "Summarizer (native): computing step took %d secs",
+                        "Summarizer (native): training took %f secs",
                         duration / 1000);
         // Return all covariance & mean
         jclass clazz = env->GetObjectClass(resultObj);

diff --git a/mllib-dal/src/main/native/service.h b/mllib-dal/src/main/native/service.h
@@ -89,18 +89,33 @@ inline void printHomegenTable(const oneapi::dal::table &table) {
     if (table.get_row_count() <= 10) {
         for (std::int64_t i = 0; i < table.get_row_count(); i++) {
             logger::print(logger::INFO, "");
-            for (std::int64_t j = 0; j < table.get_column_count(); j++) {
-                logger::print(logger::NONE, "%10f",
-                              x[i * table.get_column_count() + j]);
+            if (table.get_column_count() <= 20) {
+                for (std::int64_t j = 0; j < table.get_column_count(); j++) {
+                    logger::print(logger::NONE, "%10f",
+                                  x[i * table.get_column_count() + j]);
+                }
+            } else {
+                for (std::int64_t j = 0; j < 20; j++) {
+                    logger::print(logger::NONE, "%10f",
+                                  x[i * table.get_column_count() + j]);
+                }
             }
             logger::println(logger::NONE, "");
         }
+
     } else {
         for (std::int64_t i = 0; i < 5; i++) {
             logger::print(logger::INFO, "");
-            for (std::int64_t j = 0; j < table.get_column_count(); j++) {
-                logger::print(logger::NONE, "%10f",
-                              x[i * table.get_column_count() + j]);
+            if (table.get_column_count() <= 20) {
+                for (std::int64_t j = 0; j < table.get_column_count(); j++) {
+                    logger::print(logger::NONE, "%10f",
+                                  x[i * table.get_column_count() + j]);
+                }
+            } else {
+                for (std::int64_t j = 0; j < 20; j++) {
+                    logger::print(logger::NONE, "%10f",
+                                  x[i * table.get_column_count() + j]);
+                }
             }
             logger::println(logger::NONE, "");
         }
@@ -109,9 +124,16 @@ inline void printHomegenTable(const oneapi::dal::table &table) {
         for (std::int64_t i = table.get_row_count() - 5;
              i < table.get_row_count(); i++) {
             logger::print(logger::INFO, "");
-            for (std::int64_t j = 0; j < table.get_column_count(); j++) {
-                logger::print(logger::NONE, "%10f",
-                              x[i * table.get_column_count() + j]);
+            if (table.get_column_count() <= 20) {
+                for (std::int64_t j = 0; j < table.get_column_count(); j++) {
+                    logger::print(logger::NONE, "%10f",
+                                  x[i * table.get_column_count() + j]);
+                }
+            } else {
+                for (std::int64_t j = 0; j < 20; j++) {
+                    logger::print(logger::NONE, "%10f",
+                                  x[i * table.get_column_count() + j]);
+                }
             }
             logger::println(logger::NONE, "");
         }

diff --git a/mllib-dal/src/main/scala/com/intel/oap/mllib/CommonJob.scala b/mllib-dal/src/main/scala/com/intel/oap/mllib/CommonJob.scala
@@ -19,24 +19,17 @@ package com.intel.oap.mllib
 import com.intel.oneapi.dal.table.Common
 import org.apache.spark.TaskContext
 import org.apache.spark.rdd.RDD
+import org.apache.spark.SparkEnv
 
 object CommonJob {
   def initCCLAndSetAffinityMask(data: RDD[_],
                                       executorNum: Int,
                                       kvsIPPort: String,
                                       useDevice: String): Unit = {
         data.mapPartitionsWithIndex { (rank, table) =>
-          OneCCL.init(executorNum, rank, kvsIPPort,
+          val executorId = SparkEnv.get.executorId.toInt
+          OneCCL.init(executorNum, executorId, kvsIPPort,
             Common.ComputeDevice.getDeviceByName(useDevice).ordinal())
-          val gpuIndices = if (useDevice == "GPU") {
-              val resources = TaskContext.get().resources()
-              resources("gpu").addresses.map(_.toInt)
-          } else {
-              Array.empty[Int]
-          }
-          if (gpuIndices.nonEmpty) {
-            OneCCL.setAffinityMask(gpuIndices(0).toString())
-          }
           Iterator.empty
         }.count()
     }

diff --git a/...dal/src/main/scala/com/intel/oap/mllib/classification/RandomForestClassifierDALImpl.scala b/...dal/src/main/scala/com/intel/oap/mllib/classification/RandomForestClassifierDALImpl.scala
@@ -19,7 +19,7 @@ import com.intel.oap.mllib.Utils.getOneCCLIPPort
 import com.intel.oap.mllib.{CommonJob, OneCCL, OneDAL, Utils}
 import com.intel.oneapi.dal.table.Common
 import org.apache.spark.annotation.Since
-import org.apache.spark.TaskContext
+import org.apache.spark.{SparkEnv, TaskContext}
 import org.apache.spark.internal.Logging
 import org.apache.spark.ml.classification.DecisionTreeClassificationModel
 import org.apache.spark.ml.linalg.{Matrix, Vector}
@@ -121,7 +121,12 @@ class RandomForestClassifierDALImpl(val uid: String,
 
       logInfo(s"RandomForestClassifierDAL compute took ${durationCompute} secs")
 
-      val ret = if (rank == 0) {
+      val isRoot = if (useDevice == "GPU") {
+        SparkEnv.get.executorId.toInt == 0
+      } else {
+        rank == 0
+      }
+      val ret = if (isRoot) {
         Iterator(hashmap)
       } else {
         Iterator.empty

diff --git a/mllib-dal/src/main/scala/com/intel/oap/mllib/clustering/KMeansDALImpl.scala b/mllib-dal/src/main/scala/com/intel/oap/mllib/clustering/KMeansDALImpl.scala
@@ -19,7 +19,7 @@ package com.intel.oap.mllib.clustering
 import com.intel.oap.mllib.Utils.getOneCCLIPPort
 import com.intel.oap.mllib.{CommonJob, OneCCL, OneDAL, Utils}
 import com.intel.oneapi.dal.table.Common
-import org.apache.spark.TaskContext
+import org.apache.spark.{SparkEnv, TaskContext}
 import org.apache.spark.internal.Logging
 import org.apache.spark.ml.linalg.Vector
 import org.apache.spark.ml.util._
@@ -92,8 +92,12 @@ class KMeansDALImpl(var nClusters: Int,
         gpuIndices,
         result
       )
-
-      val ret = if (rank == 0) {
+      val isRoot = if (useDevice == "GPU") {
+        SparkEnv.get.executorId.toInt == 0
+      } else {
+        rank == 0
+      }
+      val ret = if (isRoot) {
           assert(cCentroids != 0)
           val centerVectors = if (useDevice == "GPU") {
             OneDAL.homogenTableToVectors(OneDAL.makeHomogenTable(cCentroids))

diff --git a/mllib-dal/src/main/scala/com/intel/oap/mllib/feature/PCADALImpl.scala b/mllib-dal/src/main/scala/com/intel/oap/mllib/feature/PCADALImpl.scala
@@ -20,7 +20,7 @@ import java.nio.DoubleBuffer
 import com.intel.daal.data_management.data.{HomogenNumericTable, NumericTable}
 import com.intel.oap.mllib.Utils.getOneCCLIPPort
 import com.intel.oap.mllib.{CommonJob, OneCCL, OneDAL, Service, Utils}
-import org.apache.spark.TaskContext
+import org.apache.spark.{SparkEnv, TaskContext}
 import org.apache.spark.annotation.Since
 import org.apache.spark.internal.Logging
 import org.apache.spark.ml.linalg._
@@ -86,8 +86,13 @@ class PCADALImpl(val k: Int,
         gpuIndices,
         result
       )
+      val isRoot = if (useDevice == "GPU") {
+        SparkEnv.get.executorId.toInt == 0
+      } else {
+        rank == 0
+      }
 
-      val ret = if (rank == 0) {
+      val ret = if (isRoot) {
         val principleComponents = if (useDevice == "GPU") {
           val pcNumericTable = OneDAL.makeHomogenTable(result.getPcNumericTable)
           getPrincipleComponentsFromOneAPI(pcNumericTable, k)

diff --git a/mllib-dal/src/main/scala/com/intel/oap/mllib/regression/LinearRegressionDALImpl.scala b/mllib-dal/src/main/scala/com/intel/oap/mllib/regression/LinearRegressionDALImpl.scala
@@ -19,8 +19,7 @@ package com.intel.oap.mllib.regression
 import com.intel.oap.mllib.Utils.getOneCCLIPPort
 import com.intel.oap.mllib.{CommonJob, OneCCL, OneDAL, Utils}
 import com.intel.oneapi.dal.table.Common
-import org.apache.spark.SparkException
-import org.apache.spark.TaskContext
+import org.apache.spark.{SparkEnv, SparkException, TaskContext}
 import org.apache.spark.internal.Logging
 import org.apache.spark.ml.linalg.{DenseVector, Vector}
 import org.apache.spark.ml.util._
@@ -156,7 +155,12 @@ class LinearRegressionDALImpl( val fitIntercept: Boolean,
           result
         )
 
-        val ret = if (rank == 0) {
+      val isRoot = if (useDevice == "GPU") {
+        SparkEnv.get.executorId.toInt == 0
+      } else {
+        rank == 0
+      }
+      val ret = if (isRoot) {
           val coefficientArray = if (useDevice == "GPU") {
               OneDAL.homogenTableToVectors(OneDAL.makeHomogenTable(cbeta))
             } else {

diff --git a/mllib-dal/src/main/scala/com/intel/oap/mllib/regression/RandomForestRegressorDALImpl.scala b/mllib-dal/src/main/scala/com/intel/oap/mllib/regression/RandomForestRegressorDALImpl.scala
@@ -19,7 +19,7 @@ import com.intel.oap.mllib.Utils.getOneCCLIPPort
 import com.intel.oap.mllib.classification.{LearningNode, RandomForestResult}
 import com.intel.oap.mllib.{CommonJob, OneCCL, OneDAL, Utils}
 import com.intel.oneapi.dal.table.Common
-import org.apache.spark.TaskContext
+import org.apache.spark.{SparkEnv, TaskContext}
 import org.apache.spark.internal.Logging
 import org.apache.spark.ml.classification.DecisionTreeClassificationModel
 import org.apache.spark.ml.linalg.Matrix
@@ -112,7 +112,12 @@ class RandomForestRegressorDALImpl(val uid: String,
 
       logInfo(s"RandomForestRegressorDALImpl compute took ${durationCompute} secs")
 
-      val ret = if (rank == 0) {
+      val isRoot = if (useDevice == "GPU") {
+        SparkEnv.get.executorId.toInt == 0
+      } else {
+        rank == 0
+      }
+      val ret = if (isRoot) {
         val convResultStartTime = System.nanoTime()
         val predictionNumericTable = OneDAL.homogenTableToMatrix(
           OneDAL.makeHomogenTable(result.getPredictionNumericTable))

diff --git a/mllib-dal/src/main/scala/com/intel/oap/mllib/stat/CorrelationDALImpl.scala b/mllib-dal/src/main/scala/com/intel/oap/mllib/stat/CorrelationDALImpl.scala
@@ -19,7 +19,7 @@ package com.intel.oap.mllib.stat
 import com.intel.oap.mllib.Utils.getOneCCLIPPort
 import com.intel.oap.mllib.{CommonJob, OneCCL, OneDAL, Utils}
 import com.intel.oneapi.dal.table.Common
-import org.apache.spark.TaskContext
+import org.apache.spark.{SparkEnv, TaskContext}
 import org.apache.spark.internal.Logging
 import org.apache.spark.ml.linalg.{Matrix, Vector}
 import org.apache.spark.rdd.RDD
@@ -83,7 +83,12 @@ class CorrelationDALImpl(
 
       logInfo(s"CorrelationDAL compute took ${durationCompute} secs")
 
-      val ret = if (rank == 0) {
+      val isRoot = if (useDevice == "GPU") {
+        SparkEnv.get.executorId.toInt == 0
+      } else {
+        rank == 0
+      }
+      val ret = if (isRoot) {
         val convResultStartTime = System.nanoTime()
         val correlationNumericTable = if (useDevice == "GPU") {
           OneDAL.homogenTableToMatrix(OneDAL.makeHomogenTable(result.getCorrelationNumericTable))