You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@uniffle.apache.org by "zuston (via GitHub)" <gi...@apache.org> on 2025/02/08 02:55:49 UTC
[PR] [#291] feat(client): Introduce PrefetchableClientReadHandler to support async read [incubator-uniffle]
zuston opened a new pull request, #2365:
URL: https://github.com/apache/incubator-uniffle/pull/2365
### What changes were proposed in this pull request?
1. Introduce PrefetchableClientReadHandler to support async read. And this will be disabled by default.
2. Apply for the memory/localfile/hdfs read handler
### Why are the changes needed?
Recently I found some important spark jobs are slow due to the lots of shuffle read operations. If we could support async read, the job's performance will be improved.
So this PR is the callback for #291. almost 3 years ago!
### Does this PR introduce _any_ user-facing change?
Yes. Some configs are introduced
1. `rss.client.read.prefetch.enabled`
2. `rss.client.read.prefetch.capacity`
3. `rss.client.read.prefetch.timeoutSec`
### How was this patch tested?
1. Unit tests
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#291] feat(client): Introduce PrefetchableClientReadHandler to support async read [incubator-uniffle]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on PR #2365:
URL: https://github.com/apache/incubator-uniffle/pull/2365#issuecomment-2644728083
> > > What's the relationship between it and prefetchTimeout?
> >
> >
> > If the async fetch costs too much time, I hope the sync wait could recognize this case and then fast fail
>
> PrefetchTimeout may be equal to read time out. Isn't right? They are all the one time read max time.
Yes. But I hope it could be configured (< grpc read timeout)
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#291] feat(client): Introduce PrefetchableClientReadHandler to support async read [incubator-uniffle]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on PR #2365:
URL: https://github.com/apache/incubator-uniffle/pull/2365#issuecomment-2644462164
> 1. What will it happen if the operation of read fails? Will it break the order of the data? Because it maybe ordered if mr sort data in the server side.
No, the real concurrency number is 1.
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#291] feat(client): Introduce PrefetchableClientReadHandler to support async read [incubator-uniffle]
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on PR #2365:
URL: https://github.com/apache/incubator-uniffle/pull/2365#issuecomment-2644454988
1. What will it happen if the operation of read fails? Will it break the order of the data? Because it maybe ordered if mr sort data in the server side.
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#291] feat(client): Introduce PrefetchableClientReadHandler to support async read [incubator-uniffle]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #2365:
URL: https://github.com/apache/incubator-uniffle/pull/2365#issuecomment-2644464531
## Test Results
2 328 files - 653 2 328 suites - 653 5h 3m 51s ⏱️ - 1h 24m 33s
906 tests - 195 901 ✅ - 198 2 💤 ±0 1 ❌ +1 2 🔥 + 2
10 137 runs - 3 652 10 091 ✅ - 3 668 30 💤 ±0 1 ❌ +1 15 🔥 +15
For more details on these failures and errors, see [this check](https://github.com/apache/incubator-uniffle/runs/36886314938).
Results for commit e6bad9e5. ± Comparison against base commit e5cfc4a0.
<details>
<summary>This pull request <b>removes</b> 199 and <b>adds</b> 4 tests. <i>Note that renamed tests count towards both.</i></summary>
```
org.apache.spark.shuffle.DelegationRssShuffleManagerTest ‑ testCreateFallback
org.apache.spark.shuffle.DelegationRssShuffleManagerTest ‑ testCreateInDriver
org.apache.spark.shuffle.DelegationRssShuffleManagerTest ‑ testCreateInDriverDenied
org.apache.spark.shuffle.DelegationRssShuffleManagerTest ‑ testCreateInExecutor
org.apache.spark.shuffle.DelegationRssShuffleManagerTest ‑ testDefaultIncludeExcludeProperties
org.apache.spark.shuffle.DelegationRssShuffleManagerTest ‑ testExcludeProperties
org.apache.spark.shuffle.DelegationRssShuffleManagerTest ‑ testIncludeProperties
org.apache.spark.shuffle.DelegationRssShuffleManagerTest ‑ testTryAccessCluster
org.apache.spark.shuffle.FunctionUtilsTests ‑ testOnceFunction0
org.apache.spark.shuffle.RssShuffleManagerTest ‑ testCreateShuffleManagerServer
…
```
```
org.apache.uniffle.storage.handler.impl.PrefetchableClientReadHandlerTest ‑ test_with_fetch_failure
org.apache.uniffle.storage.handler.impl.PrefetchableClientReadHandlerTest ‑ test_with_prefetch
org.apache.uniffle.storage.handler.impl.PrefetchableClientReadHandlerTest ‑ test_with_timeout
org.apache.uniffle.storage.handler.impl.PrefetchableClientReadHandlerTest ‑ test_without_prefetch
```
</details>
[test-results]:data:application/gzip;base64,H4sIAA3OpmcC/02Myw6DIBBFf8Ww7mKGqQ/6Mw3KkJD6aBBWpv9esGJdnnNvziasG3kVj0qS7G6VWKMLVzbR6+CWORnsJGFSaQ/5oaAp9FzjMOwK/+rl3rlzCqvdmDOnYO8Xf1x8nHMTAak9sEQRQOHpflWCIi7VnUsU62SGZZpcSCS46bVRXPdEDD0rLQFbAk58V1C3BnVjrTXi8wVPysZ6EwEAAA==
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#291] feat(client): Introduce PrefetchableClientReadHandler to support async read [incubator-uniffle]
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #2365:
URL: https://github.com/apache/incubator-uniffle/pull/2365#discussion_r1947415766
##########
storage/src/main/java/org/apache/uniffle/storage/handler/impl/PrefetchableClientReadHandler.java:
##########
@@ -0,0 +1,136 @@
+/*
+ * 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.uniffle.storage.handler.impl;
+
+import java.util.Optional;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.uniffle.common.ShuffleDataResult;
+import org.apache.uniffle.common.exception.RssException;
+
+public abstract class PrefetchableClientReadHandler extends AbstractClientReadHandler {
+ private static final Logger LOG = LoggerFactory.getLogger(PrefetchableClientReadHandler.class);
+
+ private boolean prefetchEnabled;
+ private int prefetchQueueCapacity;
+ private int prefetchTimeoutSec;
+ private LinkedBlockingQueue<Optional<ShuffleDataResult>> prefetchResultQueue;
+ private ExecutorService prefetchExecutors;
+ private AtomicBoolean abnormalFetchTag;
+ private AtomicBoolean finishedTag;
+
+ public PrefetchableClientReadHandler(Optional<PrefetchOption> prefetchOptional) {
+ if (prefetchOptional.isPresent()) {
+ PrefetchOption option = prefetchOptional.get();
+ if (option.capacity <= 0) {
+ throw new RssException("Illegal prefetch capacity: " + option.capacity);
+ }
+ LOG.info("Prefetch is enabled, capacity: {}", option.capacity);
+ this.prefetchEnabled = true;
+ this.prefetchQueueCapacity = option.capacity;
+ this.prefetchTimeoutSec = option.timeoutSec;
+ this.prefetchResultQueue = new LinkedBlockingQueue<>(option.capacity);
+ // todo: support multi threads to prefetch
+ this.prefetchExecutors = Executors.newFixedThreadPool(1);
+ this.abnormalFetchTag = new AtomicBoolean(false);
+ this.finishedTag = new AtomicBoolean(false);
+ } else {
+ this.prefetchEnabled = false;
+ }
+ }
+
+ public static class PrefetchOption {
+ private int capacity;
+ private int timeoutSec;
+
+ public PrefetchOption(int capacity, int timeoutSec) {
+ this.capacity = capacity;
+ this.timeoutSec = timeoutSec;
+ }
+ }
+
+ protected abstract ShuffleDataResult doReadShuffleData();
+
+ @Override
+ public ShuffleDataResult readShuffleData() {
+ if (!prefetchEnabled) {
+ return doReadShuffleData();
+ }
+
+ int free = prefetchQueueCapacity - prefetchResultQueue.size();
+ for (int i = 0; i < free; i++) {
+ prefetchExecutors.submit(
+ () -> {
+ // if it has been marked as abnormal/finished state, skip the following fetching.
+ if (abnormalFetchTag.get() || finishedTag.get()) {
+ return;
+ }
+ try {
+ ShuffleDataResult result = doReadShuffleData();
+ if (result == null) {
+ this.finishedTag.set(true);
+ }
+ prefetchResultQueue.offer(Optional.ofNullable(result));
+ } catch (Exception e) {
+ abnormalFetchTag.set(true);
+ LOG.error("Errors on doing readShuffleData", e);
+ }
+ });
+ }
+
+ long start = System.currentTimeMillis();
+ while (true) {
+ if (abnormalFetchTag.get()) {
+ throw new RssException("Fast fail due to the fetch failure");
+ }
+
+ Optional<ShuffleDataResult> optionalShuffleDataResult = prefetchResultQueue.poll();
+ if (optionalShuffleDataResult != null) {
+ if (optionalShuffleDataResult.isPresent()) {
+ return optionalShuffleDataResult.get();
+ } else {
+ return null;
+ }
+ }
+
+ if (System.currentTimeMillis() - start > prefetchTimeoutSec * 1000) {
+ throw new RssException("Unexpected duration of reading shuffle data. Fast fail!");
+ }
+
+ try {
+ Thread.sleep(10);
Review Comment:
Method `wait` may be better than sleep.
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#291] feat(client): Introduce PrefetchableClientReadHandler to support async read [incubator-uniffle]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #2365:
URL: https://github.com/apache/incubator-uniffle/pull/2365#discussion_r1947487075
##########
storage/src/main/java/org/apache/uniffle/storage/handler/impl/PrefetchableClientReadHandler.java:
##########
@@ -0,0 +1,136 @@
+/*
+ * 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.uniffle.storage.handler.impl;
+
+import java.util.Optional;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.uniffle.common.ShuffleDataResult;
+import org.apache.uniffle.common.exception.RssException;
+
+public abstract class PrefetchableClientReadHandler extends AbstractClientReadHandler {
+ private static final Logger LOG = LoggerFactory.getLogger(PrefetchableClientReadHandler.class);
+
+ private boolean prefetchEnabled;
+ private int prefetchQueueCapacity;
+ private int prefetchTimeoutSec;
+ private LinkedBlockingQueue<Optional<ShuffleDataResult>> prefetchResultQueue;
+ private ExecutorService prefetchExecutors;
+ private AtomicBoolean abnormalFetchTag;
+ private AtomicBoolean finishedTag;
+
+ public PrefetchableClientReadHandler(Optional<PrefetchOption> prefetchOptional) {
+ if (prefetchOptional.isPresent()) {
+ PrefetchOption option = prefetchOptional.get();
+ if (option.capacity <= 0) {
+ throw new RssException("Illegal prefetch capacity: " + option.capacity);
+ }
+ LOG.info("Prefetch is enabled, capacity: {}", option.capacity);
+ this.prefetchEnabled = true;
+ this.prefetchQueueCapacity = option.capacity;
+ this.prefetchTimeoutSec = option.timeoutSec;
+ this.prefetchResultQueue = new LinkedBlockingQueue<>(option.capacity);
+ // todo: support multi threads to prefetch
+ this.prefetchExecutors = Executors.newFixedThreadPool(1);
+ this.abnormalFetchTag = new AtomicBoolean(false);
+ this.finishedTag = new AtomicBoolean(false);
+ } else {
+ this.prefetchEnabled = false;
+ }
+ }
+
+ public static class PrefetchOption {
+ private int capacity;
+ private int timeoutSec;
+
+ public PrefetchOption(int capacity, int timeoutSec) {
+ this.capacity = capacity;
+ this.timeoutSec = timeoutSec;
+ }
+ }
+
+ protected abstract ShuffleDataResult doReadShuffleData();
+
+ @Override
+ public ShuffleDataResult readShuffleData() {
+ if (!prefetchEnabled) {
+ return doReadShuffleData();
+ }
+
+ int free = prefetchQueueCapacity - prefetchResultQueue.size();
+ for (int i = 0; i < free; i++) {
+ prefetchExecutors.submit(
+ () -> {
+ // if it has been marked as abnormal/finished state, skip the following fetching.
+ if (abnormalFetchTag.get() || finishedTag.get()) {
+ return;
+ }
+ try {
+ ShuffleDataResult result = doReadShuffleData();
+ if (result == null) {
+ this.finishedTag.set(true);
+ }
+ prefetchResultQueue.offer(Optional.ofNullable(result));
+ } catch (Exception e) {
+ abnormalFetchTag.set(true);
+ LOG.error("Errors on doing readShuffleData", e);
+ }
+ });
+ }
+
+ long start = System.currentTimeMillis();
+ while (true) {
+ if (abnormalFetchTag.get()) {
+ throw new RssException("Fast fail due to the fetch failure");
+ }
+
+ Optional<ShuffleDataResult> optionalShuffleDataResult = prefetchResultQueue.poll();
+ if (optionalShuffleDataResult != null) {
+ if (optionalShuffleDataResult.isPresent()) {
+ return optionalShuffleDataResult.get();
+ } else {
+ return null;
+ }
+ }
+
+ if (System.currentTimeMillis() - start > prefetchTimeoutSec * 1000) {
+ throw new RssException("Unexpected duration of reading shuffle data. Fast fail!");
+ }
+
+ try {
+ Thread.sleep(10);
Review Comment:
Removed
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#291] feat(client): Introduce PrefetchableClientReadHandler to support async read [incubator-uniffle]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston merged PR #2365:
URL: https://github.com/apache/incubator-uniffle/pull/2365
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#291] feat(client): Introduce PrefetchableClientReadHandler to support async read [incubator-uniffle]
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on PR #2365:
URL: https://github.com/apache/incubator-uniffle/pull/2365#issuecomment-2644732605
> > > > What's the relationship between it and prefetchTimeout?
> > >
> > >
> > > If the async fetch costs too much time, I hope the sync wait could recognize this case and then fast fail
> >
> >
> > PrefetchTimeout may be equal to read time out. Isn't right? They are all the one time read max time.
>
> Yes. But I hope it could be configured (< grpc read timeout)
OK.
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#291] feat(client): Introduce PrefetchableClientReadHandler to support async read [incubator-uniffle]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on PR #2365:
URL: https://github.com/apache/incubator-uniffle/pull/2365#issuecomment-2644553702
> What's the relationship between it and prefetchTimeout?
If the async fetch costs too much time, I hope the sync wait could recognize this case and then fast fail
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#291] feat(client): Introduce PrefetchableClientReadHandler to support async read [incubator-uniffle]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on code in PR #2365:
URL: https://github.com/apache/incubator-uniffle/pull/2365#discussion_r1947422031
##########
storage/src/main/java/org/apache/uniffle/storage/handler/impl/PrefetchableClientReadHandler.java:
##########
@@ -0,0 +1,136 @@
+/*
+ * 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.uniffle.storage.handler.impl;
+
+import java.util.Optional;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.uniffle.common.ShuffleDataResult;
+import org.apache.uniffle.common.exception.RssException;
+
+public abstract class PrefetchableClientReadHandler extends AbstractClientReadHandler {
+ private static final Logger LOG = LoggerFactory.getLogger(PrefetchableClientReadHandler.class);
+
+ private boolean prefetchEnabled;
+ private int prefetchQueueCapacity;
+ private int prefetchTimeoutSec;
+ private LinkedBlockingQueue<Optional<ShuffleDataResult>> prefetchResultQueue;
+ private ExecutorService prefetchExecutors;
+ private AtomicBoolean abnormalFetchTag;
+ private AtomicBoolean finishedTag;
+
+ public PrefetchableClientReadHandler(Optional<PrefetchOption> prefetchOptional) {
+ if (prefetchOptional.isPresent()) {
+ PrefetchOption option = prefetchOptional.get();
+ if (option.capacity <= 0) {
+ throw new RssException("Illegal prefetch capacity: " + option.capacity);
+ }
+ LOG.info("Prefetch is enabled, capacity: {}", option.capacity);
+ this.prefetchEnabled = true;
+ this.prefetchQueueCapacity = option.capacity;
+ this.prefetchTimeoutSec = option.timeoutSec;
+ this.prefetchResultQueue = new LinkedBlockingQueue<>(option.capacity);
+ // todo: support multi threads to prefetch
+ this.prefetchExecutors = Executors.newFixedThreadPool(1);
+ this.abnormalFetchTag = new AtomicBoolean(false);
+ this.finishedTag = new AtomicBoolean(false);
+ } else {
+ this.prefetchEnabled = false;
+ }
+ }
+
+ public static class PrefetchOption {
+ private int capacity;
+ private int timeoutSec;
+
+ public PrefetchOption(int capacity, int timeoutSec) {
+ this.capacity = capacity;
+ this.timeoutSec = timeoutSec;
+ }
+ }
+
+ protected abstract ShuffleDataResult doReadShuffleData();
+
+ @Override
+ public ShuffleDataResult readShuffleData() {
+ if (!prefetchEnabled) {
+ return doReadShuffleData();
+ }
+
+ int free = prefetchQueueCapacity - prefetchResultQueue.size();
+ for (int i = 0; i < free; i++) {
+ prefetchExecutors.submit(
+ () -> {
+ // if it has been marked as abnormal/finished state, skip the following fetching.
+ if (abnormalFetchTag.get() || finishedTag.get()) {
+ return;
+ }
+ try {
+ ShuffleDataResult result = doReadShuffleData();
+ if (result == null) {
+ this.finishedTag.set(true);
+ }
+ prefetchResultQueue.offer(Optional.ofNullable(result));
+ } catch (Exception e) {
+ abnormalFetchTag.set(true);
+ LOG.error("Errors on doing readShuffleData", e);
+ }
+ });
+ }
+
+ long start = System.currentTimeMillis();
+ while (true) {
+ if (abnormalFetchTag.get()) {
+ throw new RssException("Fast fail due to the fetch failure");
+ }
+
+ Optional<ShuffleDataResult> optionalShuffleDataResult = prefetchResultQueue.poll();
+ if (optionalShuffleDataResult != null) {
+ if (optionalShuffleDataResult.isPresent()) {
+ return optionalShuffleDataResult.get();
+ } else {
+ return null;
+ }
+ }
+
+ if (System.currentTimeMillis() - start > prefetchTimeoutSec * 1000) {
+ throw new RssException("Unexpected duration of reading shuffle data. Fast fail!");
+ }
+
+ try {
+ Thread.sleep(10);
Review Comment:
Wait is OK but introduce extra sync code. Let me take a deep think
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#291] feat(client): Introduce PrefetchableClientReadHandler to support async read [incubator-uniffle]
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on code in PR #2365:
URL: https://github.com/apache/incubator-uniffle/pull/2365#discussion_r1947437854
##########
storage/src/main/java/org/apache/uniffle/storage/handler/impl/PrefetchableClientReadHandler.java:
##########
@@ -0,0 +1,136 @@
+/*
+ * 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.uniffle.storage.handler.impl;
+
+import java.util.Optional;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.uniffle.common.ShuffleDataResult;
+import org.apache.uniffle.common.exception.RssException;
+
+public abstract class PrefetchableClientReadHandler extends AbstractClientReadHandler {
+ private static final Logger LOG = LoggerFactory.getLogger(PrefetchableClientReadHandler.class);
+
+ private boolean prefetchEnabled;
+ private int prefetchQueueCapacity;
+ private int prefetchTimeoutSec;
+ private LinkedBlockingQueue<Optional<ShuffleDataResult>> prefetchResultQueue;
+ private ExecutorService prefetchExecutors;
+ private AtomicBoolean abnormalFetchTag;
+ private AtomicBoolean finishedTag;
+
+ public PrefetchableClientReadHandler(Optional<PrefetchOption> prefetchOptional) {
+ if (prefetchOptional.isPresent()) {
+ PrefetchOption option = prefetchOptional.get();
+ if (option.capacity <= 0) {
+ throw new RssException("Illegal prefetch capacity: " + option.capacity);
+ }
+ LOG.info("Prefetch is enabled, capacity: {}", option.capacity);
+ this.prefetchEnabled = true;
+ this.prefetchQueueCapacity = option.capacity;
+ this.prefetchTimeoutSec = option.timeoutSec;
+ this.prefetchResultQueue = new LinkedBlockingQueue<>(option.capacity);
+ // todo: support multi threads to prefetch
+ this.prefetchExecutors = Executors.newFixedThreadPool(1);
+ this.abnormalFetchTag = new AtomicBoolean(false);
+ this.finishedTag = new AtomicBoolean(false);
+ } else {
+ this.prefetchEnabled = false;
+ }
+ }
+
+ public static class PrefetchOption {
+ private int capacity;
+ private int timeoutSec;
+
+ public PrefetchOption(int capacity, int timeoutSec) {
+ this.capacity = capacity;
+ this.timeoutSec = timeoutSec;
+ }
+ }
+
+ protected abstract ShuffleDataResult doReadShuffleData();
+
+ @Override
+ public ShuffleDataResult readShuffleData() {
+ if (!prefetchEnabled) {
+ return doReadShuffleData();
+ }
+
+ int free = prefetchQueueCapacity - prefetchResultQueue.size();
+ for (int i = 0; i < free; i++) {
+ prefetchExecutors.submit(
+ () -> {
+ // if it has been marked as abnormal/finished state, skip the following fetching.
+ if (abnormalFetchTag.get() || finishedTag.get()) {
+ return;
+ }
+ try {
+ ShuffleDataResult result = doReadShuffleData();
+ if (result == null) {
+ this.finishedTag.set(true);
+ }
+ prefetchResultQueue.offer(Optional.ofNullable(result));
+ } catch (Exception e) {
+ abnormalFetchTag.set(true);
+ LOG.error("Errors on doing readShuffleData", e);
+ }
+ });
+ }
+
+ long start = System.currentTimeMillis();
+ while (true) {
+ if (abnormalFetchTag.get()) {
+ throw new RssException("Fast fail due to the fetch failure");
+ }
+
+ Optional<ShuffleDataResult> optionalShuffleDataResult = prefetchResultQueue.poll();
+ if (optionalShuffleDataResult != null) {
+ if (optionalShuffleDataResult.isPresent()) {
+ return optionalShuffleDataResult.get();
+ } else {
+ return null;
+ }
+ }
+
+ if (System.currentTimeMillis() - start > prefetchTimeoutSec * 1000) {
+ throw new RssException("Unexpected duration of reading shuffle data. Fast fail!");
+ }
+
+ try {
+ Thread.sleep(10);
Review Comment:
Sleep will occupy the cpu and wait won't occupy the cpu.
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#291] feat(client): Introduce PrefetchableClientReadHandler to support async read [incubator-uniffle]
Posted by "zuston (via GitHub)" <gi...@apache.org>.
zuston commented on PR #2365:
URL: https://github.com/apache/incubator-uniffle/pull/2365#issuecomment-2644450855
cc @jerqi
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#291] feat(client): Introduce PrefetchableClientReadHandler to support async read [incubator-uniffle]
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on PR #2365:
URL: https://github.com/apache/incubator-uniffle/pull/2365#issuecomment-2644549124
Could you add document for these config options?
Do we have read timeout option? What's the relationship between it and prefetchTimeout?
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org
Re: [PR] [#291] feat(client): Introduce PrefetchableClientReadHandler to support async read [incubator-uniffle]
Posted by "jerqi (via GitHub)" <gi...@apache.org>.
jerqi commented on PR #2365:
URL: https://github.com/apache/incubator-uniffle/pull/2365#issuecomment-2644676283
> > What's the relationship between it and prefetchTimeout?
>
> If the async fetch costs too much time, I hope the sync wait could recognize this case and then fast fail
PrefetchTimeout may be equal to read time out. Isn't right? They are all the one time read max time.
--
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@uniffle.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@uniffle.apache.org
For additional commands, e-mail: issues-help@uniffle.apache.org