You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "sopel39 (via GitHub)" <gi...@apache.org> on 2024/12/13 15:00:50 UTC
[PR] Fix ParallelIterable deadlock [iceberg]
sopel39 opened a new pull request, #11781:
URL: https://github.com/apache/iceberg/pull/11781
It was observed that with high concurrency/high workload scenario cluster deadlocks due to manifest readers waiting for connection from S3 pool.
Specifically, ManifestGroup#plan will create ManifestReader per every ParallelIterable.Task. These readers will effectively hold onto S3 connection from the pool. When ParallelIterable queue is full, Task will be tabled for later use.
Consider scenario:
S3 connection pool size=1
approximateMaxQueueSize=1
workerPoolSize=1
ParallelIterable1: starts TaskP1
ParallelIterable1: TaskP1 produces result, queue gets full, TaskP1 is put on hold (holds S3 connection)
ParallelIterable2: starts TaskP2, TaskP2 is scheduled on workerPool but is blocked on S3 connection pool
ParallelIterable1: result gets consumed, TaskP1 is scheduled again
ParallelIterable1: TaskP1 waits for workerPool to be free, but TaskP2 is waiting for TaskP1 to release connection
The fix make sure Task is finished once it's started. This way limited resources like connection pool are not put on hold. Queue size might exceed strict limits, but it should still be bounded.
Fixes https://github.com/apache/iceberg/issues/11768
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix ParallelIterable deadlock [iceberg]
Posted by "sopel39 (via GitHub)" <gi...@apache.org>.
sopel39 commented on code in PR #11781:
URL: https://github.com/apache/iceberg/pull/11781#discussion_r1886884393
##########
core/src/main/java/org/apache/iceberg/util/ParallelIterable.java:
##########
@@ -257,17 +257,17 @@ private static class Task<T> implements Supplier<Optional<Task<T>>>, Closeable {
@Override
public Optional<Task<T>> get() {
try {
+ if (queue.size() >= approximateMaxQueueSize) {
Review Comment:
> so, each table scan leads to 2*cores tasks/s3 connections?
potentially, yes. but there are extra conditions like queue must be full.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix ParallelIterable deadlock [iceberg]
Posted by "findepi (via GitHub)" <gi...@apache.org>.
findepi commented on code in PR #11781:
URL: https://github.com/apache/iceberg/pull/11781#discussion_r1888434738
##########
core/src/test/java/org/apache/iceberg/util/TestParallelIterable.java:
##########
@@ -171,38 +171,55 @@ public void limitQueueSize() {
}
@Test
- public void queueSizeOne() {
Review Comment:
What's the reason to remove existing test?
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix ParallelIterable deadlock [iceberg]
Posted by "osscm (via GitHub)" <gi...@apache.org>.
osscm commented on code in PR #11781:
URL: https://github.com/apache/iceberg/pull/11781#discussion_r1885784288
##########
core/src/main/java/org/apache/iceberg/util/ParallelIterable.java:
##########
@@ -257,17 +257,17 @@ private static class Task<T> implements Supplier<Optional<Task<T>>>, Closeable {
@Override
public Optional<Task<T>> get() {
try {
+ if (queue.size() >= approximateMaxQueueSize) {
Review Comment:
so, each table scan leads to `2*cores` tasks/s3 connections?
and still can exhaust S3 connections?
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix ParallelIterable deadlock [iceberg]
Posted by "findepi (via GitHub)" <gi...@apache.org>.
findepi commented on PR #11781:
URL: https://github.com/apache/iceberg/pull/11781#issuecomment-2584202338
IIRC the OOM problem was a real production problem (cc @raunaqmorarka @dekimir @losipiuk), so I am not convinced it's OK to restore it.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix ParallelIterable deadlock [iceberg]
Posted by "sopel39 (via GitHub)" <gi...@apache.org>.
sopel39 commented on code in PR #11781:
URL: https://github.com/apache/iceberg/pull/11781#discussion_r1905856982
##########
core/src/main/java/org/apache/iceberg/util/ParallelIterable.java:
##########
@@ -257,17 +257,17 @@ private static class Task<T> implements Supplier<Optional<Task<T>>>, Closeable {
@Override
public Optional<Task<T>> get() {
try {
+ if (queue.size() >= approximateMaxQueueSize) {
+ // Yield when queue is over the size limit. Task will be resubmitted later and continue
+ // the work.
+ return Optional.of(this);
+ }
+
if (iterator == null) {
iterator = input.iterator();
}
while (iterator.hasNext()) {
- if (queue.size() >= approximateMaxQueueSize) {
Review Comment:
Disccused offline. Generally, this change can increase memory usage when manifest files produce a huge amount of entries. Hence this change sits between unbounded queue and strictly bounded queue as implemented by @findepi. In the future we could completely eliminate `ParallelIterable` as producers know exactly (or can estimate) how man entries are in each manifest file, hence it's possible to accurately schedule manifest file reads.
Alternatively, rest catalogs will provide scan planning capabilities (plus huge cache) hence engine wouldn't need to read the manifest files
##########
core/src/main/java/org/apache/iceberg/util/ParallelIterable.java:
##########
@@ -257,17 +263,21 @@ private static class Task<T> implements Supplier<Optional<Task<T>>>, Closeable {
@Override
public Optional<Task<T>> get() {
try {
+ if (queue.size() >= approximateMaxQueueSize) {
+ // Tasks might hold references (via iterator) to constrained resources
Review Comment:
Updated comment
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix ParallelIterable deadlock [iceberg]
Posted by "sopel39 (via GitHub)" <gi...@apache.org>.
sopel39 commented on PR #11781:
URL: https://github.com/apache/iceberg/pull/11781#issuecomment-2541648505
cc @findepi @RussellSpitzer @osscm
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix ParallelIterable deadlock [iceberg]
Posted by "findepi (via GitHub)" <gi...@apache.org>.
findepi commented on code in PR #11781:
URL: https://github.com/apache/iceberg/pull/11781#discussion_r1888440219
##########
core/src/test/java/org/apache/iceberg/util/TestParallelIterable.java:
##########
@@ -171,38 +171,55 @@ public void limitQueueSize() {
}
@Test
- public void queueSizeOne() {
- List<Iterable<Integer>> iterables =
- ImmutableList.of(
- () -> IntStream.range(0, 100).iterator(),
- () -> IntStream.range(0, 100).iterator(),
- () -> IntStream.range(0, 100).iterator());
+ @Timeout(10)
+ public void noDeadlock() {
+ ExecutorService executor = Executors.newFixedThreadPool(1);
+ Semaphore semaphore = new Semaphore(1);
Review Comment:
Add a code comment that this simulates a constrained resource, such as S3 connection pool that has limit on a max number of connections
##########
core/src/test/java/org/apache/iceberg/util/TestParallelIterable.java:
##########
@@ -171,38 +171,55 @@ public void limitQueueSize() {
}
@Test
- public void queueSizeOne() {
- List<Iterable<Integer>> iterables =
- ImmutableList.of(
- () -> IntStream.range(0, 100).iterator(),
- () -> IntStream.range(0, 100).iterator(),
- () -> IntStream.range(0, 100).iterator());
+ @Timeout(10)
+ public void noDeadlock() {
+ ExecutorService executor = Executors.newFixedThreadPool(1);
+ Semaphore semaphore = new Semaphore(1);
- Multiset<Integer> expectedValues =
- IntStream.range(0, 100)
- .boxed()
- .flatMap(i -> Stream.of(i, i, i))
- .collect(ImmutableMultiset.toImmutableMultiset());
+ List<Iterable<Integer>> iterablesA =
+ ImmutableList.of(
+ testIterable(
+ semaphore::acquire, semaphore::release, IntStream.range(0, 100).iterator()));
+ List<Iterable<Integer>> iterablesB =
+ ImmutableList.of(
+ testIterable(
+ semaphore::acquire, semaphore::release, IntStream.range(200, 300).iterator()));
- ExecutorService executor = Executors.newCachedThreadPool();
- ParallelIterable<Integer> parallelIterable = new ParallelIterable<>(iterables, executor, 1);
- ParallelIterator<Integer> iterator = (ParallelIterator<Integer>) parallelIterable.iterator();
+ ParallelIterable<Integer> parallelIterableA = new ParallelIterable<>(iterablesA, executor, 1);
+ ParallelIterable<Integer> parallelIterableB = new ParallelIterable<>(iterablesB, executor, 1);
Review Comment:
i have a code warning in those lines. Do you get one as well?
##########
core/src/test/java/org/apache/iceberg/util/TestParallelIterable.java:
##########
@@ -171,38 +171,55 @@ public void limitQueueSize() {
}
@Test
- public void queueSizeOne() {
- List<Iterable<Integer>> iterables =
- ImmutableList.of(
- () -> IntStream.range(0, 100).iterator(),
- () -> IntStream.range(0, 100).iterator(),
- () -> IntStream.range(0, 100).iterator());
+ @Timeout(10)
+ public void noDeadlock() {
Review Comment:
this new test indeed times out before the changes
##########
core/src/test/java/org/apache/iceberg/util/TestParallelIterable.java:
##########
@@ -148,17 +150,15 @@ public void limitQueueSize() {
.collect(ImmutableMultiset.toImmutableMultiset());
int maxQueueSize = 20;
- ExecutorService executor = Executors.newCachedThreadPool();
+ ExecutorService executor = Executors.newSingleThreadExecutor();
Review Comment:
is the change to existing test important?
or is it enough for the noDeadlock test to test there is no deadock?
##########
core/src/main/java/org/apache/iceberg/util/ParallelIterable.java:
##########
@@ -153,6 +155,7 @@ private synchronized boolean checkTasks() {
try {
Optional<Task<T>> continuation = taskFutures[i].get();
continuation.ifPresent(yieldedTasks::addLast);
+ taskFutures[i] = null;
Review Comment:
Why?
##########
core/src/main/java/org/apache/iceberg/util/ParallelIterable.java:
##########
@@ -257,17 +257,17 @@ private static class Task<T> implements Supplier<Optional<Task<T>>>, Closeable {
@Override
public Optional<Task<T>> get() {
try {
+ if (queue.size() >= approximateMaxQueueSize) {
+ // Yield when queue is over the size limit. Task will be resubmitted later and continue
+ // the work.
+ return Optional.of(this);
Review Comment:
it's very neat you solved the problem by just moving 2 lines of code around (okay, 5 including comments). Please add a code comment capturing the wisdom behind this seemingly cosmetic change.
##########
core/src/test/java/org/apache/iceberg/util/TestParallelIterable.java:
##########
@@ -171,38 +171,55 @@ public void limitQueueSize() {
}
@Test
- public void queueSizeOne() {
- List<Iterable<Integer>> iterables =
- ImmutableList.of(
- () -> IntStream.range(0, 100).iterator(),
- () -> IntStream.range(0, 100).iterator(),
- () -> IntStream.range(0, 100).iterator());
+ @Timeout(10)
+ public void noDeadlock() {
+ ExecutorService executor = Executors.newFixedThreadPool(1);
+ Semaphore semaphore = new Semaphore(1);
- Multiset<Integer> expectedValues =
- IntStream.range(0, 100)
- .boxed()
- .flatMap(i -> Stream.of(i, i, i))
- .collect(ImmutableMultiset.toImmutableMultiset());
+ List<Iterable<Integer>> iterablesA =
+ ImmutableList.of(
+ testIterable(
+ semaphore::acquire, semaphore::release, IntStream.range(0, 100).iterator()));
+ List<Iterable<Integer>> iterablesB =
+ ImmutableList.of(
+ testIterable(
+ semaphore::acquire, semaphore::release, IntStream.range(200, 300).iterator()));
- ExecutorService executor = Executors.newCachedThreadPool();
- ParallelIterable<Integer> parallelIterable = new ParallelIterable<>(iterables, executor, 1);
- ParallelIterator<Integer> iterator = (ParallelIterator<Integer>) parallelIterable.iterator();
+ ParallelIterable<Integer> parallelIterableA = new ParallelIterable<>(iterablesA, executor, 1);
+ ParallelIterable<Integer> parallelIterableB = new ParallelIterable<>(iterablesB, executor, 1);
- Multiset<Integer> actualValues = HashMultiset.create();
+ parallelIterableA.iterator().next();
+ parallelIterableB.iterator().next();
- while (iterator.hasNext()) {
- assertThat(iterator.queueSize())
- .as("iterator internal queue size")
- .isLessThanOrEqualTo(1 + iterables.size());
- actualValues.add(iterator.next());
- }
+ executor.shutdownNow();
Review Comment:
please use try finally to manage an executor in tests
(i am aware existing tests don't do this, but that's not a reason not to do better in new code)
##########
core/src/main/java/org/apache/iceberg/util/ParallelIterable.java:
##########
@@ -165,7 +168,10 @@ private synchronized boolean checkTasks() {
}
}
- taskFutures[i] = submitNextTask();
+ // submit a new task if there is space in the queue
+ if (queue.size() < maxQueueSize) {
+ taskFutures[i] = submitNextTask();
Review Comment:
Can this be 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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix ParallelIterable deadlock [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on PR #11781:
URL: https://github.com/apache/iceberg/pull/11781#issuecomment-2578314363
BTW, I like the new direction that @RussellSpitzer outlined. using byte size (instead of number of elements) is more intuitive and easier to calculate a good default to cap memory foot print. E.g., the max read bytes could be set to 1/10 of JVM heap size. 1/10 is just an example. We can estimate a proper default value based on typical file size/memory foot print ratio.
there could be an edge condition that a single manifest file is larger than the default value. In this case and if the buffer is empty or less than half full, task should be allowed.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix ParallelIterable deadlock [iceberg]
Posted by "stevenzwu (via GitHub)" <gi...@apache.org>.
stevenzwu commented on PR #11781:
URL: https://github.com/apache/iceberg/pull/11781#issuecomment-2584586732
@findepi the deadlock issue is probably the worse of the two. the deadlock issue probably needs to be addressed urgently.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix ParallelIterable deadlock [iceberg]
Posted by "sopel39 (via GitHub)" <gi...@apache.org>.
sopel39 commented on PR #11781:
URL: https://github.com/apache/iceberg/pull/11781#issuecomment-2551861574
rebased
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix ParallelIterable deadlock [iceberg]
Posted by "osscm (via GitHub)" <gi...@apache.org>.
osscm commented on code in PR #11781:
URL: https://github.com/apache/iceberg/pull/11781#discussion_r1887173077
##########
core/src/main/java/org/apache/iceberg/util/ParallelIterable.java:
##########
@@ -257,17 +257,17 @@ private static class Task<T> implements Supplier<Optional<Task<T>>>, Closeable {
@Override
public Optional<Task<T>> get() {
try {
+ if (queue.size() >= approximateMaxQueueSize) {
Review Comment:
Are we saying if queue is full then only we can reach to a condition that can lead to s3 connections are all exhausted?
Though adding task to queue does not mean a S3 connection?
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix ParallelIterable deadlock [iceberg]
Posted by "sopel39 (via GitHub)" <gi...@apache.org>.
sopel39 commented on code in PR #11781:
URL: https://github.com/apache/iceberg/pull/11781#discussion_r1887631464
##########
core/src/main/java/org/apache/iceberg/util/ParallelIterable.java:
##########
@@ -257,17 +257,17 @@ private static class Task<T> implements Supplier<Optional<Task<T>>>, Closeable {
@Override
public Optional<Task<T>> get() {
try {
+ if (queue.size() >= approximateMaxQueueSize) {
Review Comment:
In order for connection to be kept around task has to be started and then it needs to yield. If task doesn't yield then the S3 connection is fully utilized and returned afterwards
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix ParallelIterable deadlock [iceberg]
Posted by "osscm (via GitHub)" <gi...@apache.org>.
osscm commented on code in PR #11781:
URL: https://github.com/apache/iceberg/pull/11781#discussion_r1884308287
##########
core/src/main/java/org/apache/iceberg/util/ParallelIterable.java:
##########
@@ -257,17 +257,17 @@ private static class Task<T> implements Supplier<Optional<Task<T>>>, Closeable {
@Override
public Optional<Task<T>> get() {
try {
+ if (queue.size() >= approximateMaxQueueSize) {
Review Comment:
https://github.com/apache/iceberg/pull/11781/files#diff-0586c055bbd9d220727cfa60c802321693a959c0cfb3a510376fc5a5bb1dc897R99
`this.taskFutures = new CompletableFuture[2 * ThreadPools.WORKER_THREAD_POOL_SIZE];`
here the global thread pool size is driving the taskFuture array size, can this leads to more tasks then the available S3 connections?
as I believe each table scan will create a new `ParallelIterable` ?
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix ParallelIterable deadlock [iceberg]
Posted by "sopel39 (via GitHub)" <gi...@apache.org>.
sopel39 commented on code in PR #11781:
URL: https://github.com/apache/iceberg/pull/11781#discussion_r1890160296
##########
core/src/test/java/org/apache/iceberg/util/TestParallelIterable.java:
##########
@@ -171,38 +171,55 @@ public void limitQueueSize() {
}
@Test
- public void queueSizeOne() {
Review Comment:
The test is no longer viable as queue size is no longer a hard limit. The queue size will be at min the amount of items produced by single `org.apache.iceberg.util.ParallelIterable.Task`
##########
core/src/main/java/org/apache/iceberg/util/ParallelIterable.java:
##########
@@ -153,6 +155,7 @@ private synchronized boolean checkTasks() {
try {
Optional<Task<T>> continuation = taskFutures[i].get();
continuation.ifPresent(yieldedTasks::addLast);
+ taskFutures[i] = null;
Review Comment:
This is because of the
```
// submit a new task if there is space in the queue
if (queue.size() < maxQueueSize) {
```
below. The queue might be full, hence `taskFutures[i]` should remain empty as it was already added to `yieldedTasks`
##########
core/src/test/java/org/apache/iceberg/util/TestParallelIterable.java:
##########
@@ -148,17 +150,15 @@ public void limitQueueSize() {
.collect(ImmutableMultiset.toImmutableMultiset());
int maxQueueSize = 20;
- ExecutorService executor = Executors.newCachedThreadPool();
+ ExecutorService executor = Executors.newSingleThreadExecutor();
Review Comment:
`ParallelIterable` will still spawn `2 * ThreadPools.WORKER_THREAD_POOL_SIZE` tasks. These tasks won't be immediately picked up. When they are started, they are guaranteed to finish. This means that queue size will be temporary higher than `maxQueueSize`. In order to properly test the limits, we need to reduce executor size (as it will be in practice) to constraint number of concurrently running tasks.
##########
core/src/test/java/org/apache/iceberg/util/TestParallelIterable.java:
##########
@@ -171,38 +171,55 @@ public void limitQueueSize() {
}
@Test
- public void queueSizeOne() {
- List<Iterable<Integer>> iterables =
- ImmutableList.of(
- () -> IntStream.range(0, 100).iterator(),
- () -> IntStream.range(0, 100).iterator(),
- () -> IntStream.range(0, 100).iterator());
+ @Timeout(10)
+ public void noDeadlock() {
+ ExecutorService executor = Executors.newFixedThreadPool(1);
+ Semaphore semaphore = new Semaphore(1);
- Multiset<Integer> expectedValues =
- IntStream.range(0, 100)
- .boxed()
- .flatMap(i -> Stream.of(i, i, i))
- .collect(ImmutableMultiset.toImmutableMultiset());
+ List<Iterable<Integer>> iterablesA =
+ ImmutableList.of(
+ testIterable(
+ semaphore::acquire, semaphore::release, IntStream.range(0, 100).iterator()));
+ List<Iterable<Integer>> iterablesB =
+ ImmutableList.of(
+ testIterable(
+ semaphore::acquire, semaphore::release, IntStream.range(200, 300).iterator()));
- ExecutorService executor = Executors.newCachedThreadPool();
- ParallelIterable<Integer> parallelIterable = new ParallelIterable<>(iterables, executor, 1);
- ParallelIterator<Integer> iterator = (ParallelIterator<Integer>) parallelIterable.iterator();
+ ParallelIterable<Integer> parallelIterableA = new ParallelIterable<>(iterablesA, executor, 1);
+ ParallelIterable<Integer> parallelIterableB = new ParallelIterable<>(iterablesB, executor, 1);
Review Comment:
no. Language version issue?
##########
core/src/main/java/org/apache/iceberg/util/ParallelIterable.java:
##########
@@ -165,7 +168,10 @@ private synchronized boolean checkTasks() {
}
}
- taskFutures[i] = submitNextTask();
+ // submit a new task if there is space in the queue
+ if (queue.size() < maxQueueSize) {
+ taskFutures[i] = submitNextTask();
Review Comment:
no. It's just an optimization to avoid active loop (while queue is full) via executor which increases congestion.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core: Fix possible deadlock in ParallelIterable [iceberg]
Posted by "findepi (via GitHub)" <gi...@apache.org>.
findepi commented on PR #11781:
URL: https://github.com/apache/iceberg/pull/11781#issuecomment-2585320623
thank you @sopel39 for the PR and @osscm @RussellSpitzer @stevenzwu for reviews!
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix ParallelIterable deadlock [iceberg]
Posted by "sopel39 (via GitHub)" <gi...@apache.org>.
sopel39 commented on code in PR #11781:
URL: https://github.com/apache/iceberg/pull/11781#discussion_r1884535343
##########
core/src/main/java/org/apache/iceberg/util/ParallelIterable.java:
##########
@@ -257,17 +257,17 @@ private static class Task<T> implements Supplier<Optional<Task<T>>>, Closeable {
@Override
public Optional<Task<T>> get() {
try {
+ if (queue.size() >= approximateMaxQueueSize) {
+ // Yield when queue is over the size limit. Task will be resubmitted later and continue
+ // the work.
+ return Optional.of(this);
+ }
+
if (iterator == null) {
iterator = input.iterator();
}
while (iterator.hasNext()) {
- if (queue.size() >= approximateMaxQueueSize) {
Review Comment:
The queue should still be bounded albeit not strictly up to `approximateMaxQueueSize`.
> I assume we have some performance implications of not fully loading the queue?
What do you mean by not fully loading the queue? This PR guarantees that `Task` that is started will complete and not hold external resources.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix ParallelIterable deadlock [iceberg]
Posted by "findepi (via GitHub)" <gi...@apache.org>.
findepi commented on PR #11781:
URL: https://github.com/apache/iceberg/pull/11781#issuecomment-2585320135
It seems we're in agreement. The deadlock must be resolved. This PR supposedly avoids any deadlocks at the cost of increased memory usage, which is fair. I think we should not remove memory pressure protections.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix ParallelIterable deadlock [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #11781:
URL: https://github.com/apache/iceberg/pull/11781#discussion_r1901095259
##########
core/src/main/java/org/apache/iceberg/util/ParallelIterable.java:
##########
@@ -257,17 +263,21 @@ private static class Task<T> implements Supplier<Optional<Task<T>>>, Closeable {
@Override
public Optional<Task<T>> get() {
try {
+ if (queue.size() >= approximateMaxQueueSize) {
+ // Tasks might hold references (via iterator) to constrained resources
Review Comment:
Checking the code, this comment seems to be in conversation with the previous comment that was removed below, it also removes the context of the "queue / ApproximateMaxQueueSize" check.
I think it should definitely mention first that we "yield when the queue is over the size limit". The rest of this information isn't really relevant to this block of code without knowing that it previously used to allow for yielding in between elements of the iterator.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix ParallelIterable deadlock [iceberg]
Posted by "findepi (via GitHub)" <gi...@apache.org>.
findepi commented on PR #11781:
URL: https://github.com/apache/iceberg/pull/11781#issuecomment-2583924725
> As noted in my comment to Piotr, I think this is a fix to the deadlock but I think it may be better to just remove the yielding behavior all together
Would this restore OOM problem?
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix ParallelIterable deadlock [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on PR #11781:
URL: https://github.com/apache/iceberg/pull/11781#issuecomment-2584026138
> > As noted in my comment to Piotr, I think this is a fix to the deadlock but I think it may be better to just remove the yielding behavior all together
>
> Would this restore OOM problem?
Potentially this fix would also restore the OOM problem
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Core: Fix possible deadlock in ParallelIterable [iceberg]
Posted by "findepi (via GitHub)" <gi...@apache.org>.
findepi merged PR #11781:
URL: https://github.com/apache/iceberg/pull/11781
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix ParallelIterable deadlock [iceberg]
Posted by "findepi (via GitHub)" <gi...@apache.org>.
findepi commented on PR #11781:
URL: https://github.com/apache/iceberg/pull/11781#issuecomment-2585320268
Given that many approvals, it looks ready to go.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix ParallelIterable deadlock [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on PR #11781:
URL: https://github.com/apache/iceberg/pull/11781#issuecomment-2569914295
From a discussion I had with @sopel39 today;
I think we can go forward this solution but I think it will basically re-introduce the memory usage issue that we saw previously. (For some use cases)
From our discussion I believe we have been working at this from the wrong direction.
## Recap -
The original implementation basically assumed that we could read as made iterators as parallelism allowed. This led to basically unbounded memory usage. This ended up becoming an issue for systems using a lightweight coordinator now requiring them to have a huge memory footprint.
Next, to solve our issue with the memory footprint we added what is essentially a buffered read-ahead. A queue is filled with elements from the files we are reading in parallel and we check the queue depth every time we add elements to bound its size. The max queue size here is checked at every pull so we really can never go more than parallelism items over the max queue size. Unfortunately, this leads to the current deadlock issue since we can potentially yield an iterator in the middle of a file and be left only with iterators for files which cannot yet be opened because all file handles are owned by yielded iterators.
The current proposed solution is to switch checking the queue size per element and instead check only before opening a new file for the first time. This means that any file that is opened is read completely into the read-ahead queue. This fixes the dead lock issue as we will never yield in the middle of the file but possibly reintroduces the memory issue. We could potentially open up to "parallelism" files and load them all into the queue before having a chance to lead so we may be back where we started.
## Where do we go from here -
The current implementation is basically trying to solve the issue of :
*How do we read a bunch of unbounded streams in parallel without consuming too much memory?*
But this is actually a bit over general for what we are actually trying to do. Our actual problem is
*How do we read from a bunch of files without reading too many records at a time?
The key difference here is that we know exactly how long each file is *before* we open it. Instead of simply opening
every file and checking how much we've actually read in parallel, we can open only as many files as are required to fill our read ahead buffer and yield on the others.
I think what we should do is a bit like this (haven't thought this part through too much)
```
AtomicLong usedBuffer
if file not opened
Synchronized {
if (usedBuffer + file.length < MAX_BUFFER) {
usedBuffer += file.length
} else {
yield, do not yet open this file
}
Read File into Buffer
onNext {
usedBuffer --
}
```
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix ParallelIterable deadlock [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #11781:
URL: https://github.com/apache/iceberg/pull/11781#discussion_r1901098696
##########
core/src/main/java/org/apache/iceberg/util/ParallelIterable.java:
##########
@@ -257,17 +257,17 @@ private static class Task<T> implements Supplier<Optional<Task<T>>>, Closeable {
@Override
public Optional<Task<T>> get() {
try {
+ if (queue.size() >= approximateMaxQueueSize) {
+ // Yield when queue is over the size limit. Task will be resubmitted later and continue
+ // the work.
+ return Optional.of(this);
+ }
+
if (iterator == null) {
iterator = input.iterator();
}
while (iterator.hasNext()) {
- if (queue.size() >= approximateMaxQueueSize) {
Review Comment:
Previously we could yield in between elements of the iterator, now we can only yield when the iterator is exhausted. This seems like a pretty big difference?
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix ParallelIterable deadlock [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on PR #11781:
URL: https://github.com/apache/iceberg/pull/11781#issuecomment-2575944738
@findepi I'm on board with this but I want to make sure you are also happy with this. I'm unsure of whether having the ability to yield before files will really help memory pressure, so I'm slightly thinking we should just revert the yielding capabilities all together and just go back to the implementation from last year.
I'm willing to go forward though with either direction for now. In the future I think we really need a fake filesystem benchmarking test that we can use to simulate how the algorithms we write here will work since we are mostly working blind 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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix ParallelIterable deadlock [iceberg]
Posted by "sopel39 (via GitHub)" <gi...@apache.org>.
sopel39 commented on PR #11781:
URL: https://github.com/apache/iceberg/pull/11781#issuecomment-2579996415
> @sopel39 I assume you tried this change and it avoided deadlock problem. Has this been tested with the OOM scenario for large manifest files?
We tested in staging env. So far coordinator didn't show any excessive mem usage
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix ParallelIterable deadlock [iceberg]
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on code in PR #11781:
URL: https://github.com/apache/iceberg/pull/11781#discussion_r1884527019
##########
core/src/main/java/org/apache/iceberg/util/ParallelIterable.java:
##########
@@ -257,17 +257,17 @@ private static class Task<T> implements Supplier<Optional<Task<T>>>, Closeable {
@Override
public Optional<Task<T>> get() {
try {
+ if (queue.size() >= approximateMaxQueueSize) {
+ // Yield when queue is over the size limit. Task will be resubmitted later and continue
+ // the work.
+ return Optional.of(this);
+ }
+
if (iterator == null) {
iterator = input.iterator();
}
while (iterator.hasNext()) {
- if (queue.size() >= approximateMaxQueueSize) {
Review Comment:
Isn't this a pretty significant change in behavior or ParallelIterable? I assume we have some performance implications of not fully loading the queue?
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [PR] Fix ParallelIterable deadlock [iceberg]
Posted by "sopel39 (via GitHub)" <gi...@apache.org>.
sopel39 commented on code in PR #11781:
URL: https://github.com/apache/iceberg/pull/11781#discussion_r1884520727
##########
core/src/main/java/org/apache/iceberg/util/ParallelIterable.java:
##########
@@ -257,17 +257,17 @@ private static class Task<T> implements Supplier<Optional<Task<T>>>, Closeable {
@Override
public Optional<Task<T>> get() {
try {
+ if (queue.size() >= approximateMaxQueueSize) {
Review Comment:
> here the global thread pool size is driving the taskFuture array size, can this leads to more tasks then the available S3 connections?
Correct
> as I believe each table scan will create a new ParallelIterable ?
Yes.
--
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@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org