You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@celeborn.apache.org by "cfmcgrady (via GitHub)" <gi...@apache.org> on 2023/09/22 04:14:24 UTC
[GitHub] [incubator-celeborn] cfmcgrady commented on a diff in pull request #1932: [CELEBORN-980] Asynchronously delete original files to fix `ReusedExchange` bug
cfmcgrady commented on code in PR #1932:
URL: https://github.com/apache/incubator-celeborn/pull/1932#discussion_r1333856840
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/PartitionFilesSorter.java:
##########
@@ -696,4 +722,68 @@ private void readBufferBySize(ByteBuffer buffer, int toRead) throws IOException
}
}
}
+
+ public void cleanupExpiredShuffleKey(Set<String> expiredShuffleKeys) {
+ cleaner.cleanupExpiredShuffleKey(expiredShuffleKeys);
+ }
+}
+
+class PartitionFilesCleaner {
+ private static final Logger logger = LoggerFactory.getLogger(PartitionFilesCleaner.class);
+
+ private final LinkedBlockingQueue<PartitionFilesSorter.FileSorter> queue =
+ new LinkedBlockingQueue<>();
+ private final Lock lock = new ReentrantLock();
+ private final Condition notEmpty = lock.newCondition();
+
+ PartitionFilesCleaner() {
+ Thread cleaner =
+ new Thread(
+ () -> {
+ while (true) {
+ try {
+ lock.lockInterruptibly();
+ while (queue.isEmpty()) {
+ notEmpty.await(500, TimeUnit.MILLISECONDS);
+ }
+ Iterator<PartitionFilesSorter.FileSorter> it = queue.iterator();
+ while (it.hasNext()) {
+ PartitionFilesSorter.FileSorter sorter = it.next();
+ try {
+ if (sorter.getOriginFileInfo().isStreamsEmpty()) {
Review Comment:
> If there is no non skew streams, the origin file can be deleted.
it's not, only when the range open stream request(skew stream) will trigger `FileSorter.sort()`, and then the `FileSorter` instance will be added to the cleanup queue.
> I think it's good to record all stream in fileinfo, I'll need it in supporting memory storage.
yeah, I also read the proposal you posted on the mailing list, in the current implementation, record all activity streams in the fileinfo already done.
##########
worker/src/main/java/org/apache/celeborn/service/deploy/worker/storage/PartitionFilesSorter.java:
##########
@@ -696,4 +722,68 @@ private void readBufferBySize(ByteBuffer buffer, int toRead) throws IOException
}
}
}
+
+ public void cleanupExpiredShuffleKey(Set<String> expiredShuffleKeys) {
+ cleaner.cleanupExpiredShuffleKey(expiredShuffleKeys);
+ }
+}
+
+class PartitionFilesCleaner {
+ private static final Logger logger = LoggerFactory.getLogger(PartitionFilesCleaner.class);
+
+ private final LinkedBlockingQueue<PartitionFilesSorter.FileSorter> queue =
+ new LinkedBlockingQueue<>();
+ private final Lock lock = new ReentrantLock();
+ private final Condition notEmpty = lock.newCondition();
+
+ PartitionFilesCleaner() {
+ Thread cleaner =
+ new Thread(
+ () -> {
+ while (true) {
+ try {
+ lock.lockInterruptibly();
+ while (queue.isEmpty()) {
+ notEmpty.await(500, TimeUnit.MILLISECONDS);
+ }
+ Iterator<PartitionFilesSorter.FileSorter> it = queue.iterator();
+ while (it.hasNext()) {
+ PartitionFilesSorter.FileSorter sorter = it.next();
+ try {
+ if (sorter.getOriginFileInfo().isStreamsEmpty()) {
Review Comment:
> If there is no non skew streams, the origin file can be deleted.
it's not, only when the range open stream request(skew stream) will trigger `FileSorter.sort()`, and then the `FileSorter` instance will be added to the cleanup queue.
> I think it's good to record all stream in fileinfo, I'll need it in supporting memory storage.
yeah, I also read the proposal you posted on the mailing list, in the current implementation, record all activity streams in the fileinfo already done.
--
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