You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@celeborn.apache.org by zh...@apache.org on 2023/06/02 06:58:38 UTC
[incubator-celeborn] 19/20: [CELEBORN-624] StorageManager should only remove expired app dirs (#1531)
This is an automated email from the ASF dual-hosted git repository.
zhongqiangchen pushed a commit to branch branch-0.3
in repository https://gitbox.apache.org/repos/asf/incubator-celeborn.git
commit d81c0c07bc40998bf5546b76083417909ad35970
Author: Angerszhuuuu <an...@gmail.com>
AuthorDate: Fri Jun 2 11:33:33 2023 +0800
[CELEBORN-624] StorageManager should only remove expired app dirs (#1531)
---
.../org/apache/celeborn/common/CelebornConf.scala | 11 ----------
docs/configuration/worker.md | 1 -
.../deploy/worker/storage/StorageManager.scala | 24 +++++++++-------------
3 files changed, 10 insertions(+), 26 deletions(-)
diff --git a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
index 3fcbf6f29..6b936abb0 100644
--- a/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
+++ b/common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala
@@ -600,7 +600,6 @@ class CelebornConf(loadDefaults: Boolean) extends Cloneable with Logging with Se
def workerFetchIoThreads: Option[Int] = get(WORKER_FETCH_IO_THREADS)
def workerReplicateIoThreads: Option[Int] = get(WORKER_REPLICATE_IO_THREADS)
def registerWorkerTimeout: Long = get(WORKER_REGISTER_TIMEOUT)
- def workerApplicationDataKeepAliveTime: Long = get(WORKER_APPLICATION_DATA_KEEP_ALIVE_TIME)
def workerWorkingDir: String = get(WORKER_WORKING_DIR)
def workerCloseIdleConnections: Boolean = get(WORKER_CLOSE_IDLE_CONNECTIONS)
def workerReplicateFastFailDuration: Long = get(WORKER_REPLICATE_FAST_FAIL_DURATION)
@@ -1905,16 +1904,6 @@ object CelebornConf extends Logging {
.bytesConf(ByteUnit.BYTE)
.createWithDefaultString("5G")
- val WORKER_APPLICATION_DATA_KEEP_ALIVE_TIME: ConfigEntry[Long] =
- buildConf("celeborn.worker.storage.applicationData.keepAliveTime")
- .withAlternative("celeborn.worker.noneEmptyDirExpireDuration")
- .withAlternative("rss.expire.nonEmptyDir.duration")
- .categories("worker")
- .doc("If a non-empty application shuffle data dir have not been operated during le duration time, will mark this application as expired.")
- .version("0.3.0")
- .timeConf(TimeUnit.MILLISECONDS)
- .createWithDefaultString("1d")
-
val WORKER_CHECK_FILE_CLEAN_MAX_RETRIES: ConfigEntry[Int] =
buildConf("celeborn.worker.storage.checkDirsEmpty.maxRetries")
.withAlternative("celeborn.worker.disk.checkFileClean.maxRetries")
diff --git a/docs/configuration/worker.md b/docs/configuration/worker.md
index 067f05cba..88736db5a 100644
--- a/docs/configuration/worker.md
+++ b/docs/configuration/worker.md
@@ -86,7 +86,6 @@ license: |
| celeborn.worker.sortPartition.reservedMemoryPerPartition | 1mb | Reserved memory when sorting a shuffle file off-heap. | 0.3.0 |
| celeborn.worker.sortPartition.threads | <undefined> | PartitionSorter's thread counts. | 0.3.0 |
| celeborn.worker.sortPartition.timeout | 220s | Timeout for a shuffle file to sort. | 0.3.0 |
-| celeborn.worker.storage.applicationData.keepAliveTime | 1d | If a non-empty application shuffle data dir have not been operated during le duration time, will mark this application as expired. | 0.3.0 |
| celeborn.worker.storage.checkDirsEmpty.maxRetries | 3 | The number of retries for a worker to check if the working directory is cleaned up before registering with the master. | 0.3.0 |
| celeborn.worker.storage.checkDirsEmpty.timeout | 1000ms | The wait time per retry for a worker to check if the working directory is cleaned up before registering with the master. | 0.3.0 |
| celeborn.worker.storage.dirs | <undefined> | Directory list to store shuffle data. It's recommended to configure one directory on each disk. Storage size limit can be set for each directory. For the sake of performance, there should be no more than 2 flush threads on the same disk partition if you are using HDD, and should be 8 or more flush threads on the same disk partition if you are using SSD. For example: `dir1[:capacity=][:disktype=][:flushthread=],dir2[:capacity=][:disktyp [...]
diff --git a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/StorageManager.scala b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/StorageManager.scala
index 7ca4e83e0..3c0267fa1 100644
--- a/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/StorageManager.scala
+++ b/worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/StorageManager.scala
@@ -192,7 +192,7 @@ final private[worker] class StorageManager(conf: CelebornConf, workerSource: Abs
this.db = null
}
}
- cleanupExpiredAppDirs(System.currentTimeMillis(), conf.workerGracefulShutdown)
+ cleanupExpiredAppDirs()
if (!checkIfWorkingDirCleaned) {
logWarning(
"Worker still has residual files in the working directory before registering with Master, " +
@@ -488,7 +488,6 @@ final private[worker] class StorageManager(conf: CelebornConf, workerSource: Abs
}
}
- private val applicationDataKeepAliveTime = conf.workerApplicationDataKeepAliveTime
private val storageScheduler =
ThreadUtils.newDaemonSingleThreadScheduledExecutor("storage-scheduler")
@@ -496,9 +495,8 @@ final private[worker] class StorageManager(conf: CelebornConf, workerSource: Abs
new Runnable {
override def run(): Unit = {
try {
- // Clean up dirs which has not been modified
- // in the past {{noneEmptyExpireDurationsMs}}.
- cleanupExpiredAppDirs(System.currentTimeMillis() - applicationDataKeepAliveTime)
+ // Clean up dirs which it's application is expired.
+ cleanupExpiredAppDirs()
} catch {
case exception: Exception =>
logWarning(s"Cleanup expired shuffle data exception: ${exception.getMessage}")
@@ -509,19 +507,17 @@ final private[worker] class StorageManager(conf: CelebornConf, workerSource: Abs
30,
TimeUnit.MINUTES)
- private def cleanupExpiredAppDirs(expireTime: Long, isGracefulShutdown: Boolean = false): Unit = {
+ private def cleanupExpiredAppDirs(): Unit = {
val appIds = shuffleKeySet().asScala.map(key => Utils.splitShuffleKey(key)._1)
disksSnapshot().filter(_.status != DiskStatus.IO_HANG).foreach { diskInfo =>
diskInfo.dirs.foreach {
case workingDir if workingDir.exists() =>
workingDir.listFiles().foreach { appDir =>
- // Don't delete shuffleKey's data recovered from levelDB when restart with graceful shutdown
- if (!(isGracefulShutdown && appIds.contains(appDir.getName))) {
- if (appDir.lastModified() < expireTime) {
- val threadPool = diskOperators.get(diskInfo.mountPoint)
- deleteDirectory(appDir, threadPool)
- logInfo(s"Delete expired app dir $appDir.")
- }
+ // Don't delete shuffleKey's data that exist correct shuffle file info.
+ if (!appIds.contains(appDir.getName)) {
+ val threadPool = diskOperators.get(diskInfo.mountPoint)
+ deleteDirectory(appDir, threadPool)
+ logInfo(s"Delete expired app dir $appDir.")
}
}
// workingDir not exist when initializing worker on new disk
@@ -535,7 +531,7 @@ final private[worker] class StorageManager(conf: CelebornConf, workerSource: Abs
val iter = hadoopFs.listFiles(hdfsWorkPath, false)
while (iter.hasNext) {
val fileStatus = iter.next()
- if (fileStatus.getModificationTime < expireTime) {
+ if (!appIds.contains(fileStatus.getPath.getName)) {
hadoopFs.delete(fileStatus.getPath, true)
}
}