[incubator-celeborn] branch main updated: [CELEBORN-599] Consolidate calculation of mount point (#1505)

[ch...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

chengpan pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git


The following commit(s) were added to refs/heads/main by this push:
     new de97ad26c [CELEBORN-599] Consolidate calculation of mount point (#1505)
de97ad26c is described below

commit de97ad26ce15943232f2a326452a5f9b5bb18611
Author: Leo Li <ly...@163.com>
AuthorDate: Tue May 23 14:06:02 2023 +0800

    [CELEBORN-599] Consolidate calculation of mount point (#1505)
    
    * [CELEBORN-599] Fix worker dirs get mount point
    
    * update
    
    * update
    
    ---------
    
    Co-authored-by: liyihe <li...@bigo.sg>
---
 .../apache/celeborn/common/meta/DeviceInfo.scala   | 43 +++++++++++++---------
 .../celeborn/common/meta/DeviceInfoSuite.scala     | 41 +++++++++++++++++++++
 2 files changed, 67 insertions(+), 17 deletions(-)

diff --git a/common/src/main/scala/org/apache/celeborn/common/meta/DeviceInfo.scala b/common/src/main/scala/org/apache/celeborn/common/meta/DeviceInfo.scala
index be63f1859..5dbb8d2c4 100644
--- a/common/src/main/scala/org/apache/celeborn/common/meta/DeviceInfo.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/meta/DeviceInfo.scala
@@ -19,7 +19,6 @@ package org.apache.celeborn.common.meta
 
 import java.io.File
 import java.util
-import java.util.concurrent.ConcurrentHashMap
 
 import scala.collection.JavaConverters._
 import scala.collection.mutable.ListBuffer
@@ -227,22 +226,27 @@ object DeviceInfo {
     val retDeviceInfos = JavaUtils.newConcurrentHashMap[String, DeviceInfo]()
     val retDiskInfos = JavaUtils.newConcurrentHashMap[String, DiskInfo]()
 
-    workingDirs.groupBy { f =>
-      getMountPoint(f._1.getAbsolutePath, mountPointToDeviceInfo.keySet())
+    workingDirs.groupBy { case (dir, _, _, _) =>
+      getMountPoint(dir.getCanonicalPath, mountPointToDeviceInfo.keySet())
     }.foreach {
       case (mountPoint, dirs) =>
-        val deviceInfo = mountPointToDeviceInfo.get(mountPoint)
-        val diskInfo = new DiskInfo(
-          mountPoint,
-          dirs.map(_._1).toList,
-          deviceInfo,
-          conf)
-        val (_, maxUsableSpace, threadCount, storageType) = dirs(0)
-        diskInfo.configuredUsableSpace = maxUsableSpace
-        diskInfo.threadCount = threadCount
-        diskInfo.storageType = storageType
-        deviceInfo.addDiskInfo(diskInfo)
-        retDiskInfos.put(mountPoint, diskInfo)
+        if (mountPoint.nonEmpty) {
+          val deviceInfo = mountPointToDeviceInfo.get(mountPoint)
+          val diskInfo = new DiskInfo(
+            mountPoint,
+            dirs.map(_._1).toList,
+            deviceInfo,
+            conf)
+          val (_, maxUsableSpace, threadCount, storageType) = dirs(0)
+          diskInfo.configuredUsableSpace = maxUsableSpace
+          diskInfo.threadCount = threadCount
+          diskInfo.storageType = storageType
+          deviceInfo.addDiskInfo(diskInfo)
+          retDiskInfos.put(mountPoint, diskInfo)
+        } else {
+          logger.warn(
+            s"Can't find mount point for ${dirs.map(_._1.getCanonicalPath).mkString(",")}")
+        }
     }
     deviceNameToDeviceInfo.asScala.foreach {
       case (_, deviceInfo) =>
@@ -262,8 +266,13 @@ object DeviceInfo {
     var curMount = ""
     mountPoints.asScala.foreach(mount => {
       if (absPath.startsWith(mount) && mount.length > curMax) {
-        curMax = mount.length
-        curMount = mount
+        if (absPath.length == mount.length) {
+          return mount
+        } else if (absPath.length > mount.length &&
+          (mount == "/" || absPath.substring(mount.length, mount.length + 1) == "/")) {
+          curMax = mount.length
+          curMount = mount
+        }
       }
     })
     curMount
diff --git a/common/src/test/scala/org/apache/celeborn/common/meta/DeviceInfoSuite.scala b/common/src/test/scala/org/apache/celeborn/common/meta/DeviceInfoSuite.scala
new file mode 100644
index 000000000..250c20c3c
--- /dev/null
+++ b/common/src/test/scala/org/apache/celeborn/common/meta/DeviceInfoSuite.scala
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.
+ */
+
+package org.apache.celeborn.common.meta
+
+import java.util
+
+import org.apache.celeborn.CelebornFunSuite
+
+class DeviceInfoSuite extends CelebornFunSuite {
+
+  test("test getMountPoint") {
+    val mountPoints = new util.HashSet[String]()
+    mountPoints.add("/")
+    mountPoints.add("/mnt/disk1")
+    mountPoints.add("/mnt/disk2")
+    mountPoints.add("/data")
+
+    assert(DeviceInfo.getMountPoint("/mnt/disk1/data", mountPoints) === "/mnt/disk1")
+    assert(DeviceInfo.getMountPoint("/mnt/disk1/data", mountPoints) === "/mnt/disk1")
+    assert(DeviceInfo.getMountPoint("/mnt/disk2/data", mountPoints) === "/mnt/disk2")
+    assert(DeviceInfo.getMountPoint("/mnt/disk3/data", mountPoints) === "/")
+    assert(DeviceInfo.getMountPoint("/data", mountPoints) === "/data")
+    assert(DeviceInfo.getMountPoint("/data/data", mountPoints) === "/data")
+    assert(DeviceInfo.getMountPoint("/data1/data", mountPoints) === "/")
+  }
+}