diff --git a/modules/control-utility/src/test/java/org/apache/ignite/testsuites/IgniteControlUtilityTestSuite2.java b/modules/control-utility/src/test/java/org/apache/ignite/testsuites/IgniteControlUtilityTestSuite2.java index eae0f1e86dcbb..589dd53dcebc9 100644 --- a/modules/control-utility/src/test/java/org/apache/ignite/testsuites/IgniteControlUtilityTestSuite2.java +++ b/modules/control-utility/src/test/java/org/apache/ignite/testsuites/IgniteControlUtilityTestSuite2.java @@ -34,6 +34,7 @@ import org.apache.ignite.util.GridCommandHandlerScheduleIndexRebuildTest; import org.apache.ignite.util.GridCommandHandlerTracingConfigurationTest; import org.apache.ignite.util.IdleVerifyDumpTest; +import org.apache.ignite.util.IoTestCommandTest; import org.apache.ignite.util.MetricCommandTest; import org.apache.ignite.util.PerformanceStatisticsCommandTest; import org.apache.ignite.util.SystemViewCommandTest; @@ -64,6 +65,7 @@ SystemViewCommandTest.class, MetricCommandTest.class, + IoTestCommandTest.class, PerformanceStatisticsCommandTest.class, CacheMetricsCommandTest.class, diff --git a/modules/control-utility/src/test/java/org/apache/ignite/util/IoTestCommandTest.java b/modules/control-utility/src/test/java/org/apache/ignite/util/IoTestCommandTest.java new file mode 100644 index 0000000000000..18fa9545ff64e --- /dev/null +++ b/modules/control-utility/src/test/java/org/apache/ignite/util/IoTestCommandTest.java @@ -0,0 +1,109 @@ +/* + * 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.util; + +import org.apache.ignite.internal.IgniteEx; +import org.apache.ignite.internal.management.io.IoTestCommand; +import org.junit.Test; + +import static org.apache.ignite.internal.commandline.CommandHandler.EXIT_CODE_INVALID_ARGUMENTS; +import static org.apache.ignite.internal.commandline.CommandHandler.EXIT_CODE_OK; +import static org.apache.ignite.util.SystemViewCommandTest.NODE_ID; + +/** + * Tests for the {@link IoTestCommand}. + */ +public class IoTestCommandTest extends GridCommandHandlerAbstractTest { + /** {@inheritDoc} */ + @Override protected void afterTest() throws Exception { + super.afterTest(); + + stopAllGrids(); + } + + /** */ + @Test + public void testCommunication() throws Exception { + IgniteEx srv = startGrids(2); + + executeCommand( + EXIT_CODE_OK, + "--io-test", "communication", + NODE_ID, srv.localNode().id().toString(), + "--warmup", "0", + "--duration", "100", + "--threads", "1", + "--payload-size", "32" + ); + + String output = String.valueOf(lastOperationResult); + + assertTrue(output, output.contains("Communication SPI test")); + assertTrue(output, output.contains("Source node: " + srv.localNode().id())); + assertTrue(output, output.contains("Payload: 32 bytes each way")); + assertTrue(output, output.contains("Message handling: system pool")); + assertTrue(output, output.contains("Target node: " + grid(1).localNode().id())); + assertFalse(output, output.contains("Target node: " + srv.localNode().id())); + assertTrue(output, output.contains("End-to-end RTT (ms):")); + assertTrue(output, output.contains("Node-local delays (ms, min/avg/max):")); + assertTrue(output, output.contains("Source request pre-serialization delay:")); + assertTrue(output, output.contains("Target request dispatch delay:")); + assertTrue(output, output.contains("Estimated one-way message delay (ms, min/avg/max):")); + assertTrue(output, output.contains("Clock assumption: synchronized wall clocks")); + assertTrue(output, output.contains("Request (serialization start -> deserialization complete):")); + assertTrue(output, output.contains("Response (serialization start -> deserialization complete):")); + } + + /** */ + @Test + public void testDiscovery() throws Exception { + startGrids(3); + + executeCommand( + EXIT_CODE_OK, + "--io-test", "discovery", + "--samples", "3", + "--interval", "10", + "--payload-size", "32" + ); + + String output = String.valueOf(lastOperationResult); + + assertTrue(output, output.contains("TcpDiscoverySpi ring test")); + assertTrue(output, output.contains("Coordinator: " + grid(0).localNode().id())); + assertTrue(output, output.contains("Samples: 3 | Inter-sample delay: 10 ms")); + assertTrue(output, output.contains("Request application payload: 32 bytes")); + assertTrue(output, output.contains("Server ring path:")); + assertTrue(output, output.contains(grid(0).localNode().id().toString())); + assertTrue(output, output.contains(grid(1).localNode().id().toString())); + assertTrue(output, output.contains(grid(2).localNode().id().toString())); + assertTrue(output, output.contains("Request ring latency (submission -> local ACK, ms):")); + assertTrue(output, output.contains("Estimated per-hop one-way message delay (ms, min/avg/max):")); + assertTrue(output, output.contains("Clock assumption: synchronized wall clocks")); + } + + /** */ + @Test + public void testInvalidPayloadSize() { + executeCommand( + EXIT_CODE_INVALID_ARGUMENTS, + "--io-test", "discovery", + "--payload-size", "65537" + ); + } +} 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 69e92e29ce0fe..862c98b96f6a8 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 @@ -34,6 +34,8 @@ import org.apache.ignite.internal.managers.deployment.GridDeploymentInfoBean; import org.apache.ignite.internal.managers.deployment.GridDeploymentRequest; import org.apache.ignite.internal.managers.deployment.GridDeploymentResponse; +import org.apache.ignite.internal.managers.discovery.IoTestDiscoveryAckMessage; +import org.apache.ignite.internal.managers.discovery.IoTestDiscoveryMessage; import org.apache.ignite.internal.managers.encryption.ChangeCacheEncryptionRequest; import org.apache.ignite.internal.managers.encryption.EncryptionDataBagItem; import org.apache.ignite.internal.managers.encryption.GenerateEncryptionKeyRequest; @@ -479,6 +481,8 @@ public CoreMessagesProvider(Marshaller dfltMarsh, Marshaller schemaAwareMarsh, C withNoSchemaResolvedClassLoader(CacheJoinNodeDiscoveryData.class); withNoSchemaResolvedClassLoader(CacheReconnectInfo.class); withNoSchemaResolvedClassLoader(ClusterCacheGroupRecoveryData.class); + withNoSchema(IoTestDiscoveryMessage.class); + withNoSchema(IoTestDiscoveryAckMessage.class); // [10000 - 10200]: Transaction and lock related messages. Most of them originally comes from Communication. msgIdx = 10000; diff --git a/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java b/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java index 063774e7b13b5..e778aa9214a64 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java @@ -3382,26 +3382,6 @@ public void dumpDebugInfo() { } } - /** - * @param node Node. - * @param payload Message payload. - * @param procFromNioThread If {@code true} message is processed from NIO thread. - * @return Response future. - */ - public IgniteInternalFuture sendIoTest(ClusterNode node, byte[] payload, boolean procFromNioThread) { - return ctx.io().sendIoTest(node, payload, procFromNioThread); - } - - /** - * @param nodes Nodes. - * @param payload Message payload. - * @param procFromNioThread If {@code true} message is processed from NIO thread. - * @return Response future. - */ - public IgniteInternalFuture sendIoTest(List nodes, byte[] payload, boolean procFromNioThread) { - return ctx.io().sendIoTest(nodes, payload, procFromNioThread); - } - /** Registers configuration system view. */ private void registerConfigurationSystemView() { ctx.systemView().registerInnerCollectionView( diff --git a/modules/core/src/main/java/org/apache/ignite/internal/IgniteMXBeanImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/IgniteMXBeanImpl.java index fc4749be1d3f5..655c83a711d4d 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/IgniteMXBeanImpl.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/IgniteMXBeanImpl.java @@ -24,6 +24,7 @@ import javax.management.JMException; import org.apache.ignite.IgniteCheckedException; import org.apache.ignite.IgniteException; +import org.apache.ignite.IgniteLogger; import org.apache.ignite.cluster.ClusterNode; import org.apache.ignite.cluster.ClusterState; import org.apache.ignite.internal.util.typedef.internal.A; @@ -298,6 +299,7 @@ public IgniteMXBeanImpl(IgniteKernal kernal) { } /** {@inheritDoc} */ + @Deprecated @Override public void runIoTest( long warmup, long duration, @@ -307,8 +309,30 @@ public IgniteMXBeanImpl(IgniteKernal kernal) { int payLoadSize, boolean procFromNioThread ) { - ctx.io().runIoTest(warmup, duration, threads, maxLatency, rangesCnt, payLoadSize, - procFromNioThread, new ArrayList(ctx.cluster().get().forServers().forRemotes().nodes())); + IgniteInternalFuture fut = ctx.io().ioTest().runIoTest( + warmup, + duration, + threads, + maxLatency, + rangesCnt, + payLoadSize, + procFromNioThread, + new ArrayList<>(ctx.cluster().get().forServers().forRemotes().nodes()) + ); + + fut.listen(f -> { + IgniteLogger log = ctx.log(ctx.io().getClass()); + + try { + String res = f.get(); + + if (log.isInfoEnabled()) + log.info(res); + } + catch (IgniteCheckedException e) { + U.error(log, "IO test failed.", e); + } + }); } /** {@inheritDoc} */ diff --git a/modules/core/src/main/java/org/apache/ignite/internal/management/IgniteCommandRegistry.java b/modules/core/src/main/java/org/apache/ignite/internal/management/IgniteCommandRegistry.java index f7c9ea9c57fcb..b3cc0c5b8c19d 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/management/IgniteCommandRegistry.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/management/IgniteCommandRegistry.java @@ -30,6 +30,7 @@ import org.apache.ignite.internal.management.diagnostic.DiagnosticCommand; import org.apache.ignite.internal.management.encryption.EncryptionCommand; import org.apache.ignite.internal.management.event.EventCommand; +import org.apache.ignite.internal.management.io.IoTestCommand; import org.apache.ignite.internal.management.kill.KillCommand; import org.apache.ignite.internal.management.meta.MetaCommand; import org.apache.ignite.internal.management.metric.MetricCommand; @@ -77,7 +78,8 @@ public IgniteCommandRegistry() { new PerformanceStatisticsCommand(), new CdcCommand(), new ConsistencyCommand(), - new EventCommand() + new EventCommand(), + new IoTestCommand() ); U.loadService(CommandsProvider.class).forEach(p -> p.commands().forEach(this::register)); diff --git a/modules/core/src/main/java/org/apache/ignite/internal/management/io/IoTestCommand.java b/modules/core/src/main/java/org/apache/ignite/internal/management/io/IoTestCommand.java new file mode 100644 index 0000000000000..354fedf2d9d94 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/management/io/IoTestCommand.java @@ -0,0 +1,31 @@ +/* + * 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.management.io; + +import org.apache.ignite.internal.management.api.CommandRegistryImpl; + +/** */ +public class IoTestCommand extends CommandRegistryImpl { + /** */ + public IoTestCommand() { + super( + new IoTestCommunicationCommand(), + new IoTestDiscoveryCommand() + ); + } +} diff --git a/modules/core/src/main/java/org/apache/ignite/internal/management/io/IoTestCommunicationCommand.java b/modules/core/src/main/java/org/apache/ignite/internal/management/io/IoTestCommunicationCommand.java new file mode 100644 index 0000000000000..4b38f643a2312 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/management/io/IoTestCommunicationCommand.java @@ -0,0 +1,61 @@ +/* + * 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.management.io; + +import java.util.Collection; +import java.util.function.Consumer; +import org.apache.ignite.cluster.ClusterNode; +import org.apache.ignite.internal.management.api.ComputeCommand; +import org.apache.ignite.internal.util.typedef.F; +import org.jetbrains.annotations.Nullable; + +import static org.apache.ignite.internal.management.api.CommandUtils.node; + +/** */ +public class IoTestCommunicationCommand implements ComputeCommand { + /** {@inheritDoc} */ + @Override public String description() { + return "Tests Communication SPI latency to all remote server nodes."; + } + + /** {@inheritDoc} */ + @Override public Class argClass() { + return IoTestCommunicationCommandArg.class; + } + + /** {@inheritDoc} */ + @Override public Class taskClass() { + return IoTestCommunicationTask.class; + } + + /** {@inheritDoc} */ + @Override public @Nullable Collection nodes( + Collection nodes, + IoTestCommunicationCommandArg arg + ) { + return node(arg.nodeId(), nodes); + } + + /** {@inheritDoc} */ + @Override public void printResult(IoTestCommunicationCommandArg arg, String res, Consumer printer) { + if (F.isEmpty(res)) + printer.accept("Failed to run test. See logs for details."); + else + printer.accept(res); + } +} diff --git a/modules/core/src/main/java/org/apache/ignite/internal/management/io/IoTestCommunicationCommandArg.java b/modules/core/src/main/java/org/apache/ignite/internal/management/io/IoTestCommunicationCommandArg.java new file mode 100644 index 0000000000000..c0a54dd75b7b3 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/management/io/IoTestCommunicationCommandArg.java @@ -0,0 +1,180 @@ +/* + * 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.management.io; + +import java.util.UUID; +import java.util.concurrent.TimeUnit; +import org.apache.ignite.internal.Order; +import org.apache.ignite.internal.dto.IgniteDataTransferObject; +import org.apache.ignite.internal.management.api.Argument; +import org.apache.ignite.internal.util.typedef.internal.A; + +/** */ +public class IoTestCommunicationCommandArg extends IgniteDataTransferObject { + /** */ + private static final long serialVersionUID = 0; + + /** Maximum number of test threads. */ + private static final int MAX_THREADS = 64; + + /** Maximum number of histogram ranges. */ + private static final int MAX_RANGES_COUNT = 1_000; + + /** Maximum payload size. */ + private static final int MAX_PAYLOAD_SIZE = 1024 * 1024; + + /** Maximum warmup or test duration. */ + private static final long MAX_PHASE_DURATION = TimeUnit.HOURS.toMillis(1); + + /** */ + @Order(0) + @Argument(description = "Source node ID.") + UUID nodeId; + + /** */ + @Order(1) + @Argument(optional = true, description = "Warmup duration (millis, max 1 hour).") + long warmup = TimeUnit.SECONDS.toMillis(5); + + /** */ + @Order(2) + @Argument(optional = true, description = "Test duration (millis, max 1 hour).") + long duration = TimeUnit.SECONDS.toMillis(30); + + /** */ + @Order(3) + @Argument(optional = true, description = "Number of test threads (max 64).") + int threads = 1; + + /** */ + @Order(4) + @Argument(optional = true, description = "RTT histogram upper bound (nanos).") + long maxLatency = TimeUnit.MILLISECONDS.toNanos(100); + + /** */ + @Order(5) + @Argument(optional = true, description = "Ranges count for RTT histogram (max 1000).") + int rangesCnt = 5; + + /** */ + @Order(6) + @Argument(optional = true, description = "Payload size in each direction (bytes, max 1 MiB).") + int payloadSize; + + /** */ + @Order(7) + @Argument(optional = true, description = "Process requests and responses in NIO threads.") + boolean procFromNioThread; + + /** */ + public UUID nodeId() { + return nodeId; + } + + /** */ + public void nodeId(UUID nodeId) { + this.nodeId = nodeId; + } + + /** */ + public long warmup() { + return warmup; + } + + /** */ + public void warmup(long warmup) { + A.ensure(warmup >= 0 && warmup <= MAX_PHASE_DURATION, + "warmup must be between 0 and " + MAX_PHASE_DURATION); + + this.warmup = warmup; + } + + /** */ + public long duration() { + return duration; + } + + /** */ + public void duration(long duration) { + A.ensure(duration > 0 && duration <= MAX_PHASE_DURATION, + "duration must be between 1 and " + MAX_PHASE_DURATION); + + this.duration = duration; + } + + /** */ + public int threads() { + return threads; + } + + /** */ + public void threads(int threads) { + A.ensure(threads > 0 && threads <= MAX_THREADS, + "threads must be between 1 and " + MAX_THREADS); + + this.threads = threads; + } + + /** */ + public long maxLatency() { + return maxLatency; + } + + /** */ + public void maxLatency(long maxLatency) { + A.ensure(maxLatency > 0, "maxLatency must be > 0"); + + this.maxLatency = maxLatency; + } + + /** */ + public int rangesCnt() { + return rangesCnt; + } + + /** */ + public void rangesCnt(int rangesCnt) { + A.ensure(rangesCnt > 0 && rangesCnt <= MAX_RANGES_COUNT, + "rangesCnt must be between 1 and " + MAX_RANGES_COUNT); + + this.rangesCnt = rangesCnt; + } + + /** */ + public int payloadSize() { + return payloadSize; + } + + /** */ + public void payloadSize(int payloadSize) { + A.ensure(payloadSize >= 0 && payloadSize <= MAX_PAYLOAD_SIZE, + "payloadSize must be between 0 and " + MAX_PAYLOAD_SIZE); + + this.payloadSize = payloadSize; + } + + /** */ + public boolean procFromNioThread() { + return procFromNioThread; + } + + /** */ + public void procFromNioThread(boolean procFromNioThread) { + this.procFromNioThread = procFromNioThread; + } +} diff --git a/modules/core/src/main/java/org/apache/ignite/internal/management/io/IoTestCommunicationTask.java b/modules/core/src/main/java/org/apache/ignite/internal/management/io/IoTestCommunicationTask.java new file mode 100644 index 0000000000000..7f7628c13b7c9 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/management/io/IoTestCommunicationTask.java @@ -0,0 +1,114 @@ +/* + * 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.management.io; + +import java.util.ArrayList; +import java.util.List; +import org.apache.ignite.IgniteCheckedException; +import org.apache.ignite.IgniteException; +import org.apache.ignite.cluster.ClusterNode; +import org.apache.ignite.internal.IgniteInternalFuture; +import org.apache.ignite.internal.processors.task.GridInternal; +import org.apache.ignite.internal.visor.VisorJob; +import org.apache.ignite.internal.visor.VisorOneNodeTask; + +/** */ +@GridInternal +public class IoTestCommunicationTask extends VisorOneNodeTask { + /** Serial version uid. */ + private static final long serialVersionUID = 0L; + + /** {@inheritDoc} */ + @Override protected VisorJob job(IoTestCommunicationCommandArg arg) { + return new IoTestJob(arg, false); + } + + /** */ + private static class IoTestJob extends VisorJob { + /** */ + private static final long serialVersionUID = 0L; + + /** Running test. */ + private transient volatile IgniteInternalFuture testFut; + + /** + * Create job with specified argument. + * + * @param arg Job argument. + * @param debug Flag indicating whether debug information should be printed into node log. + */ + protected IoTestJob(IoTestCommunicationCommandArg arg, boolean debug) { + super(arg, debug); + } + + /** {@inheritDoc} */ + @Override protected String run(IoTestCommunicationCommandArg arg) throws IgniteException { + List nodes = new ArrayList<>(ignite.cluster().forServers().forRemotes().nodes()); + + if (nodes.isEmpty()) + throw new IgniteException("No remote server nodes found."); + + testFut = ignite.context().io().ioTest().runIoTest( + arg.warmup(), + arg.duration(), + arg.threads(), + arg.maxLatency(), + arg.rangesCnt(), + arg.payloadSize(), + arg.procFromNioThread(), + nodes + ); + + try { + if (isCancelled()) + testFut.cancel(); + + return testFut.get(); + } + catch (IgniteCheckedException e) { + try { + testFut.cancel(); + } + catch (IgniteCheckedException cancelErr) { + e.addSuppressed(cancelErr); + } + + throw new IgniteException("Communication SPI test failed.", e); + } + finally { + testFut = null; + } + } + + /** {@inheritDoc} */ + @Override public void cancel() { + super.cancel(); + + IgniteInternalFuture fut = testFut; + + if (fut != null) { + try { + fut.cancel(); + } + catch (IgniteCheckedException ignored) { + // No-op. + } + } + } + } +} diff --git a/modules/core/src/main/java/org/apache/ignite/internal/management/io/IoTestDiscoveryCommand.java b/modules/core/src/main/java/org/apache/ignite/internal/management/io/IoTestDiscoveryCommand.java new file mode 100644 index 0000000000000..a816955a016f1 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/management/io/IoTestDiscoveryCommand.java @@ -0,0 +1,61 @@ +/* + * 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.management.io; + +import java.util.Collection; +import java.util.function.Consumer; +import org.apache.ignite.cluster.ClusterNode; +import org.apache.ignite.internal.management.api.ComputeCommand; +import org.apache.ignite.internal.util.typedef.F; +import org.jetbrains.annotations.Nullable; + +import static org.apache.ignite.internal.management.api.CommandUtils.coordinatorOrNull; + +/** */ +public class IoTestDiscoveryCommand implements ComputeCommand { + /** {@inheritDoc} */ + @Override public String description() { + return "Tests how custom messages traverse the TcpDiscoverySpi ring."; + } + + /** {@inheritDoc} */ + @Override public Class argClass() { + return IoTestDiscoveryCommandArg.class; + } + + /** {@inheritDoc} */ + @Override public Class taskClass() { + return IoTestDiscoveryTask.class; + } + + /** {@inheritDoc} */ + @Override public @Nullable Collection nodes( + Collection nodes, + IoTestDiscoveryCommandArg arg + ) { + return coordinatorOrNull(nodes); + } + + /** {@inheritDoc} */ + @Override public void printResult(IoTestDiscoveryCommandArg arg, String res, Consumer printer) { + if (F.isEmpty(res)) + printer.accept("Failed to run test. See logs for details."); + else + printer.accept(res); + } +} diff --git a/modules/core/src/main/java/org/apache/ignite/internal/management/io/IoTestDiscoveryCommandArg.java b/modules/core/src/main/java/org/apache/ignite/internal/management/io/IoTestDiscoveryCommandArg.java new file mode 100644 index 0000000000000..f678c2ffa2745 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/management/io/IoTestDiscoveryCommandArg.java @@ -0,0 +1,96 @@ +/* + * 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.management.io; + +import java.util.concurrent.TimeUnit; +import org.apache.ignite.internal.Order; +import org.apache.ignite.internal.dto.IgniteDataTransferObject; +import org.apache.ignite.internal.management.api.Argument; +import org.apache.ignite.internal.util.typedef.internal.A; + +/** Arguments of the discovery IO test. */ +public class IoTestDiscoveryCommandArg extends IgniteDataTransferObject { + /** */ + private static final long serialVersionUID = 0; + + /** Maximum payload size. */ + private static final int MAX_PAYLOAD_SIZE = 64 * 1024; + + /** Maximum number of samples. */ + private static final int MAX_SAMPLES = 100; + + /** Minimum interval between samples. */ + private static final long MIN_INTERVAL = 10; + + /** Maximum interval between samples. */ + private static final long MAX_INTERVAL = TimeUnit.MINUTES.toMillis(1); + + /** */ + @Order(0) + @Argument(optional = true, description = "Number of samples (max 100).") + int samples = 10; + + /** */ + @Order(1) + @Argument(optional = true, description = "Interval between samples (millis, 10 to 60000).") + long interval = 100; + + /** */ + @Order(2) + @Argument(optional = true, description = "Payload size (bytes, max 64 KiB).") + int payloadSize = 100; + + /** */ + public int samples() { + return samples; + } + + /** */ + public void samples(int samples) { + A.ensure(samples > 0 && samples <= MAX_SAMPLES, + "samples must be between 1 and " + MAX_SAMPLES); + + this.samples = samples; + } + + /** */ + public long interval() { + return interval; + } + + /** */ + public void interval(long interval) { + A.ensure(interval >= MIN_INTERVAL && interval <= MAX_INTERVAL, + "interval must be between " + MIN_INTERVAL + " and " + MAX_INTERVAL); + + this.interval = interval; + } + + /** */ + public int payloadSize() { + return payloadSize; + } + + /** */ + public void payloadSize(int payloadSize) { + A.ensure(payloadSize >= 0 && payloadSize <= MAX_PAYLOAD_SIZE, + "payloadSize must be between 0 and " + MAX_PAYLOAD_SIZE); + + this.payloadSize = payloadSize; + } +} diff --git a/modules/core/src/main/java/org/apache/ignite/internal/management/io/IoTestDiscoveryTask.java b/modules/core/src/main/java/org/apache/ignite/internal/management/io/IoTestDiscoveryTask.java new file mode 100644 index 0000000000000..ee5d2866c2830 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/management/io/IoTestDiscoveryTask.java @@ -0,0 +1,61 @@ +/* + * 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.management.io; + +import org.apache.ignite.IgniteException; +import org.apache.ignite.internal.processors.task.GridInternal; +import org.apache.ignite.internal.visor.VisorJob; +import org.apache.ignite.internal.visor.VisorOneNodeTask; + +/** */ +@GridInternal +public class IoTestDiscoveryTask extends VisorOneNodeTask { + /** Serial version uid. */ + private static final long serialVersionUID = 0L; + + /** {@inheritDoc} */ + @Override protected VisorJob job(IoTestDiscoveryCommandArg arg) { + return new IoTestJob(arg, false); + } + + /** */ + private static class IoTestJob extends VisorJob { + /** */ + private static final long serialVersionUID = 0L; + + /** + * Create job with specified argument. + * + * @param arg Job argument. + * @param debug Flag indicating whether debug information should be printed into node log. + */ + protected IoTestJob(IoTestDiscoveryCommandArg arg, boolean debug) { + super(arg, debug); + } + + /** {@inheritDoc} */ + @Override protected String run(IoTestDiscoveryCommandArg arg) throws IgniteException { + return ignite.context().discovery().ioTest().runTest( + arg.samples(), + arg.interval(), + arg.payloadSize(), + this::isCancelled + ); + } + } +} diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoManager.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoManager.java index 34b4f45488a6b..1ce28fc0dda72 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoManager.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoManager.java @@ -33,7 +33,6 @@ import java.nio.channels.FileChannel; import java.nio.channels.SocketChannel; import java.nio.channels.WritableByteChannel; -import java.time.Instant; import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; @@ -41,28 +40,21 @@ import java.util.HashMap; import java.util.HashSet; import java.util.Iterator; -import java.util.List; import java.util.Map; import java.util.Map.Entry; import java.util.Objects; import java.util.Queue; import java.util.Set; import java.util.UUID; -import java.util.concurrent.BrokenBarrierException; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentLinkedDeque; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.ConcurrentMap; -import java.util.concurrent.CyclicBarrier; import java.util.concurrent.Executor; import java.util.concurrent.ExecutorService; import java.util.concurrent.RejectedExecutionException; -import java.util.concurrent.ThreadLocalRandom; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; -import java.util.concurrent.atomic.AtomicLong; -import java.util.concurrent.atomic.AtomicReference; -import java.util.concurrent.atomic.LongAdder; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReadWriteLock; import java.util.concurrent.locks.ReentrantLock; @@ -104,11 +96,7 @@ import org.apache.ignite.internal.processors.tracing.SpanTags; import org.apache.ignite.internal.thread.context.Scope; import org.apache.ignite.internal.util.GridBoundedConcurrentLinkedHashSet; -import org.apache.ignite.internal.util.IgniteUtils; import org.apache.ignite.internal.util.StripedCompositeReadWriteLock; -import org.apache.ignite.internal.util.future.GridFinishedFuture; -import org.apache.ignite.internal.util.future.GridFutureAdapter; -import org.apache.ignite.internal.util.lang.IgnitePair; import org.apache.ignite.internal.util.tostring.GridToStringExclude; import org.apache.ignite.internal.util.tostring.GridToStringInclude; import org.apache.ignite.internal.util.typedef.F; @@ -164,7 +152,6 @@ import static org.apache.ignite.internal.processors.tracing.SpanType.COMMUNICATION_REGULAR_PROCESS; import static org.apache.ignite.internal.processors.tracing.messages.TraceableMessagesTable.traceName; import static org.apache.ignite.internal.thread.pool.IgniteThreadPoolExecutor.newCachedThreadPool; -import static org.apache.ignite.internal.thread.pool.IgniteThreadPoolExecutor.newFixedThreadPool; import static org.apache.ignite.internal.util.lang.ClusterNodeFunc.localNode; import static org.apache.ignite.internal.util.lang.ClusterNodeFunc.remoteNodes; import static org.apache.ignite.internal.util.nio.GridNioBackPressureControl.threadProcessingMessage; @@ -351,18 +338,15 @@ public class GridIoManager extends GridManagerAdapter> /** Stopping flag. */ private volatile boolean stopping; - /** */ - private final AtomicReference> ioTestMap = new AtomicReference<>(); - - /** */ - private final AtomicLong ioTestId = new AtomicLong(); - /** */ private final TcpCommunicationInverseConnectionHandler invConnHandler = new TcpCommunicationInverseConnectionHandler(); /** No-op runnable. */ private static final IgniteRunnable NOOP = () -> {}; + /** */ + private IoTestHandler ioTestHnd; + /** * @param ctx Grid kernal context. */ @@ -490,44 +474,7 @@ public void resetMetrics() { if (log.isDebugEnabled()) log.debug(startInfo()); - addMessageListener(GridTopic.TOPIC_IO_TEST, new GridMessageListener() { - @Override public void onMessage(UUID nodeId, Object msg, byte plc) { - ClusterNode node = ctx.discovery().node(nodeId); - - if (node == null) - return; - - IgniteIoTestMessage msg0 = (IgniteIoTestMessage)msg; - - msg0.senderNodeId(nodeId); - - if (msg0.request()) { - IgniteIoTestMessage res = new IgniteIoTestMessage(msg0.id(), false, null); - - res.flags(msg0.flags()); - res.onRequestProcessed(); - - res.copyDataFromRequest(msg0); - - try { - sendToGridTopic(node, GridTopic.TOPIC_IO_TEST, res, GridIoPolicy.SYSTEM_POOL); - } - catch (IgniteCheckedException e) { - U.error(log, "Failed to send IO test response [msg=" + msg0 + "]", e); - } - } - else { - IoTestFuture fut = ioTestMap().get(msg0.id()); - - msg0.onResponseProcessed(); - - if (fut == null) - U.warn(log, "Failed to find IO test future [msg=" + msg0 + ']'); - else - fut.onResponse(msg0); - } - } - }); + ioTestHnd = new IoTestHandler(ctx); } /** {@inheritDoc} */ @@ -537,357 +484,6 @@ public void resetMetrics() { return super.onReconnected(clusterRestarted); } - /** - * @param nodes Nodes. - * @param payload Payload. - * @param procFromNioThread If {@code true} message is processed from NIO thread. - * @return Response future. - */ - public IgniteInternalFuture sendIoTest(List nodes, byte[] payload, boolean procFromNioThread) { - long id = ioTestId.getAndIncrement(); - - IoTestFuture fut = new IoTestFuture(id, nodes.size()); - - IgniteIoTestMessage msg = new IgniteIoTestMessage(id, true, payload); - - msg.processFromNioThread(procFromNioThread); - - ioTestMap().put(id, fut); - - for (int i = 0; i < nodes.size(); i++) { - ClusterNode node = nodes.get(i); - - try { - sendToGridTopic(node, GridTopic.TOPIC_IO_TEST, msg, GridIoPolicy.SYSTEM_POOL); - } - catch (IgniteCheckedException e) { - ioTestMap().remove(msg.id()); - - return new GridFinishedFuture(e); - } - } - - return fut; - } - - /** - * @param node Node. - * @param payload Payload. - * @param procFromNioThread If {@code true} message is processed from NIO thread. - * @return Response future. - */ - public IgniteInternalFuture> sendIoTest( - ClusterNode node, - byte[] payload, - boolean procFromNioThread - ) { - long id = ioTestId.getAndIncrement(); - - IoTestFuture fut = new IoTestFuture(id, 1); - - IgniteIoTestMessage msg = new IgniteIoTestMessage(id, true, payload); - - msg.processFromNioThread(procFromNioThread); - - ioTestMap().put(id, fut); - - try { - sendToGridTopic(node, GridTopic.TOPIC_IO_TEST, msg, GridIoPolicy.SYSTEM_POOL); - } - catch (IgniteCheckedException e) { - ioTestMap().remove(msg.id()); - - return new GridFinishedFuture(e); - } - - return fut; - } - - /** - * @return IO test futures map. - */ - private ConcurrentHashMap ioTestMap() { - ConcurrentHashMap map = ioTestMap.get(); - - if (map == null) { - if (!ioTestMap.compareAndSet(null, map = new ConcurrentHashMap<>())) - map = ioTestMap.get(); - } - - return map; - } - - /** - * @param warmup Warmup duration in milliseconds. - * @param duration Test duration in milliseconds. - * @param threads Thread count. - * @param latencyLimit Max latency in nanoseconds. - * @param rangesCnt Ranges count in resulting histogram. - * @param payLoadSize Payload size in bytes. - * @param procFromNioThread {@code True} to process requests in NIO threads. - * @param nodes Nodes participating in test. - */ - public void runIoTest( - final long warmup, - final long duration, - final int threads, - final long latencyLimit, - final int rangesCnt, - final int payLoadSize, - final boolean procFromNioThread, - final List nodes - ) { - ExecutorService svc = newFixedThreadPool("io-latency-inspector", ctx.igniteInstanceName(), threads + 1); - - final AtomicBoolean warmupFinished = new AtomicBoolean(); - final AtomicBoolean done = new AtomicBoolean(); - final CyclicBarrier bar = new CyclicBarrier(threads + 1); - final LongAdder cnt = new LongAdder(); - final long sleepDuration = 5000; - final byte[] payLoad = new byte[payLoadSize]; - final Map[] res = new Map[threads]; - - boolean failed = true; - - try { - svc.execute(new Runnable() { - @Override public void run() { - boolean failed = true; - - try { - bar.await(); - - long start = System.currentTimeMillis(); - - if (log.isInfoEnabled()) - log.info("IO test started " + - "[warmup=" + warmup + - ", duration=" + duration + - ", threads=" + threads + - ", latencyLimit=" + latencyLimit + - ", rangesCnt=" + rangesCnt + - ", payLoadSize=" + payLoadSize + - ", procFromNioThreads=" + procFromNioThread + ']' - ); - - for (;;) { - if (!warmupFinished.get() && System.currentTimeMillis() - start > warmup) { - if (log.isInfoEnabled()) - log.info("IO test warmup finished."); - - warmupFinished.set(true); - - start = System.currentTimeMillis(); - } - - if (warmupFinished.get() && System.currentTimeMillis() - start > duration) { - if (log.isInfoEnabled()) - log.info("IO test finished, will wait for all threads to finish."); - - done.set(true); - - bar.await(); - - failed = false; - - break; - } - - if (log.isInfoEnabled()) - log.info("IO test [opsCnt/sec=" + (cnt.sumThenReset() * 1000 / sleepDuration) + - ", warmup=" + !warmupFinished.get() + - ", elapsed=" + (System.currentTimeMillis() - start) + ']'); - - Thread.sleep(sleepDuration); - } - - // At this point all threads have finished the test and - // stored data to the resulting array of maps. - // Need to iterate it over and sum values for all threads. - printIoTestResults(res); - } - catch (InterruptedException | BrokenBarrierException e) { - U.error(log, "IO test failed.", e); - } - finally { - if (failed) - bar.reset(); - } - } - }); - - for (int i = 0; i < threads; i++) { - final int i0 = i; - - res[i] = U.newHashMap(nodes.size()); - - svc.execute(new Runnable() { - @Override public void run() { - boolean failed = true; - ThreadLocalRandom rnd = ThreadLocalRandom.current(); - int size = nodes.size(); - Map res0 = res[i0]; - - try { - boolean warmupFinished0 = false; - - bar.await(); - - for (;;) { - if (done.get()) - break; - - if (!warmupFinished0) - warmupFinished0 = warmupFinished.get(); - - ClusterNode node = nodes.get(rnd.nextInt(size)); - - List msgs = sendIoTest(node, payLoad, procFromNioThread).get(); - - cnt.increment(); - - for (IgniteIoTestMessage msg : msgs) { - UUID nodeId = msg.senderNodeId(); - - assert nodeId != null; - - IoTestThreadLocalNodeResults nodeRes = res0.get(nodeId); - - if (nodeRes == null) - res0.put(nodeId, - nodeRes = new IoTestThreadLocalNodeResults(rangesCnt, latencyLimit)); - - nodeRes.onResult(msg); - } - } - - bar.await(); - - failed = false; - } - catch (Exception e) { - U.error(log, "IO test worker thread failed.", e); - } - finally { - if (failed) - bar.reset(); - } - } - }); - } - - failed = false; - } - finally { - if (failed) - U.shutdownNow(GridIoManager.class, svc, log); - } - } - - /** - * @param rawRes Resulting map. - */ - private void printIoTestResults( - Map[] rawRes - ) { - Map res = new HashMap<>(); - - for (Map r : rawRes) { - for (Entry e : r.entrySet()) { - IoTestNodeResults r0 = res.get(e.getKey()); - - if (r0 == null) - res.put(e.getKey(), r0 = new IoTestNodeResults()); - - r0.add(e.getValue()); - } - } - - StringBuilder b = new StringBuilder(U.nl()) - .append("IO test results (round-trip count per each latency bin).") - .append(U.nl()); - - for (Entry e : res.entrySet()) { - ClusterNode node = ctx.discovery().node(e.getKey()); - - long binLatencyMcs = e.getValue().binLatencyMcs(); - - b.append("Node ID: ").append(e.getKey()).append(" (addrs=") - .append(node != null ? node.addresses().toString() : "n/a") - .append(", binLatency=").append(binLatencyMcs).append("mcs") - .append(')').append(U.nl()); - - b.append("Latency bin, mcs | Count exclusive | Percentage exclusive | " + - "Count inclusive | Percentage inclusive ").append(U.nl()); - - long[] nodeRes = e.getValue().resLatency; - - long sum = 0; - - for (int i = 0; i < nodeRes.length; i++) - sum += nodeRes[i]; - - long curSum = 0; - - for (int i = 0; i < nodeRes.length; i++) { - curSum += nodeRes[i]; - - if (i < nodeRes.length - 1) - b.append(String.format("<%11d mcs | %15d | %19.6f%% | %15d | %19.6f%%\n", - (i + 1) * binLatencyMcs, - nodeRes[i], (100.0 * nodeRes[i]) / sum, - curSum, (100.0 * curSum) / sum)); - else - b.append(String.format(">%11d mcs | %15d | %19.6f%% | %15d | %19.6f%%\n", - i * binLatencyMcs, - nodeRes[i], (100.0 * nodeRes[i]) / sum, - curSum, (100.0 * curSum) / sum)); - } - - b.append(U.nl()).append("Total latency (ns): ").append(U.nl()) - .append(String.format("%15d", e.getValue().totalLatency)).append(U.nl()); - - b.append(U.nl()).append("Max latencies (ns):").append(U.nl()); - format(b, e.getValue().maxLatency); - - b.append(U.nl()).append("Max request send queue times (ns):").append(U.nl()); - format(b, e.getValue().maxReqSendQueueTime); - - b.append(U.nl()).append("Max request receive queue times (ns):").append(U.nl()); - format(b, e.getValue().maxReqRcvQueueTime); - - b.append(U.nl()).append("Max response send queue times (ns):").append(U.nl()); - format(b, e.getValue().maxResSendQueueTime); - - b.append(U.nl()).append("Max response receive queue times (ns):").append(U.nl()); - format(b, e.getValue().maxResRcvQueueTime); - - b.append(U.nl()).append("Max request wire times (millis):").append(U.nl()); - format(b, e.getValue().maxReqWireTimeMillis); - - b.append(U.nl()).append("Max response wire times (millis):").append(U.nl()); - format(b, e.getValue().maxResWireTimeMillis); - - b.append(U.nl()); - } - - if (log.isInfoEnabled()) - log.info(b.toString()); - } - - /** - * @param b Builder. - * @param pairs Pairs to format. - */ - private static void format(StringBuilder b, Collection> pairs) { - for (IgnitePair p : pairs) { - b.append(String.format("%15d", p.get1())) - .append(" ") - .append(IgniteUtils.DEBUG_DATE_FMT.format(Instant.ofEpochMilli(p.get2()))) - .append(U.nl()); - } - } - /** {@inheritDoc} */ @Override public void onKernalStart0() throws IgniteCheckedException { discoLsnr = new GridLocalEventListener() { @@ -1092,6 +688,9 @@ private TcpCommunicationSpi getTcpCommunicationSpi() { /** {@inheritDoc} */ @SuppressWarnings("BusyWait") @Override public void onKernalStop0(boolean cancel) { + if (ioTestHnd != null) + ioTestHnd.stop(); + // No more communication messages. getSpi().setListener(null); @@ -1374,17 +973,6 @@ private void processRegularMessage( MTC.span().addLog(() -> "Regular process queued"); - if (msg.topicOrdinal() == TOPIC_IO_TEST.ordinal()) { - IgniteIoTestMessage msg0 = (IgniteIoTestMessage)msg.message(); - - if (msg0.processFromNioThread()) - c.run(); - else - ctx.pools().getStripedExecutorService().execute(-1, c); - - return; - } - final int part = msg.partition(); // Store partition to avoid possible recalculation. if (plc == GridIoPolicy.SYSTEM_POOL && part != GridIoMessage.STRIPE_DISABLED_PART) { @@ -2988,6 +2576,11 @@ public void dumpStats() { X.println(">>> discoWaitMapSize: " + waitMap.size()); } + /** @return IO test handler. */ + public IoTestHandler ioTest() { + return ioTestHnd; + } + /** * Read context holds all the information about current transfer read from channel process. */ @@ -3973,271 +3566,6 @@ public UUID nodeId() { } } - /** - * - */ - private class IoTestFuture extends GridFutureAdapter> { - /** */ - private final long id; - - /** */ - private final int cntr; - - /** */ - private final List ress; - - /** - * @param id ID. - * @param cntr Counter. - */ - IoTestFuture(long id, int cntr) { - assert cntr > 0 : cntr; - - this.id = id; - this.cntr = cntr; - - ress = new ArrayList<>(cntr); - } - - /** - * - */ - void onResponse(IgniteIoTestMessage res) { - boolean complete; - - synchronized (this) { - ress.add(res); - - complete = cntr == ress.size(); - } - - if (complete) - onDone(ress); - } - - /** {@inheritDoc} */ - @Override public boolean onDone(List res, @Nullable Throwable err) { - if (super.onDone(res, err)) { - ioTestMap().remove(id); - - return true; - } - - return false; - } - - /** {@inheritDoc} */ - @Override public String toString() { - return S.toString(IoTestFuture.class, this); - } - } - - /** - * - */ - private static class IoTestThreadLocalNodeResults { - /** */ - private final long[] resLatency; - - /** */ - private final int rangesCnt; - - /** */ - private long totalLatency; - - /** */ - private long maxLatency; - - /** */ - private long maxLatencyTs; - - /** */ - private long maxReqSendQueueTime; - - /** */ - private long maxReqSendQueueTimeTs; - - /** */ - private long maxReqRcvQueueTime; - - /** */ - private long maxReqRcvQueueTimeTs; - - /** */ - private long maxResSendQueueTime; - - /** */ - private long maxResSendQueueTimeTs; - - /** */ - private long maxResRcvQueueTime; - - /** */ - private long maxResRcvQueueTimeTs; - - /** */ - private long maxReqWireTimeMillis; - - /** */ - private long maxReqWireTimeTs; - - /** */ - private long maxResWireTimeMillis; - - /** */ - private long maxResWireTimeTs; - - /** */ - private final long latencyLimit; - - /** - * @param rangesCnt Ranges count. - * @param latencyLimit - */ - public IoTestThreadLocalNodeResults(int rangesCnt, long latencyLimit) { - this.rangesCnt = rangesCnt; - this.latencyLimit = latencyLimit; - - resLatency = new long[rangesCnt + 1]; - } - - /** - * @param msg - */ - public void onResult(IgniteIoTestMessage msg) { - long now = System.currentTimeMillis(); - - long latency = msg.responseProcessedTs() - msg.requestCreateTs(); - - int idx = latency >= latencyLimit ? - rangesCnt /* Timed out. */ : - (int)Math.floor((1.0 * latency) / ((1.0 * latencyLimit) / rangesCnt)); - - resLatency[idx]++; - - totalLatency += latency; - - if (maxLatency < latency) { - maxLatency = latency; - maxLatencyTs = now; - } - - long reqSndQueueTime = msg.requestSendTs() - msg.requestCreateTs(); - - if (maxReqSendQueueTime < reqSndQueueTime) { - maxReqSendQueueTime = reqSndQueueTime; - maxReqSendQueueTimeTs = now; - } - - long reqRcvQueueTime = msg.requestProcessTs() - msg.requestReceiveTs(); - - if (maxReqRcvQueueTime < reqRcvQueueTime) { - maxReqRcvQueueTime = reqRcvQueueTime; - maxReqRcvQueueTimeTs = now; - } - - long resSndQueueTime = msg.responseSendTs() - msg.requestProcessTs(); - - if (maxResSendQueueTime < resSndQueueTime) { - maxResSendQueueTime = resSndQueueTime; - maxResSendQueueTimeTs = now; - } - - long resRcvQueueTime = msg.responseProcessedTs() - msg.responseReceiveTs(); - - if (maxResRcvQueueTime < resRcvQueueTime) { - maxResRcvQueueTime = resRcvQueueTime; - maxResRcvQueueTimeTs = now; - } - - long reqWireTimeMillis = msg.requestReceivedTsMillis() - msg.requestSendTsMillis(); - - if (maxReqWireTimeMillis < reqWireTimeMillis) { - maxReqWireTimeMillis = reqWireTimeMillis; - maxReqWireTimeTs = now; - } - - long resWireTimeMillis = msg.responseReceivedTsMillis() - msg.requestSendTsMillis(); - - if (maxResWireTimeMillis < resWireTimeMillis) { - maxResWireTimeMillis = resWireTimeMillis; - maxResWireTimeTs = now; - } - } - } - - /** - * - */ - private static class IoTestNodeResults { - /** */ - private long latencyLimit; - - /** */ - private long[] resLatency; - - /** */ - private long totalLatency; - - /** */ - private Collection> maxLatency = new ArrayList<>(); - - /** */ - private Collection> maxReqSendQueueTime = new ArrayList<>(); - - /** */ - private Collection> maxReqRcvQueueTime = new ArrayList<>(); - - /** */ - private Collection> maxResSendQueueTime = new ArrayList<>(); - - /** */ - private Collection> maxResRcvQueueTime = new ArrayList<>(); - - /** */ - private Collection> maxReqWireTimeMillis = new ArrayList<>(); - - /** */ - private Collection> maxResWireTimeMillis = new ArrayList<>(); - - /** - * @param res Node results to add. - */ - public void add(IoTestThreadLocalNodeResults res) { - if (resLatency == null) { - resLatency = res.resLatency.clone(); - latencyLimit = res.latencyLimit; - } - else { - assert latencyLimit == res.latencyLimit; - assert resLatency.length == res.resLatency.length; - - for (int i = 0; i < resLatency.length; i++) - resLatency[i] += res.resLatency[i]; - } - - totalLatency += res.totalLatency; - - maxLatency.add(F.pair(res.maxLatency, res.maxLatencyTs)); - maxReqSendQueueTime.add(F.pair(res.maxReqSendQueueTime, res.maxReqSendQueueTimeTs)); - maxReqRcvQueueTime.add(F.pair(res.maxReqRcvQueueTime, res.maxReqRcvQueueTimeTs)); - maxResSendQueueTime.add(F.pair(res.maxResSendQueueTime, res.maxResSendQueueTimeTs)); - maxResRcvQueueTime.add(F.pair(res.maxResRcvQueueTime, res.maxResRcvQueueTimeTs)); - maxReqWireTimeMillis.add(F.pair(res.maxReqWireTimeMillis, res.maxReqWireTimeTs)); - maxResWireTimeMillis.add(F.pair(res.maxResWireTimeMillis, res.maxResWireTimeTs)); - } - - /** - * @return Bin latency in microseconds. - */ - public long binLatencyMcs() { - if (resLatency == null) - throw new IllegalStateException(); - - return latencyLimit / (1000 * (resLatency.length - 1)); - } - } - /** * Responsible for handling network situation where server cannot open connection to client and * has to ask client to establish a connection to specific server. diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/IgniteIoTestMessage.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/IgniteIoTestMessage.java index d95ead4f902de..6544b99ce42ca 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/IgniteIoTestMessage.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/IgniteIoTestMessage.java @@ -17,311 +17,187 @@ package org.apache.ignite.internal.managers.communication; -import java.util.UUID; import org.apache.ignite.IgniteCheckedException; import org.apache.ignite.internal.MarshallableMessage; import org.apache.ignite.internal.Order; import org.apache.ignite.internal.util.typedef.internal.S; import org.apache.ignite.marshaller.Marshaller; -/** - * - */ +/** Communication SPI test message. */ public class IgniteIoTestMessage implements MarshallableMessage { - /** */ - private static final byte FLAG_PROC_FROM_NIO = 1; - - /** */ + /** Test ID. */ @Order(0) long id; - /** */ + /** Process message in NIO thread. */ @Order(1) - byte flags; + boolean procFromNioThread; - /** */ + /** Request flag. */ @Order(2) boolean req; - /** */ + /** Payload. */ @Order(3) byte[] payload; - /** */ + /** Request creation timestamp from the source node monotonic clock. */ @Order(4) long reqCreateTs; - /** */ + /** Request serialization-start timestamp from the source node monotonic clock. */ @Order(5) long reqSndTs; - /** */ + /** Request serialization-start timestamp from the source node wall clock. */ @Order(6) long reqSndTsMillis; - /** */ + /** Request deserialization-completion timestamp from the target node monotonic clock. */ @Order(7) long reqRcvTs; - /** */ + /** Request deserialization-completion timestamp from the target node wall clock. */ @Order(8) long reqRcvTsMillis; - /** */ + /** Request listener invocation timestamp from the target node monotonic clock. */ @Order(9) long reqProcTs; - /** */ + /** Response serialization-start timestamp from the target node monotonic clock. */ @Order(10) long resSndTs; - /** */ + /** Response serialization-start timestamp from the target node wall clock. */ @Order(11) long resSndTsMillis; - /** */ - @Order(12) + /** Response deserialization-completion timestamp from the source node monotonic clock. */ long resRcvTs; - /** */ - @Order(13) + /** Response deserialization-completion timestamp from the source node wall clock. */ long resRcvTsMillis; - /** */ - @Order(14) + /** Response listener invocation timestamp from the source node monotonic clock. */ long resProcTs; - /** */ - private UUID sndNodeId; - - /** - * - */ + /** Required by the message factory. */ public IgniteIoTestMessage() { // No-op. } - /** - * @param id Message ID. - * @param req Request flag. - * @param payload Payload. - */ - public IgniteIoTestMessage(long id, boolean req, byte[] payload) { + /** Request constructor. */ + public IgniteIoTestMessage(long id, byte[] payload, boolean procFromNioThread) { this.id = id; - this.req = req; this.payload = payload; + this.procFromNioThread = procFromNioThread; + req = true; reqCreateTs = System.nanoTime(); } - /** - * @return {@code True} if message should be processed from NIO thread - * (otherwise message is submitted to system pool). - */ - public boolean processFromNioThread() { - return isFlag(FLAG_PROC_FROM_NIO); - } - - /** - * @param procFromNioThread {@code True} if message should be processed from NIO thread. - */ - public void processFromNioThread(boolean procFromNioThread) { - setFlag(procFromNioThread, FLAG_PROC_FROM_NIO); - } - - /** - * @param flags Flags. - */ - public void flags(byte flags) { - this.flags = flags; - } - - /** - * @return Flags. - */ - public byte flags() { - return flags; - } - - /** - * Sets flag mask. - * - * @param flag Set or clear. - * @param mask Mask. - */ - private void setFlag(boolean flag, int mask) { - flags = flag ? (byte)(flags | mask) : (byte)(flags & ~mask); + /** Response constructor. */ + public IgniteIoTestMessage(IgniteIoTestMessage req) { + id = req.id; + payload = req.payload; + procFromNioThread = req.procFromNioThread; + reqCreateTs = req.reqCreateTs; + reqSndTs = req.reqSndTs; + reqSndTsMillis = req.reqSndTsMillis; + reqRcvTs = req.reqRcvTs; + reqRcvTsMillis = req.reqRcvTsMillis; + reqProcTs = req.reqProcTs; } - /** - * Reads flag mask. - * - * @param mask Mask to read. - * @return Flag value. - */ - private boolean isFlag(int mask) { - return (flags & mask) != 0; + /** @return {@code True} to process this message in NIO thread. */ + public boolean processFromNioThread() { + return procFromNioThread; } - /** - * @return {@code true} if this is request. - */ + /** @return {@code True} if this is a request. */ public boolean request() { return req; } - /** - * @return ID. - */ - public long id() { + /** @return Test ID. */ + public long testId() { return id; } - /** - * @return Request create timestamp. - */ - public long requestCreateTs() { - return reqCreateTs; - } - - /** - * @return Request send timestamp. - */ - public long requestSendTs() { - return reqSndTs; + /** Records request listener invocation. */ + void onRequestProcessed() { + reqProcTs = System.nanoTime(); } - /** - * @return Request receive timestamp. - */ - public long requestReceiveTs() { - return reqRcvTs; + /** Records response listener invocation. */ + void onResponseProcessed() { + resProcTs = System.nanoTime(); } - /** - * @return Request process started timestamp. - */ - public long requestProcessTs() { - return reqProcTs; + /** @return End-to-end RTT from request creation to response listener invocation, in nanoseconds. */ + long roundTripNanos() { + return resProcTs - reqCreateTs; } - /** - * @return Response send timestamp. - */ - public long responseSendTs() { - return resSndTs; + /** @return Delay from request creation to request serialization start, in nanoseconds. */ + long requestSendQueueNanos() { + return reqSndTs - reqCreateTs; } - /** - * @return Response receive timestamp. - */ - public long responseReceiveTs() { - return resRcvTs; + /** @return Delay from request deserialization completion to request listener invocation, in nanoseconds. */ + long requestReceiveQueueNanos() { + return reqProcTs - reqRcvTs; } - /** - * @return Request send timestamp (millis). - */ - public long requestSendTsMillis() { - return reqSndTsMillis; + /** @return Delay from request listener invocation to response serialization start, in nanoseconds. */ + long responseSendQueueNanos() { + return resSndTs - reqProcTs; } - /** - * @return Request received timestamp (millis). - */ - public long requestReceivedTsMillis() { - return reqRcvTsMillis; + /** @return Delay from response deserialization completion to response listener invocation, in nanoseconds. */ + long responseReceiveQueueNanos() { + return resProcTs - resRcvTs; } /** - * @return Response received timestamp (millis). + * @return Estimated one-way request delay from serialization start on the source to deserialization completion on + * the target, in milliseconds. Requires synchronized wall clocks. */ - public long responseReceivedTsMillis() { - return resRcvTsMillis; + long requestWireTimeMillis() { + return reqRcvTsMillis - reqSndTsMillis; } /** - * This method is called to initialize tracing variables. - * TODO: introduce direct message lifecycle API? + * @return Estimated one-way response delay from serialization start on the target to deserialization completion on + * the source, in milliseconds. Requires synchronized wall clocks. */ - public void onAfterRead() { - if (req && reqRcvTs == 0) { - reqRcvTs = System.nanoTime(); - - reqRcvTsMillis = System.currentTimeMillis(); - } - - if (!req && resRcvTs == 0) { - resRcvTs = System.nanoTime(); - - resRcvTsMillis = System.currentTimeMillis(); - } + long responseWireTimeMillis() { + return resRcvTsMillis - resSndTsMillis; } - /** - * This method is called to initialize tracing variables. - * TODO: introduce direct message lifecycle API? - */ - public void onBeforeWrite() { + /** Records the start of the first serialization attempt. */ + void onBeforeWrite() { if (req && reqSndTs == 0) { reqSndTs = System.nanoTime(); - reqSndTsMillis = System.currentTimeMillis(); } - - if (!req && resSndTs == 0) { + else if (!req && resSndTs == 0) { resSndTs = System.nanoTime(); - resSndTsMillis = System.currentTimeMillis(); } } - /** - * - */ - public void copyDataFromRequest(IgniteIoTestMessage req) { - reqCreateTs = req.reqCreateTs; - - reqSndTs = req.reqSndTs; - reqSndTsMillis = req.reqSndTsMillis; - - reqRcvTs = req.reqRcvTs; - reqRcvTsMillis = req.reqRcvTsMillis; - } - - /** - * - */ - public void onRequestProcessed() { - reqProcTs = System.nanoTime(); - } - - /** - * - */ - public void onResponseProcessed() { - resProcTs = System.nanoTime(); - } - - /** - * @return Response processed timestamp. - */ - public long responseProcessedTs() { - return resProcTs; - } - - /** - * @return Sender node ID. - */ - public UUID senderNodeId() { - return sndNodeId; - } - - /** - * @param sndNodeId Sender node ID. - */ - public void senderNodeId(UUID sndNodeId) { - this.sndNodeId = sndNodeId; + /** Records the completion of deserialization. */ + void onAfterRead() { + if (req && reqRcvTs == 0) { + reqRcvTs = System.nanoTime(); + reqRcvTsMillis = System.currentTimeMillis(); + } + else if (!req && resRcvTs == 0) { + resRcvTs = System.nanoTime(); + resRcvTsMillis = System.currentTimeMillis(); + } } /** {@inheritDoc} */ diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/IoTestHandler.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/IoTestHandler.java new file mode 100644 index 0000000000000..36e5aac736d24 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/IoTestHandler.java @@ -0,0 +1,605 @@ +/* + * 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.managers.communication; + +import java.util.ArrayList; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.TreeMap; +import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; +import org.apache.ignite.IgniteCheckedException; +import org.apache.ignite.IgniteLogger; +import org.apache.ignite.cluster.ClusterNode; +import org.apache.ignite.internal.GridKernalContext; +import org.apache.ignite.internal.GridTopic; +import org.apache.ignite.internal.IgniteInternalFuture; +import org.apache.ignite.internal.NodeStoppingException; +import org.apache.ignite.internal.util.future.GridCompoundFuture; +import org.apache.ignite.internal.util.future.GridFinishedFuture; +import org.apache.ignite.internal.util.future.GridFutureAdapter; +import org.apache.ignite.internal.util.typedef.internal.A; +import org.apache.ignite.internal.util.typedef.internal.LT; +import org.apache.ignite.internal.util.typedef.internal.S; +import org.apache.ignite.internal.util.typedef.internal.U; +import org.jetbrains.annotations.Nullable; + +import static org.apache.ignite.internal.thread.pool.IgniteThreadPoolExecutor.newFixedThreadPool; + +/** Communication SPI test handler. */ +public class IoTestHandler { + /** Test ID generator. */ + private static final AtomicLong ID_GEN = new AtomicLong(); + + /** Kernal context. */ + private final GridKernalContext ctx; + + /** Logger. */ + private final IgniteLogger log; + + /** Pending test requests. */ + private final ConcurrentHashMap ioTests = new ConcurrentHashMap<>(); + + /** Stop flag. */ + private volatile boolean stopping; + + /** Test-running flag. */ + private final AtomicBoolean testRunning = new AtomicBoolean(); + + /** Active test. */ + private volatile IoTestRunFuture activeTest; + + /** Constructor. */ + public IoTestHandler(GridKernalContext ctx) { + this.ctx = ctx; + log = ctx.log(getClass()); + + ctx.io().addMessageListener(GridTopic.TOPIC_IO_TEST, (nodeId, msg, plc) -> { + IgniteIoTestMessage msg0 = (IgniteIoTestMessage)msg; + + if (msg0.request()) { + msg0.onRequestProcessed(); + + try { + ctx.io().sendToGridTopic( + nodeId, + GridTopic.TOPIC_IO_TEST, + new IgniteIoTestMessage(msg0), + GridIoPolicy.SYSTEM_POOL + ); + } + catch (Exception e) { + LT.warn(log, "Failed to send IO test response [nodeId=" + nodeId + "]", e); + } + } + else { + msg0.onResponseProcessed(); + + IoTestFuture fut = ioTests.get(msg0.testId()); + + if (fut != null) + fut.onDone(msg0); + else if (log.isDebugEnabled()) + log.debug("Failed to find IO test future [msg=" + msg0 + ']'); + } + }); + } + + /** + * Sends one test request to every node. + * + * @param nodes Nodes. + * @param payload Payload. + * @param procFromNioThread Process messages in NIO threads. + * @return Aggregate response future. + */ + public GridCompoundFuture sendIoTest( + List nodes, + byte[] payload, + boolean procFromNioThread + ) { + GridCompoundFuture resFut = new GridCompoundFuture<>(); + + nodes.forEach(node -> resFut.add(sendIoTest(node, payload, procFromNioThread))); + + resFut.markInitialized(); + + return resFut; + } + + /** + * Sends a test request. + * + * @param node Node. + * @param payload Payload. + * @param procFromNioThread Process messages in NIO threads. + * @return Response future. + */ + public IgniteInternalFuture sendIoTest( + ClusterNode node, + byte[] payload, + boolean procFromNioThread + ) { + if (stopping) + return new GridFinishedFuture<>(stoppingException()); + + long id = ID_GEN.getAndIncrement(); + + IoTestFuture fut = new IoTestFuture(id); + + ioTests.put(id, fut); + + if (stopping) + fut.onDone(stoppingException()); + else { + try { + ctx.io().sendToGridTopic( + node, + GridTopic.TOPIC_IO_TEST, + new IgniteIoTestMessage(id, payload, procFromNioThread), + GridIoPolicy.SYSTEM_POOL + ); + } + catch (IgniteCheckedException | RuntimeException e) { + fut.onDone(e); + } + } + + return fut; + } + + /** + * Runs a latency test against the supplied nodes. + * + * @param warmup Warmup duration in milliseconds. + * @param duration Test duration in milliseconds. + * @param threads Thread count. + * @param latencyLimit RTT histogram upper bound in nanoseconds. + * @param rangesCnt RTT histogram range count. + * @param payloadSize Payload size in bytes. + * @param procFromNioThread Process messages in NIO threads. + * @param nodes Nodes participating in the test. + * @return Test result future. + */ + public IgniteInternalFuture runIoTest( + long warmup, + long duration, + int threads, + long latencyLimit, + int rangesCnt, + int payloadSize, + boolean procFromNioThread, + List nodes + ) { + A.notEmpty(nodes, "nodes"); + + if (stopping) + return new GridFinishedFuture<>(stoppingException()); + + A.ensure(testRunning.compareAndSet(false, true), "Communication IO test is already running."); + + List testNodes = new ArrayList<>(nodes); + ExecutorService svc; + + try { + svc = newFixedThreadPool("io-latency-inspector", ctx.igniteInstanceName(), threads); + } + catch (RuntimeException | Error e) { + testRunning.set(false); + + throw e; + } + + AtomicBoolean finished = new AtomicBoolean(); + IoTestRunFuture testRes = new IoTestRunFuture(svc, finished); + AtomicInteger remaining = new AtomicInteger(threads); + AtomicInteger firstSampleIdx = new AtomicInteger(); + byte[] payload = new byte[payloadSize]; + Map results = new ConcurrentHashMap<>(); + long startNanos = System.nanoTime(); + long warmupNanos = TimeUnit.MILLISECONDS.toNanos(warmup); + long totalNanos = warmupNanos + TimeUnit.MILLISECONDS.toNanos(duration); + long responseTimeout = Math.max(1, ctx.config().getFailureDetectionTimeout()); + + activeTest = testRes; + + if (stopping) { + testRes.onDone(stoppingException()); + + return testRes; + } + + try { + for (int i = 0; i < threads; i++) { + int workerIdx = i; + + svc.execute(() -> { + long targetIdx = workerIdx; + + try { + while (!finished.get() && elapsedNanos(startNanos) < warmupNanos) { + ClusterNode node = testNodes.get((int)(targetIdx++ % testNodes.size())); + + sendAndMeasure(node, payload, procFromNioThread, responseTimeout); + } + + for (int idx; !finished.get() && elapsedNanos(startNanos) < totalNanos && + (idx = firstSampleIdx.getAndIncrement()) < testNodes.size(); ) { + recordResult(results, testNodes.get(idx), payload, procFromNioThread, + responseTimeout, rangesCnt, latencyLimit); + } + + while (!finished.get() && elapsedNanos(startNanos) < totalNanos) + recordResult(results, testNodes.get((int)(targetIdx++ % testNodes.size())), payload, + procFromNioThread, + responseTimeout, rangesCnt, latencyLimit); + } + catch (Exception e) { + testRes.onDone(e); + } + finally { + if (remaining.decrementAndGet() == 0 && !finished.get()) { + if (results.size() != testNodes.size()) { + testRes.onDone(new IgniteCheckedException( + "Communication SPI test duration is too short to sample every target node " + + "[targets=" + testNodes.size() + ", sampled=" + results.size() + ']')); + } + else { + try { + testRes.onDone(formatResults(results, payloadSize, warmup, duration, threads, + procFromNioThread)); + } + catch (RuntimeException e) { + testRes.onDone(e); + } + } + } + } + }); + } + } + catch (RuntimeException e) { + testRes.onDone(e); + } + + return testRes; + } + + /** Stops this handler and completes pending requests. */ + void stop() { + stopping = true; + + NodeStoppingException err = stoppingException(); + + ioTests.values().forEach(fut -> fut.onDone(err)); + + IoTestRunFuture test = activeTest; + + if (test != null) + test.onDone(err); + } + + /** Records one round-trip result. */ + private void recordResult( + Map results, + ClusterNode node, + byte[] payload, + boolean procFromNioThread, + long responseTimeout, + int rangesCnt, + long latencyLimit + ) throws IgniteCheckedException { + IgniteIoTestMessage res = sendAndMeasure(node, payload, procFromNioThread, responseTimeout); + + results.computeIfAbsent(node.id(), ignored -> new IoTestNodeResults(rangesCnt, latencyLimit)) + .onResult(res); + } + + /** Sends a request and measures its round-trip time on the local node. */ + private IgniteIoTestMessage sendAndMeasure( + ClusterNode node, + byte[] payload, + boolean procFromNioThread, + long responseTimeout + ) throws IgniteCheckedException { + IgniteInternalFuture fut = sendIoTest(node, payload, procFromNioThread); + + try { + return fut.get(responseTimeout); + } + catch (IgniteCheckedException e) { + throw new IgniteCheckedException("Communication SPI test request failed [nodeId=" + node.id() + + ", addresses=" + node.addresses() + ']', e); + } + finally { + if (!fut.isDone()) + fut.cancel(); + } + } + + /** Returns monotonic elapsed time. */ + private static long elapsedNanos(long startNanos) { + return System.nanoTime() - startNanos; + } + + /** Formats test results. */ + private String formatResults( + Map rawResults, + int payloadSize, + long warmup, + long duration, + int threads, + boolean procFromNioThread + ) { + Map results = new TreeMap<>(rawResults); + + ClusterNode source = ctx.discovery().localNode(); + StringBuilder b = new StringBuilder("Communication SPI test").append(U.nl()) + .append("Source node: ").append(source.id()).append(" [addresses=").append(source.addresses()).append(']') + .append(U.nl()) + .append("Payload: ").append(payloadSize).append(" bytes each way").append(U.nl()) + .append("Message handling: ").append(procFromNioThread ? "NIO thread" : "system pool").append(U.nl()) + .append("Warmup: ").append(warmup).append(" ms | Duration: ").append(duration) + .append(" ms | Threads: ").append(threads).append(U.nl()); + + for (Map.Entry entry : results.entrySet()) { + ClusterNode node = ctx.discovery().node(entry.getKey()); + IoTestNodeResults nodeResults = entry.getValue(); + + b.append(U.nl()) + .append("Target node: ").append(entry.getKey()).append(" [addresses=") + .append(node != null ? node.addresses() : "n/a") + .append(']').append(U.nl()) + .append("Samples: ").append(nodeResults.count).append(U.nl()) + .append(String.format(Locale.ROOT, "End-to-end RTT (ms): min=%.3f, avg=%.3f, max=%.3f%n", + nodeResults.minLatency / 1_000_000.0, + nodeResults.totalLatency / (double)nodeResults.count / 1_000_000, + nodeResults.maxLatency / 1_000_000.0)) + .append("Node-local delays (ms, min/avg/max):").append(U.nl()); + + appendTiming(b, "Source request pre-serialization delay", nodeResults.reqSndQueue, 1_000_000); + appendTiming(b, "Target request dispatch delay", nodeResults.reqRcvQueue, 1_000_000); + appendTiming(b, "Target response pre-serialization delay", nodeResults.resSndQueue, 1_000_000); + appendTiming(b, "Source response dispatch delay", nodeResults.resRcvQueue, 1_000_000); + + b.append("Estimated one-way message delay (ms, min/avg/max):").append(U.nl()) + .append(" Clock assumption: synchronized wall clocks; negative values indicate clock offset or ") + .append("wall-clock adjustment.") + .append(U.nl()); + + appendTiming(b, "Request (serialization start -> deserialization complete)", nodeResults.reqWireTime, 1); + appendTiming(b, "Response (serialization start -> deserialization complete)", nodeResults.resWireTime, 1); + + b.append("RTT histogram:").append(U.nl()); + + for (int i = 0; i < nodeResults.resLatency.length; i++) { + long bucketCount = nodeResults.resLatency[i]; + + if (bucketCount == 0) + continue; + + double lowerBound = nodeResults.binUpperBound(i) / 1_000_000.0; + String range = i < nodeResults.resLatency.length - 1 + ? String.format(Locale.ROOT, "[%.3f, %.3f) ms", lowerBound, + nodeResults.binUpperBound(i + 1) / 1_000_000.0) + : String.format(Locale.ROOT, "[%.3f, +inf) ms", lowerBound); + + b.append(String.format(Locale.ROOT, " %-31s %d (%.2f%%)%n", + range + ':', + bucketCount, + 100.0 * bucketCount / nodeResults.count)); + } + } + + return b.toString(); + } + + /** Appends min/average/max timing values. */ + private static void appendTiming(StringBuilder b, String name, TimingStats timing, double divisor) { + b.append(String.format(Locale.ROOT, " %s: min=%.3f, avg=%.3f, max=%.3f%n", + name, timing.min / divisor, timing.average() / divisor, timing.max / divisor)); + } + + /** Creates a node-stopping error. */ + private NodeStoppingException stoppingException() { + return new NodeStoppingException("IO test has been cancelled because the local node is stopping: " + + ctx.localNodeId()); + } + + /** Pending request future. */ + private class IoTestFuture extends GridFutureAdapter { + /** Test ID. */ + private final long id; + + /** Constructor. */ + IoTestFuture(long id) { + this.id = id; + } + + /** {@inheritDoc} */ + @Override protected boolean onDone( + @Nullable IgniteIoTestMessage res, + @Nullable Throwable err, + boolean cancel + ) { + if (super.onDone(res, err, cancel)) { + ioTests.remove(id, this); + + return true; + } + + return false; + } + + /** {@inheritDoc} */ + @Override public boolean cancel() { + return onCancelled(); + } + + /** {@inheritDoc} */ + @Override public String toString() { + return S.toString(IoTestFuture.class, this); + } + } + + /** Running test future. */ + private class IoTestRunFuture extends GridFutureAdapter { + /** Test executor. */ + private final ExecutorService svc; + + /** Finished flag shared with workers. */ + private final AtomicBoolean finished; + + /** Constructor. */ + IoTestRunFuture(ExecutorService svc, AtomicBoolean finished) { + this.svc = svc; + this.finished = finished; + } + + /** {@inheritDoc} */ + @Override protected boolean onDone(@Nullable String res, @Nullable Throwable err, boolean cancel) { + if (!finished.compareAndSet(false, true)) + return false; + + if (cancel || err != null) + svc.shutdownNow(); + else + svc.shutdown(); + + activeTest = null; + testRunning.set(false); + + return super.onDone(res, err, cancel); + } + + /** {@inheritDoc} */ + @Override public boolean cancel() { + return onCancelled(); + } + } + + /** Aggregated node results. */ + private static class IoTestNodeResults { + /** Histogram. */ + private final long[] resLatency; + + /** Histogram range count. */ + private final int rangesCnt; + + /** Maximum expected latency. */ + private final long latencyLimit; + + /** Total latency. */ + private long totalLatency; + + /** Minimum latency. */ + private long minLatency = Long.MAX_VALUE; + + /** Maximum latency. */ + private long maxLatency; + + /** Sample count. */ + private long count; + + /** Request send queue statistics. */ + private final TimingStats reqSndQueue = new TimingStats(); + + /** Request receive queue statistics. */ + private final TimingStats reqRcvQueue = new TimingStats(); + + /** Response send queue statistics. */ + private final TimingStats resSndQueue = new TimingStats(); + + /** Response receive queue statistics. */ + private final TimingStats resRcvQueue = new TimingStats(); + + /** Approximate request transfer statistics. */ + private final TimingStats reqWireTime = new TimingStats(); + + /** Approximate response transfer statistics. */ + private final TimingStats resWireTime = new TimingStats(); + + /** Constructor. */ + IoTestNodeResults(int rangesCnt, long latencyLimit) { + this.rangesCnt = rangesCnt; + this.latencyLimit = latencyLimit; + + resLatency = new long[rangesCnt + 1]; + } + + /** Adds a sample. */ + synchronized void onResult(IgniteIoTestMessage msg) { + long latency = msg.roundTripNanos(); + int idx = latency >= latencyLimit + ? rangesCnt + : (int)(latency * (double)rangesCnt / latencyLimit); + + resLatency[idx]++; + totalLatency += latency; + minLatency = Math.min(minLatency, latency); + maxLatency = Math.max(maxLatency, latency); + count++; + + reqSndQueue.add(msg.requestSendQueueNanos()); + reqRcvQueue.add(msg.requestReceiveQueueNanos()); + resSndQueue.add(msg.responseSendQueueNanos()); + resRcvQueue.add(msg.responseReceiveQueueNanos()); + reqWireTime.add(msg.requestWireTimeMillis()); + resWireTime.add(msg.responseWireTimeMillis()); + } + + /** Returns the upper bound of a histogram bin. */ + double binUpperBound(int bin) { + return latencyLimit * (double)bin / rangesCnt; + } + } + + /** Min/average/max accumulator. */ + private static class TimingStats { + /** Minimum. */ + private long min = Long.MAX_VALUE; + + /** Maximum. */ + private long max = Long.MIN_VALUE; + + /** Sum. */ + private double total; + + /** Count. */ + private long count; + + /** Adds a value. */ + void add(long val) { + min = Math.min(min, val); + max = Math.max(max, val); + total += val; + count++; + } + + /** @return Average. */ + double average() { + return total / count; + } + } +} diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java index 23cc502b17775..da3675804e4e7 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/GridDiscoveryManager.java @@ -304,6 +304,9 @@ public class GridDiscoveryManager extends GridManagerAdapter { /** Local node compatibility consistent ID. */ private Serializable consistentId; + /** */ + private IoTestDiscoveryHandler ioTestHnd; + /** @param ctx Context. */ public GridDiscoveryManager(GridKernalContext ctx) { super(ctx, ctx.config().getDiscoverySpi()); @@ -531,6 +534,8 @@ private void updateClientNodes(UUID leftNodeId) { if (ctx.config().getCommunicationFailureResolver() != null) ctx.resource().injectGeneric(ctx.config().getCommunicationFailureResolver()); + ioTestHnd = new IoTestDiscoveryHandler(ctx); + // Shared reference between DiscoverySpiListener and DiscoverySpiDataExchange. AtomicReference> lastStateChangeEvtLsnrFutRef = new AtomicReference<>(); @@ -2708,6 +2713,11 @@ public ClusterNode historicalNode(UUID nodeId) { return null; } + /** @return IO test handler. */ + public IoTestDiscoveryHandler ioTest() { + return ioTestHnd; + } + /** Network segments checker. */ private class SegmentChecker extends IgniteAsyncObjectHandler { /** */ diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/IoTestDiscoveryAckMessage.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/IoTestDiscoveryAckMessage.java new file mode 100644 index 0000000000000..e910a6acc0fa1 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/IoTestDiscoveryAckMessage.java @@ -0,0 +1,69 @@ +/* + * 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.managers.discovery; + +import java.util.ArrayList; +import java.util.List; +import java.util.UUID; +import org.apache.ignite.internal.Order; +import org.apache.ignite.lang.IgniteUuid; +import org.apache.ignite.plugin.extensions.communication.MessageFactory; +import org.apache.ignite.spi.discovery.DiscoverySpiCustomMessage; +import org.jetbrains.annotations.Nullable; + +/** Acknowledgement for {@link IoTestDiscoveryMessage}. */ +public class IoTestDiscoveryAckMessage extends DiscoveryServerOnlyCustomMessage { + /** Request message ID. */ + @Order(0) + IgniteUuid requestId; + + /** Ordered server node path. */ + @Order(1) + List path; + + /** Estimated one-way delay for every completed ring hop. */ + @Order(2) + List hopTimesMillis; + + /** Request ring latency from submission to local acknowledgement processing, measured on the coordinator. */ + long ringTimeNanos; + + /** Empty constructor for {@link MessageFactory}. */ + public IoTestDiscoveryAckMessage() { + // No-op. + } + + /** @param msg Request. */ + public IoTestDiscoveryAckMessage(IoTestDiscoveryMessage msg) { + super(IgniteUuid.randomUuid()); + + requestId = msg.id(); + path = new ArrayList<>(msg.path); + hopTimesMillis = new ArrayList<>(msg.hopTimesMillis); + } + + /** @return Request message ID. */ + public IgniteUuid requestId() { + return requestId; + } + + /** {@inheritDoc} */ + @Override public @Nullable DiscoverySpiCustomMessage ackMessage() { + return null; + } +} diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/IoTestDiscoveryHandler.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/IoTestDiscoveryHandler.java new file mode 100644 index 0000000000000..a426a40aa5d95 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/IoTestDiscoveryHandler.java @@ -0,0 +1,291 @@ +/* + * 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.managers.discovery; + +import java.util.ArrayList; +import java.util.List; +import java.util.Locale; +import java.util.LongSummaryStatistics; +import java.util.UUID; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.function.BooleanSupplier; +import org.apache.ignite.IgniteCheckedException; +import org.apache.ignite.IgniteException; +import org.apache.ignite.IgniteLogger; +import org.apache.ignite.internal.GridKernalContext; +import org.apache.ignite.internal.IgniteFutureTimeoutCheckedException; +import org.apache.ignite.internal.util.future.GridFutureAdapter; +import org.apache.ignite.internal.util.typedef.internal.A; +import org.apache.ignite.internal.util.typedef.internal.U; +import org.apache.ignite.lang.IgniteUuid; +import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi; + +/** Runs a bounded latency test through the discovery ring. */ +public class IoTestDiscoveryHandler { + /** Cancellation polling interval. */ + private static final long CANCEL_POLL_INTERVAL_MILLIS = 100; + + /** */ + private final GridKernalContext ctx; + + /** */ + private final IgniteLogger log; + + /** Pending sample. */ + private volatile IoTestDiscoveryFuture pendingTest; + + /** Ensures that only one test runs on the coordinator. */ + private final AtomicBoolean testRunning = new AtomicBoolean(); + + /** @param ctx Kernal context. */ + public IoTestDiscoveryHandler(GridKernalContext ctx) { + this.ctx = ctx; + log = ctx.log(getClass()); + + ctx.discovery().setCustomEventListener(IoTestDiscoveryMessage.class, (topVer, snd, msg) -> + msg.onProcessed(ctx.localNodeId())); + + ctx.discovery().setCustomEventListener(IoTestDiscoveryAckMessage.class, (topVer, snd, msg) -> { + if (!U.isLocalNodeCoordinator(ctx.discovery())) + return; + + IoTestDiscoveryFuture fut = pendingTest; + + if (fut != null && fut.requestId.equals(msg.requestId())) + fut.onAck(msg); + else if (log.isDebugEnabled()) + log.debug("Ignoring unknown discovery IO test acknowledgement: " + msg.requestId()); + }); + } + + /** + * @param samples Number of samples. + * @param intervalMillis Interval between samples. + * @param payloadSize Payload size. + * @param cancelled Cancellation flag. + * @return Test report. + */ + public String runTest(int samples, long intervalMillis, int payloadSize, BooleanSupplier cancelled) { + A.ensure(ctx.discovery().getInjectedDiscoverySpi() instanceof TcpDiscoverySpi, + "Discovery IO test requires TcpDiscoverySpi."); + A.ensure(U.isLocalNodeCoordinator(ctx.discovery()), "Should be executed on the coordinator node."); + A.ensure(ctx.discovery().aliveServerNodes().size() > 1, + "Discovery IO test requires at least two server nodes."); + A.notNull(cancelled, "cancelled"); + A.ensure(testRunning.compareAndSet(false, true), "Discovery IO test is already running."); + + try { + byte[] payload = new byte[payloadSize]; + long timeout = ctx.config().getNetworkTimeout(); + List ringTimes = new ArrayList<>(samples); + List hopTimes = new ArrayList<>(); + List path = null; + + for (int i = 0; i < samples; i++) { + ensureNotCancelled(cancelled); + + IoTestDiscoveryFuture fut = send(payload); + IoTestDiscoveryAckMessage ack; + + try { + ack = await(fut, timeout, cancelled); + } + catch (IgniteCheckedException e) { + throw new IgniteException("Discovery IO test sample timed out or failed.", e); + } + finally { + if (pendingTest == fut) + pendingTest = null; + } + + if (path == null) + path = new ArrayList<>(ack.path); + else if (!path.equals(ack.path)) + throw new IgniteException("Discovery ring path changed during the test."); + + if (ack.hopTimesMillis.size() != path.size()) + throw new IgniteException("Incomplete discovery ring timing data."); + + while (hopTimes.size() < path.size()) + hopTimes.add(new LongSummaryStatistics()); + + for (int hop = 0; hop < path.size(); hop++) + hopTimes.get(hop).accept(ack.hopTimesMillis.get(hop)); + + ringTimes.add(ack.ringTimeNanos); + + if (i + 1 < samples) + sleep(intervalMillis, cancelled); + } + + return formatSummary(payloadSize, intervalMillis, ringTimes, path, hopTimes); + } + finally { + testRunning.set(false); + } + } + + /** Sends one test message. */ + private IoTestDiscoveryFuture send(byte[] payload) { + IoTestDiscoveryMessage msg = new IoTestDiscoveryMessage(payload); + IoTestDiscoveryFuture fut = new IoTestDiscoveryFuture(msg.id()); + + pendingTest = fut; + + try { + ctx.discovery().sendCustomEvent(msg); + } + catch (IgniteCheckedException | RuntimeException e) { + fut.onDone(e); + } + + return fut; + } + + /** Waits for one sample while observing job cancellation. */ + private static IoTestDiscoveryAckMessage await( + IoTestDiscoveryFuture fut, + long timeout, + BooleanSupplier cancelled + ) throws IgniteCheckedException { + long startNanos = System.nanoTime(); + long remaining = Math.max(1, timeout); + + while (true) { + ensureNotCancelled(cancelled); + + try { + IoTestDiscoveryAckMessage res = fut.get(Math.min(remaining, CANCEL_POLL_INTERVAL_MILLIS)); + + ensureNotCancelled(cancelled); + + return res; + } + catch (IgniteFutureTimeoutCheckedException e) { + ensureNotCancelled(cancelled); + + remaining = timeout - TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos); + + if (remaining <= 0) + throw e; + } + } + } + + /** Sleeps between samples while observing job cancellation. */ + private static void sleep(long millis, BooleanSupplier cancelled) { + for (long remaining = millis; remaining > 0; ) { + ensureNotCancelled(cancelled); + + long delay = Math.min(remaining, CANCEL_POLL_INTERVAL_MILLIS); + + try { + U.sleep(delay); + } + catch (IgniteCheckedException e) { + throw new IgniteException("Discovery IO test was interrupted.", e); + } + + remaining -= delay; + } + } + + /** Fails the test if its management job was cancelled. */ + private static void ensureNotCancelled(BooleanSupplier cancelled) { + if (cancelled.getAsBoolean()) + throw new IgniteException("Discovery IO test was cancelled."); + } + + /** Formats a compact report. */ + private String formatSummary( + int payloadSize, + long intervalMillis, + List ringTimes, + List path, + List hopTimes + ) { + ringTimes.sort(Long::compare); + + StringBuilder sb = new StringBuilder(); + + sb.append("TcpDiscoverySpi ring test\n"); + sb.append("Coordinator: ").append(ctx.localNodeId()).append('\n'); + sb.append("Samples: ").append(ringTimes.size()).append(" | Inter-sample delay: ").append(intervalMillis) + .append(" ms\n"); + sb.append("Request application payload: ").append(payloadSize).append(" bytes\n"); + sb.append("Server ring path: "); + + for (UUID nodeId : path) + sb.append(nodeId).append(" -> "); + + sb.append(ctx.localNodeId()).append('\n'); + sb.append(String.format(Locale.ROOT, + "Request ring latency (submission -> local ACK, ms): min=%.3f, p50=%.3f, p95=%.3f, max=%.3f%n", + toMillis(ringTimes.get(0)), + toMillis(percentile(ringTimes, 50)), + toMillis(percentile(ringTimes, 95)), + toMillis(ringTimes.get(ringTimes.size() - 1)))); + sb.append("Estimated per-hop one-way message delay (ms, min/avg/max):\n"); + sb.append(" Clock assumption: synchronized wall clocks; negative values indicate clock offset or ") + .append("wall-clock adjustment.\n"); + + for (int hop = 0; hop < path.size(); hop++) { + LongSummaryStatistics timing = hopTimes.get(hop); + + sb.append(String.format(Locale.ROOT, " %s -> %s: min=%d, avg=%.3f, max=%d%n", + path.get(hop), path.get((hop + 1) % path.size()), + timing.getMin(), timing.getAverage(), timing.getMax())); + } + + return sb.toString(); + } + + /** Returns the nearest-rank percentile. */ + private static long percentile(List sorted, int percentile) { + int idx = (int)Math.ceil(sorted.size() * percentile / 100.0) - 1; + + return sorted.get(idx); + } + + /** Converts nanoseconds to milliseconds. */ + private static double toMillis(long nanos) { + return nanos / 1_000_000.0; + } + + /** Pending discovery test. */ + private static class IoTestDiscoveryFuture extends GridFutureAdapter { + /** Request ID. */ + private final IgniteUuid requestId; + + /** Local start timestamp. */ + private final long startNanos = System.nanoTime(); + + /** @param requestId Request ID. */ + IoTestDiscoveryFuture(IgniteUuid requestId) { + this.requestId = requestId; + } + + /** Completes this sample from the discovery listener. */ + void onAck(IoTestDiscoveryAckMessage ack) { + ack.ringTimeNanos = System.nanoTime() - startNanos; + + onDone(ack); + } + } +} diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/IoTestDiscoveryMessage.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/IoTestDiscoveryMessage.java new file mode 100644 index 0000000000000..cf90d106bb842 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/IoTestDiscoveryMessage.java @@ -0,0 +1,88 @@ +/* + * 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.managers.discovery; + +import java.util.ArrayList; +import java.util.List; +import java.util.UUID; +import org.apache.ignite.IgniteCheckedException; +import org.apache.ignite.internal.MarshallableMessage; +import org.apache.ignite.internal.Order; +import org.apache.ignite.lang.IgniteUuid; +import org.apache.ignite.marshaller.Marshaller; +import org.apache.ignite.plugin.extensions.communication.MessageFactory; +import org.apache.ignite.spi.discovery.DiscoverySpiCustomMessage; +import org.jetbrains.annotations.Nullable; + +/** Mutable message used to record a Discovery SPI ring path. */ +public class IoTestDiscoveryMessage extends DiscoveryServerOnlyCustomMessage implements MarshallableMessage { + /** Payload. */ + @Order(0) + byte[] payload; + + /** Ordered server node path. */ + @Order(1) + List path; + + /** Current hop serialization-start timestamp from the sender wall clock. */ + @Order(2) + long hopSendTsMillis; + + /** Estimated one-way delay from serialization start to deserialization completion for every completed ring hop. */ + @Order(3) + List hopTimesMillis; + + /** Empty constructor for {@link MessageFactory}. */ + public IoTestDiscoveryMessage() { + // No-op. + } + + /** @param payload Payload. */ + public IoTestDiscoveryMessage(byte[] payload) { + super(IgniteUuid.randomUuid()); + + this.payload = payload; + path = new ArrayList<>(); + hopTimesMillis = new ArrayList<>(); + } + + /** @param nodeId Node that processed this message. */ + public void onProcessed(UUID nodeId) { + path.add(nodeId); + } + + /** {@inheritDoc} */ + @Override public void prepareMarshal(Marshaller marsh) throws IgniteCheckedException { + hopSendTsMillis = System.currentTimeMillis(); + } + + /** {@inheritDoc} */ + @Override public void finishUnmarshal(Marshaller marsh, ClassLoader clsLdr) throws IgniteCheckedException { + hopTimesMillis.add(System.currentTimeMillis() - hopSendTsMillis); + } + + /** {@inheritDoc} */ + @Override public boolean isMutable() { + return true; + } + + /** {@inheritDoc} */ + @Override public @Nullable DiscoverySpiCustomMessage ackMessage() { + return new IoTestDiscoveryAckMessage(this); + } +} diff --git a/modules/core/src/main/java/org/apache/ignite/mxbean/IgniteMXBean.java b/modules/core/src/main/java/org/apache/ignite/mxbean/IgniteMXBean.java index 06fe8ee935e6b..dc9343106fe33 100644 --- a/modules/core/src/main/java/org/apache/ignite/mxbean/IgniteMXBean.java +++ b/modules/core/src/main/java/org/apache/ignite/mxbean/IgniteMXBean.java @@ -627,7 +627,9 @@ public boolean pingNodeByAddress( * @param rangesCnt Ranges count in resulting histogram. * @param payLoadSize Payload size in bytes. * @param procFromNioThread {@code True} to process requests in NIO threads. + * @deprecated Use the {@code io-test} command instead. */ + @Deprecated @MXBeanDescription("Runs IO latency test against all remote server nodes in cluster.") void runIoTest( @MXBeanParameter(name = "warmup", description = "Warmup duration (millis).") diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java index 49b2f6e47e2c0..31ae169a364c8 100644 --- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java +++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java @@ -6182,8 +6182,9 @@ private void processCustomMessage(TcpDiscoveryCustomEventMessage msg, boolean wa DiscoverySpiCustomMessage nextMsg = customMsg.ackMessage(); if (nextMsg != null) { - TcpDiscoveryCustomEventMessage ackMsg = new TcpDiscoveryCustomEventMessage( - getLocalNodeId(), nextMsg); + TcpDiscoveryCustomEventMessage ackMsg = nextMsg instanceof DiscoveryServerOnlyCustomMessage + ? new TcpDiscoveryServerOnlyCustomEventMessage(getLocalNodeId(), nextMsg) + : new TcpDiscoveryCustomEventMessage(getLocalNodeId(), nextMsg); ackMsg.topologyVersion(msg.topologyVersion()); ackMsg.opCtxMsg = operationCtxDispatcher.collectDistributedAttributeValues(); diff --git a/modules/core/src/test/java/org/apache/ignite/internal/managers/communication/IgniteIoTestMessagesTest.java b/modules/core/src/test/java/org/apache/ignite/internal/managers/communication/IgniteIoTestMessagesTest.java index 46b0adfd4598b..3e0a50820ec62 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/managers/communication/IgniteIoTestMessagesTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/managers/communication/IgniteIoTestMessagesTest.java @@ -26,9 +26,7 @@ import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest; import org.junit.Test; -/** - * - */ +/** Tests Communication SPI test messages. */ public class IgniteIoTestMessagesTest extends GridCommonAbstractTest { /** {@inheritDoc} */ @Override protected void beforeTestsStarted() throws Exception { @@ -42,23 +40,23 @@ public class IgniteIoTestMessagesTest extends GridCommonAbstractTest { /** */ @Test - public void testIoTestMessages() { + public void testIoTestMessages() throws Exception { + byte[] payload = new byte[1024]; + for (Ignite node : G.allGrids()) { IgniteKernal ignite = (IgniteKernal)node; - + IoTestHandler ioTest = ignite.context().io().ioTest(); List rmts = new ArrayList<>(ignite.cluster().forRemotes().nodes()); assertEquals(4, rmts.size()); for (ClusterNode rmt : rmts) { - ignite.sendIoTest(rmt, new byte[1024], false); - - ignite.sendIoTest(rmt, new byte[1024], true); - - ignite.sendIoTest(rmts, new byte[1024], false); - - ignite.sendIoTest(rmts, new byte[1024], true); + ioTest.sendIoTest(rmt, payload, false).get(); + ioTest.sendIoTest(rmt, payload, true).get(); } + + ioTest.sendIoTest(rmts, payload, false).get(); + ioTest.sendIoTest(rmts, payload, true).get(); } } } diff --git a/modules/core/src/test/java/org/apache/ignite/internal/thread/context/OperationContextAttributePropagationTest.java b/modules/core/src/test/java/org/apache/ignite/internal/thread/context/OperationContextAttributePropagationTest.java index 17bcf34b58a2d..26a464f37543c 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/thread/context/OperationContextAttributePropagationTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/thread/context/OperationContextAttributePropagationTest.java @@ -211,8 +211,8 @@ private void checkOperationContextCommunicationTransmission( to.context().io().addMessageListener(TOPIC_IO_TEST, lsnr); try { - from.context().io().sendIoTest(node(from, to), null, false); - from.context().io().sendIoTest(node(from, to), null, true); + from.context().io().ioTest().sendIoTest(node(from, to), null, false); + from.context().io().ioTest().sendIoTest(node(from, to), null, true); assertTrue(rcvLatch.await(getTestTimeout(), MILLISECONDS)); } diff --git a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationInverseConnectionEstablishingTest.java b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationInverseConnectionEstablishingTest.java index beed8d787a073..755ac29c194c8 100644 --- a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationInverseConnectionEstablishingTest.java +++ b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTcpCommunicationInverseConnectionEstablishingTest.java @@ -222,7 +222,7 @@ public void testClientReconnectDuringInverseConnection() throws Exception { IgniteInternalFuture fut = GridTestUtils.runAsync(() -> { ClusterNode clientNode = grid(3).context().discovery().node(clientNodeId); - grid(3).context().io().sendIoTest(clientNode, new byte[10], false); + grid(3).context().io().ioTest().sendIoTest(clientNode, new byte[10], false); }); doSleep(2000L); // Client failover timeout is 8 seconds. @@ -308,7 +308,7 @@ public void testClientSkipsInverseConnectionResponse() throws Exception { CommunicationWorkerThreadUtils.onNodeLeft(spi, clientNode.consistentId(), clientNode.id()); IgniteInternalFuture fut = GridTestUtils.runAsync(() -> - srv.context().io().sendIoTest(clientNode, new byte[10], false).get() + srv.context().io().ioTest().sendIoTest(clientNode, new byte[10], false).get() ); assertTrue(GridTestUtils.waitForCondition(fut::isDone, 30_000)); @@ -360,7 +360,7 @@ public void testClientSkippingInverseConnResponseIsForciblyFailed() throws Excep CommunicationWorkerThreadUtils.onNodeLeft(spi, clientNode.consistentId(), clientNode.id()); IgniteInternalFuture fut = GridTestUtils.runAsync(() -> - srv.context().io().sendIoTest(clientNode, new byte[10], false).get() + srv.context().io().ioTest().sendIoTest(clientNode, new byte[10], false).get() ); assertTrue(GridTestUtils.waitForCondition(clientFailedEvtFlag::get, 10_000)); diff --git a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTotallyUnreachableClientTest.java b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTotallyUnreachableClientTest.java index 333048c1339c8..44cfbc547a31e 100644 --- a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTotallyUnreachableClientTest.java +++ b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/GridTotallyUnreachableClientTest.java @@ -83,7 +83,7 @@ public void testTotallyUnreachableClient() throws Exception { ClusterNode clientNode1 = client1.localNode(); IgniteInternalFuture fut = GridTestUtils.runAsync(() -> - srv.context().io().sendIoTest(clientNode1, new byte[10], false).get() + srv.context().io().ioTest().sendIoTest(clientNode1, new byte[10], false).get() ); fut.get(30, TimeUnit.SECONDS); @@ -94,7 +94,7 @@ public void testTotallyUnreachableClient() throws Exception { GridTestUtils.assertThrowsAnyCause(log, () -> { return GridTestUtils.runAsync(() -> - client2.context().io().sendIoTest(clientNode1, new byte[10], false).get() + client2.context().io().ioTest().sendIoTest(clientNode1, new byte[10], false).get() ).get(30, TimeUnit.SECONDS); }, IgniteSpiException.class, "Cannot send"); } diff --git a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/IgniteTcpCommunicationHandshakeWaitTest.java b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/IgniteTcpCommunicationHandshakeWaitTest.java index f480a6901ccea..132f30d778962 100644 --- a/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/IgniteTcpCommunicationHandshakeWaitTest.java +++ b/modules/core/src/test/java/org/apache/ignite/spi/communication/tcp/IgniteTcpCommunicationHandshakeWaitTest.java @@ -100,7 +100,7 @@ public void testHandshakeOnNodeJoining() throws Exception { assertEquals(3, nodes.size()); - return ignite.context().io().sendIoTest(new ArrayList<>(nodes), null, true).get(); + return ignite.context().io().ioTest().sendIoTest(new ArrayList<>(nodes), null, true).get(); }); startGrid("srv3"); diff --git a/modules/yardstick/src/main/java/org/apache/ignite/yardstick/cache/IgniteIoTestSendAllBenchmark.java b/modules/yardstick/src/main/java/org/apache/ignite/yardstick/cache/IgniteIoTestSendAllBenchmark.java index d77c4a636da87..10fad1e69bd0b 100644 --- a/modules/yardstick/src/main/java/org/apache/ignite/yardstick/cache/IgniteIoTestSendAllBenchmark.java +++ b/modules/yardstick/src/main/java/org/apache/ignite/yardstick/cache/IgniteIoTestSendAllBenchmark.java @@ -25,7 +25,7 @@ public class IgniteIoTestSendAllBenchmark extends IgniteIoTestAbstractBenchmark { /** {@inheritDoc} */ @Override public boolean test(Map ctx) throws Exception { - ignite.sendIoTest(targetNodes, null, false).get(); + ignite.context().io().ioTest().sendIoTest(targetNodes, null, false).get(); return true; } diff --git a/modules/yardstick/src/main/java/org/apache/ignite/yardstick/cache/IgniteIoTestSendRandomBenchmark.java b/modules/yardstick/src/main/java/org/apache/ignite/yardstick/cache/IgniteIoTestSendRandomBenchmark.java index ac3a07034a3a6..092e5ff0e1933 100644 --- a/modules/yardstick/src/main/java/org/apache/ignite/yardstick/cache/IgniteIoTestSendRandomBenchmark.java +++ b/modules/yardstick/src/main/java/org/apache/ignite/yardstick/cache/IgniteIoTestSendRandomBenchmark.java @@ -28,7 +28,7 @@ public class IgniteIoTestSendRandomBenchmark extends IgniteIoTestAbstractBenchma @Override public boolean test(Map ctx) throws Exception { ClusterNode node = targetNodes.get(nextRandom(targetNodes.size())); - ignite.sendIoTest(node, null, false).get(); + ignite.context().io().ioTest().sendIoTest(node, null, false).get(); return true; } diff --git a/modules/yardstick/src/main/java/org/apache/ignite/yardstick/io/IgniteIoTestSendAllBenchmark.java b/modules/yardstick/src/main/java/org/apache/ignite/yardstick/io/IgniteIoTestSendAllBenchmark.java index 9011910aafe7a..e7f8b2b170411 100644 --- a/modules/yardstick/src/main/java/org/apache/ignite/yardstick/io/IgniteIoTestSendAllBenchmark.java +++ b/modules/yardstick/src/main/java/org/apache/ignite/yardstick/io/IgniteIoTestSendAllBenchmark.java @@ -25,7 +25,7 @@ public class IgniteIoTestSendAllBenchmark extends IgniteIoTestAbstractBenchmark { /** {@inheritDoc} */ @Override public boolean test(Map ctx) throws Exception { - ignite.sendIoTest(targetNodes, null, false).get(); + ignite.context().io().ioTest().sendIoTest(targetNodes, null, false).get(); return true; } diff --git a/modules/yardstick/src/main/java/org/apache/ignite/yardstick/io/IgniteIoTestSendRandomBenchmark.java b/modules/yardstick/src/main/java/org/apache/ignite/yardstick/io/IgniteIoTestSendRandomBenchmark.java index 88368e0cd3107..5b1f209d4c197 100644 --- a/modules/yardstick/src/main/java/org/apache/ignite/yardstick/io/IgniteIoTestSendRandomBenchmark.java +++ b/modules/yardstick/src/main/java/org/apache/ignite/yardstick/io/IgniteIoTestSendRandomBenchmark.java @@ -28,7 +28,7 @@ public class IgniteIoTestSendRandomBenchmark extends IgniteIoTestAbstractBenchma @Override public boolean test(Map ctx) throws Exception { ClusterNode node = targetNodes.get(nextRandom(targetNodes.size())); - ignite.sendIoTest(node, null, false).get(); + ignite.context().io().ioTest().sendIoTest(node, null, false).get(); return true; }