This is an automated email from the ASF dual-hosted git repository.
yashmayya pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new 12e7af1dc18 Skip block copies for spool stages whose workers are all
remote (#19428)
12e7af1dc18 is described below
commit 12e7af1dc18d50981b82eb1f7b08ecb545c3558c
Author: Yash Mayya <[email protected]>
AuthorDate: Mon Sep 21 14:52:34 2026 -0700
Skip block copies for spool stages whose workers are all remote (#19428)
---
.../pinot/query/mailbox/GrpcSendingMailbox.java | 9 ++-
.../query/mailbox/InMemorySendingMailbox.java | 6 ++
.../apache/pinot/query/mailbox/SendingMailbox.java | 31 ++++++++-
.../runtime/operator/MailboxSendOperator.java | 3 +-
.../runtime/operator/exchange/BlockExchange.java | 34 ++++++++++
.../operator/exchange/BroadcastExchange.java | 32 +++++-----
.../query/mailbox/GrpcSendingMailboxTest.java | 12 ++++
.../query/mailbox/InMemorySendingMailboxTest.java | 12 ++++
.../operator/exchange/BlockExchangeTest.java | 22 +++++++
.../operator/exchange/BroadcastExchangeTest.java | 73 +++++++++++++++++++---
10 files changed, 204 insertions(+), 30 deletions(-)
diff --git
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/GrpcSendingMailbox.java
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/GrpcSendingMailbox.java
index d8483a1e835..90ae69c8d0e 100644
---
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/GrpcSendingMailbox.java
+++
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/GrpcSendingMailbox.java
@@ -116,9 +116,12 @@ public class GrpcSendingMailbox implements SendingMailbox {
return false;
}
- /// NOTE: [org.apache.pinot.query.runtime.operator.exchange.BlockExchange]
implementations rely on this method
- /// serializing the block synchronously on the calling thread: once it
returns, the block's contents may be handed
- /// by reference to a local receiver that mutates them.
+ @Override
+ public boolean deliversByReference() {
+ // Blocks are serialized within send(MseBlock.Data), so the receiver never
reads this instance
+ return false;
+ }
+
@Override
public void send(MseBlock.Data data) {
QueryThreadContext.checkTerminationAndSampleUsage(SEND_SCOPE);
diff --git
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/InMemorySendingMailbox.java
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/InMemorySendingMailbox.java
index 5b94e6ea5a7..c9cef3552d3 100644
---
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/InMemorySendingMailbox.java
+++
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/InMemorySendingMailbox.java
@@ -62,6 +62,12 @@ public class InMemorySendingMailbox implements
SendingMailbox {
return true;
}
+ @Override
+ public boolean deliversByReference() {
+ // Blocks are offered to the receiving mailbox as they are, so the
receiver reads the same instance
+ return true;
+ }
+
@Override
public void send(MseBlock.Data data) {
QueryThreadContext.checkTerminationAndSampleUsage(SEND_SCOPE);
diff --git
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/SendingMailbox.java
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/SendingMailbox.java
index a5b8f9260b1..790837aa83b 100644
---
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/SendingMailbox.java
+++
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/mailbox/SendingMailbox.java
@@ -33,14 +33,41 @@ import org.apache.pinot.segment.spi.memory.DataBuffer;
/// - One call to [#cancel(Throwable)] if the sender wants to cancel the
receiver
public interface SendingMailbox extends AutoCloseable {
- /// Returns whether the mailbox is sending data to a local receiver, where
blocks can be directly passed to the
- /// receiver.
+ /// Returns whether blocks can be passed to this mailbox whole, instead of
being split into smaller blocks that
+ /// respect the maximum content size of a mailbox message.
+ ///
+ /// This says nothing about whether the receiver ends up with a reference to
the block: see
+ /// [#deliversByReference()].
boolean isLocal();
+ /// Returns whether a receiver may keep a reference to the blocks sent to
this mailbox, instead of reading their
+ /// contents within [#send(MseBlock.Data)]. The answer must not change
during the life of the mailbox, so callers
+ /// can read it once.
+ ///
+ /// A mailbox that returns `true` here also returns `true` from
[#isLocal()], because a block can only be given by
+ /// reference to a receiver in this process. The reverse does not hold: a
local mailbox can still read the block
+ /// within [#send(MseBlock.Data)] and pass on something else.
+ ///
+ /// A caller that sends the same block instance to more than one mailbox,
when anything downstream mutates the
+ /// contents of that block, must:
+ ///
+ /// 1. Give a copy of the block to all the mailboxes that return `true`
here, except one.
+ /// 2. Give the original block to that remaining mailbox last, once every
other mailbox has returned from
+ /// [#send(MseBlock.Data)].
+ ///
+ /// Step 2 matters as much as step 1. A receiver of the original block can
mutate it as soon as `send` gives it the
+ /// block, which would corrupt the block for a mailbox that is still reading
it.
+ ///
+ /// See [org.apache.pinot.query.runtime.operator.exchange.BroadcastExchange].
+ boolean deliversByReference();
+
/// Sends a data block to the receiver. Note that SendingMailbox are
required to acquire resources lazily in this
/// call, and they should **not** acquire any resources when they are
created. This method should throw if there was
/// an error sending the data, since that would allow
/// [org.apache.pinot.query.runtime.operator.exchange.BlockExchange] to exit
early.
+ ///
+ /// Implementations that return `false` from [#deliversByReference()] must
finish reading the block before this
+ /// method returns, because the caller may then pass the same block to a
receiver that mutates it.
void send(MseBlock.Data data);
/// Sends an EOS block to the receiver. Note that SendingMailbox are
required to acquire resources lazily in this
diff --git
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MailboxSendOperator.java
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MailboxSendOperator.java
index 363dc124470..e90d70cee26 100644
---
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MailboxSendOperator.java
+++
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/MailboxSendOperator.java
@@ -89,7 +89,8 @@ public class MailboxSendOperator extends MultiStageOperator {
/// 1. One inner exchange is created for each receiver stage, using the
method mentioned above and keeping the
/// distribution type specified in the [MailboxSendNode].
/// 2. Then, a single outer broadcast exchange is created to fan out the
data to all the inner exchanges. It copies
- /// blocks that carry aggregation intermediate results so that no two
receiver stages share them.
+ /// blocks that carry aggregation intermediate results, so that no two
receiver stages share them. Stages whose
+ /// workers all read the block within
[SendingMailbox#send(MseBlock.Data)] do not need a copy.
///
/// @see BlockExchange#asSendingMailbox(String)
private static BlockExchange getBlockExchange(OpChainExecutionContext ctx,
MailboxSendNode node,
diff --git
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/exchange/BlockExchange.java
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/exchange/BlockExchange.java
index 56c058c324c..4ad02f16843 100644
---
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/exchange/BlockExchange.java
+++
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/exchange/BlockExchange.java
@@ -43,6 +43,7 @@ public abstract class BlockExchange implements AutoCloseable {
private final List<SendingMailbox> _sendingMailboxes;
private final BlockSplitter _splitter;
private final Function<List<SendingMailbox>, Integer> _statsIndexChooser;
+ private final boolean _deliversByReference;
protected static final Function<List<SendingMailbox>, Integer>
RANDOM_INDEX_CHOOSER =
(mailboxes) -> ThreadLocalRandom.current().nextInt(mailboxes.size());
@@ -86,6 +87,21 @@ public abstract class BlockExchange implements AutoCloseable
{
_sendingMailboxes = sendingMailboxes;
_splitter = splitter;
_statsIndexChooser = statsIndexChooser;
+ _deliversByReference = anyDeliversByReference(sendingMailboxes);
+ }
+
+ /// Returns whether any of the given mailboxes delivers blocks by reference.
Mailboxes are fixed when the exchange
+ /// is created, and each of them gives a constant answer, so this is
computed once.
+ ///
+ /// The answer therefore counts mailboxes that terminate early later on.
That only makes a caller take a copy it
+ /// did not need. It never makes a caller skip a copy it did need.
+ private static boolean anyDeliversByReference(List<SendingMailbox>
sendingMailboxes) {
+ for (SendingMailbox sendingMailbox : sendingMailboxes) {
+ if (sendingMailbox.deliversByReference()) {
+ return true;
+ }
+ }
+ return false;
}
/// API to send a block to the destination mailboxes.
@@ -164,6 +180,12 @@ public abstract class BlockExchange implements
AutoCloseable {
}
}
+ /// Sends the block to the destinations, following the distribution strategy
of this exchange.
+ ///
+ /// Implementations must finish reading the block before this method
returns. [BlockExchangeSendingMailbox] reports
+ /// that it does not deliver blocks by reference on that basis, so an
implementation that queued a block for
+ /// another thread would break [SendingMailbox#deliversByReference()] for
every exchange that decorates it, and
+ /// silently corrupt the aggregation intermediate results that
[BroadcastExchange] shares between destinations.
protected abstract void route(List<SendingMailbox> destinations,
MseBlock.Data block);
@Override
@@ -213,9 +235,21 @@ public abstract class BlockExchange implements
AutoCloseable {
@Override
public boolean isLocal() {
+ // Blocks are handed to the decorated exchange whole, and splitting them
is left to that exchange.
+ // TODO(#19427): the decorated exchange is currently built with
BlockSplitter#NO_OP, so blocks sent
+ // through a multi-send node are never split. See
MailboxSendOperator#getBlockExchange.
return true;
}
+ @Override
+ public boolean deliversByReference() {
+ // The decorated exchange passes blocks to its own mailboxes, so this
mailbox delivers by reference only if
+ // any of those does. The question is whether a receiver keeps the block
alive after route returns, not whether
+ // the decorated exchange shares one block between its own mailboxes: an
inner HashExchange gives each of its
+ // mailboxes a different block, but the rows of those blocks still hold
the cells of this one.
+ return _deliversByReference;
+ }
+
@Override
public void send(MseBlock.Data data) {
if (LOGGER.isTraceEnabled()) {
diff --git
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/exchange/BroadcastExchange.java
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/exchange/BroadcastExchange.java
index 0d902df14bf..11695ab0f13 100644
---
a/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/exchange/BroadcastExchange.java
+++
b/pinot-query-runtime/src/main/java/org/apache/pinot/query/runtime/operator/exchange/BroadcastExchange.java
@@ -28,14 +28,14 @@ import
org.apache.pinot.query.runtime.blocks.RowHeapDataBlock;
/// Broadcast blocks to all the destinations.
///
-/// This is the only exchange that routes the same block to more than one
destination, and local (same-JVM) mailboxes
-/// deliver on-heap blocks by reference. Blocks that
+/// This is the only exchange that routes the same block to more than one
destination, and some mailboxes
+/// [deliver blocks by reference][SendingMailbox#deliversByReference()].
Blocks that
/// [carry aggregation intermediate
results][RowHeapDataBlock#containsObjectColumns()] cannot be shared this way:
/// downstream operators mutate those objects in place when they merge them or
extract final results, so two
-/// receivers on the same server would corrupt them. For such blocks, [#route]
gives every local destination except
-/// the first its own [copy][RowHeapDataBlock#copyObjectColumns()]. Remote
destinations only read the block to
-/// serialize it, before any local receiver can mutate it, so they do not need
copies. Blocks without OBJECT columns
-/// are shared by reference with all the destinations.
+/// receivers of the same block would corrupt them. For such blocks, [#route]
gives every destination that delivers
+/// by reference, except the first, its own
[copy][RowHeapDataBlock#copyObjectColumns()]. The other destinations read
+/// the block within [SendingMailbox#send(MseBlock.Data)], before any receiver
can mutate it, so they do not need
+/// copies. Blocks without OBJECT columns are shared by reference with all the
destinations.
///
/// This also protects multi-send (spool) nodes, which fan each block out to
the exchanges of their receiver stages
/// through this exchange (see [BlockExchange#asSendingMailbox]).
@@ -60,26 +60,26 @@ class BroadcastExchange extends BlockExchange {
}
return;
}
- // Send a copy to every active local destination except the first one,
which receives the original block without
- // copying. Remote destinations serialize the original block on this
thread, and the copies are also made on this
- // thread, so all reads of the original block finish before it is handed
to a local receiver that can start
- // mutating it.
+ // Send a copy to every active destination that delivers by reference,
except the first one, which receives the
+ // original block without copying. The other destinations read the
original block within send, and the copies are
+ // also made on this thread, so all reads of the original block finish
before it is handed to a receiver that can
+ // start mutating it.
RowHeapDataBlock rowHeapBlock = block.asRowHeap();
- SendingMailbox firstLocalDestination = null;
+ SendingMailbox firstByReferenceDestination = null;
for (SendingMailbox mailbox : destinations) {
if (mailbox.isEarlyTerminated()) {
continue;
}
- if (!mailbox.isLocal()) {
+ if (!mailbox.deliversByReference()) {
sendBlock(mailbox, block);
- } else if (firstLocalDestination == null) {
- firstLocalDestination = mailbox;
+ } else if (firstByReferenceDestination == null) {
+ firstByReferenceDestination = mailbox;
} else {
sendBlock(mailbox, rowHeapBlock.copyObjectColumns());
}
}
- if (firstLocalDestination != null) {
- sendBlock(firstLocalDestination, block);
+ if (firstByReferenceDestination != null) {
+ sendBlock(firstByReferenceDestination, block);
}
}
}
diff --git
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/mailbox/GrpcSendingMailboxTest.java
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/mailbox/GrpcSendingMailboxTest.java
index 57f074352cf..9241f45b754 100644
---
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/mailbox/GrpcSendingMailboxTest.java
+++
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/mailbox/GrpcSendingMailboxTest.java
@@ -80,6 +80,18 @@ public class GrpcSendingMailboxTest {
}
}
+ @Test
+ public void doesNotDeliverBlocksByReference() {
+ GrpcSendingMailbox mailbox =
+ new GrpcSendingMailbox("test-mailbox",
Mockito.mock(ChannelManager.class), "localhost", 0, Long.MAX_VALUE,
+ new StatMap<>(MailboxSendOperator.StatKey.class), 4 * 1024 * 1024,
true);
+
+ // Blocks are serialized within send(MseBlock.Data), so senders that share
one block between mailboxes do not
+ // need to give this one a copy. See BroadcastExchange.
+ Assert.assertFalse(mailbox.deliversByReference());
+ Assert.assertFalse(mailbox.isLocal());
+ }
+
/// Regression test for the lazy-initialization data race on
`_contentObserver`. Before the fix, both `sendInternal`
/// and `cancel` had an unsynchronized `if (_contentObserver == null) {
_contentObserver = getContentObserver(); }`
/// pattern. `_contentObserver` is `volatile` so individual reads/writes are
atomic, but two threads racing through
diff --git
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/mailbox/InMemorySendingMailboxTest.java
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/mailbox/InMemorySendingMailboxTest.java
index a2b4541350a..c550d37af5b 100644
---
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/mailbox/InMemorySendingMailboxTest.java
+++
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/mailbox/InMemorySendingMailboxTest.java
@@ -49,4 +49,16 @@ public class InMemorySendingMailboxTest {
Mockito.verifyNoInteractions(mailboxService);
}
}
+
+ @Test
+ public void deliversBlocksByReference() {
+ InMemorySendingMailbox mailbox =
+ new InMemorySendingMailbox("test-mailbox",
Mockito.mock(MailboxService.class), Long.MAX_VALUE,
+ new StatMap<>(MailboxSendOperator.StatKey.class));
+
+ // Blocks are offered to the receiving mailbox as they are, so senders
that share one block between mailboxes
+ // must give this one a copy. See BroadcastExchange.
+ Assert.assertTrue(mailbox.deliversByReference());
+ Assert.assertTrue(mailbox.isLocal());
+ }
}
diff --git
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/operator/exchange/BlockExchangeTest.java
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/operator/exchange/BlockExchangeTest.java
index c2087ecfdbf..907766e21e3 100644
---
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/operator/exchange/BlockExchangeTest.java
+++
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/operator/exchange/BlockExchangeTest.java
@@ -165,6 +165,28 @@ public class BlockExchangeTest {
Assert.assertEquals(sentBlocks.get(1).asRowHeap().getRows(),
outBlockTwo.getRows());
}
+ @Test
+ public void shouldDeliverByReferenceWhenAnyDestinationDoes() {
+ // Given: an exchange whose destinations are one mailbox that serializes
blocks and one that does not
+ when(_mailbox2.deliversByReference()).thenReturn(true);
+ BlockExchange exchange = new TestBlockExchange(List.of(_mailbox1,
_mailbox2));
+
+ // Then: the exchange exposed as a mailbox reports that it delivers by
reference
+ Assert.assertTrue(exchange.asSendingMailbox("1").deliversByReference());
+ }
+
+ @Test
+ public void shouldNotDeliverByReferenceWhenNoDestinationDoes() {
+ // Given: an exchange whose destinations all serialize the blocks they are
sent
+ BlockExchange exchange = new TestBlockExchange(List.of(_mailbox1,
_mailbox2));
+
+ // Then: the exchange exposed as a mailbox reports that it does not
deliver by reference. It still takes blocks
+ // whole, because splitting them is left to the exchange it decorates
+ SendingMailbox sendingMailbox = exchange.asSendingMailbox("1");
+ Assert.assertFalse(sendingMailbox.deliversByReference());
+ Assert.assertTrue(sendingMailbox.isLocal());
+ }
+
private static class TestBlockExchange extends BlockExchange {
protected TestBlockExchange(List<SendingMailbox> destinations) {
this(destinations, (block, size) -> Iterators.singletonIterator(block));
diff --git
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/operator/exchange/BroadcastExchangeTest.java
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/operator/exchange/BroadcastExchangeTest.java
index 7255149df72..c370d5a52e4 100644
---
a/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/operator/exchange/BroadcastExchangeTest.java
+++
b/pinot-query-runtime/src/test/java/org/apache/pinot/query/runtime/operator/exchange/BroadcastExchangeTest.java
@@ -63,8 +63,11 @@ public class BroadcastExchangeTest {
@BeforeMethod
public void setUp() {
_mocks = MockitoAnnotations.openMocks(this);
+ // In-memory mailboxes take blocks whole and hand them to the receiver by
reference
Mockito.when(_mailbox1.isLocal()).thenReturn(true);
Mockito.when(_mailbox2.isLocal()).thenReturn(true);
+ Mockito.when(_mailbox1.deliversByReference()).thenReturn(true);
+ Mockito.when(_mailbox2.deliversByReference()).thenReturn(true);
_aggFunction = mockFunnelAggFunction();
}
@@ -89,7 +92,7 @@ public class BroadcastExchangeTest {
// getAggFunctions() is deprecated, but the copy must preserve it for
downstream serialization, so assert on it
@SuppressWarnings("deprecation")
@Test
- public void shouldCopyBlocksWithObjectColumnsForAllLocalDestinationsButOne()
{
+ public void
shouldCopyBlocksWithObjectColumnsForAllByReferenceDestinationsButOne() {
// Given:
PriorityQueue<FunnelStepEvent> stepEvents = stepEvents(new
FunnelStepEvent(1000L, 0),
new FunnelStepEvent(2000L, 1));
@@ -124,16 +127,16 @@ public class BroadcastExchangeTest {
}
@Test
- public void shouldSendOriginalBlockToRemoteDestinationsBeforeTheLocalOne() {
- // Given: a remote destination between two local ones
- SendingMailbox remoteMailbox = Mockito.mock(SendingMailbox.class);
+ public void
shouldSendOriginalBlockToOtherDestinationsBeforeTheByReferenceOne() {
+ // Given: a destination that reads the block within send, between two that
deliver it by reference
+ SendingMailbox remoteMailbox = remoteMailbox();
RowHeapDataBlock block = funnelBlock(stepEvents(new FunnelStepEvent(1000L,
0)));
// When:
route(block, remoteMailbox, _mailbox1, _mailbox2);
- // Then: the remote destination serializes the original, the first local
destination receives the original by
- // reference after all other reads of it, and the extra local destination
receives a copy
+ // Then: the remote destination serializes the original, the first
by-reference destination receives the original
+ // after all other reads of it, and the extra by-reference destination
receives a copy
assertSame(capturedBlock(remoteMailbox), block);
assertSame(capturedBlock(_mailbox1), block);
assertNotSame(capturedBlock(_mailbox2), block);
@@ -143,12 +146,52 @@ public class BroadcastExchangeTest {
inOrder.verify(_mailbox1).send(Mockito.any(MseBlock.Data.class));
}
+ @Test
+ public void shouldNotCopyBlocksForSpoolStagesWithoutByReferenceWorkers() {
+ // Given: two receiver stages of a multi-send (spool) node, each fanning
out to workers on other servers
+ SendingMailbox remoteMailbox1 = remoteMailbox();
+ SendingMailbox remoteMailbox2 = remoteMailbox();
+ RowHeapDataBlock block = funnelBlock(stepEvents(new FunnelStepEvent(1000L,
0)));
+
+ // When:
+ route(block, spoolStageMailbox(remoteMailbox1),
spoolStageMailbox(remoteMailbox2));
+
+ // Then: no copies are made, because every worker of both stages
serializes the block
+ Mockito.verify(_aggFunction,
Mockito.never()).serializeIntermediateResult(Mockito.any());
+ assertSame(capturedBlock(remoteMailbox1), block);
+ assertSame(capturedBlock(remoteMailbox2), block);
+ }
+
+ @Test
+ public void shouldCopyBlocksForSpoolStagesWithAByReferenceWorker() {
+ // Given: two receiver stages of a multi-send (spool) node. The second one
has a worker on this server, next to
+ // a worker on another server
+ SendingMailbox remoteMailbox = remoteMailbox();
+ RowHeapDataBlock block = funnelBlock(stepEvents(new FunnelStepEvent(1000L,
0)));
+
+ // When:
+ route(block, spoolStageMailbox(_mailbox1),
spoolStageMailbox(remoteMailbox, _mailbox2));
+
+ // Then: one copy is made, because a single by-reference worker makes the
whole stage share the block
+ Mockito.verify(_aggFunction,
Mockito.times(1)).serializeIntermediateResult(Mockito.any());
+ assertSame(capturedBlock(_mailbox1), block);
+ MseBlock.Data copiedBlock = capturedBlock(_mailbox2);
+ assertNotSame(copiedBlock, block);
+ // Within that stage the copy is shared again: only one of its workers
keeps a reference to it
+ assertSame(capturedBlock(remoteMailbox), copiedBlock);
+ // The original block reaches its receiver last, after every other
destination has read it
+ InOrder inOrder = Mockito.inOrder(remoteMailbox, _mailbox2, _mailbox1);
+ inOrder.verify(remoteMailbox).send(Mockito.any(MseBlock.Data.class));
+ inOrder.verify(_mailbox2).send(Mockito.any(MseBlock.Data.class));
+ inOrder.verify(_mailbox1).send(Mockito.any(MseBlock.Data.class));
+ }
+
@Test
@SuppressWarnings("unchecked")
public void shouldShareBlocksWithObjectColumnsWithRemoteOnlyDestinations() {
// Given: only remote destinations, which serialize the block instead of
delivering it by reference
- SendingMailbox remoteMailbox1 = Mockito.mock(SendingMailbox.class);
- SendingMailbox remoteMailbox2 = Mockito.mock(SendingMailbox.class);
+ SendingMailbox remoteMailbox1 = remoteMailbox();
+ SendingMailbox remoteMailbox2 = remoteMailbox();
RowHeapDataBlock block = funnelBlock(stepEvents(new FunnelStepEvent(1000L,
0)));
// When:
@@ -255,6 +298,20 @@ public class BroadcastExchangeTest {
new BroadcastExchange(destinationList,
BlockSplitter.NO_OP).route(destinationList, block);
}
+ /// Wraps the mailboxes of one receiver stage the way a multi-send (spool)
node does: an inner exchange per stage,
+ /// exposed to the outer exchange as a single sending mailbox.
+ private static SendingMailbox spoolStageMailbox(SendingMailbox...
innerMailboxes) {
+ return new BroadcastExchange(List.of(innerMailboxes),
BlockSplitter.NO_OP).asSendingMailbox("1");
+ }
+
+ /// A mailbox to a worker on another server. It reads the block within send
instead of giving it to the receiver.
+ private static SendingMailbox remoteMailbox() {
+ SendingMailbox mailbox = Mockito.mock(SendingMailbox.class);
+ Mockito.when(mailbox.isLocal()).thenReturn(false);
+ Mockito.when(mailbox.deliversByReference()).thenReturn(false);
+ return mailbox;
+ }
+
private static MseBlock.Data capturedBlock(SendingMailbox mailbox) {
ArgumentCaptor<MseBlock.Data> captor =
ArgumentCaptor.forClass(MseBlock.Data.class);
Mockito.verify(mailbox, Mockito.times(1)).send(captor.capture());
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
