diff --git a/modules/core/src/main/java/org/apache/ignite/internal/CoreMessagesProvider.java b/modules/core/src/main/java/org/apache/ignite/internal/CoreMessagesProvider.java index a460ed789d24b..66be5c98df724 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/CoreMessagesProvider.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/CoreMessagesProvider.java @@ -210,6 +210,7 @@ import org.apache.ignite.internal.processors.datastreamer.DataStreamerEntry; import org.apache.ignite.internal.processors.datastreamer.DataStreamerRequest; import org.apache.ignite.internal.processors.datastreamer.DataStreamerResponse; +import org.apache.ignite.internal.processors.datastreamer.StreamReceiverMessage; import org.apache.ignite.internal.processors.marshaller.MappedName; import org.apache.ignite.internal.processors.marshaller.MappingAcceptedMessage; import org.apache.ignite.internal.processors.marshaller.MappingProposedMessage; @@ -662,6 +663,7 @@ public CoreMessagesProvider(Marshaller dfltMarsh, Marshaller schemaAwareMarsh) { register(DataStreamerEntry.class); register(DataStreamerRequest.class); register(DataStreamerResponse.class); + register(StreamReceiverMessage.class); // [11900 - 12000]: Metrics, monitoring messages. msgIdx = 11900; diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamProcessor.java index 464a74d82ee25..6e2d787a893b0 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamProcessor.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamProcessor.java @@ -27,6 +27,7 @@ import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException; import org.apache.ignite.internal.managers.communication.GridIoManager; import org.apache.ignite.internal.managers.communication.GridMessageListener; +import org.apache.ignite.internal.managers.communication.MessageMarshalling; import org.apache.ignite.internal.managers.deployment.GridDeployment; import org.apache.ignite.internal.processors.GridProcessorAdapter; import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion; @@ -48,7 +49,6 @@ import org.apache.ignite.lang.IgniteClosure; import org.apache.ignite.lang.IgniteFuture; import org.apache.ignite.lang.IgniteInClosure; -import org.apache.ignite.marshaller.Marshaller; import org.apache.ignite.stream.StreamReceiver; import org.jetbrains.annotations.Nullable; @@ -68,9 +68,6 @@ public class DataStreamProcessor extends GridProcessorAdapter { /** Data Streamer flusher. */ private final DataStreamerFlusher flusher = new DataStreamerFlusher(); - /** Marshaller. */ - private final Marshaller marsh; - /** * @param ctx Kernal context. */ @@ -86,8 +83,6 @@ public DataStreamProcessor(GridKernalContext ctx) { } }); } - - marsh = ctx.marshaller(); } /** {@inheritDoc} */ @@ -240,7 +235,10 @@ private void processRequest(final UUID nodeId, final DataStreamerRequest req) { StreamReceiver updater; try { - updater = U.unmarshal(marsh, req.updaterBytes(), U.resolveClassLoader(clsLdr, ctx.config())); + // Read here, not on the inbound pass: the deployment class loader is known only at this point. + MessageMarshalling.unmarshal(req, ctx, null, U.resolveClassLoader(clsLdr, ctx.config())); + + updater = req.updater(); if (updater != null) ctx.resource().injectGeneric(updater); diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java index 1040f06ddf83b..a8bc307096e4b 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java @@ -152,11 +152,8 @@ public class DataStreamerImpl implements IgniteDataStreamer, Delayed /** Amount of permissions should be available to continue new data processing. */ private static final int REMAP_SEMAPHORE_PERMISSIONS_COUNT = Integer.MAX_VALUE; - /** Cache receiver. */ - private StreamReceiver rcvr = ISOLATED_UPDATER; - - /** */ - private byte[] updaterBytes; + /** Cache receiver, in the message that carries it to the remote nodes. */ + private volatile StreamReceiverMessage rcvrMsg = new StreamReceiverMessage(ISOLATED_UPDATER); /** IO policy resovler for data load request. */ private IgniteClosure ioPlcRslvr; @@ -489,12 +486,18 @@ public IgniteInternalFuture internalFuture() { @Override public void receiver(StreamReceiver rcvr) { A.notNull(rcvr, "rcvr"); - this.rcvr = rcvr; + rcvrMsg = new StreamReceiverMessage(rcvr); + } + + /** @return Cache receiver. */ + @SuppressWarnings("unchecked") + private StreamReceiver receiver() { + return (StreamReceiver)rcvrMsg.receiver(); } /** {@inheritDoc} */ @Override public boolean allowOverwrite() { - return rcvr != ISOLATED_UPDATER; + return receiver() != ISOLATED_UPDATER; } /** {@inheritDoc} */ @@ -507,7 +510,7 @@ public IgniteInternalFuture internalFuture() { if (node == null) throw new CacheException("Failed to get node for cache: " + cacheName); - rcvr = allow ? DataStreamerCacheUpdaters.individual() : ISOLATED_UPDATER; + rcvrMsg = new StreamReceiverMessage(allow ? DataStreamerCacheUpdaters.individual() : ISOLATED_UPDATER); } /** {@inheritDoc} */ @@ -655,7 +658,7 @@ public IgniteFuture addDataInternal(Collection e lock(false); - if (rcvr instanceof IsolatedUpdater && inconsistencyWarned.compareAndSet(false, true)) + if (receiver() instanceof IsolatedUpdater && inconsistencyWarned.compareAndSet(false, true)) log.warning(WRN_INCONSISTENT_UPDATES); try { @@ -886,6 +889,8 @@ private void load0( assert key != null; if (initPda) { + StreamReceiver rcvr = receiver(); + if (cacheObjCtx.addDeploymentInfo()) jobPda = new DataStreamerPda(key.value(cacheObjCtx, false), entry.getValue() != null ? entry.getValue().value(cacheObjCtx, false) : null, @@ -1850,7 +1855,7 @@ else if (!topFut.isDone()) false, skipStore, keepBinary, - rcvr), + receiver()), plc); locFuts.add(callFut); @@ -1943,12 +1948,6 @@ private void submit( if (val != null) val.marshal(cacheObjCtx); } - - if (updaterBytes == null) { - assert rcvr != null; - - updaterBytes = U.marshal(ctx, rcvr); - } } catch (IgniteCheckedException e) { U.error(log, "Failed to marshal.", e); @@ -1991,11 +1990,13 @@ private void submit( if (topVer == null) topVer = ctx.cache().context().exchange().readyAffinityVersion(); + StreamReceiverMessage rcvrMsg0 = rcvrMsg; + DataStreamerRequest req = new DataStreamerRequest( reqId, topicId, cacheName, - updaterBytes, + rcvrMsg0, entries, true, skipStore, @@ -2007,7 +2008,7 @@ private void submit( dep != null ? dep.classLoaderId() : null, dep == null, topVer, - (rcvr == ISOLATED_UPDATER) ? partId : NO_STRIPE); + (rcvrMsg0.receiver() == ISOLATED_UPDATER) ? partId : NO_STRIPE); try { ctx.io().sendToGridTopic(node, TOPIC_DATASTREAM, req, plc); diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerRequest.java index 88af958a50373..f846c465599a3 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerRequest.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerRequest.java @@ -26,16 +26,18 @@ import org.apache.ignite.internal.StripedMessage; import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion; import org.apache.ignite.internal.processors.cache.GridCacheUtils; +import org.apache.ignite.internal.util.tostring.GridToStringExclude; import org.apache.ignite.internal.util.tostring.GridToStringInclude; import org.apache.ignite.internal.util.typedef.internal.S; import org.apache.ignite.lang.IgniteUuid; import org.apache.ignite.plugin.extensions.communication.CacheIdAware; +import org.apache.ignite.stream.StreamReceiver; import org.jetbrains.annotations.NotNull; import org.jetbrains.annotations.Nullable; import static org.apache.ignite.internal.GridTopic.TOPIC_DATASTREAM; -/** */ +/** Batch of streamed entries. The receiver it carries is a user class, hence the deferred unmarshalling. */ public class DataStreamerRequest implements DeferredUnmarshalMessage, CacheIdAware, StripedMessage { /** */ @Order(0) @@ -49,10 +51,10 @@ public class DataStreamerRequest implements DeferredUnmarshalMessage, CacheIdAwa @Order(2) String cacheName; - /** */ - // TODO: Refactor bytes serialization - IGNITE-27977 + /** Cache receiver, in the message that carries it. Out of {@code toString()}: it is a user object. */ + @GridToStringExclude @Order(3) - byte[] updaterBytes; + StreamReceiverMessage updaterMsg; /** Entries to update. */ @Order(4) @@ -112,7 +114,7 @@ public DataStreamerRequest() { * @param reqId Request ID. * @param resTopicId Response topic ID. * @param cacheName Cache name. - * @param updaterBytes Cache receiver. + * @param updaterMsg Cache receiver, in the message that carries it. * @param entries Entries to put. * @param ignoreDepOwnership Ignore ownership. * @param skipStore Skip store flag. @@ -130,7 +132,7 @@ public DataStreamerRequest( long reqId, IgniteUuid resTopicId, @Nullable String cacheName, - byte[] updaterBytes, + StreamReceiverMessage updaterMsg, Collection entries, boolean ignoreDepOwnership, boolean skipStore, @@ -149,7 +151,7 @@ public DataStreamerRequest( this.reqId = reqId; this.resTopicId = resTopicId; this.cacheName = cacheName; - this.updaterBytes = updaterBytes; + this.updaterMsg = updaterMsg; this.entries = entries; this.ignoreDepOwnership = ignoreDepOwnership; this.skipStore = skipStore; @@ -180,8 +182,8 @@ String cacheName() { } /** @return Updater. */ - byte[] updaterBytes() { - return updaterBytes; + StreamReceiver updater() { + return updaterMsg != null ? updaterMsg.receiver() : null; } /** @return Entries to update. */ diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/StreamReceiverMessage.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/StreamReceiverMessage.java new file mode 100644 index 0000000000000..076bd8fe384e5 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/StreamReceiverMessage.java @@ -0,0 +1,59 @@ +/* + * 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.ignite.internal.processors.datastreamer; + +import org.apache.ignite.internal.Marshalled; +import org.apache.ignite.internal.Order; +import org.apache.ignite.internal.UseBinaryMarshaller; +import org.apache.ignite.plugin.extensions.communication.Message; +import org.apache.ignite.stream.StreamReceiver; + +/** + * The receiver of a streamer on its way to the nodes that own the data: a user object here, its serialized form on + * the wire. One instance serves every batch of a streamer, so the receiver is marshalled once and the batches share + * the result; a streamer given another receiver builds another instance. + */ +@UseBinaryMarshaller +public class StreamReceiverMessage implements Message { + /** */ + @Marshalled("rcvrBytes") + StreamReceiver rcvr; + + /** + * Serialized {@link #rcvr}, written by whichever batch is marshalled first and read by the rest. Those batches + * leave on different threads, hence the {@code volatile}: a reader seeing the reference before the contents would + * skip the marshalling and send a half-written array. + */ + @Order(0) + volatile byte[] rcvrBytes; + + /** Empty constructor. */ + public StreamReceiverMessage() { + // No-op. + } + + /** @param rcvr Receiver. */ + StreamReceiverMessage(StreamReceiver rcvr) { + this.rcvr = rcvr; + } + + /** @return Receiver. */ + StreamReceiver receiver() { + return rcvr; + } +} diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java index 2bc777180e0ff..5e350aafc1f95 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java @@ -24,6 +24,7 @@ import java.util.Map; import java.util.Random; import java.util.concurrent.Callable; +import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.CountDownLatch; import java.util.concurrent.CyclicBarrier; import java.util.concurrent.TimeUnit; @@ -88,6 +89,16 @@ public class DataStreamerImplSelfTest extends GridCommonAbstractTest { /** Indicates whether we need to make the topology stale */ private static boolean needStaleTop = false; + /** Receiver carriers of the streamer requests sent since the current test started. */ + private static final Collection sentReceivers = new ConcurrentLinkedQueue<>(); + + /** {@inheritDoc} */ + @Override protected void beforeTest() throws Exception { + super.beforeTest(); + + sentReceivers.clear(); + } + /** {@inheritDoc} */ @Override protected void afterTest() throws Exception { super.afterTest(); @@ -142,6 +153,36 @@ public void testCloseWithCancellation() throws Exception { assertTrue(fut.isDone()); } + /** + * The receiver does not change between batches, so it is marshalled once: every request carries the very bytes + * produced for the first one. + * + * @throws Exception If failed. + */ + @Test + public void testReceiverMarshalledOncePerStreamer() throws Exception { + cnt = 0; + + startGrids(2); + + try (IgniteDataStreamer ldr = grid(0).dataStreamer(DEFAULT_CACHE_NAME)) { + ldr.perNodeBufferSize(1); + + for (int i = 0; i < KEYS_COUNT; i++) + ldr.addData(i, i); + } + + assertTrue("Expected more than one request to a remote node, got " + sentReceivers.size(), + sentReceivers.size() > 1); + + StreamReceiverMessage first = F.first(sentReceivers); + + assertNotNull(first.rcvrBytes); + + for (StreamReceiverMessage rcvr : sentReceivers) + assertTrue("The receiver was marshalled more than once", first.rcvrBytes == rcvr.rcvrBytes); + } + /** * Test inconsistency log warning of the streamer. Default receiver goes first and is set again after a consistent * receiver. The warning must appear only once. @@ -670,6 +711,12 @@ private CacheConfiguration cacheConfiguration() { private static class StaleTopologyCommunicationSpi extends TcpCommunicationSpi { /** {@inheritDoc} */ @Override public void sendMessage(ClusterNode node, Message msg, IgniteInClosure ackC) { + Message sentMsg = msg instanceof GridIoMessage ? ((GridIoMessage)msg).message() : null; + + // The message is already marshalled at this point, so the serialized receiver is in place. + if (sentMsg instanceof DataStreamerRequest) + sentReceivers.add(((DataStreamerRequest)sentMsg).updaterMsg); + // Send stale topology only in the first request to avoid indefinitely getting failures. if (needStaleTop) { if (msg instanceof GridIoMessage) { @@ -692,7 +739,7 @@ private static class StaleTopologyCommunicationSpi extends TcpCommunicationSpi { req.requestId(), req.resTopicId, req.cacheName(), - req.updaterBytes(), + new StreamReceiverMessage(req.updater()), req.entries(), req.ignoreDeploymentOwnership(), req.skipStore(),