You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@celeborn.apache.org by "zhongqiangczq (via GitHub)" <gi...@apache.org> on 2023/02/03 03:51:36 UTC
[GitHub] [incubator-celeborn] zhongqiangczq opened a new pull request, #1202: [CELEBORN-243] [REVERT]fix bug that os's disk usage is low but celebo…
zhongqiangczq opened a new pull request, #1202:
URL: https://github.com/apache/incubator-celeborn/pull/1202
…rn thinks that it's high_disk_usage
<!--
Thanks for sending a pull request! Here are some tips for you:
- Make sure the PR title start w/ a JIRA ticket, e.g. '[CELEBORN-XXXX] Your PR title ...'.
- Be sure to keep the PR description updated to reflect all changes.
- Please write your PR title to summarize what this PR proposes.
- If possible, provide a concise example to reproduce the issue for a faster review.
-->
### What changes were proposed in this pull request?
Pre pr doesn't solve the problem.
It donesn't effect that woker cleanup the expired shuffle keys by using partitionLocationInfo to getAllMasterLocations,
because partitionLocationInfo's shufflekeys have been removed by commitFiles.
So, this pr do the following things as below:
1. workingDirWriters use a map to record the map between filename and filewriter.
2. StorageManger reverse fileinfos to get the expired files and then remove the filewriter
### Why are the changes needed?
### Does this PR introduce _any_ user-facing change?
### How was this patch tested?
@waitinfuture @FMX @RexXiong
--
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
[GitHub] [incubator-celeborn] zhongqiangczq commented on a diff in pull request #1202: [CELEBORN-243] [REWORK]fix bug that os's disk usage is low but celebo…
Posted by "zhongqiangczq (via GitHub)" <gi...@apache.org>.
zhongqiangczq commented on code in PR #1202:
URL: https://github.com/apache/incubator-celeborn/pull/1202#discussion_r1095444363
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/StorageManager.scala:
##########
@@ -317,9 +318,7 @@ final private[worker] class StorageManager(conf: CelebornConf, workerSource: Abs
}
fileInfos.computeIfAbsent(shuffleKey, newMapFunc).put(fileName, fileInfo)
- hdfsWriters.synchronized {
- hdfsWriters.add(hdfsWriter)
- }
+ hdfsWriters.put(fileName, hdfsWriter)
Review Comment:
hdfswriters role is just as workingDirWriters's value. so no need align
--
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
[GitHub] [incubator-celeborn] codecov[bot] commented on pull request #1202: [CELEBORN-243] [REVERT]fix bug that os's disk usage is low but celebo…
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #1202:
URL: https://github.com/apache/incubator-celeborn/pull/1202#issuecomment-1414853724
# [Codecov](https://codecov.io/gh/apache/incubator-celeborn/pull/1202?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#1202](https://codecov.io/gh/apache/incubator-celeborn/pull/1202?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (665d409) into [main](https://codecov.io/gh/apache/incubator-celeborn/commit/ced08a1d89740a1eaace03fde91ee9cb6319a22f?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (ced08a1) will **decrease** coverage by `0.08%`.
> The diff coverage is `0.00%`.
```diff
@@ Coverage Diff @@
## main #1202 +/- ##
============================================
- Coverage 26.88% 26.79% -0.08%
+ Complexity 777 773 -4
============================================
Files 205 206 +1
Lines 17433 17494 +61
Branches 1899 1904 +5
============================================
+ Hits 4685 4686 +1
- Misses 12425 12481 +56
- Partials 323 327 +4
```
| [Impacted Files](https://codecov.io/gh/apache/incubator-celeborn/pull/1202?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...born/service/deploy/worker/storage/FileWriter.java](https://codecov.io/gh/apache/incubator-celeborn/pull/1202?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-d29ya2VyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9jZWxlYm9ybi9zZXJ2aWNlL2RlcGxveS93b3JrZXIvc3RvcmFnZS9GaWxlV3JpdGVyLmphdmE=) | `36.50% <0.00%> (ø)` | |
| [...service/deploy/worker/storage/StorageManager.scala](https://codecov.io/gh/apache/incubator-celeborn/pull/1202?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-d29ya2VyL3NyYy9tYWluL3NjYWxhL29yZy9hcGFjaGUvY2VsZWJvcm4vc2VydmljZS9kZXBsb3kvd29ya2VyL3N0b3JhZ2UvU3RvcmFnZU1hbmFnZXIuc2NhbGE=) | `0.00% <0.00%> (ø)` | |
| [...celeborn/service/deploy/master/SlotsAllocator.java](https://codecov.io/gh/apache/incubator-celeborn/pull/1202?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-bWFzdGVyL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9jZWxlYm9ybi9zZXJ2aWNlL2RlcGxveS9tYXN0ZXIvU2xvdHNBbGxvY2F0b3IuamF2YQ==) | `69.27% <0.00%> (-2.45%)` | :arrow_down: |
| [...apache/celeborn/service/deploy/master/Master.scala](https://codecov.io/gh/apache/incubator-celeborn/pull/1202?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-bWFzdGVyL3NyYy9tYWluL3NjYWxhL29yZy9hcGFjaGUvY2VsZWJvcm4vc2VydmljZS9kZXBsb3kvbWFzdGVyL01hc3Rlci5zY2FsYQ==) | `0.00% <0.00%> (ø)` | |
| [...ache/celeborn/common/identity/UserIdentifier.scala](https://codecov.io/gh/apache/incubator-celeborn/pull/1202?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y29tbW9uL3NyYy9tYWluL3NjYWxhL29yZy9hcGFjaGUvY2VsZWJvcm4vY29tbW9uL2lkZW50aXR5L1VzZXJJZGVudGlmaWVyLnNjYWxh) | `0.00% <0.00%> (ø)` | |
| [...eleborn/common/metrics/source/AbstractSource.scala](https://codecov.io/gh/apache/incubator-celeborn/pull/1202?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y29tbW9uL3NyYy9tYWluL3NjYWxhL29yZy9hcGFjaGUvY2VsZWJvcm4vY29tbW9uL21ldHJpY3Mvc291cmNlL0Fic3RyYWN0U291cmNlLnNjYWxh) | `0.00% <0.00%> (ø)` | |
| [...mon/metrics/source/ResourceConsumptionSource.scala](https://codecov.io/gh/apache/incubator-celeborn/pull/1202?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y29tbW9uL3NyYy9tYWluL3NjYWxhL29yZy9hcGFjaGUvY2VsZWJvcm4vY29tbW9uL21ldHJpY3Mvc291cmNlL1Jlc291cmNlQ29uc3VtcHRpb25Tb3VyY2Uuc2NhbGE=) | `0.00% <0.00%> (ø)` | |
| [...cala/org/apache/celeborn/common/CelebornConf.scala](https://codecov.io/gh/apache/incubator-celeborn/pull/1202?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-Y29tbW9uL3NyYy9tYWluL3NjYWxhL29yZy9hcGFjaGUvY2VsZWJvcm4vY29tbW9uL0NlbGVib3JuQ29uZi5zY2FsYQ==) | `80.61% <0.00%> (+0.03%)` | :arrow_up: |
:mega: We’re building smart automated test selection to slash your CI/CD build times. [Learn more](https://about.codecov.io/iterative-testing/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
--
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
[GitHub] [incubator-celeborn] waitinfuture commented on pull request #1202: [CELEBORN-243] [REWORK]fix bug that os's disk usage is low but celebo…
Posted by "waitinfuture (via GitHub)" <gi...@apache.org>.
waitinfuture commented on PR #1202:
URL: https://github.com/apache/incubator-celeborn/pull/1202#issuecomment-1416702198
This PR actually fixes [CELEBORN-249](https://issues.apache.org/jira/projects/CELEBORN/issues/CELEBORN-249?filter=allopenissues)
--
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
[GitHub] [incubator-celeborn] waitinfuture merged pull request #1202: [CELEBORN-243] [REWORK]fix bug that os's disk usage is low but celebo…
Posted by "waitinfuture (via GitHub)" <gi...@apache.org>.
waitinfuture merged PR #1202:
URL: https://github.com/apache/incubator-celeborn/pull/1202
--
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
[GitHub] [incubator-celeborn] zhongqiangczq commented on a diff in pull request #1202: [CELEBORN-243] [REWORK]fix bug that os's disk usage is low but celebo…
Posted by "zhongqiangczq (via GitHub)" <gi...@apache.org>.
zhongqiangczq commented on code in PR #1202:
URL: https://github.com/apache/incubator-celeborn/pull/1202#discussion_r1095765575
##########
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:
it's removed after destoryed
--
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
[GitHub] [incubator-celeborn] RexXiong commented on a diff in pull request #1202: [CELEBORN-243] [REVERT]fix bug that os's disk usage is low but celebo…
Posted by "RexXiong (via GitHub)" <gi...@apache.org>.
RexXiong commented on code in PR #1202:
URL: https://github.com/apache/incubator-celeborn/pull/1202#discussion_r1095407363
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/StorageManager.scala:
##########
@@ -317,9 +318,7 @@ final private[worker] class StorageManager(conf: CelebornConf, workerSource: Abs
}
fileInfos.computeIfAbsent(shuffleKey, newMapFunc).put(fileName, fileInfo)
- hdfsWriters.synchronized {
- hdfsWriters.add(hdfsWriter)
- }
+ hdfsWriters.put(fileName, hdfsWriter)
Review Comment:
current hdfsWriters and workingDirWriters use different map key, can we align this?
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/StorageManager.scala:
##########
@@ -424,6 +421,24 @@ 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
+ logDebug(s"Cleanup expired shuffle before ${workingDirWriters.get(dir).size()}")
+ if (workingDirWriters.get(dir) != null) {
+ val fileWriter = workingDirWriters.get(dir).get(fileInfo.getFile.getAbsolutePath)
+ if (fileWriter != null) {
+ fileWriter.destroy(new IOException(
+ s"Destroy FileWriter ${fileWriter} caused by shuffle ${shuffleKey} expired."))
+ logDebug(s"Cleanup expired filelength:${fileWriter.getFileInfo.getFileLength}")
+ }
+ }
+ if (workingDirWriters.get(dir).size() == 0) {
+ workingDirWriters.remove(dir)
Review Comment:
if workingDirWriters.get(dir) can be null value, there will be a NPE
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/StorageManager.scala:
##########
@@ -424,6 +421,24 @@ 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
+ logDebug(s"Cleanup expired shuffle before ${workingDirWriters.get(dir).size()}")
+ if (workingDirWriters.get(dir) != null) {
+ val fileWriter = workingDirWriters.get(dir).get(fileInfo.getFile.getAbsolutePath)
+ if (fileWriter != null) {
+ fileWriter.destroy(new IOException(
+ s"Destroy FileWriter ${fileWriter} caused by shuffle ${shuffleKey} expired."))
+ logDebug(s"Cleanup expired filelength:${fileWriter.getFileInfo.getFileLength}")
+ }
+ }
+ if (workingDirWriters.get(dir).size() == 0) {
+ workingDirWriters.remove(dir)
Review Comment:
there will be a concurrent problem if we put a new filewriter while we remove this dir.
--
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
[GitHub] [incubator-celeborn] zhongqiangczq commented on a diff in pull request #1202: [CELEBORN-243] [REWORK]fix bug that os's disk usage is low but celebo…
Posted by "zhongqiangczq (via GitHub)" <gi...@apache.org>.
zhongqiangczq commented on code in PR #1202:
URL: https://github.com/apache/incubator-celeborn/pull/1202#discussion_r1095457716
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/StorageManager.scala:
##########
@@ -424,6 +421,24 @@ 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
+ logDebug(s"Cleanup expired shuffle before ${workingDirWriters.get(dir).size()}")
+ if (workingDirWriters.get(dir) != null) {
+ val fileWriter = workingDirWriters.get(dir).get(fileInfo.getFile.getAbsolutePath)
+ if (fileWriter != null) {
+ fileWriter.destroy(new IOException(
+ s"Destroy FileWriter ${fileWriter} caused by shuffle ${shuffleKey} expired."))
+ logDebug(s"Cleanup expired filelength:${fileWriter.getFileInfo.getFileLength}")
+ }
+ }
+ if (workingDirWriters.get(dir).size() == 0) {
+ workingDirWriters.remove(dir)
Review Comment:
yes. it my be null. now fixed
--
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
[GitHub] [incubator-celeborn] zhongqiangczq commented on a diff in pull request #1202: [CELEBORN-243] [REWORK]fix bug that os's disk usage is low but celebo…
Posted by "zhongqiangczq (via GitHub)" <gi...@apache.org>.
zhongqiangczq commented on code in PR #1202:
URL: https://github.com/apache/incubator-celeborn/pull/1202#discussion_r1095687806
##########
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:
this method is still needed for removing expired filewriter.
expired filewriter will reuse destory method to call hook
--
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
[GitHub] [incubator-celeborn] zhongqiangczq commented on a diff in pull request #1202: [CELEBORN-243] [REWORK]fix bug that os's disk usage is low but celebo…
Posted by "zhongqiangczq (via GitHub)" <gi...@apache.org>.
zhongqiangczq commented on code in PR #1202:
URL: https://github.com/apache/incubator-celeborn/pull/1202#discussion_r1095767278
##########
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:
now use shuffleFiles to traverse the fileinfos to delete all the expired writers
--
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
[GitHub] [incubator-celeborn] zhongqiangczq commented on a diff in pull request #1202: [CELEBORN-243] [REWORK]fix bug that os's disk usage is low but celebo…
Posted by "zhongqiangczq (via GitHub)" <gi...@apache.org>.
zhongqiangczq commented on code in PR #1202:
URL: https://github.com/apache/incubator-celeborn/pull/1202#discussion_r1095456008
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/StorageManager.scala:
##########
@@ -424,6 +421,24 @@ 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
+ logDebug(s"Cleanup expired shuffle before ${workingDirWriters.get(dir).size()}")
+ if (workingDirWriters.get(dir) != null) {
+ val fileWriter = workingDirWriters.get(dir).get(fileInfo.getFile.getAbsolutePath)
+ if (fileWriter != null) {
+ fileWriter.destroy(new IOException(
+ s"Destroy FileWriter ${fileWriter} caused by shuffle ${shuffleKey} expired."))
+ logDebug(s"Cleanup expired filelength:${fileWriter.getFileInfo.getFileLength}")
+ }
+ }
+ if (workingDirWriters.get(dir).size() == 0) {
+ workingDirWriters.remove(dir)
Review Comment:
yes. it my be null. now fixed
--
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
[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…
Posted by "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
[GitHub] [incubator-celeborn] zhongqiangczq commented on a diff in pull request #1202: [CELEBORN-243] [REWORK]fix bug that os's disk usage is low but celebo…
Posted by "zhongqiangczq (via GitHub)" <gi...@apache.org>.
zhongqiangczq commented on code in PR #1202:
URL: https://github.com/apache/incubator-celeborn/pull/1202#discussion_r1095766350
##########
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:
now dir replace to mountPointWorkingdir, because mountPointWorkingdir is consit of mountpoint + workingdir
--
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
[GitHub] [incubator-celeborn] zhongqiangczq commented on a diff in pull request #1202: [CELEBORN-243] [REWORK]fix bug that os's disk usage is low but celebo…
Posted by "zhongqiangczq (via GitHub)" <gi...@apache.org>.
zhongqiangczq commented on code in PR #1202:
URL: https://github.com/apache/incubator-celeborn/pull/1202#discussion_r1095444363
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/StorageManager.scala:
##########
@@ -317,9 +318,7 @@ final private[worker] class StorageManager(conf: CelebornConf, workerSource: Abs
}
fileInfos.computeIfAbsent(shuffleKey, newMapFunc).put(fileName, fileInfo)
- hdfsWriters.synchronized {
- hdfsWriters.add(hdfsWriter)
- }
+ hdfsWriters.put(fileName, hdfsWriter)
Review Comment:
hdfs doesn't need record diskinfo as local dir.
hdfswriters role is just as workingDirWriters's value. so no need align
--
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
[GitHub] [incubator-celeborn] zhongqiangczq commented on a diff in pull request #1202: [CELEBORN-243] [REWORK]fix bug that os's disk usage is low but celebo…
Posted by "zhongqiangczq (via GitHub)" <gi...@apache.org>.
zhongqiangczq commented on code in PR #1202:
URL: https://github.com/apache/incubator-celeborn/pull/1202#discussion_r1095758329
##########
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:
now it's deleted.
--
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
[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…
Posted by "waitinfuture (via GitHub)" <gi...@apache.org>.
waitinfuture commented on code in PR #1202:
URL: https://github.com/apache/incubator-celeborn/pull/1202#discussion_r1095786402
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/StorageManager.scala:
##########
@@ -424,7 +420,35 @@ final private[worker] class StorageManager(conf: CelebornConf, workerSource: Abs
expiredShuffleKeys.asScala.foreach { shuffleKey =>
logInfo(s"Cleanup expired shuffle $shuffleKey.")
if (fileInfos.containsKey(shuffleKey)) {
- val hdfsInfos = fileInfos.remove(shuffleKey).asScala.filter(_._2.isHdfs)
+ val removedFileInfos = fileInfos.remove(shuffleKey)
+ val hdfsInfos = new mutable.HashMap[String, FileInfo]
+ if (removedFileInfos != null) {
+ removedFileInfos.asScala.foreach {
+ case (fileName, fileInfo) => {
+ if (fileInfo.isHdfs) {
+ hdfsInfos += (fileName -> fileInfo)
Review Comment:
need refine
--
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