You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@celeborn.apache.org by "oOKevinOo (via GitHub)" <gi...@apache.org> on 2023/11/08 08:26:57 UTC
[PR] support file to memory cache [incubator-celeborn]
oOKevinOo opened a new pull request, #2081:
URL: https://github.com/apache/incubator-celeborn/pull/2081
<!--
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?
Support for small file caching to improve shuffle performance,The architecture is as follows:
### Why are the changes needed?
Support for small file caching to improve shuffle performance
### Does this PR introduce _any_ user-facing change?
None
### How was this patch tested?
--
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] support file to memory cache [incubator-celeborn]
Posted by "waitinfuture (via GitHub)" <gi...@apache.org>.
waitinfuture commented on code in PR #2081:
URL: https://github.com/apache/incubator-celeborn/pull/2081#discussion_r1386523255
##########
common/src/main/java/org/apache/celeborn/common/util/MemCacheManager.java:
##########
@@ -0,0 +1,70 @@
+package org.apache.celeborn.common.util;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicLong;
+
+import io.netty.buffer.ByteBuf;
+
+import org.apache.celeborn.common.CelebornConf;
+
+public class MemCacheManager {
+ private static MemCacheManager memCacheManager;
+ private ConcurrentMap<String, ByteBuf> caches = new ConcurrentHashMap<>();
+ private CelebornConf conf;
+ private long maxCacheSize;
+ private boolean cacheEnable;
+ private AtomicLong currentCacheSize = new AtomicLong(0);
+
+ public MemCacheManager(CelebornConf conf) {
+ this.conf = conf;
Review Comment:
`this.conf` seems to be useless
##########
common/src/main/java/org/apache/celeborn/common/util/MemCacheManager.java:
##########
@@ -0,0 +1,70 @@
+package org.apache.celeborn.common.util;
Review Comment:
Please add license header for the new added file
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/PartitionFilesSorter.java:
##########
@@ -610,11 +611,25 @@ public void sort() throws InterruptedException {
}
private void initializeFiles() throws IOException {
+ MemCacheManager memCacheManager = MemCacheManager.getMemCacheManager();
Review Comment:
We can get `CelebornConf` by calling `conf.getCelebornConf()` here.
##########
common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala:
##########
@@ -2057,6 +2060,22 @@ object CelebornConf extends Logging {
.bytesConf(ByteUnit.BYTE)
.createWithDefaultString("2g")
+ val FILE_CACHE_MAX_SIZE: ConfigEntry[Long] =
+ buildConf("celeborn.shuffle.cache.max")
+ .categories("worker")
+ .doc("Max size for shuffle file cache to memory")
+ .version("0.4.0")
+ .bytesConf(ByteUnit.BYTE)
+ .createWithDefaultString("10g")
+
+ val FILE_CACHE_ENABLE: ConfigEntry[Boolean] =
+ buildConf("celeborn.shuffle.cache.enable")
+ .categories("worker")
+ .doc("memory cache enable")
+ .version("0.4.0")
Review Comment:
0.3.2
##########
common/src/main/java/org/apache/celeborn/common/util/MemCacheManager.java:
##########
@@ -0,0 +1,70 @@
+package org.apache.celeborn.common.util;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicLong;
+
+import io.netty.buffer.ByteBuf;
+
+import org.apache.celeborn.common.CelebornConf;
+
+public class MemCacheManager {
+ private static MemCacheManager memCacheManager;
+ private ConcurrentMap<String, ByteBuf> caches = new ConcurrentHashMap<>();
+ private CelebornConf conf;
+ private long maxCacheSize;
+ private boolean cacheEnable;
+ private AtomicLong currentCacheSize = new AtomicLong(0);
+
+ public MemCacheManager(CelebornConf conf) {
+ this.conf = conf;
+ maxCacheSize = this.conf.fileCacheMaxSize();
+ cacheEnable = this.conf.fileCacheEnable();
+ }
+
+ public void putCache(String key, ByteBuf cache) {
+ int cacheSize = cache.readableBytes();
+ caches.put(key, cache);
+ currentCacheSize.getAndAdd(cacheSize);
+ }
+
+ public boolean canCache(int cacheSize) {
+ return cacheEnable && (maxCacheSize > currentCacheSize.get() + cacheSize);
+ }
+
+ public void removeCache(String key) {
+ ByteBuf cache = caches.remove(key);
Review Comment:
Seems need to release the cache to avoid netty memory leak.
##########
common/src/main/java/org/apache/celeborn/common/util/MemCacheManager.java:
##########
@@ -0,0 +1,70 @@
+package org.apache.celeborn.common.util;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicLong;
+
+import io.netty.buffer.ByteBuf;
+
+import org.apache.celeborn.common.CelebornConf;
+
+public class MemCacheManager {
+ private static MemCacheManager memCacheManager;
+ private ConcurrentMap<String, ByteBuf> caches = new ConcurrentHashMap<>();
+ private CelebornConf conf;
+ private long maxCacheSize;
+ private boolean cacheEnable;
+ private AtomicLong currentCacheSize = new AtomicLong(0);
+
+ public MemCacheManager(CelebornConf conf) {
+ this.conf = conf;
+ maxCacheSize = this.conf.fileCacheMaxSize();
+ cacheEnable = this.conf.fileCacheEnable();
+ }
+
+ public void putCache(String key, ByteBuf cache) {
+ int cacheSize = cache.readableBytes();
+ caches.put(key, cache);
+ currentCacheSize.getAndAdd(cacheSize);
+ }
+
+ public boolean canCache(int cacheSize) {
+ return cacheEnable && (maxCacheSize > currentCacheSize.get() + cacheSize);
+ }
+
+ public void removeCache(String key) {
+ ByteBuf cache = caches.remove(key);
+ currentCacheSize.getAndAdd(-1 * cache.readableBytes());
+ }
+
+ public boolean contains(String key) {
+ if (!cacheEnable) return false;
+ return caches.containsKey(key);
+ }
+
+ public ByteBuf getCache(String key) {
+ return caches.get(key);
+ }
+
+ public static synchronized MemCacheManager getMemCacheManager(CelebornConf conf) {
+ if (memCacheManager == null) {
Review Comment:
ditto
##########
common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala:
##########
@@ -2057,6 +2060,22 @@ object CelebornConf extends Logging {
.bytesConf(ByteUnit.BYTE)
.createWithDefaultString("2g")
+ val FILE_CACHE_MAX_SIZE: ConfigEntry[Long] =
+ buildConf("celeborn.shuffle.cache.max")
+ .categories("worker")
+ .doc("Max size for shuffle file cache to memory")
+ .version("0.4.0")
Review Comment:
0.3.2
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/ChunkStreamManager.java:
##########
@@ -97,22 +99,20 @@ public void chunkBeingSent(long streamId) {
StreamState streamState = streams.get(streamId);
if (streamState != null) {
streamState.chunksBeingTransferred++;
+ chunksBeingTransferredNum.incrementAndGet();
Review Comment:
The current logic is referred to Spark's implementation. I'm not sure how much it will benefit with this change.
##########
common/src/main/java/org/apache/celeborn/common/util/MemCacheManager.java:
##########
@@ -0,0 +1,70 @@
+package org.apache.celeborn.common.util;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicLong;
+
+import io.netty.buffer.ByteBuf;
+
+import org.apache.celeborn.common.CelebornConf;
+
+public class MemCacheManager {
+ private static MemCacheManager memCacheManager;
+ private ConcurrentMap<String, ByteBuf> caches = new ConcurrentHashMap<>();
+ private CelebornConf conf;
+ private long maxCacheSize;
+ private boolean cacheEnable;
+ private AtomicLong currentCacheSize = new AtomicLong(0);
+
+ public MemCacheManager(CelebornConf conf) {
+ this.conf = conf;
+ maxCacheSize = this.conf.fileCacheMaxSize();
+ cacheEnable = this.conf.fileCacheEnable();
+ }
+
+ public void putCache(String key, ByteBuf cache) {
+ int cacheSize = cache.readableBytes();
+ caches.put(key, cache);
+ currentCacheSize.getAndAdd(cacheSize);
+ }
+
+ public boolean canCache(int cacheSize) {
+ return cacheEnable && (maxCacheSize > currentCacheSize.get() + cacheSize);
+ }
+
+ public void removeCache(String key) {
+ ByteBuf cache = caches.remove(key);
+ currentCacheSize.getAndAdd(-1 * cache.readableBytes());
+ }
+
+ public boolean contains(String key) {
+ if (!cacheEnable) return false;
+ return caches.containsKey(key);
+ }
+
+ public ByteBuf getCache(String key) {
+ return caches.get(key);
+ }
+
+ public static synchronized MemCacheManager getMemCacheManager(CelebornConf conf) {
+ if (memCacheManager == null) {
+ memCacheManager = new MemCacheManager(conf);
+ }
+ return memCacheManager;
+ }
+
+ public static synchronized MemCacheManager getMemCacheManager() {
Review Comment:
nit: recommend to write this way
```
public static MemCacheManager getMemCacheManager(CelebornConf conf) {
if (memCacheManager == null) {
synchronized (MemCacheManager.class) {
if (memCacheManager == null) {
memCacheManager = new MemCacheManager(conf);
}
}
}
return memCacheManager;
}
```
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/FileWriter.java:
##########
@@ -154,19 +155,27 @@ public void decrementPendingWrites() {
@GuardedBy("flushLock")
protected void flush(boolean finalFlush) throws IOException {
+ MemCacheManager cacheManager = MemCacheManager.getMemCacheManager();
Review Comment:
We can make `cacheManager` a member of `FileWriter`, and initialize in the constructor of `FileWriter`, we can get `CelebornConf` there.
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/FileWriter.java:
##########
@@ -154,19 +155,27 @@ public void decrementPendingWrites() {
@GuardedBy("flushLock")
protected void flush(boolean finalFlush) throws IOException {
+ MemCacheManager cacheManager = MemCacheManager.getMemCacheManager();
// flushBuffer == null here means writer already closed
if (flushBuffer != null) {
int numBytes = flushBuffer.readableBytes();
if (numBytes != 0) {
notifier.checkException();
- notifier.numPendingFlushes.incrementAndGet();
- FlushTask task = null;
- if (channel != null) {
- task = new LocalFlushTask(flushBuffer, channel, notifier);
- } else if (fileInfo.isHdfs()) {
- task = new HdfsFlushTask(flushBuffer, fileInfo.getHdfsPath(), notifier);
+ if (finalFlush && fileInfo.getFileLength() == 0 && cacheManager.canCache(numBytes)) {
Review Comment:
Multiple thread can call `flush` simutaneously. We need synchronize `cacheManager.canCache(numBytes)`. Maybe we can add a method `cacheManager#tryCache` which returns true if caches successfully.
##########
common/src/main/java/org/apache/celeborn/common/meta/FileManagedBuffers.java:
##########
@@ -56,6 +66,26 @@ public ManagedBuffer chunk(int chunkIndex, int offset, int len) {
final long chunkLength = offsets[chunkIndex + 1] - chunkOffset;
assert offset < chunkLength;
long length = Math.min(chunkLength - offset, len);
+ MemCacheManager memCacheManager = MemCacheManager.getMemCacheManager();
+ chunkTotal++;
+ if (memCacheManager.contains(file.getAbsolutePath())) {
+ String filePath = file.getAbsolutePath();
+ ByteBuf fileBuffer = memCacheManager.getCache(filePath);
+ ByteBuf buffer = fileBuffer.slice((int) (chunkOffset + offset), (int) length);
Review Comment:
Seems we need to call `buffer.retain` here to support retried read. cc @RexXiong @FMX
##########
common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala:
##########
@@ -2057,6 +2060,22 @@ object CelebornConf extends Logging {
.bytesConf(ByteUnit.BYTE)
.createWithDefaultString("2g")
+ val FILE_CACHE_MAX_SIZE: ConfigEntry[Long] =
+ buildConf("celeborn.shuffle.cache.max")
+ .categories("worker")
+ .doc("Max size for shuffle file cache to memory")
+ .version("0.4.0")
+ .bytesConf(ByteUnit.BYTE)
+ .createWithDefaultString("10g")
Review Comment:
10g is too big, for some users' environments total direct memory is below 10g. maybe 512m is better
##########
common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala:
##########
@@ -2057,6 +2060,22 @@ object CelebornConf extends Logging {
.bytesConf(ByteUnit.BYTE)
.createWithDefaultString("2g")
+ val FILE_CACHE_MAX_SIZE: ConfigEntry[Long] =
+ buildConf("celeborn.shuffle.cache.max")
+ .categories("worker")
+ .doc("Max size for shuffle file cache to memory")
+ .version("0.4.0")
+ .bytesConf(ByteUnit.BYTE)
+ .createWithDefaultString("10g")
+
+ val FILE_CACHE_ENABLE: ConfigEntry[Boolean] =
+ buildConf("celeborn.shuffle.cache.enable")
+ .categories("worker")
+ .doc("memory cache enable")
+ .version("0.4.0")
+ .booleanConf
+ .createWithDefault(true)
Review Comment:
Better defaults to false at first
--
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] support file to memory cache [incubator-celeborn]
Posted by "cfmcgrady (via GitHub)" <gi...@apache.org>.
cfmcgrady commented on PR #2081:
URL: https://github.com/apache/incubator-celeborn/pull/2081#issuecomment-1801705550
> I think we can merge this first, since this PR has a good result for small shuffle files
then the PR target branch should be `main` instead of `branch-0.3`?
--
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] support file to memory cache [incubator-celeborn]
Posted by "codecov[bot] (via GitHub)" <gi...@apache.org>.
codecov[bot] commented on PR #2081:
URL: https://github.com/apache/incubator-celeborn/pull/2081#issuecomment-1803136427
## [Codecov](https://app.codecov.io/gh/apache/incubator-celeborn/pull/2081?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
> Merging [#2081](https://app.codecov.io/gh/apache/incubator-celeborn/pull/2081?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) (8c01f77) into [branch-0.3](https://app.codecov.io/gh/apache/incubator-celeborn/commit/697571ee8fb0ad1a29a101122f2f2bc9ce4965aa?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) (697571e) will **decrease** coverage by `0.16%`.
> Report is 7 commits behind head on branch-0.3.
> The diff coverage is `36.77%`.
```diff
@@ Coverage Diff @@
## branch-0.3 #2081 +/- ##
==============================================
- Coverage 46.17% 46.01% -0.16%
==============================================
Files 165 168 +3
Lines 10542 10608 +66
Branches 984 993 +9
==============================================
+ Hits 4867 4880 +13
- Misses 5355 5402 +47
- Partials 320 326 +6
```
| [Files](https://app.codecov.io/gh/apache/incubator-celeborn/pull/2081?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [...pache/celeborn/common/meta/FileManagedBuffers.java](https://app.codecov.io/gh/apache/incubator-celeborn/pull/2081?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9jZWxlYm9ybi9jb21tb24vbWV0YS9GaWxlTWFuYWdlZEJ1ZmZlcnMuamF2YQ==) | `78.95% <0.00%> (ø)` | |
| [...cala/org/apache/celeborn/common/CelebornConf.scala](https://app.codecov.io/gh/apache/incubator-celeborn/pull/2081?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29tbW9uL3NyYy9tYWluL3NjYWxhL29yZy9hcGFjaGUvY2VsZWJvcm4vY29tbW9uL0NlbGVib3JuQ29uZi5zY2FsYQ==) | `87.46% <93.34%> (+0.03%)` | :arrow_up: |
| [...g/apache/celeborn/common/util/MemCacheManager.java](https://app.codecov.io/gh/apache/incubator-celeborn/pull/2081?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9jZWxlYm9ybi9jb21tb24vdXRpbC9NZW1DYWNoZU1hbmFnZXIuamF2YQ==) | `40.75% <40.75%> (ø)` | |
| [...org/apache/celeborn/common/meta/MemoryBuffers.java](https://app.codecov.io/gh/apache/incubator-celeborn/pull/2081?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-Y29tbW9uL3NyYy9tYWluL2phdmEvb3JnL2FwYWNoZS9jZWxlYm9ybi9jb21tb24vbWV0YS9NZW1vcnlCdWZmZXJzLmphdmE=) | `0.00% <0.00%> (ø)` | |
... and [8 files with indirect coverage changes](https://app.codecov.io/gh/apache/incubator-celeborn/pull/2081/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
:mega: Codecov offers a browser extension for seamless coverage viewing on GitHub. Try it in [Chrome](https://chrome.google.com/webstore/detail/codecov/gedikamndpbemklijjkncpnolildpbgo) or [Firefox](https://addons.mozilla.org/en-US/firefox/addon/codecov/) today!
--
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-1118] Cache files in memory [incubator-celeborn]
Posted by "waitinfuture (via GitHub)" <gi...@apache.org>.
waitinfuture commented on code in PR #2081:
URL: https://github.com/apache/incubator-celeborn/pull/2081#discussion_r1389187870
##########
common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala:
##########
@@ -2057,6 +2060,22 @@ object CelebornConf extends Logging {
.bytesConf(ByteUnit.BYTE)
.createWithDefaultString("2g")
+ val FILE_CACHE_MAX_SIZE: ConfigEntry[Long] =
+ buildConf("celeborn.shuffle.cache.max")
+ .categories("worker")
+ .doc("Max size for shuffle file cache to memory")
+ .version("0.3.2")
+ .bytesConf(ByteUnit.BYTE)
+ .createWithDefaultString("256m")
+
+ val FILE_CACHE_ENABLE: ConfigEntry[Boolean] =
+ buildConf("celeborn.shuffle.cache.enable")
Review Comment:
celeborn.worker.memoryCache.enabled
##########
common/src/main/java/org/apache/celeborn/common/meta/MemoryBuffers.java:
##########
@@ -0,0 +1,79 @@
+/*
+ * 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.List;
+
+import io.netty.buffer.ByteBuf;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.celeborn.common.network.buffer.ManagedBuffer;
+import org.apache.celeborn.common.network.buffer.NettyManagedBuffer;
+import org.apache.celeborn.common.network.util.TransportConf;
+import org.apache.celeborn.common.util.MemCacheManager;
+
+public class MemoryBuffers implements Buffers {
+ private static final Logger logger = LoggerFactory.getLogger(FileManagedBuffers.class);
Review Comment:
`MemoryBuffers.class`
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/FileWriter.java:
##########
@@ -159,14 +162,21 @@ protected void flush(boolean finalFlush) throws IOException {
int numBytes = flushBuffer.readableBytes();
if (numBytes != 0) {
notifier.checkException();
- notifier.numPendingFlushes.incrementAndGet();
- FlushTask task = null;
- if (channel != null) {
- task = new LocalFlushTask(flushBuffer, channel, notifier);
- } else if (fileInfo.isHdfs()) {
- task = new HdfsFlushTask(flushBuffer, fileInfo.getHdfsPath(), notifier);
+ if (finalFlush && fileInfo.getFileLength() == 0 && cacheManager.canCache(numBytes)) {
Review Comment:
Need to check whether it's local disk here `channel != null`
##########
common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala:
##########
@@ -2057,6 +2060,22 @@ object CelebornConf extends Logging {
.bytesConf(ByteUnit.BYTE)
.createWithDefaultString("2g")
+ val FILE_CACHE_MAX_SIZE: ConfigEntry[Long] =
+ buildConf("celeborn.shuffle.cache.max")
Review Comment:
celeborn.worker.memoryCache.capacity
##########
worker/src/main/scala/org/apache/celeborn/service/deploy/worker/storage/StorageManager.scala:
##########
@@ -558,6 +560,10 @@ final private[worker] class StorageManager(conf: CelebornConf, workerSource: Abs
if (removedFileInfos != null) {
removedFileInfos.asScala.foreach {
case (_, fileInfo) =>
+ if (memCacheManager.contains(fileInfo.getFilePath)) {
Review Comment:
Multiple threads can simultaneously call `cleanupExpiredShuffleKey`. IMO we can change `removeCache` as follows:
```
public void removeCache(String key) {
ByteBuf cache = caches.remove(key);
if (cache != null) {
currentCacheSize.getAndAdd(-1 * cache.readableBytes());
cache.release();
}
}
```
Here we can just call `removeCache` without checking contains.
##########
common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala:
##########
@@ -2057,6 +2060,22 @@ object CelebornConf extends Logging {
.bytesConf(ByteUnit.BYTE)
.createWithDefaultString("2g")
+ val FILE_CACHE_MAX_SIZE: ConfigEntry[Long] =
Review Comment:
WORKER_MEMORYCACHE_CAPACITY
##########
common/src/main/java/org/apache/celeborn/common/meta/MemoryBuffers.java:
##########
@@ -0,0 +1,79 @@
+/*
+ * 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.List;
+
+import io.netty.buffer.ByteBuf;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.celeborn.common.network.buffer.ManagedBuffer;
+import org.apache.celeborn.common.network.buffer.NettyManagedBuffer;
+import org.apache.celeborn.common.network.util.TransportConf;
+import org.apache.celeborn.common.util.MemCacheManager;
+
+public class MemoryBuffers implements Buffers {
+ private static final Logger logger = LoggerFactory.getLogger(FileManagedBuffers.class);
+
+ private final long[] offsets;
+ private final int numChunks;
+ private final String filePath;
+
+ private final TransportConf conf;
Review Comment:
`conf` is not used
##########
common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala:
##########
@@ -913,6 +913,9 @@ class CelebornConf(loadDefaults: Boolean) extends Cloneable with Logging with Se
def partitionSplitMinimumSize: Long = get(WORKER_PARTITION_SPLIT_MIN_SIZE)
def partitionSplitMaximumSize: Long = get(WORKER_PARTITION_SPLIT_MAX_SIZE)
+ def fileCacheMaxSize: Long = get(FILE_CACHE_MAX_SIZE)
+ def fileCacheEnable: Boolean = get(FILE_CACHE_ENABLE)
Review Comment:
workerMemoryCacheEnabled
##########
common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala:
##########
@@ -913,6 +913,9 @@ class CelebornConf(loadDefaults: Boolean) extends Cloneable with Logging with Se
def partitionSplitMinimumSize: Long = get(WORKER_PARTITION_SPLIT_MIN_SIZE)
def partitionSplitMaximumSize: Long = get(WORKER_PARTITION_SPLIT_MAX_SIZE)
+ def fileCacheMaxSize: Long = get(FILE_CACHE_MAX_SIZE)
Review Comment:
workerMemoryCacheCapacity
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/PartitionFilesSorter.java:
##########
@@ -611,10 +614,23 @@ public void sort() throws InterruptedException {
private void initializeFiles() throws IOException {
if (isHdfs) {
+ if (memCacheManager.contains(originFilePath)) {
Review Comment:
Since this PR only affects local disk, we can remove this code block
##########
common/src/main/scala/org/apache/celeborn/common/CelebornConf.scala:
##########
@@ -2057,6 +2060,22 @@ object CelebornConf extends Logging {
.bytesConf(ByteUnit.BYTE)
.createWithDefaultString("2g")
+ val FILE_CACHE_MAX_SIZE: ConfigEntry[Long] =
+ buildConf("celeborn.shuffle.cache.max")
+ .categories("worker")
+ .doc("Max size for shuffle file cache to memory")
+ .version("0.3.2")
+ .bytesConf(ByteUnit.BYTE)
+ .createWithDefaultString("256m")
+
+ val FILE_CACHE_ENABLE: ConfigEntry[Boolean] =
Review Comment:
WORKER_MEMORYCACHE_ENABLED
--
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-1118] Cache files in memory [incubator-celeborn]
Posted by "oOKevinOo (via GitHub)" <gi...@apache.org>.
oOKevinOo closed pull request #2081: [CELEBORN-1118] Cache files in memory
URL: https://github.com/apache/incubator-celeborn/pull/2081
--
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] support file to memory cache [incubator-celeborn]
Posted by "oOKevinOo (via GitHub)" <gi...@apache.org>.
oOKevinOo commented on PR #2081:
URL: https://github.com/apache/incubator-celeborn/pull/2081#issuecomment-1803112148
> > > Hi, I've created a Jira ticket for you, https://issues.apache.org/jira/browse/CELEBORN-1118.
> >
> >
> > Is this PR a subtask of [Support memory file storage](https://www.mail-archive.com/dev@celeborn.apache.org/msg00613.html)?
>
> Not really. We discussed with @oOKevinOo offline, he implemented this PR without aware of https://www.mail-archive.com/dev@celeborn.apache.org/msg00613.html . I think we can merge this first, since this PR has a good result for small shuffle files, @oOKevinOo could you share some of your test? The memory storage will take longer time before GA.
We are currently facing serious shuffle read performance issues, but the disk performance is far from expectations. Through service investigation, we suspect that small files have a significant impact on disk performance, so we raised this PR. This is our indicator data based on the pressure measurement of this PR:
> > > Hi, I've created a Jira ticket for you, https://issues.apache.org/jira/browse/CELEBORN-1118.
> >
> >
> > Is this PR a subtask of [Support memory file storage](https://www.mail-archive.com/dev@celeborn.apache.org/msg00613.html)?
>
> Not really. We discussed with @oOKevinOo offline, he implemented this PR without aware of https://www.mail-archive.com/dev@celeborn.apache.org/msg00613.html . I think we can merge this first, since this PR has a good result for small shuffle files, @oOKevinOo could you share some of your test? The memory storage will take longer time before GA.
> > > Hi, I've created a Jira ticket for you, https://issues.apache.org/jira/browse/CELEBORN-1118.
> >
> >
> > Is this PR a subtask of [Support memory file storage](https://www.mail-archive.com/dev@celeborn.apache.org/msg00613.html)?
>
> Not really. We discussed with @oOKevinOo offline, he implemented this PR without aware of https://www.mail-archive.com/dev@celeborn.apache.org/msg00613.html . I think we can merge this first, since this PR has a good result for small shuffle files, @oOKevinOo could you share some of your test? The memory storage will take longer time before GA.
i tick into test discription
--
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] support file to memory cache [incubator-celeborn]
Posted by "cfmcgrady (via GitHub)" <gi...@apache.org>.
cfmcgrady commented on PR #2081:
URL: https://github.com/apache/incubator-celeborn/pull/2081#issuecomment-1801494092
> Hi, I've created a Jira ticket for you, https://issues.apache.org/jira/browse/CELEBORN-1118.
Is this PR a subtask of [Support memory file storage](https://www.mail-archive.com/dev@celeborn.apache.org/msg00613.html)?
--
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] support file to memory cache [incubator-celeborn]
Posted by "waitinfuture (via GitHub)" <gi...@apache.org>.
waitinfuture commented on PR #2081:
URL: https://github.com/apache/incubator-celeborn/pull/2081#issuecomment-1801429922
Hi @oOKevinOo , thanks for this PR! please run `UPDATE=1 build/mvn clean test -pl common -am -Dtest=none -DwildcardSuites=org.apache.celeborn.ConfigurationSuite` and `dev/reformat`, see https://github.com/apache/incubator-celeborn/blob/main/CONTRIBUTING.md
--
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-1118] Cache files in memory [incubator-celeborn]
Posted by "waitinfuture (via GitHub)" <gi...@apache.org>.
waitinfuture commented on PR #2081:
URL: https://github.com/apache/incubator-celeborn/pull/2081#issuecomment-1805213067
> > I think we can merge this first, since this PR has a good result for small shuffle files
>
> then the PR target branch should be `main` instead of `branch-0.3`?
Right, I also think we should merge to main. @oOKevinOo said there are some conflicts with main. I will have a check and try to resolve the conflicts.
--
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] support file to memory cache [incubator-celeborn]
Posted by "waitinfuture (via GitHub)" <gi...@apache.org>.
waitinfuture commented on PR #2081:
URL: https://github.com/apache/incubator-celeborn/pull/2081#issuecomment-1801682258
> > Hi, I've created a Jira ticket for you, https://issues.apache.org/jira/browse/CELEBORN-1118.
>
> Is this PR a subtask of [Support memory file storage](https://www.mail-archive.com/dev@celeborn.apache.org/msg00613.html)?
Not really. We discussed with @oOKevinOo offline, he implemented this PR without aware of https://www.mail-archive.com/dev@celeborn.apache.org/msg00613.html . I think we can merge this first, since this PR has very good benefit for small shuffle files, @oOKevinOo could you share some of your test? The memory storage will take longer time before GA.
--
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] support file to memory cache [incubator-celeborn]
Posted by "FMX (via GitHub)" <gi...@apache.org>.
FMX commented on PR #2081:
URL: https://github.com/apache/incubator-celeborn/pull/2081#issuecomment-1801410691
Hi, I've created a Jira ticket for you, https://issues.apache.org/jira/browse/CELEBORN-1118.
--
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