JackieTien97 commented on code in PR #18244:
URL: https://github.com/apache/iotdb/pull/18244#discussion_r3611859593
##########
iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/execution/exchange/SharedTsBlockQueueTest.java:
##########
@@ -139,119 +139,47 @@ public void concurrencyTest() {
queue.setMaxBytesCanReserve(Long.MAX_VALUE);
ExecutorService executor = Executors.newFixedThreadPool(2);
- AtomicReference<Integer> numOfTimesSenderBlocked = new
AtomicReference<>(0);
- AtomicReference<Integer> numOfTimesReceiverBlocked = new
AtomicReference<>(0);
- AtomicReference<Integer> numOfTsBlocksToSend = new AtomicReference<>(1000);
- AtomicReference<Integer> numOfTsBlocksToReceive = new
AtomicReference<>(1000);
- executor.submit(
- new SendTask(
- queue, mockTsBlockSize, numOfTsBlocksToSend,
numOfTimesSenderBlocked, executor));
- executor.submit(
- new ReceiveTask(queue, numOfTsBlocksToReceive,
numOfTimesReceiverBlocked, executor));
-
- while (numOfTsBlocksToSend.get() != 0 && numOfTsBlocksToReceive.get() !=
0) {
- String message =
- String.format(
- "Sender %d: %d, Receiver %d: %d",
- numOfTimesSenderBlocked.get(),
- numOfTsBlocksToSend.get(),
- numOfTimesReceiverBlocked.get(),
- numOfTsBlocksToReceive.get());
- System.out.println(message);
- try {
- Thread.sleep(10L);
- } catch (InterruptedException e) {
- Assert.fail(e.getMessage());
- }
- }
- }
-
- private static class SendTask implements Runnable {
-
- private final SharedTsBlockQueue queue;
- private final long mockTsBlockSize;
- private final AtomicReference<Integer> numOfTsBlocksToSend;
- private final AtomicReference<Integer> numOfTimesBlocked;
- private final ExecutorService executor;
-
- public SendTask(
- SharedTsBlockQueue queue,
- long mockTsBlockSize,
- AtomicReference<Integer> numOfTsBlocksToSend,
- AtomicReference<Integer> numOfTimesBlocked,
- ExecutorService executor) {
- this.queue = Validate.notNull(queue);
- Validate.isTrue(mockTsBlockSize > 0L);
- this.mockTsBlockSize = mockTsBlockSize;
- this.numOfTsBlocksToSend = Validate.notNull(numOfTsBlocksToSend);
- this.numOfTimesBlocked = Validate.notNull(numOfTimesBlocked);
- this.executor = Validate.notNull(executor);
- }
-
- @Override
- public void run() {
- ListenableFuture<Void> blockedOnMemory = null;
- while (numOfTsBlocksToSend.get() > 0) {
- synchronized (queue) {
- blockedOnMemory =
queue.add(Utils.createMockTsBlock(mockTsBlockSize));
- }
- numOfTsBlocksToSend.updateAndGet(v -> v - 1);
- if (!blockedOnMemory.isDone()) {
- break;
- }
- }
-
- if (blockedOnMemory != null) {
- numOfTimesBlocked.updateAndGet(v -> v + 1);
- blockedOnMemory.addListener(
- new SendTask(queue, mockTsBlockSize, numOfTsBlocksToSend,
numOfTimesBlocked, executor),
- executor);
- } else {
- synchronized (queue) {
- queue.setNoMoreTsBlocks(true);
- }
- }
- }
- }
-
- private static class ReceiveTask implements Runnable {
-
- private final SharedTsBlockQueue queue;
- private final AtomicReference<Integer> numOfTsBlocksToReceive;
- private final AtomicReference<Integer> numOfTimesBlocked;
- private final ExecutorService executor;
-
- public ReceiveTask(
- SharedTsBlockQueue queue,
- AtomicReference<Integer> numOfTsBlocksToReceive,
- AtomicReference<Integer> numOfTimesBlocked,
- ExecutorService executor) {
- this.queue = Validate.notNull(queue);
- this.numOfTsBlocksToReceive = Validate.notNull(numOfTsBlocksToReceive);
- this.numOfTimesBlocked = Validate.notNull(numOfTimesBlocked);
- this.executor = Validate.notNull(executor);
- }
-
- @Override
- public void run() {
- ListenableFuture<Void> blocked = null;
- while (numOfTsBlocksToReceive.get() > 0) {
- synchronized (queue) {
- blocked = queue.isBlocked();
- if (blocked.isDone()) {
- queue.remove();
- numOfTsBlocksToReceive.updateAndGet(v -> v - 1);
- } else {
- break;
- }
- }
- }
-
- if (blocked != null) {
- numOfTimesBlocked.updateAndGet(v -> v + 1);
- blocked.addListener(
- new ReceiveTask(queue, numOfTsBlocksToReceive, numOfTimesBlocked,
executor), executor);
- }
+ try {
+ Future<Void> sender =
+ executor.submit(
+ () -> {
+ for (int i = 0; i < numOfTsBlocks; i++) {
+ ListenableFuture<Void> blockedOnMemory;
+ synchronized (queue) {
+ blockedOnMemory =
queue.add(Utils.createMockTsBlock(mockTsBlockSize));
+ }
+ blockedOnMemory.get();
+ }
+ synchronized (queue) {
+ queue.setNoMoreTsBlocks(true);
+ }
+ return null;
+ });
+ Future<Void> receiver =
+ executor.submit(
+ () -> {
+ for (int i = 0; i < numOfTsBlocks; i++) {
+ ListenableFuture<Void> blocked;
+ synchronized (queue) {
+ blocked = queue.isBlocked();
+ }
+ blocked.get();
+ synchronized (queue) {
+ queue.remove();
+ }
+ }
+ return null;
+ });
+
+ sender.get(30, TimeUnit.SECONDS);
Review Comment:
Could we wait for whichever worker completes first here? If the receiver
fails before the sender finishes, it stops freeing queue memory; the sender can
then block once the pool fills, so this Future.get call reports a 30-second
TimeoutException and receiver.get is never reached. That hides the receiver's
actual exception, even though this change aims to propagate worker failures.
Using an ExecutorCompletionService with a shared deadline (and cancelling the
sibling task on failure) would make this fail fast and preserve the root cause.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
