anton-vinogradov commented on code in PR #13454: URL: https://github.com/apache/ignite/pull/13454#discussion_r3752042315
########## modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerReceiverMessage.java: ########## @@ -0,0 +1,68 @@ +/* + * 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; +import org.jetbrains.annotations.Nullable; + +/** DataStreamer cache receiver/updater message. */ +@UseBinaryMarshaller +public class DataStreamerReceiverMessage implements Message { + /** Custom cache receiver/updater; {@code null} when {@link #builtIn} is effective. */ + @Marshalled("rcvrBytes") + @Nullable StreamReceiver<?, ?> rcvr; + + /** Serialized {@link #rcvr}. */ Review Comment: Added — `Serialized {@link #rcvr}; {@code null} when {@link #builtIn} is effective.` ########## modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerReceiverMessage.java: ########## @@ -0,0 +1,68 @@ +/* + * 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; +import org.jetbrains.annotations.Nullable; + +/** DataStreamer cache receiver/updater message. */ +@UseBinaryMarshaller +public class DataStreamerReceiverMessage implements Message { + /** Custom cache receiver/updater; {@code null} when {@link #builtIn} is effective. */ + @Marshalled("rcvrBytes") + @Nullable StreamReceiver<?, ?> rcvr; + + /** Serialized {@link #rcvr}. */ + @Order(0) + volatile @Nullable byte[] rcvrBytes; Review Comment: Yes, only here. `rcvr` and `builtIn` are set in a constructor, and on the receiving side by the unmarshalling that happens before the message reaches the processing thread — both publish safely. `rcvrBytes` is the one written late: the first batch to be marshalled fills it, and the other batches, on other threads, read it. Without the keyword a reader could see the reference before the contents and send a half-written array. ########## modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerReceiverMessage.java: ########## @@ -0,0 +1,68 @@ +/* + * 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; +import org.jetbrains.annotations.Nullable; + +/** DataStreamer cache receiver/updater message. */ +@UseBinaryMarshaller +public class DataStreamerReceiverMessage implements Message { + /** Custom cache receiver/updater; {@code null} when {@link #builtIn} is effective. */ + @Marshalled("rcvrBytes") + @Nullable StreamReceiver<?, ?> rcvr; + + /** Serialized {@link #rcvr}. */ + @Order(0) + volatile @Nullable byte[] rcvrBytes; + + /** A built-in updater every node has; {@code null} when {@link #rcvr} is effective. */ + @Order(1) + @Nullable DataStreamerBuiltInUpdater builtIn; + + /** Empty constructor for serialization purposes. */ + public DataStreamerReceiverMessage() { + // No-op. + } + + /** @param rcvr Custom receiver. */ + DataStreamerReceiverMessage(StreamReceiver<?, ?> rcvr) { + assert DataStreamerBuiltInUpdater.of(rcvr) == null : "A built-in updater travels by name: " + rcvr; + + this.rcvr = rcvr; + } + + /** @param builtIn Built-in updater every node has, named rather than sent. */ Review Comment: Reworded — `Built-in updater, sent as a name rather than as a copy.` ########## modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java: ########## @@ -146,17 +146,14 @@ public class DataStreamerImpl<K, V> implements IgniteDataStreamer<K, V>, Delayed */ private final Map<Long, ThreadBuffer> threadBufMap = new ConcurrentHashMap<>(); - /** Isolated receiver. */ - private static final StreamReceiver ISOLATED_UPDATER = new IsolatedUpdater(); + /** Default, Isolated receiver. */ + static final StreamReceiver ISOLATED_UPDATER = new IsolatedUpdater(); /** 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<K, V> rcvr = ISOLATED_UPDATER; - - /** */ - private byte[] updaterBytes; + /** Cache receiver in the message that carries it; {@code null} while none is set. */ + private volatile DataStreamerReceiverMessage rcvrMsg; Review Comment: It can, and that is the state before any receiver is set — the Isolated updater is used meanwhile. Marked the field and said so: `/** Message of the cache receiver; {@code null} until a receiver is set, the Isolated updater being used so far. */` ########## modules/core/src/main/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImpl.java: ########## @@ -489,12 +486,29 @@ public IgniteInternalFuture<?> internalFuture() { @Override public void receiver(StreamReceiver<K, V> rcvr) { A.notNull(rcvr, "rcvr"); - this.rcvr = rcvr; + DataStreamerBuiltInUpdater builtIn = DataStreamerBuiltInUpdater.of(rcvr); + + rcvrMsg = builtIn != null ? builtIn.message() : new DataStreamerReceiverMessage(rcvr); + } + + /** @return Message of the receiver in use, a new one naming the Isolated updater until a receiver is set. */ Review Comment: Done — custom/built-in and "sent as a name" in the streamer javadocs too. ########## modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java: ########## @@ -142,6 +148,71 @@ 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 { + startGridsAndStream(new TestReceiver()); + + assertEquals("The receiver was marshalled more than once", 1, sentUpdaters.size()); + + assertTrue("The receiver was named instead of sent", F.first(sentUpdaters) instanceof byte[]); Review Comment: Same in the test: the assertions read "Expected ISOLATED to be sent as a name" and "A custom receiver must be sent, not named". ########## modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java: ########## @@ -88,6 +91,9 @@ public class DataStreamerImplSelfTest extends GridCommonAbstractTest { /** Indicates whether we need to make the topology stale */ private static boolean needStaleTop = false; + /** Distinct updaters sent since the current test started: the serialized bytes, or the built-in constant. */ + private final Set<Object> sentUpdaters = Collections.synchronizedSet(new HashSet<>()); Review Comment: Not needed — a plain `HashSet` already tells the updaters apart. `byte[]` inherits identity equality, so two marshallings of the same receiver are two elements, while the same array is one; and the built-in ones are enum constants. That is exactly the "distinct" we want, and it keeps the assertions to a single equality. ########## modules/core/src/test/java/org/apache/ignite/internal/processors/datastreamer/DataStreamerImplSelfTest.java: ########## @@ -142,6 +148,71 @@ 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 { + startGridsAndStream(new TestReceiver()); + + assertEquals("The receiver was marshalled more than once", 1, sentUpdaters.size()); + + assertTrue("The receiver was named instead of sent", F.first(sentUpdaters) instanceof byte[]); + } + + /** + * Every built-in updater is named rather than sent, and the data still lands. + * + * @throws Exception If failed. + */ + @Test + public void testBuiltInUpdaterIsNotSent() throws Exception { + for (DataStreamerBuiltInUpdater builtIn : DataStreamerBuiltInUpdater.values()) { + startGridsAndStream(builtIn.updater()); + + assertEquals("Expected " + builtIn + " to be named, not sent", Collections.singleton(builtIn), + sentUpdaters); + + IgniteCache<Object, Object> cache = grid(1).cache(DEFAULT_CACHE_NAME); + + for (int i = 0; i < KEYS_COUNT; i++) + assertEquals(i, cache.get(i)); + + stopAllGrids(); + } + } + + /** + * Starts two nodes and streams {@link #KEYS_COUNT} entries from the first one, a request per entry, collecting + * the updaters they carry. Waits for the partition map first: until it is ready every partition is primary here, + * and a streamer that overwrites sends nothing to the remote node. + * + * @param rcvr Receiver to stream with. + * @throws Exception If failed. + */ + @SuppressWarnings("unchecked") + private void startGridsAndStream(StreamReceiver<?, ?> rcvr) throws Exception { + cnt = 0; + + startGrids(2); + + awaitPartitionMapExchange(); Review Comment: Yes, for the same reason as before: until the partition map is ready every partition is primary on the node we stream from, so a streamer that overwrites — which is the case as soon as a receiver is set — sends nothing to the remote node and the test sees no requests at all. It is stated in the helper javadoc. -- 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]
