[GitHub] [incubator-celeborn] waitinfuture commented on a diff in pull request #1202: [CELEBORN-243] [REWORK]fix bug that os's disk usage is low but celebo…

["waitinfuture (via GitHub)" <gi...@apache.org>]

waitinfuture commented on code in PR #1202:
URL: https://github.com/apache/incubator-celeborn/pull/1202#discussion_r1095631664


##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/Worker.scala:
##########
@@ -415,13 +415,11 @@ private[celeborn] class Worker(
         val fileWriter = partition.asInstanceOf[WorkingPartition].getFileWriter
         fileWriter.destroy(new IOException(
           s"Destroy FileWriter ${fileWriter} caused by shuffle ${shuffleKey} expired."))
-        removeExpiredWorkingDirWriters(fileWriter)

Review Comment:
   I think we can delete line 414 to 423, cc @RexXiong 



##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/FileWriter.java:
##########
@@ -308,13 +308,11 @@ public synchronized void destroy(IOException ioException) {
     }
   }
 
-  public void registerDestroyHook(List<FileWriter> fileWriters) {
+  public void registerDestroyHook(ConcurrentHashMap<String, FileWriter> fileWriters) {

Review Comment:
   do we need this method?



##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/StorageManager.scala:
##########
@@ -424,6 +421,20 @@ final private[worker] class StorageManager(conf: CelebornConf, workerSource: Abs
     expiredShuffleKeys.asScala.foreach { shuffleKey =>
       logInfo(s"Cleanup expired shuffle $shuffleKey.")
       if (fileInfos.containsKey(shuffleKey)) {
+        fileInfos.get(shuffleKey).asScala.foreach {
+          case (_, fileInfo) => {
+            val dir = fileInfo.getFile.getParentFile.getParentFile.getParentFile

Review Comment:
   mountPoint



##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/StorageManager.scala:
##########
@@ -424,6 +421,20 @@ final private[worker] class StorageManager(conf: CelebornConf, workerSource: Abs
     expiredShuffleKeys.asScala.foreach { shuffleKey =>
       logInfo(s"Cleanup expired shuffle $shuffleKey.")
       if (fileInfos.containsKey(shuffleKey)) {
+        fileInfos.get(shuffleKey).asScala.foreach {
+          case (_, fileInfo) => {
+            val dir = fileInfo.getFile.getParentFile.getParentFile.getParentFile
+            val writers = workingDirWriters.get(dir)
+            if (writers != null) {
+              val fileWriter = writers.get(fileInfo.getFilePath)
+              if (fileWriter != null) {
+                fileWriter.destroy(new IOException(

Review Comment:
   remove the fileWriter form writers here



##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/StorageManager.scala:
##########
@@ -424,6 +421,20 @@ final private[worker] class StorageManager(conf: CelebornConf, workerSource: Abs
     expiredShuffleKeys.asScala.foreach { shuffleKey =>
       logInfo(s"Cleanup expired shuffle $shuffleKey.")
       if (fileInfos.containsKey(shuffleKey)) {
+        fileInfos.get(shuffleKey).asScala.foreach {

Review Comment:
   ```val shuffleFiles = fileInfos.remove(shuffleKey)``` to reuse



-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@celeborn.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org