Vladsz83 commented on code in PR #13095: URL: https://github.com/apache/ignite/pull/13095#discussion_r3656497810
########## modules/core/src/test/java/org/apache/ignite/plugin/extensions/communication/MessageMarshalOnceTest.java: ########## @@ -0,0 +1,256 @@ +/* + * 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.plugin.extensions.communication; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import org.apache.ignite.IgniteException; +import org.apache.ignite.cluster.ClusterNode; +import org.apache.ignite.configuration.IgniteConfiguration; +import org.apache.ignite.internal.CoreMessagesProvider; +import org.apache.ignite.internal.GridKernalContext; +import org.apache.ignite.internal.GridTopic; +import org.apache.ignite.internal.TestRecordingCommunicationSpi; +import org.apache.ignite.internal.managers.communication.GridIoMessage; +import org.apache.ignite.internal.managers.communication.GridIoPolicy; +import org.apache.ignite.internal.processors.cache.CacheObjectContext; +import org.apache.ignite.internal.processors.cache.GridCacheMessage; +import org.apache.ignite.lang.IgniteInClosure; +import org.apache.ignite.plugin.AbstractTestPluginProvider; +import org.apache.ignite.plugin.ExtensionRegistry; +import org.apache.ignite.plugin.PluginContext; +import org.apache.ignite.spi.IgniteSpiException; +import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi; +import org.apache.ignite.testframework.GridTestUtils; +import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest; +import org.junit.Test; + +/** + * Verifies the marshal-once-before-fan-out contract of the collection {@code send}: a message broadcast to N remote + * nodes is marshalled <b>exactly once</b> (prepared before the fan-out and reused for every destination), not once per + * destination. A counting marshaller is registered for a test message; broadcasting it to {@link #RMT_CNT} remote nodes + * must produce {@code RMT_CNT} sends but a single {@code marshal}. Counting the sends keeps the check honest (a + * lone send would also marshal once). The unmarshal-once counterpart lives in {@link MessageUnmarshalOnceTest}. + * + * <p>The second scenario covers the send-retry path: a cache message whose first send attempt fails is retried, + * and the retry must not re-marshal the payload (it is prepared once by {@code GridCacheIoManager} before the + * retry loop, while {@code GridIoManager} marshals only the wrapper). + */ +public class MessageMarshalOnceTest extends GridCommonAbstractTest { Review Comment: Join with `MessageUnmarshalOnceTest`? ########## modules/core/src/test/java/org/apache/ignite/plugin/extensions/communication/MessageMarshalOnceTest.java: ########## @@ -0,0 +1,256 @@ +/* + * 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.plugin.extensions.communication; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import org.apache.ignite.IgniteException; +import org.apache.ignite.cluster.ClusterNode; +import org.apache.ignite.configuration.IgniteConfiguration; +import org.apache.ignite.internal.CoreMessagesProvider; +import org.apache.ignite.internal.GridKernalContext; +import org.apache.ignite.internal.GridTopic; +import org.apache.ignite.internal.TestRecordingCommunicationSpi; +import org.apache.ignite.internal.managers.communication.GridIoMessage; +import org.apache.ignite.internal.managers.communication.GridIoPolicy; +import org.apache.ignite.internal.processors.cache.CacheObjectContext; +import org.apache.ignite.internal.processors.cache.GridCacheMessage; +import org.apache.ignite.lang.IgniteInClosure; +import org.apache.ignite.plugin.AbstractTestPluginProvider; +import org.apache.ignite.plugin.ExtensionRegistry; +import org.apache.ignite.plugin.PluginContext; +import org.apache.ignite.spi.IgniteSpiException; +import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi; +import org.apache.ignite.testframework.GridTestUtils; +import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest; +import org.junit.Test; + +/** + * Verifies the marshal-once-before-fan-out contract of the collection {@code send}: a message broadcast to N remote + * nodes is marshalled <b>exactly once</b> (prepared before the fan-out and reused for every destination), not once per + * destination. A counting marshaller is registered for a test message; broadcasting it to {@link #RMT_CNT} remote nodes + * must produce {@code RMT_CNT} sends but a single {@code marshal}. Counting the sends keeps the check honest (a + * lone send would also marshal once). The unmarshal-once counterpart lives in {@link MessageUnmarshalOnceTest}. + * + * <p>The second scenario covers the send-retry path: a cache message whose first send attempt fails is retried, + * and the retry must not re-marshal the payload (it is prepared once by {@code GridCacheIoManager} before the + * retry loop, while {@code GridIoManager} marshals only the wrapper). + */ +public class MessageMarshalOnceTest extends GridCommonAbstractTest { + /** Direct type for the test message, past the core range. */ + private static final short TYPE = (short)(CoreMessagesProvider.MAX_MESSAGE_ID + 1); + + /** Direct type for the retry-check cache message. */ + private static final short RETRY_TYPE = (short)(CoreMessagesProvider.MAX_MESSAGE_ID + 2); + + /** Number of remote destinations to broadcast to (a single marshal must serve all of them). */ + private static final int RMT_CNT = 4; + + /** Counts {@code marshal} invocations of {@link MarshalOnceCheckMessage} across the JVM. */ + private static final AtomicInteger MARSHAL_CNT = new AtomicInteger(); + + /** Counts {@code marshal} invocations of {@link RetryCheckMessage} across the JVM. */ + private static final AtomicInteger RETRY_MARSHAL_CNT = new AtomicInteger(); + + /** When {@code true}, nodes are configured with the SPI failing the first send of {@link RetryCheckMessage}. */ + private boolean failFirstSend; + + /** {@inheritDoc} */ + @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception { + IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName); + + cfg.setCommunicationSpi(failFirstSend ? new FailFirstSendSpi() : new TestRecordingCommunicationSpi()); + + cfg.setPluginProviders(new AbstractTestPluginProvider() { + @Override public String name() { + return "marshal-once-test"; + } + + @Override public void initExtensions(PluginContext ctx, ExtensionRegistry registry) { + registry.registerExtension(MessageFactoryProvider.class, factory -> { + factory.register(TYPE, MarshalOnceCheckMessage::new, new Serializer(), new CountingMarshaller()); + factory.register(RETRY_TYPE, RetryCheckMessage::new, new RetrySerializer(), new RetryCountingMarshaller()); + }); + } + }); + + return cfg; + } + + /** {@inheritDoc} */ + @Override protected void afterTest() throws Exception { + stopAllGrids(); + + super.afterTest(); + } + + /** @throws Exception If failed. */ + @Test + public void testBroadcastMarshalsExactlyOnce() throws Exception { + startGrids(RMT_CNT + 1); + + TestRecordingCommunicationSpi commSpi = TestRecordingCommunicationSpi.spi(grid(0)); + + commSpi.record(MarshalOnceCheckMessage.class); + + List<ClusterNode> rmts = new ArrayList<>(RMT_CNT); + + for (int i = 1; i <= RMT_CNT; i++) + rmts.add(grid(i).localNode()); + + MARSHAL_CNT.set(0); + + grid(0).context().io().sendToGridTopic(rmts, GridTopic.TOPIC_IO_TEST, new MarshalOnceCheckMessage(), + GridIoPolicy.PUBLIC_POOL); + + assertEquals("Broadcast must be sent to all " + RMT_CNT + " nodes", RMT_CNT, commSpi.recordedMessages(true).size()); + + assertEquals("A message broadcast to " + RMT_CNT + " nodes must be marshalled exactly once, not per destination", + 1, MARSHAL_CNT.get()); + } + + /** @throws Exception If failed. */ + @Test + public void testRetryMarshalsExactlyOnce() throws Exception { + failFirstSend = true; + + startGrids(2); + + RETRY_MARSHAL_CNT.set(0); Review Comment: Can it be not 0 here? ########## modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java: ########## @@ -1184,7 +1198,29 @@ private void processDhtTxPrepareRequest(final UUID nodeId, final GridDhtTxPrepar if (nearTx != null) res.nearEvicted(nearTx.evicted()); - List<IgniteTxKey> writesCacheMissed = req.nearWritesCacheMissed(); + List<IgniteTxKey> writesCacheMissed = new ArrayList<>(); Review Comment: Why this change? ########## modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/message/CalciteMessageUnmarshalThreadIntegrationTest.java: ########## @@ -0,0 +1,92 @@ +/* + * 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.query.calcite.message; + +import java.io.IOException; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; +import java.io.Serializable; +import java.util.List; +import java.util.Queue; +import java.util.concurrent.ConcurrentLinkedQueue; +import org.apache.ignite.internal.processors.query.calcite.exec.task.AbstractQueryTaskExecutor; +import org.apache.ignite.internal.processors.query.calcite.integration.AbstractBasicIntegrationTest; +import org.junit.Test; + +/** + * Verifies that payload of calcite messages is unmarshalled on the query task executor, not on a NIO thread: probe + * objects travel as {@code QueryStartRequest} parameters and as {@code QueryBatchMessage} rows, and record the thread + * that deserializes them. + */ +public class CalciteMessageUnmarshalThreadIntegrationTest extends AbstractBasicIntegrationTest { Review Comment: Can we put it in `CalciteCommunicationMessageSerializationTest`? ########## modules/codegen/src/main/java/org/apache/ignite/internal/MessageProcessor.java: ########## @@ -97,21 +110,32 @@ public class MessageProcessor extends AbstractProcessor { static final String[] SKIP_MESSAGES = { Review Comment: Maybe we should do smth. like `registerExclusionMessages(Class<? extents Message> msgCls)` instead ########## modules/core/src/test/java/org/apache/ignite/plugin/extensions/communication/MessageMarshalOnceTest.java: ########## @@ -0,0 +1,256 @@ +/* + * 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.plugin.extensions.communication; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import org.apache.ignite.IgniteException; +import org.apache.ignite.cluster.ClusterNode; +import org.apache.ignite.configuration.IgniteConfiguration; +import org.apache.ignite.internal.CoreMessagesProvider; +import org.apache.ignite.internal.GridKernalContext; +import org.apache.ignite.internal.GridTopic; +import org.apache.ignite.internal.TestRecordingCommunicationSpi; +import org.apache.ignite.internal.managers.communication.GridIoMessage; +import org.apache.ignite.internal.managers.communication.GridIoPolicy; +import org.apache.ignite.internal.processors.cache.CacheObjectContext; +import org.apache.ignite.internal.processors.cache.GridCacheMessage; +import org.apache.ignite.lang.IgniteInClosure; +import org.apache.ignite.plugin.AbstractTestPluginProvider; +import org.apache.ignite.plugin.ExtensionRegistry; +import org.apache.ignite.plugin.PluginContext; +import org.apache.ignite.spi.IgniteSpiException; +import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi; +import org.apache.ignite.testframework.GridTestUtils; +import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest; +import org.junit.Test; + +/** + * Verifies the marshal-once-before-fan-out contract of the collection {@code send}: a message broadcast to N remote + * nodes is marshalled <b>exactly once</b> (prepared before the fan-out and reused for every destination), not once per + * destination. A counting marshaller is registered for a test message; broadcasting it to {@link #RMT_CNT} remote nodes + * must produce {@code RMT_CNT} sends but a single {@code marshal}. Counting the sends keeps the check honest (a + * lone send would also marshal once). The unmarshal-once counterpart lives in {@link MessageUnmarshalOnceTest}. + * + * <p>The second scenario covers the send-retry path: a cache message whose first send attempt fails is retried, + * and the retry must not re-marshal the payload (it is prepared once by {@code GridCacheIoManager} before the + * retry loop, while {@code GridIoManager} marshals only the wrapper). + */ +public class MessageMarshalOnceTest extends GridCommonAbstractTest { + /** Direct type for the test message, past the core range. */ + private static final short TYPE = (short)(CoreMessagesProvider.MAX_MESSAGE_ID + 1); + + /** Direct type for the retry-check cache message. */ + private static final short RETRY_TYPE = (short)(CoreMessagesProvider.MAX_MESSAGE_ID + 2); + + /** Number of remote destinations to broadcast to (a single marshal must serve all of them). */ + private static final int RMT_CNT = 4; + + /** Counts {@code marshal} invocations of {@link MarshalOnceCheckMessage} across the JVM. */ + private static final AtomicInteger MARSHAL_CNT = new AtomicInteger(); + + /** Counts {@code marshal} invocations of {@link RetryCheckMessage} across the JVM. */ + private static final AtomicInteger RETRY_MARSHAL_CNT = new AtomicInteger(); + + /** When {@code true}, nodes are configured with the SPI failing the first send of {@link RetryCheckMessage}. */ + private boolean failFirstSend; + + /** {@inheritDoc} */ + @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception { + IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName); + + cfg.setCommunicationSpi(failFirstSend ? new FailFirstSendSpi() : new TestRecordingCommunicationSpi()); + + cfg.setPluginProviders(new AbstractTestPluginProvider() { + @Override public String name() { + return "marshal-once-test"; + } + + @Override public void initExtensions(PluginContext ctx, ExtensionRegistry registry) { + registry.registerExtension(MessageFactoryProvider.class, factory -> { + factory.register(TYPE, MarshalOnceCheckMessage::new, new Serializer(), new CountingMarshaller()); + factory.register(RETRY_TYPE, RetryCheckMessage::new, new RetrySerializer(), new RetryCountingMarshaller()); + }); + } + }); + + return cfg; + } + + /** {@inheritDoc} */ + @Override protected void afterTest() throws Exception { + stopAllGrids(); + + super.afterTest(); + } + + /** @throws Exception If failed. */ + @Test + public void testBroadcastMarshalsExactlyOnce() throws Exception { + startGrids(RMT_CNT + 1); + + TestRecordingCommunicationSpi commSpi = TestRecordingCommunicationSpi.spi(grid(0)); + + commSpi.record(MarshalOnceCheckMessage.class); + + List<ClusterNode> rmts = new ArrayList<>(RMT_CNT); + + for (int i = 1; i <= RMT_CNT; i++) + rmts.add(grid(i).localNode()); + + MARSHAL_CNT.set(0); Review Comment: Can it be not 0 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
