You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@celeborn.apache.org by "FMX (via GitHub)" <gi...@apache.org> on 2023/12/28 09:54:01 UTC
[PR] [CELEBORN-1201] Optimize memory usage of cache in partition sorter [incubator-celeborn]
FMX opened a new pull request, #2194:
URL: https://github.com/apache/incubator-celeborn/pull/2194
### What changes were proposed in this pull request?
Add a cache in partition sorted and limit its max size.
### Why are the changes needed?
To reduce memory consumption in partition sort by tweak the index cache.
### Does this PR introduce _any_ user-facing change?
NO.
### How was this patch tested?
GA and cluster.
--
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
Re: [PR] [CELEBORN-1201] Optimize memory usage of cache in partition sorter [incubator-celeborn]
Posted by "FMX (via GitHub)" <gi...@apache.org>.
FMX commented on code in PR #2194:
URL: https://github.com/apache/incubator-celeborn/pull/2194#discussion_r1440475833
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/PartitionFilesSorter.java:
##########
@@ -295,7 +313,7 @@ public void close(int exitKind) {
}
}
}
- cachedIndexMaps.clear();
+ indexCache.cleanUp();
Review Comment:
Yes, should call `invalidateAll` here.
--
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
Re: [PR] [CELEBORN-1201] Optimize memory usage of cache in partition sorter [incubator-celeborn]
Posted by "waitinfuture (via GitHub)" <gi...@apache.org>.
waitinfuture commented on code in PR #2194:
URL: https://github.com/apache/incubator-celeborn/pull/2194#discussion_r1440404979
##########
common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala:
##########
@@ -2418,6 +2420,22 @@ object CelebornConf extends Logging {
.intConf
.createOptional
+ val PARTITION_SORTER_INDEX_CACHE_MAX_WEIGHT: ConfigEntry[Long] =
+ buildConf("celeborn.worker.sortPartition.indexCache.maxWeight")
+ .categories("worker")
+ .doc("PartitionSorter's cache max weight for index buffer.")
+ .version("0.4.1")
+ .bytesConf(ByteUnit.BYTE)
+ .createWithDefaultString("10m")
Review Comment:
Since weight is now the total number of `ShuffleBlockInfo`s, it's unreasonable to use bytes here. Better to use long and default to say 10000.
--
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
Re: [PR] [CELEBORN-1201] Optimize memory usage of cache in partition sorter [incubator-celeborn]
Posted by "waitinfuture (via GitHub)" <gi...@apache.org>.
waitinfuture commented on code in PR #2194:
URL: https://github.com/apache/incubator-celeborn/pull/2194#discussion_r1440473774
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/PartitionFilesSorter.java:
##########
@@ -295,7 +313,7 @@ public void close(int exitKind) {
}
}
}
- cachedIndexMaps.clear();
+ indexCache.cleanUp();
Review Comment:
should call `invalidateAll`?
--
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
Re: [PR] [CELEBORN-1201] Optimize memory usage of cache in partition sorter [incubator-celeborn]
Posted by "FMX (via GitHub)" <gi...@apache.org>.
FMX commented on PR #2194:
URL: https://github.com/apache/incubator-celeborn/pull/2194#issuecomment-1875220795
All comments are addressed. PTAL. @waitinfuture @ErikFang
--
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
Re: [PR] [CELEBORN-1201] Optimize memory usage of cache in partition sorter [incubator-celeborn]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #2194:
URL: https://github.com/apache/incubator-celeborn/pull/2194#issuecomment-1871019688
## [Codecov](https://app.codecov.io/gh/apache/incubator-celeborn/pull/2194?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
All modified and coverable lines are covered by tests :white_check_mark:
> Comparison is base [(`6691568`)](https://app.codecov.io/gh/apache/incubator-celeborn/commit/669156824271caec03f2f1f62c48c02cd870e684?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) 47.26% compared to head [(`004eca4`)](https://app.codecov.io/gh/apache/incubator-celeborn/pull/2194?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) 47.22%.
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## main #2194 +/- ##
==========================================
- Coverage 47.26% 47.22% -0.04%
==========================================
Files 185 185
Lines 11717 11732 +15
Branches 1043 1043
==========================================
+ Hits 5537 5539 +2
- Misses 5813 5825 +12
- Partials 367 368 +1
```
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/incubator-celeborn/pull/2194?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
--
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
Re: [PR] [CELEBORN-1201] Optimize memory usage of cache in partition sorter [incubator-celeborn]
Posted by "mridulm (via GitHub)" <gi...@apache.org>.
mridulm commented on code in PR #2194:
URL: https://github.com/apache/incubator-celeborn/pull/2194#discussion_r1442141463
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/PartitionFilesSorter.java:
##########
@@ -465,43 +501,54 @@ public DiskFileInfo resolve(
int startMapIndex,
int endMapIndex)
throws IOException {
+ String indexCacheName = shuffleKey + "-" + fileId;
Map<Integer, List<ShuffleBlockInfo>> indexMap;
- if (cachedIndexMaps.containsKey(shuffleKey)
- && cachedIndexMaps.get(shuffleKey).containsKey(fileId)) {
- indexMap = cachedIndexMaps.get(shuffleKey).get(fileId);
- } else {
- FileChannel indexChannel = null;
- FSDataInputStream hdfsIndexStream = null;
- boolean isHdfs = Utils.isHdfsPath(indexFilePath);
- int indexSize = 0;
- try {
- if (isHdfs) {
- hdfsIndexStream = StorageManager.hadoopFs().open(new Path(indexFilePath));
- indexSize =
- (int) StorageManager.hadoopFs().getFileStatus(new Path(indexFilePath)).getLen();
- } else {
- indexChannel = FileChannelUtils.openReadableFileChannel(indexFilePath);
- File indexFile = new File(indexFilePath);
- indexSize = (int) indexFile.length();
- }
- ByteBuffer indexBuf = ByteBuffer.allocate(indexSize);
- if (isHdfs) {
- readStreamFully(hdfsIndexStream, indexBuf, indexFilePath);
- } else {
- readChannelFully(indexChannel, indexBuf, indexFilePath);
- }
- indexBuf.rewind();
- indexMap = ShuffleBlockInfoUtils.parseShuffleBlockInfosFromByteBuffer(indexBuf);
- Map<String, Map<Integer, List<ShuffleBlockInfo>>> cacheMap =
- cachedIndexMaps.computeIfAbsent(shuffleKey, v -> JavaUtils.newConcurrentHashMap());
- cacheMap.put(fileId, indexMap);
- } catch (Exception e) {
- logger.error("Read sorted shuffle file index " + indexFilePath + " error, detail: ", e);
- throw new IOException("Read sorted shuffle file index failed.", e);
- } finally {
- IOUtils.closeQuietly(indexChannel, null);
- IOUtils.closeQuietly(hdfsIndexStream, null);
- }
+ try {
+ indexMap =
+ indexCache.get(
+ indexCacheName,
+ () -> {
+ FileChannel indexChannel = null;
+ FSDataInputStream hdfsIndexStream = null;
+ boolean isHdfs = Utils.isHdfsPath(indexFilePath);
+ int indexSize = 0;
+ try {
+ if (isHdfs) {
+ hdfsIndexStream = StorageManager.hadoopFs().open(new Path(indexFilePath));
+ indexSize =
+ (int)
+ StorageManager.hadoopFs()
+ .getFileStatus(new Path(indexFilePath))
+ .getLen();
+ } else {
+ indexChannel = FileChannelUtils.openReadableFileChannel(indexFilePath);
+ File indexFile = new File(indexFilePath);
+ indexSize = (int) indexFile.length();
+ }
+ ByteBuffer indexBuf = ByteBuffer.allocate(indexSize);
+ if (isHdfs) {
+ readStreamFully(hdfsIndexStream, indexBuf, indexFilePath);
+ } else {
+ readChannelFully(indexChannel, indexBuf, indexFilePath);
+ }
+ indexBuf.rewind();
+ Map<Integer, List<ShuffleBlockInfo>> tIndexMap =
+ ShuffleBlockInfoUtils.parseShuffleBlockInfosFromByteBuffer(indexBuf);
+ Set<String> indexCacheItemsSet =
+ indexCacheNames.computeIfAbsent(shuffleKey, v -> new HashSet<>());
Review Comment:
QQ: this was not done right ? Trying to understand if this is still an issue.
--
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
Re: [PR] [CELEBORN-1201] Optimize memory usage of cache in partition sorter [incubator-celeborn]
Posted by "waitinfuture (via GitHub)" <gi...@apache.org>.
waitinfuture commented on code in PR #2194:
URL: https://github.com/apache/incubator-celeborn/pull/2194#discussion_r1440208807
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/PartitionFilesSorter.java:
##########
@@ -465,43 +501,54 @@ public DiskFileInfo resolve(
int startMapIndex,
int endMapIndex)
throws IOException {
+ String indexCacheName = shuffleKey + "-" + fileId;
Review Comment:
`fileId` is already `shuffleKey - fileName`, no need to prepend `shuffleKey` again
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/PartitionFilesSorter.java:
##########
@@ -129,6 +131,35 @@ public PartitionFilesSorter(
ThreadUtils.newDaemonCachedThreadPool(
"worker-file-sorter-execute", conf.partitionSorterThreads(), 120);
+ indexCache =
+ CacheBuilder.newBuilder()
+ .concurrencyLevel(conf.partitionSorterThreads())
+ .expireAfterAccess(conf.partitionSorterIndexExpire(), TimeUnit.MILLISECONDS)
+ .maximumWeight(indexCacheMaxWeight)
+ .weigher(
+ (key, cache) -> {
+ // estimated memory usage
+ int weight = 0;
Review Comment:
It's very hard to estimate real memory usage inside JVM, I doubt whether this piece of code correctly calculates real memory usage. IMO @ErikFang 's suggestion is simpler and better, using number of `ShuffleBlockInfo` is enough and easy to tune the configuration.
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/PartitionFilesSorter.java:
##########
@@ -465,43 +501,54 @@ public DiskFileInfo resolve(
int startMapIndex,
int endMapIndex)
throws IOException {
+ String indexCacheName = shuffleKey + "-" + fileId;
Map<Integer, List<ShuffleBlockInfo>> indexMap;
- if (cachedIndexMaps.containsKey(shuffleKey)
- && cachedIndexMaps.get(shuffleKey).containsKey(fileId)) {
- indexMap = cachedIndexMaps.get(shuffleKey).get(fileId);
- } else {
- FileChannel indexChannel = null;
- FSDataInputStream hdfsIndexStream = null;
- boolean isHdfs = Utils.isHdfsPath(indexFilePath);
- int indexSize = 0;
- try {
- if (isHdfs) {
- hdfsIndexStream = StorageManager.hadoopFs().open(new Path(indexFilePath));
- indexSize =
- (int) StorageManager.hadoopFs().getFileStatus(new Path(indexFilePath)).getLen();
- } else {
- indexChannel = FileChannelUtils.openReadableFileChannel(indexFilePath);
- File indexFile = new File(indexFilePath);
- indexSize = (int) indexFile.length();
- }
- ByteBuffer indexBuf = ByteBuffer.allocate(indexSize);
- if (isHdfs) {
- readStreamFully(hdfsIndexStream, indexBuf, indexFilePath);
- } else {
- readChannelFully(indexChannel, indexBuf, indexFilePath);
- }
- indexBuf.rewind();
- indexMap = ShuffleBlockInfoUtils.parseShuffleBlockInfosFromByteBuffer(indexBuf);
- Map<String, Map<Integer, List<ShuffleBlockInfo>>> cacheMap =
- cachedIndexMaps.computeIfAbsent(shuffleKey, v -> JavaUtils.newConcurrentHashMap());
- cacheMap.put(fileId, indexMap);
- } catch (Exception e) {
- logger.error("Read sorted shuffle file index " + indexFilePath + " error, detail: ", e);
- throw new IOException("Read sorted shuffle file index failed.", e);
- } finally {
- IOUtils.closeQuietly(indexChannel, null);
- IOUtils.closeQuietly(hdfsIndexStream, null);
- }
+ try {
+ indexMap =
+ indexCache.get(
+ indexCacheName,
+ () -> {
+ FileChannel indexChannel = null;
+ FSDataInputStream hdfsIndexStream = null;
+ boolean isHdfs = Utils.isHdfsPath(indexFilePath);
+ int indexSize = 0;
+ try {
+ if (isHdfs) {
+ hdfsIndexStream = StorageManager.hadoopFs().open(new Path(indexFilePath));
+ indexSize =
+ (int)
+ StorageManager.hadoopFs()
+ .getFileStatus(new Path(indexFilePath))
+ .getLen();
+ } else {
+ indexChannel = FileChannelUtils.openReadableFileChannel(indexFilePath);
+ File indexFile = new File(indexFilePath);
+ indexSize = (int) indexFile.length();
+ }
+ ByteBuffer indexBuf = ByteBuffer.allocate(indexSize);
+ if (isHdfs) {
+ readStreamFully(hdfsIndexStream, indexBuf, indexFilePath);
+ } else {
+ readChannelFully(indexChannel, indexBuf, indexFilePath);
+ }
+ indexBuf.rewind();
+ Map<Integer, List<ShuffleBlockInfo>> tIndexMap =
+ ShuffleBlockInfoUtils.parseShuffleBlockInfosFromByteBuffer(indexBuf);
+ Set<String> indexCacheItemsSet =
+ indexCacheNames.computeIfAbsent(shuffleKey, v -> new HashSet<>());
Review Comment:
I think we need to make it a concurrent set because it will be accessed by `cleanup` in other threads.
--
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
Re: [PR] [CELEBORN-1201] Optimize memory usage of cache in partition sorter [incubator-celeborn]
Posted by "waitinfuture (via GitHub)" <gi...@apache.org>.
waitinfuture closed pull request #2194: [CELEBORN-1201] Optimize memory usage of cache in partition sorter
URL: https://github.com/apache/incubator-celeborn/pull/2194
--
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
Re: [PR] [CELEBORN-1201] Optimize memory usage of cache in partition sorter [incubator-celeborn]
Posted by "ErikFang (via GitHub)" <gi...@apache.org>.
ErikFang commented on code in PR #2194:
URL: https://github.com/apache/incubator-celeborn/pull/2194#discussion_r1439542916
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/PartitionFilesSorter.java:
##########
@@ -127,6 +129,13 @@ public PartitionFilesSorter(
ThreadUtils.newDaemonCachedThreadPool(
"worker-file-sorter-execute", conf.partitionSorterThreads(), 120);
+ indexCache =
+ CacheBuilder.newBuilder()
+ .concurrencyLevel(conf.partitionSorterThreads())
+ .expireAfterAccess(conf.partitionSorterIndexExpire(), TimeUnit.MILLISECONDS)
+ .maximumSize(indexCacheSize)
Review Comment:
how about using maximumWeight() instead and number of ShuffleBlockInfos as weight?
```suggestion
.maximumWeight(indexCacheSize)
.weigher(new Weigher<String, Map<Integer, List<ShuffleBlockInfo>>>() {
@Override
public int weigh(String key, Map<Integer, List<ShuffleBlockInfo>> infoMap) {
return infoMap.values().stream().mapToInt(List::size).sum();
}
})
```
--
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