You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@geode.apache.org by GitBox <gi...@apache.org> on 2021/10/01 23:22:26 UTC
[GitHub] [geode] Bill opened a new pull request #6930: P2P messaging concurrency test
Bill opened a new pull request #6930:
URL: https://github.com/apache/geode/pull/6930
A new DUnit-based concurrency test that stresses P2P/Direct Channel messaging.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@geode.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [geode] Bill commented on a change in pull request #6930: P2P messaging concurrency test
Posted by GitBox <gi...@apache.org>.
Bill commented on a change in pull request #6930:
URL: https://github.com/apache/geode/pull/6930#discussion_r734857336
##########
File path: geode-core/src/distributedTest/java/org/apache/geode/distributed/internal/P2PMessagingConcurrencyDUnitTest.java
##########
@@ -0,0 +1,314 @@
+/*
+ * 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.geode.distributed.internal;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.security.GeneralSecurityException;
+import java.util.Properties;
+import java.util.Random;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.LongAdder;
+
+import org.jetbrains.annotations.NotNull;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import org.apache.geode.cache.CacheFactory;
+import org.apache.geode.cache.ssl.CertStores;
+import org.apache.geode.cache.ssl.CertificateBuilder;
+import org.apache.geode.cache.ssl.CertificateMaterial;
+import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.internal.serialization.DeserializationContext;
+import org.apache.geode.internal.serialization.SerializationContext;
+import org.apache.geode.test.dunit.rules.ClusterStartupRule;
+import org.apache.geode.test.dunit.rules.MemberVM;
+import org.apache.geode.test.junit.categories.MembershipTest;
+import org.apache.geode.test.version.VersionManager;
+
+/**
+ * Tests one-way P2P messaging between two peers. A shared,
+ * ordered connection is used and many concurrent tasks
+ * compete on the sending side. Tests with TLS enabled
+ * to exercise ByteBufferSharing and friends.
+ */
+@Category({MembershipTest.class})
+public class P2PMessagingConcurrencyDUnitTest {
+
+ // how many messages will each sender generate?
+ private static final int MESSAGES_PER_SENDER = 1_000;
+
+ // number of concurrent (sending) tasks to run
+ private static final int SENDER_COUNT = 10;
+
+ // (exclusive) upper bound of random message size, in bytes
+ private static final int LARGEST_MESSAGE_BOUND = 32 * 1024 + 2; // 32KiB + 2
+
+ // random seed
+ private static final int RANDOM_SEED = 1234;
+
+ /*
+ At the time this comment was written, ClusterStartupRule was ignoring the vmCount.
+ Nevertheless since we need only 3, we're specifying it.
+ */
+ @Rule
+ public final ClusterStartupRule clusterStartupRule = new ClusterStartupRule(3);
+
+ private MemberVM sender;
+ private MemberVM receiver;
+
+ /*
+ * bytes sent on sender JVM, bytes received on receiver JVM
+ * (not used in test JVM)
+ */
+ private static LongAdder bytesTransferredAdder;
+
+ @Before
+ public void before() throws GeneralSecurityException, IOException {
+ final Properties configuration = gemFireConfiguration();
+
+ final MemberVM locator =
+ clusterStartupRule.startLocatorVM(0, 0, VersionManager.CURRENT_VERSION,
+ x -> x.withProperties(configuration).withConnectionToLocator()
+ .withoutClusterConfigurationService().withoutManagementRestService());
+
+ sender = clusterStartupRule.startServerVM(1, configuration, locator.getPort());
+ receiver = clusterStartupRule.startServerVM(2, configuration, locator.getPort());
+ }
+
+ @Test
+ public void testP2PMessagingWithTLS() {
+
+ final InternalDistributedMember receiverMember =
+ receiver.invoke(() -> {
+
+ bytesTransferredAdder = new LongAdder();
+
+ final ClusterDistributionManager cdm = getCDM();
+ final InternalDistributedMember localMember = cdm.getDistribution().getLocalMember();
+ return localMember;
+
+ });
+
+ sender.invoke(() -> {
+
+ bytesTransferredAdder = new LongAdder();
+
+ final ClusterDistributionManager cdm = getCDM();
+ final Random random = new Random(RANDOM_SEED);
+ final AtomicInteger nextSenderId = new AtomicInteger();
+
+ /*
+ When this comment was written the nThreads parameter to the thread pool
+ constructor was SENDER_COUNT. When SENDER_COUNT is much larger than the
+ number of CPUs that is counterproductive. In an ideal world we'd want
+ only as many threads as CPUs here. OTOH the P2P messaging system at the
+ time this comment was written, used blocking I/O, so we were not, as it
+ turns out, living in that ideal world.
+ */
+ final ExecutorService executor = Executors.newFixedThreadPool(SENDER_COUNT);
Review comment:
I experimented w/ `CyclicBarrier` but it didn't seem to be an improvement. That thing is mainly beneficial when you want to do some (single) task after the barrier is reached—common example is summarizing results. And it's useful when you want to coordinate N tasks, let 'em all finish, then let em run again. In my case neither of those things were needed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@geode.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [geode] Bill commented on a change in pull request #6930: GEODE-9763: DUnit Test of Concurrent P2P Messaging with TLS Enabled
Posted by GitBox <gi...@apache.org>.
Bill commented on a change in pull request #6930:
URL: https://github.com/apache/geode/pull/6930#discussion_r734898674
##########
File path: geode-core/src/distributedTest/java/org/apache/geode/distributed/internal/P2PMessagingConcurrencyDUnitTest.java
##########
@@ -0,0 +1,314 @@
+/*
+ * 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.geode.distributed.internal;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.security.GeneralSecurityException;
+import java.util.Properties;
+import java.util.Random;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.LongAdder;
+
+import org.jetbrains.annotations.NotNull;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import org.apache.geode.cache.CacheFactory;
+import org.apache.geode.cache.ssl.CertStores;
+import org.apache.geode.cache.ssl.CertificateBuilder;
+import org.apache.geode.cache.ssl.CertificateMaterial;
+import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.internal.serialization.DeserializationContext;
+import org.apache.geode.internal.serialization.SerializationContext;
+import org.apache.geode.test.dunit.rules.ClusterStartupRule;
+import org.apache.geode.test.dunit.rules.MemberVM;
+import org.apache.geode.test.junit.categories.MembershipTest;
+import org.apache.geode.test.version.VersionManager;
+
+/**
+ * Tests one-way P2P messaging between two peers. A shared,
+ * ordered connection is used and many concurrent tasks
+ * compete on the sending side. Tests with TLS enabled
+ * to exercise ByteBufferSharing and friends.
+ */
+@Category({MembershipTest.class})
+public class P2PMessagingConcurrencyDUnitTest {
+
+ // how many messages will each sender generate?
+ private static final int MESSAGES_PER_SENDER = 1_000;
+
+ // number of concurrent (sending) tasks to run
+ private static final int SENDER_COUNT = 10;
+
+ // (exclusive) upper bound of random message size, in bytes
+ private static final int LARGEST_MESSAGE_BOUND = 32 * 1024 + 2; // 32KiB + 2
+
+ // random seed
+ private static final int RANDOM_SEED = 1234;
+
+ /*
+ At the time this comment was written, ClusterStartupRule was ignoring the vmCount.
+ Nevertheless since we need only 3, we're specifying it.
+ */
+ @Rule
+ public final ClusterStartupRule clusterStartupRule = new ClusterStartupRule(3);
+
+ private MemberVM sender;
+ private MemberVM receiver;
+
+ /*
+ * bytes sent on sender JVM, bytes received on receiver JVM
+ * (not used in test JVM)
+ */
+ private static LongAdder bytesTransferredAdder;
+
+ @Before
+ public void before() throws GeneralSecurityException, IOException {
+ final Properties configuration = gemFireConfiguration();
+
+ final MemberVM locator =
+ clusterStartupRule.startLocatorVM(0, 0, VersionManager.CURRENT_VERSION,
+ x -> x.withProperties(configuration).withConnectionToLocator()
+ .withoutClusterConfigurationService().withoutManagementRestService());
+
+ sender = clusterStartupRule.startServerVM(1, configuration, locator.getPort());
+ receiver = clusterStartupRule.startServerVM(2, configuration, locator.getPort());
+ }
+
+ @Test
+ public void testP2PMessagingWithTLS() {
+
+ final InternalDistributedMember receiverMember =
+ receiver.invoke(() -> {
+
+ bytesTransferredAdder = new LongAdder();
+
+ final ClusterDistributionManager cdm = getCDM();
+ final InternalDistributedMember localMember = cdm.getDistribution().getLocalMember();
+ return localMember;
+
+ });
+
+ sender.invoke(() -> {
+
+ bytesTransferredAdder = new LongAdder();
+
+ final ClusterDistributionManager cdm = getCDM();
+ final Random random = new Random(RANDOM_SEED);
+ final AtomicInteger nextSenderId = new AtomicInteger();
+
+ /*
+ When this comment was written the nThreads parameter to the thread pool
+ constructor was SENDER_COUNT. When SENDER_COUNT is much larger than the
+ number of CPUs that is counterproductive. In an ideal world we'd want
+ only as many threads as CPUs here. OTOH the P2P messaging system at the
+ time this comment was written, used blocking I/O, so we were not, as it
+ turns out, living in that ideal world.
+ */
+ final ExecutorService executor = Executors.newFixedThreadPool(SENDER_COUNT);
Review comment:
I experimented with the `DistributedExecutorServiceRule` and while the test would benefit from the thread-dump-on-hangs, it comes at the cost of introducing starting an extra JVM for an (unsecured) locator that I don't want/need. So I stuck with my own executor service for the test for now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@geode.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [geode] Bill commented on a change in pull request #6930: GEODE-9763: DUnit Test of Concurrent P2P Messaging with TLS Enabled
Posted by GitBox <gi...@apache.org>.
Bill commented on a change in pull request #6930:
URL: https://github.com/apache/geode/pull/6930#discussion_r736805913
##########
File path: geode-core/src/distributedTest/java/org/apache/geode/distributed/internal/P2PMessagingConcurrencyDUnitTest.java
##########
@@ -0,0 +1,314 @@
+/*
+ * 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.geode.distributed.internal;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.security.GeneralSecurityException;
+import java.util.Properties;
+import java.util.Random;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.LongAdder;
+
+import org.jetbrains.annotations.NotNull;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import org.apache.geode.cache.CacheFactory;
+import org.apache.geode.cache.ssl.CertStores;
+import org.apache.geode.cache.ssl.CertificateBuilder;
+import org.apache.geode.cache.ssl.CertificateMaterial;
+import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.internal.serialization.DeserializationContext;
+import org.apache.geode.internal.serialization.SerializationContext;
+import org.apache.geode.test.dunit.rules.ClusterStartupRule;
+import org.apache.geode.test.dunit.rules.MemberVM;
+import org.apache.geode.test.junit.categories.MembershipTest;
+import org.apache.geode.test.version.VersionManager;
+
+/**
+ * Tests one-way P2P messaging between two peers. A shared,
+ * ordered connection is used and many concurrent tasks
+ * compete on the sending side. Tests with TLS enabled
+ * to exercise ByteBufferSharing and friends.
+ */
+@Category({MembershipTest.class})
+public class P2PMessagingConcurrencyDUnitTest {
+
+ // how many messages will each sender generate?
+ private static final int MESSAGES_PER_SENDER = 1_000;
+
+ // number of concurrent (sending) tasks to run
+ private static final int SENDER_COUNT = 10;
+
+ // (exclusive) upper bound of random message size, in bytes
+ private static final int LARGEST_MESSAGE_BOUND = 32 * 1024 + 2; // 32KiB + 2
+
+ // random seed
+ private static final int RANDOM_SEED = 1234;
+
+ /*
+ At the time this comment was written, ClusterStartupRule was ignoring the vmCount.
+ Nevertheless since we need only 3, we're specifying it.
+ */
+ @Rule
+ public final ClusterStartupRule clusterStartupRule = new ClusterStartupRule(3);
+
+ private MemberVM sender;
+ private MemberVM receiver;
+
+ /*
+ * bytes sent on sender JVM, bytes received on receiver JVM
+ * (not used in test JVM)
+ */
+ private static LongAdder bytesTransferredAdder;
+
+ @Before
+ public void before() throws GeneralSecurityException, IOException {
+ final Properties configuration = gemFireConfiguration();
+
+ final MemberVM locator =
+ clusterStartupRule.startLocatorVM(0, 0, VersionManager.CURRENT_VERSION,
+ x -> x.withProperties(configuration).withConnectionToLocator()
+ .withoutClusterConfigurationService().withoutManagementRestService());
+
+ sender = clusterStartupRule.startServerVM(1, configuration, locator.getPort());
+ receiver = clusterStartupRule.startServerVM(2, configuration, locator.getPort());
+ }
+
+ @Test
+ public void testP2PMessagingWithTLS() {
+
+ final InternalDistributedMember receiverMember =
+ receiver.invoke(() -> {
+
+ bytesTransferredAdder = new LongAdder();
+
+ final ClusterDistributionManager cdm = getCDM();
+ final InternalDistributedMember localMember = cdm.getDistribution().getLocalMember();
+ return localMember;
+
+ });
+
+ sender.invoke(() -> {
+
+ bytesTransferredAdder = new LongAdder();
+
+ final ClusterDistributionManager cdm = getCDM();
+ final Random random = new Random(RANDOM_SEED);
+ final AtomicInteger nextSenderId = new AtomicInteger();
+
+ /*
+ When this comment was written the nThreads parameter to the thread pool
+ constructor was SENDER_COUNT. When SENDER_COUNT is much larger than the
+ number of CPUs that is counterproductive. In an ideal world we'd want
+ only as many threads as CPUs here. OTOH the P2P messaging system at the
+ time this comment was written, used blocking I/O, so we were not, as it
+ turns out, living in that ideal world.
+ */
+ final ExecutorService executor = Executors.newFixedThreadPool(SENDER_COUNT);
Review comment:
I'm using the `DistributedExecutorServiceRule` now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@geode.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [geode] Bill merged pull request #6930: GEODE-9763: DUnit Test of Concurrent P2P Messaging with TLS Enabled
Posted by GitBox <gi...@apache.org>.
Bill merged pull request #6930:
URL: https://github.com/apache/geode/pull/6930
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@geode.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [geode] Bill commented on a change in pull request #6930: GEODE-9763: DUnit Test of Concurrent P2P Messaging with TLS Enabled
Posted by GitBox <gi...@apache.org>.
Bill commented on a change in pull request #6930:
URL: https://github.com/apache/geode/pull/6930#discussion_r736006712
##########
File path: geode-core/src/distributedTest/java/org/apache/geode/distributed/internal/P2PMessagingConcurrencyDUnitTest.java
##########
@@ -0,0 +1,314 @@
+/*
+ * 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.geode.distributed.internal;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.security.GeneralSecurityException;
+import java.util.Properties;
+import java.util.Random;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.LongAdder;
+
+import org.jetbrains.annotations.NotNull;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import org.apache.geode.cache.CacheFactory;
+import org.apache.geode.cache.ssl.CertStores;
+import org.apache.geode.cache.ssl.CertificateBuilder;
+import org.apache.geode.cache.ssl.CertificateMaterial;
+import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.internal.serialization.DeserializationContext;
+import org.apache.geode.internal.serialization.SerializationContext;
+import org.apache.geode.test.dunit.rules.ClusterStartupRule;
+import org.apache.geode.test.dunit.rules.MemberVM;
+import org.apache.geode.test.junit.categories.MembershipTest;
+import org.apache.geode.test.version.VersionManager;
+
+/**
+ * Tests one-way P2P messaging between two peers. A shared,
+ * ordered connection is used and many concurrent tasks
+ * compete on the sending side. Tests with TLS enabled
+ * to exercise ByteBufferSharing and friends.
+ */
+@Category({MembershipTest.class})
+public class P2PMessagingConcurrencyDUnitTest {
+
+ // how many messages will each sender generate?
+ private static final int MESSAGES_PER_SENDER = 1_000;
+
+ // number of concurrent (sending) tasks to run
+ private static final int SENDER_COUNT = 10;
+
+ // (exclusive) upper bound of random message size, in bytes
+ private static final int LARGEST_MESSAGE_BOUND = 32 * 1024 + 2; // 32KiB + 2
+
+ // random seed
+ private static final int RANDOM_SEED = 1234;
+
+ /*
+ At the time this comment was written, ClusterStartupRule was ignoring the vmCount.
+ Nevertheless since we need only 3, we're specifying it.
+ */
+ @Rule
+ public final ClusterStartupRule clusterStartupRule = new ClusterStartupRule(3);
+
+ private MemberVM sender;
+ private MemberVM receiver;
+
+ /*
+ * bytes sent on sender JVM, bytes received on receiver JVM
+ * (not used in test JVM)
+ */
+ private static LongAdder bytesTransferredAdder;
+
+ @Before
+ public void before() throws GeneralSecurityException, IOException {
+ final Properties configuration = gemFireConfiguration();
+
+ final MemberVM locator =
+ clusterStartupRule.startLocatorVM(0, 0, VersionManager.CURRENT_VERSION,
+ x -> x.withProperties(configuration).withConnectionToLocator()
+ .withoutClusterConfigurationService().withoutManagementRestService());
+
+ sender = clusterStartupRule.startServerVM(1, configuration, locator.getPort());
+ receiver = clusterStartupRule.startServerVM(2, configuration, locator.getPort());
+ }
+
+ @Test
+ public void testP2PMessagingWithTLS() {
+
+ final InternalDistributedMember receiverMember =
+ receiver.invoke(() -> {
+
+ bytesTransferredAdder = new LongAdder();
+
+ final ClusterDistributionManager cdm = getCDM();
+ final InternalDistributedMember localMember = cdm.getDistribution().getLocalMember();
+ return localMember;
+
+ });
+
+ sender.invoke(() -> {
+
+ bytesTransferredAdder = new LongAdder();
+
+ final ClusterDistributionManager cdm = getCDM();
+ final Random random = new Random(RANDOM_SEED);
+ final AtomicInteger nextSenderId = new AtomicInteger();
+
+ /*
+ When this comment was written the nThreads parameter to the thread pool
+ constructor was SENDER_COUNT. When SENDER_COUNT is much larger than the
+ number of CPUs that is counterproductive. In an ideal world we'd want
+ only as many threads as CPUs here. OTOH the P2P messaging system at the
+ time this comment was written, used blocking I/O, so we were not, as it
+ turns out, living in that ideal world.
+ */
+ final ExecutorService executor = Executors.newFixedThreadPool(SENDER_COUNT);
+
+ final CountDownLatch startLatch = new CountDownLatch(SENDER_COUNT);
+ final CountDownLatch stopLatch = new CountDownLatch(SENDER_COUNT);
+ final LongAdder failedRecipientCount = new LongAdder();
+
+ final Runnable doSending = () -> {
+ final int senderId = nextSenderId.getAndIncrement();
+ try {
+ startLatch.countDown();
+ startLatch.await();
+ } catch (final InterruptedException e) {
+ throw new RuntimeException("doSending failed", e);
+ }
+ final int firstMessageId = senderId * SENDER_COUNT;
+ for (int messageId = firstMessageId; messageId < firstMessageId
+ + MESSAGES_PER_SENDER; messageId++) {
+ final TestMessage msg = new TestMessage(receiverMember, random, messageId);
+
+ /*
+ * HERE is the Geode API entrypoint we intend to test (putOutgoing()).
+ */
+ final Set<InternalDistributedMember> failedRecipients = cdm.putOutgoing(msg);
+
+ if (failedRecipients != null) {
+ failedRecipientCount.add(failedRecipients.size());
+ }
+ }
+ stopLatch.countDown();
+ };
+
+ for (int i = 0; i < SENDER_COUNT; ++i) {
+ executor.submit(doSending);
+ }
+
+ stopLatch.await();
+
+ stop(executor);
+
+ assertThat(failedRecipientCount.sum()).as("message delivery failed N times").isZero();
+
+ });
+
+ final long bytesSent = sender.invoke(() -> bytesTransferredAdder.sum());
+ final long bytesReceived = receiver.invoke(() -> bytesTransferredAdder.sum());
+
+ assertThat(bytesReceived).as("bytes received != bytes sent").isEqualTo(bytesSent);
+ }
+
+ private static void stop(final ExecutorService executor) {
+ executor.shutdown();
+ try {
+ if (!executor.awaitTermination(800, TimeUnit.MILLISECONDS)) {
+ executor.shutdownNow();
+ }
+ } catch (InterruptedException e) {
+ executor.shutdownNow();
+ }
+ }
+
+ private static ClusterDistributionManager getCDM() {
+ return (ClusterDistributionManager) ((InternalCache) CacheFactory.getAnyInstance())
+ .getDistributionManager();
+ }
+
+ private static class TestMessage extends DistributionMessage {
+
+ /*
+ When this comment was written, messageId wasn't used for anything.
+ The field was added during a misguided attempt to add SHA-256
+ digest verification on sender and receiver. Then I figured out
+ that there's no way to parallelize that (for the sender) so
+ I settled for merely validating the number of bytes transferred.
+ Left the field here in case it comes in handy later.
+ */
+ private volatile int messageId;
+ private volatile Random random;
+
+ TestMessage(final InternalDistributedMember receiver,
+ final Random random, final int messageId) {
+ setRecipient(receiver);
+ this.random = random;
+ this.messageId = messageId;
+ }
+
+ // necessary for deserialization
+ public TestMessage() {
+ random = null;
+ messageId = 0;
+ }
+
+ @Override
+ public int getProcessorType() {
+ return OperationExecutors.STANDARD_EXECUTOR;
+ }
+
+ @Override
+ protected void process(final ClusterDistributionManager dm) {}
+
+ @Override
+ public void toData(final DataOutput out, final SerializationContext context)
+ throws IOException {
+ super.toData(out, context);
+
+ out.writeInt(messageId);
+
+ final int length = random.nextInt(LARGEST_MESSAGE_BOUND);
+
+ out.writeInt(length);
+
+ final byte[] payload = new byte[length];
+ random.nextBytes(payload);
+
+ out.write(payload);
+
+ /*
+ * the LongAdder should ensure that we don't introduce any (much)
+ * synchronization with other concurrent tasks here
+ */
+ bytesTransferredAdder.add(length);
+ }
+
+ @Override
+ public void fromData(final DataInput in, final DeserializationContext context)
+ throws IOException, ClassNotFoundException {
+ super.fromData(in, context);
+
+ final int messageId = in.readInt();
+
+ final int length = in.readInt();
+
+ final byte[] payload = new byte[length];
+
+ in.readFully(payload);
+
+ bytesTransferredAdder.add(length);
+ }
+
+ @Override
+ public int getDSFID() {
+ return NO_FIXED_ID; // for testing only!
+ }
+ }
+
+ @NotNull
+ private static Properties gemFireConfiguration()
+ throws GeneralSecurityException, IOException {
+
+ final Properties props = securityProperties();
+
+ /*
+ * This is something we intend to test!
+ * Send all messages, from all threads, on a single socket per recipient.
+ * maintenance tip: to see what kind of connection you're getting you can
+ * uncomment logging over in DirectChannel.sendToMany()
+ */
+ props.put("conserve-sockets", "true"); // careful: if you set a boolean it doesn't take hold!
Review comment:
resolved
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@geode.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [geode] Bill commented on a change in pull request #6930: P2P messaging concurrency test
Posted by GitBox <gi...@apache.org>.
Bill commented on a change in pull request #6930:
URL: https://github.com/apache/geode/pull/6930#discussion_r725402508
##########
File path: geode-core/src/distributedTest/java/org/apache/geode/distributed/internal/P2PMessagingConcurrencyDUnitTest.java
##########
@@ -0,0 +1,314 @@
+/*
+ * 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.geode.distributed.internal;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.security.GeneralSecurityException;
+import java.util.Properties;
+import java.util.Random;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.LongAdder;
+
+import org.jetbrains.annotations.NotNull;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import org.apache.geode.cache.CacheFactory;
+import org.apache.geode.cache.ssl.CertStores;
+import org.apache.geode.cache.ssl.CertificateBuilder;
+import org.apache.geode.cache.ssl.CertificateMaterial;
+import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.internal.serialization.DeserializationContext;
+import org.apache.geode.internal.serialization.SerializationContext;
+import org.apache.geode.test.dunit.rules.ClusterStartupRule;
+import org.apache.geode.test.dunit.rules.MemberVM;
+import org.apache.geode.test.junit.categories.MembershipTest;
+import org.apache.geode.test.version.VersionManager;
+
+/**
+ * Tests one-way P2P messaging between two peers. A shared,
+ * ordered connection is used and many concurrent tasks
+ * compete on the sending side. Tests with TLS enabled
+ * to exercise ByteBufferSharing and friends.
+ */
+@Category({MembershipTest.class})
+public class P2PMessagingConcurrencyDUnitTest {
+
+ // how many messages will each sender generate?
+ private static final int MESSAGES_PER_SENDER = 1_000;
+
+ // number of concurrent (sending) tasks to run
+ private static final int SENDER_COUNT = 10;
+
+ // (exclusive) upper bound of random message size, in bytes
+ private static final int LARGEST_MESSAGE_BOUND = 32 * 1024 + 2; // 32KiB + 2
+
+ // random seed
+ private static final int RANDOM_SEED = 1234;
+
+ /*
+ At the time this comment was written, ClusterStartupRule was ignoring the vmCount.
+ Nevertheless since we need only 3, we're specifying it.
+ */
+ @Rule
+ public final ClusterStartupRule clusterStartupRule = new ClusterStartupRule(3);
+
+ private MemberVM sender;
+ private MemberVM receiver;
+
+ /*
+ * bytes sent on sender JVM, bytes received on receiver JVM
+ * (not used in test JVM)
+ */
+ private static LongAdder bytesTransferredAdder;
+
+ @Before
+ public void before() throws GeneralSecurityException, IOException {
+ final Properties configuration = gemFireConfiguration();
+
+ final MemberVM locator =
+ clusterStartupRule.startLocatorVM(0, 0, VersionManager.CURRENT_VERSION,
+ x -> x.withProperties(configuration).withConnectionToLocator()
+ .withoutClusterConfigurationService().withoutManagementRestService());
+
+ sender = clusterStartupRule.startServerVM(1, configuration, locator.getPort());
+ receiver = clusterStartupRule.startServerVM(2, configuration, locator.getPort());
+ }
+
+ @Test
+ public void testP2PMessagingWithTLS() {
+
+ final InternalDistributedMember receiverMember =
+ receiver.invoke(() -> {
+
+ bytesTransferredAdder = new LongAdder();
+
+ final ClusterDistributionManager cdm = getCDM();
+ final InternalDistributedMember localMember = cdm.getDistribution().getLocalMember();
+ return localMember;
+
+ });
+
+ sender.invoke(() -> {
+
+ bytesTransferredAdder = new LongAdder();
+
+ final ClusterDistributionManager cdm = getCDM();
+ final Random random = new Random(RANDOM_SEED);
+ final AtomicInteger nextSenderId = new AtomicInteger();
+
+ /*
+ When this comment was written the nThreads parameter to the thread pool
+ constructor was SENDER_COUNT. When SENDER_COUNT is much larger than the
+ number of CPUs that is counterproductive. In an ideal world we'd want
+ only as many threads as CPUs here. OTOH the P2P messaging system at the
+ time this comment was written, used blocking I/O, so we were not, as it
+ turns out, living in that ideal world.
+ */
+ final ExecutorService executor = Executors.newFixedThreadPool(SENDER_COUNT);
Review comment:
@kirklund and I looked into `DistributedExecutorServiceRule` together and decided it had a problem that needed fixing before it would be suitable for use here: the constructor doesn't take a thread count, rather a VM count only. Once there is a constructor that takes thread count, maybe it'll be suitable.
Will see how I like `CyclicBarrier` though. That might make things a little more terse…
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@geode.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [geode] Bill commented on pull request #6930: GEODE-9763: DUnit Test of Concurrent P2P Messaging with TLS Enabled
Posted by GitBox <gi...@apache.org>.
Bill commented on pull request #6930:
URL: https://github.com/apache/geode/pull/6930#issuecomment-952369843
That CI failure is due to [GEODE-9622](https://issues.apache.org/jira/browse/GEODE-9622).
Proceeding with merge!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@geode.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [geode] kirklund commented on a change in pull request #6930: P2P messaging concurrency test
Posted by GitBox <gi...@apache.org>.
kirklund commented on a change in pull request #6930:
URL: https://github.com/apache/geode/pull/6930#discussion_r733127317
##########
File path: geode-core/src/test/java/org/apache/geode/internal/net/ByteBufferConcurrencyTest.java
##########
@@ -15,6 +15,7 @@
package org.apache.geode.internal.net;
+import static java.lang.Thread.yield;
Review comment:
Well, I did some digging and recommendation is to only use `Thread.yield` for debugging or testing so maybe it's ok...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@geode.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [geode] Bill commented on a change in pull request #6930: GEODE-9763: DUnit Test of Concurrent P2P Messaging with TLS Enabled
Posted by GitBox <gi...@apache.org>.
Bill commented on a change in pull request #6930:
URL: https://github.com/apache/geode/pull/6930#discussion_r734898414
##########
File path: geode-core/src/test/java/org/apache/geode/internal/net/ByteBufferConcurrencyTest.java
##########
@@ -15,6 +15,7 @@
package org.apache.geode.internal.net;
+import static java.lang.Thread.yield;
Review comment:
Well I changed it to sleep so that I could make the hold time (for the resource) configurable in the test. When trying to expose concurrency bugs it's nice to be able to introduce longer hold times that we might get with yield.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@geode.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [geode] Bill commented on a change in pull request #6930: GEODE-9763: DUnit Test of Concurrent P2P Messaging with TLS Enabled
Posted by GitBox <gi...@apache.org>.
Bill commented on a change in pull request #6930:
URL: https://github.com/apache/geode/pull/6930#discussion_r736006893
##########
File path: geode-core/src/test/java/org/apache/geode/internal/net/ByteBufferConcurrencyTest.java
##########
@@ -40,126 +41,145 @@
@LoopRunnerConfig(count = 100)
public class ByteBufferConcurrencyTest {
- private BufferPool poolMock;
-
@Test
- public void concurrentDestructAndOpenCloseShouldReturnToPoolOnce(ParallelExecutor executor)
+ public void concurrentDestructAndOpenCloseShouldReturnToPoolOnce(final ParallelExecutor executor)
throws Exception {
- poolMock = mock(BufferPool.class);
- ByteBuffer someBuffer = ByteBuffer.allocate(1);
- ByteBufferVendor sharing =
+ final BufferPool poolMock = mock(BufferPool.class);
+ final ByteBuffer someBuffer = ByteBuffer.allocate(1);
+ final ByteBufferVendor vendor =
new ByteBufferVendor(someBuffer, BufferPool.BufferType.TRACKED_SENDER,
poolMock);
- executor.inParallel(() -> {
- sharing.destruct();
- });
- executor.inParallel(() -> {
- try {
- try (ByteBufferSharing localSharing = sharing.open()) {
- localSharing.getBuffer();
- }
+
+ final RunnableWithException useBuffer = () -> {
+ try (final ByteBufferSharing sharing = vendor.open()) {
+ useBuffer(sharing);
} catch (IOException e) {
- // It's ok to get an IOException if the sharing was destroyed before this runs
+ // It's ok to get an IOException if the sharing was destruct()ed before this runs
}
+ };
+
+ for (int i = 0; i < 10; ++i) {
+ executor.inParallel(useBuffer);
+ }
+ executor.inParallel(() -> {
+ vendor.destruct();
});
+ for (int i = 0; i < 10; ++i) {
+ executor.inParallel(useBuffer);
+ }
+
executor.execute();
verify(poolMock, times(1)).releaseBuffer(any(), any());
}
+ private void useBuffer(final ByteBufferSharing sharing) throws IOException {
+ sharing.getBuffer();
+ yield(); //
Review comment:
sleeping now
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@geode.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [geode] Bill commented on a change in pull request #6930: P2P messaging concurrency test
Posted by GitBox <gi...@apache.org>.
Bill commented on a change in pull request #6930:
URL: https://github.com/apache/geode/pull/6930#discussion_r732295269
##########
File path: geode-core/src/test/java/org/apache/geode/internal/net/ByteBufferConcurrencyTest.java
##########
@@ -15,6 +15,7 @@
package org.apache.geode.internal.net;
+import static java.lang.Thread.yield;
Review comment:
corrected.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@geode.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [geode] kirklund commented on a change in pull request #6930: P2P messaging concurrency test
Posted by GitBox <gi...@apache.org>.
kirklund commented on a change in pull request #6930:
URL: https://github.com/apache/geode/pull/6930#discussion_r725227239
##########
File path: geode-core/src/test/java/org/apache/geode/internal/net/ByteBufferConcurrencyTest.java
##########
@@ -15,6 +15,7 @@
package org.apache.geode.internal.net;
+import static java.lang.Thread.yield;
Review comment:
You might want to avoid using `Thread.yield`. See the javadocs. The general recommendation from various sources is to avoid using `yield`. Maybe use `Thread.sleep` or even park that thread on a `CountDownLatch` that you later open in the test and again in tearDown (just in case).
##########
File path: geode-core/src/distributedTest/java/org/apache/geode/distributed/internal/P2PMessagingConcurrencyDUnitTest.java
##########
@@ -0,0 +1,314 @@
+/*
+ * 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.geode.distributed.internal;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.security.GeneralSecurityException;
+import java.util.Properties;
+import java.util.Random;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.LongAdder;
+
+import org.jetbrains.annotations.NotNull;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import org.apache.geode.cache.CacheFactory;
+import org.apache.geode.cache.ssl.CertStores;
+import org.apache.geode.cache.ssl.CertificateBuilder;
+import org.apache.geode.cache.ssl.CertificateMaterial;
+import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.internal.serialization.DeserializationContext;
+import org.apache.geode.internal.serialization.SerializationContext;
+import org.apache.geode.test.dunit.rules.ClusterStartupRule;
+import org.apache.geode.test.dunit.rules.MemberVM;
+import org.apache.geode.test.junit.categories.MembershipTest;
+import org.apache.geode.test.version.VersionManager;
+
+/**
+ * Tests one-way P2P messaging between two peers. A shared,
+ * ordered connection is used and many concurrent tasks
+ * compete on the sending side. Tests with TLS enabled
+ * to exercise ByteBufferSharing and friends.
+ */
+@Category({MembershipTest.class})
+public class P2PMessagingConcurrencyDUnitTest {
+
+ // how many messages will each sender generate?
+ private static final int MESSAGES_PER_SENDER = 1_000;
+
+ // number of concurrent (sending) tasks to run
+ private static final int SENDER_COUNT = 10;
+
+ // (exclusive) upper bound of random message size, in bytes
+ private static final int LARGEST_MESSAGE_BOUND = 32 * 1024 + 2; // 32KiB + 2
+
+ // random seed
+ private static final int RANDOM_SEED = 1234;
+
+ /*
+ At the time this comment was written, ClusterStartupRule was ignoring the vmCount.
+ Nevertheless since we need only 3, we're specifying it.
+ */
+ @Rule
+ public final ClusterStartupRule clusterStartupRule = new ClusterStartupRule(3);
+
+ private MemberVM sender;
+ private MemberVM receiver;
+
+ /*
+ * bytes sent on sender JVM, bytes received on receiver JVM
+ * (not used in test JVM)
+ */
+ private static LongAdder bytesTransferredAdder;
+
+ @Before
+ public void before() throws GeneralSecurityException, IOException {
+ final Properties configuration = gemFireConfiguration();
+
+ final MemberVM locator =
+ clusterStartupRule.startLocatorVM(0, 0, VersionManager.CURRENT_VERSION,
+ x -> x.withProperties(configuration).withConnectionToLocator()
+ .withoutClusterConfigurationService().withoutManagementRestService());
+
+ sender = clusterStartupRule.startServerVM(1, configuration, locator.getPort());
+ receiver = clusterStartupRule.startServerVM(2, configuration, locator.getPort());
+ }
+
+ @Test
+ public void testP2PMessagingWithTLS() {
+
+ final InternalDistributedMember receiverMember =
+ receiver.invoke(() -> {
+
+ bytesTransferredAdder = new LongAdder();
+
+ final ClusterDistributionManager cdm = getCDM();
+ final InternalDistributedMember localMember = cdm.getDistribution().getLocalMember();
+ return localMember;
+
+ });
+
+ sender.invoke(() -> {
+
+ bytesTransferredAdder = new LongAdder();
+
+ final ClusterDistributionManager cdm = getCDM();
+ final Random random = new Random(RANDOM_SEED);
+ final AtomicInteger nextSenderId = new AtomicInteger();
+
+ /*
+ When this comment was written the nThreads parameter to the thread pool
+ constructor was SENDER_COUNT. When SENDER_COUNT is much larger than the
+ number of CPUs that is counterproductive. In an ideal world we'd want
+ only as many threads as CPUs here. OTOH the P2P messaging system at the
+ time this comment was written, used blocking I/O, so we were not, as it
+ turns out, living in that ideal world.
+ */
+ final ExecutorService executor = Executors.newFixedThreadPool(SENDER_COUNT);
+
+ final CountDownLatch startLatch = new CountDownLatch(SENDER_COUNT);
+ final CountDownLatch stopLatch = new CountDownLatch(SENDER_COUNT);
+ final LongAdder failedRecipientCount = new LongAdder();
+
+ final Runnable doSending = () -> {
+ final int senderId = nextSenderId.getAndIncrement();
+ try {
+ startLatch.countDown();
+ startLatch.await();
+ } catch (final InterruptedException e) {
+ throw new RuntimeException("doSending failed", e);
Review comment:
I don't think this will propagate to the test failure if the thread is running in a local `ExecutorService`. The `ExecutorServiceRule` and `DistributedExecutorServiceRule` should propagate all such exception and show multiple stacks in the test failure if needed.
However, I generally use `ErrorCollector` or `DistributedErrorCollector` rules:
```
} catch (final InterruptedException e) {
errorCollector.addError(e);
```
`ErrorCollector` includes some other methods like `checkThat` which performs an assertion using a Hamcrest matcher.
##########
File path: geode-core/src/test/java/org/apache/geode/internal/net/ByteBufferConcurrencyTest.java
##########
@@ -40,126 +41,145 @@
@LoopRunnerConfig(count = 100)
public class ByteBufferConcurrencyTest {
- private BufferPool poolMock;
-
@Test
- public void concurrentDestructAndOpenCloseShouldReturnToPoolOnce(ParallelExecutor executor)
+ public void concurrentDestructAndOpenCloseShouldReturnToPoolOnce(final ParallelExecutor executor)
throws Exception {
- poolMock = mock(BufferPool.class);
- ByteBuffer someBuffer = ByteBuffer.allocate(1);
- ByteBufferVendor sharing =
+ final BufferPool poolMock = mock(BufferPool.class);
+ final ByteBuffer someBuffer = ByteBuffer.allocate(1);
+ final ByteBufferVendor vendor =
new ByteBufferVendor(someBuffer, BufferPool.BufferType.TRACKED_SENDER,
poolMock);
- executor.inParallel(() -> {
- sharing.destruct();
- });
- executor.inParallel(() -> {
- try {
- try (ByteBufferSharing localSharing = sharing.open()) {
- localSharing.getBuffer();
- }
+
+ final RunnableWithException useBuffer = () -> {
+ try (final ByteBufferSharing sharing = vendor.open()) {
+ useBuffer(sharing);
} catch (IOException e) {
- // It's ok to get an IOException if the sharing was destroyed before this runs
+ // It's ok to get an IOException if the sharing was destruct()ed before this runs
}
+ };
+
+ for (int i = 0; i < 10; ++i) {
+ executor.inParallel(useBuffer);
+ }
+ executor.inParallel(() -> {
+ vendor.destruct();
});
+ for (int i = 0; i < 10; ++i) {
+ executor.inParallel(useBuffer);
+ }
+
executor.execute();
verify(poolMock, times(1)).releaseBuffer(any(), any());
}
+ private void useBuffer(final ByteBufferSharing sharing) throws IOException {
+ sharing.getBuffer();
+ yield(); //
Review comment:
A `Thread.sleep` with short value should provide the same intention here if you decide to avoid `yield`.
`Thread.sleep` supposedly has a guaranteed effect, whereas `yield` is a hint that may actually be ignored by the JVM.
While it's probably ok to use `yield` in a test, I would recommend avoiding it just to so others don't see it and start trying to use it more or even in product code.
##########
File path: geode-core/src/distributedTest/java/org/apache/geode/distributed/internal/P2PMessagingConcurrencyDUnitTest.java
##########
@@ -0,0 +1,314 @@
+/*
+ * 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.geode.distributed.internal;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.security.GeneralSecurityException;
+import java.util.Properties;
+import java.util.Random;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.LongAdder;
+
+import org.jetbrains.annotations.NotNull;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import org.apache.geode.cache.CacheFactory;
+import org.apache.geode.cache.ssl.CertStores;
+import org.apache.geode.cache.ssl.CertificateBuilder;
+import org.apache.geode.cache.ssl.CertificateMaterial;
+import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.internal.serialization.DeserializationContext;
+import org.apache.geode.internal.serialization.SerializationContext;
+import org.apache.geode.test.dunit.rules.ClusterStartupRule;
+import org.apache.geode.test.dunit.rules.MemberVM;
+import org.apache.geode.test.junit.categories.MembershipTest;
+import org.apache.geode.test.version.VersionManager;
+
+/**
+ * Tests one-way P2P messaging between two peers. A shared,
+ * ordered connection is used and many concurrent tasks
+ * compete on the sending side. Tests with TLS enabled
+ * to exercise ByteBufferSharing and friends.
+ */
+@Category({MembershipTest.class})
+public class P2PMessagingConcurrencyDUnitTest {
+
+ // how many messages will each sender generate?
+ private static final int MESSAGES_PER_SENDER = 1_000;
+
+ // number of concurrent (sending) tasks to run
+ private static final int SENDER_COUNT = 10;
+
+ // (exclusive) upper bound of random message size, in bytes
+ private static final int LARGEST_MESSAGE_BOUND = 32 * 1024 + 2; // 32KiB + 2
+
+ // random seed
+ private static final int RANDOM_SEED = 1234;
+
+ /*
+ At the time this comment was written, ClusterStartupRule was ignoring the vmCount.
+ Nevertheless since we need only 3, we're specifying it.
+ */
+ @Rule
+ public final ClusterStartupRule clusterStartupRule = new ClusterStartupRule(3);
+
+ private MemberVM sender;
+ private MemberVM receiver;
+
+ /*
+ * bytes sent on sender JVM, bytes received on receiver JVM
+ * (not used in test JVM)
+ */
+ private static LongAdder bytesTransferredAdder;
+
+ @Before
+ public void before() throws GeneralSecurityException, IOException {
+ final Properties configuration = gemFireConfiguration();
+
+ final MemberVM locator =
+ clusterStartupRule.startLocatorVM(0, 0, VersionManager.CURRENT_VERSION,
+ x -> x.withProperties(configuration).withConnectionToLocator()
+ .withoutClusterConfigurationService().withoutManagementRestService());
+
+ sender = clusterStartupRule.startServerVM(1, configuration, locator.getPort());
+ receiver = clusterStartupRule.startServerVM(2, configuration, locator.getPort());
+ }
+
+ @Test
+ public void testP2PMessagingWithTLS() {
+
+ final InternalDistributedMember receiverMember =
+ receiver.invoke(() -> {
+
+ bytesTransferredAdder = new LongAdder();
+
+ final ClusterDistributionManager cdm = getCDM();
+ final InternalDistributedMember localMember = cdm.getDistribution().getLocalMember();
+ return localMember;
+
+ });
+
+ sender.invoke(() -> {
+
+ bytesTransferredAdder = new LongAdder();
+
+ final ClusterDistributionManager cdm = getCDM();
+ final Random random = new Random(RANDOM_SEED);
+ final AtomicInteger nextSenderId = new AtomicInteger();
+
+ /*
+ When this comment was written the nThreads parameter to the thread pool
+ constructor was SENDER_COUNT. When SENDER_COUNT is much larger than the
+ number of CPUs that is counterproductive. In an ideal world we'd want
+ only as many threads as CPUs here. OTOH the P2P messaging system at the
+ time this comment was written, used blocking I/O, so we were not, as it
+ turns out, living in that ideal world.
+ */
+ final ExecutorService executor = Executors.newFixedThreadPool(SENDER_COUNT);
+
+ final CountDownLatch startLatch = new CountDownLatch(SENDER_COUNT);
+ final CountDownLatch stopLatch = new CountDownLatch(SENDER_COUNT);
+ final LongAdder failedRecipientCount = new LongAdder();
+
+ final Runnable doSending = () -> {
+ final int senderId = nextSenderId.getAndIncrement();
+ try {
+ startLatch.countDown();
+ startLatch.await();
+ } catch (final InterruptedException e) {
+ throw new RuntimeException("doSending failed", e);
+ }
+ final int firstMessageId = senderId * SENDER_COUNT;
+ for (int messageId = firstMessageId; messageId < firstMessageId
+ + MESSAGES_PER_SENDER; messageId++) {
+ final TestMessage msg = new TestMessage(receiverMember, random, messageId);
+
+ /*
+ * HERE is the Geode API entrypoint we intend to test (putOutgoing()).
+ */
+ final Set<InternalDistributedMember> failedRecipients = cdm.putOutgoing(msg);
+
+ if (failedRecipients != null) {
+ failedRecipientCount.add(failedRecipients.size());
+ }
+ }
+ stopLatch.countDown();
+ };
+
+ for (int i = 0; i < SENDER_COUNT; ++i) {
+ executor.submit(doSending);
+ }
+
+ stopLatch.await();
+
+ stop(executor);
+
+ assertThat(failedRecipientCount.sum()).as("message delivery failed N times").isZero();
+
+ });
+
+ final long bytesSent = sender.invoke(() -> bytesTransferredAdder.sum());
+ final long bytesReceived = receiver.invoke(() -> bytesTransferredAdder.sum());
+
+ assertThat(bytesReceived).as("bytes received != bytes sent").isEqualTo(bytesSent);
+ }
+
+ private static void stop(final ExecutorService executor) {
+ executor.shutdown();
+ try {
+ if (!executor.awaitTermination(800, TimeUnit.MILLISECONDS)) {
+ executor.shutdownNow();
+ }
+ } catch (InterruptedException e) {
+ executor.shutdownNow();
+ }
+ }
+
+ private static ClusterDistributionManager getCDM() {
+ return (ClusterDistributionManager) ((InternalCache) CacheFactory.getAnyInstance())
+ .getDistributionManager();
+ }
+
+ private static class TestMessage extends DistributionMessage {
+
+ /*
+ When this comment was written, messageId wasn't used for anything.
+ The field was added during a misguided attempt to add SHA-256
+ digest verification on sender and receiver. Then I figured out
+ that there's no way to parallelize that (for the sender) so
+ I settled for merely validating the number of bytes transferred.
+ Left the field here in case it comes in handy later.
+ */
+ private volatile int messageId;
+ private volatile Random random;
+
+ TestMessage(final InternalDistributedMember receiver,
+ final Random random, final int messageId) {
+ setRecipient(receiver);
+ this.random = random;
+ this.messageId = messageId;
+ }
+
+ // necessary for deserialization
+ public TestMessage() {
+ random = null;
+ messageId = 0;
+ }
+
+ @Override
+ public int getProcessorType() {
+ return OperationExecutors.STANDARD_EXECUTOR;
+ }
+
+ @Override
+ protected void process(final ClusterDistributionManager dm) {}
+
+ @Override
+ public void toData(final DataOutput out, final SerializationContext context)
+ throws IOException {
+ super.toData(out, context);
+
+ out.writeInt(messageId);
+
+ final int length = random.nextInt(LARGEST_MESSAGE_BOUND);
+
+ out.writeInt(length);
+
+ final byte[] payload = new byte[length];
+ random.nextBytes(payload);
+
+ out.write(payload);
+
+ /*
+ * the LongAdder should ensure that we don't introduce any (much)
+ * synchronization with other concurrent tasks here
+ */
+ bytesTransferredAdder.add(length);
+ }
+
+ @Override
+ public void fromData(final DataInput in, final DeserializationContext context)
+ throws IOException, ClassNotFoundException {
+ super.fromData(in, context);
+
+ final int messageId = in.readInt();
+
+ final int length = in.readInt();
+
+ final byte[] payload = new byte[length];
+
+ in.readFully(payload);
+
+ bytesTransferredAdder.add(length);
+ }
+
+ @Override
+ public int getDSFID() {
+ return NO_FIXED_ID; // for testing only!
+ }
+ }
+
+ @NotNull
+ private static Properties gemFireConfiguration()
+ throws GeneralSecurityException, IOException {
+
+ final Properties props = securityProperties();
+
+ /*
+ * This is something we intend to test!
+ * Send all messages, from all threads, on a single socket per recipient.
+ * maintenance tip: to see what kind of connection you're getting you can
+ * uncomment logging over in DirectChannel.sendToMany()
+ */
+ props.put("conserve-sockets", "true"); // careful: if you set a boolean it doesn't take hold!
Review comment:
For `Properties` the `setProperty` method is preferred over `put` which is inherited from `Map`.
##########
File path: geode-core/src/distributedTest/java/org/apache/geode/distributed/internal/P2PMessagingConcurrencyDUnitTest.java
##########
@@ -0,0 +1,314 @@
+/*
+ * 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.geode.distributed.internal;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+import java.security.GeneralSecurityException;
+import java.util.Properties;
+import java.util.Random;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.LongAdder;
+
+import org.jetbrains.annotations.NotNull;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+
+import org.apache.geode.cache.CacheFactory;
+import org.apache.geode.cache.ssl.CertStores;
+import org.apache.geode.cache.ssl.CertificateBuilder;
+import org.apache.geode.cache.ssl.CertificateMaterial;
+import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
+import org.apache.geode.internal.cache.InternalCache;
+import org.apache.geode.internal.serialization.DeserializationContext;
+import org.apache.geode.internal.serialization.SerializationContext;
+import org.apache.geode.test.dunit.rules.ClusterStartupRule;
+import org.apache.geode.test.dunit.rules.MemberVM;
+import org.apache.geode.test.junit.categories.MembershipTest;
+import org.apache.geode.test.version.VersionManager;
+
+/**
+ * Tests one-way P2P messaging between two peers. A shared,
+ * ordered connection is used and many concurrent tasks
+ * compete on the sending side. Tests with TLS enabled
+ * to exercise ByteBufferSharing and friends.
+ */
+@Category({MembershipTest.class})
+public class P2PMessagingConcurrencyDUnitTest {
+
+ // how many messages will each sender generate?
+ private static final int MESSAGES_PER_SENDER = 1_000;
+
+ // number of concurrent (sending) tasks to run
+ private static final int SENDER_COUNT = 10;
+
+ // (exclusive) upper bound of random message size, in bytes
+ private static final int LARGEST_MESSAGE_BOUND = 32 * 1024 + 2; // 32KiB + 2
+
+ // random seed
+ private static final int RANDOM_SEED = 1234;
+
+ /*
+ At the time this comment was written, ClusterStartupRule was ignoring the vmCount.
+ Nevertheless since we need only 3, we're specifying it.
+ */
+ @Rule
+ public final ClusterStartupRule clusterStartupRule = new ClusterStartupRule(3);
+
+ private MemberVM sender;
+ private MemberVM receiver;
+
+ /*
+ * bytes sent on sender JVM, bytes received on receiver JVM
+ * (not used in test JVM)
+ */
+ private static LongAdder bytesTransferredAdder;
+
+ @Before
+ public void before() throws GeneralSecurityException, IOException {
+ final Properties configuration = gemFireConfiguration();
+
+ final MemberVM locator =
+ clusterStartupRule.startLocatorVM(0, 0, VersionManager.CURRENT_VERSION,
+ x -> x.withProperties(configuration).withConnectionToLocator()
+ .withoutClusterConfigurationService().withoutManagementRestService());
+
+ sender = clusterStartupRule.startServerVM(1, configuration, locator.getPort());
+ receiver = clusterStartupRule.startServerVM(2, configuration, locator.getPort());
+ }
+
+ @Test
+ public void testP2PMessagingWithTLS() {
+
+ final InternalDistributedMember receiverMember =
+ receiver.invoke(() -> {
+
+ bytesTransferredAdder = new LongAdder();
+
+ final ClusterDistributionManager cdm = getCDM();
+ final InternalDistributedMember localMember = cdm.getDistribution().getLocalMember();
+ return localMember;
+
+ });
+
+ sender.invoke(() -> {
+
+ bytesTransferredAdder = new LongAdder();
+
+ final ClusterDistributionManager cdm = getCDM();
+ final Random random = new Random(RANDOM_SEED);
+ final AtomicInteger nextSenderId = new AtomicInteger();
+
+ /*
+ When this comment was written the nThreads parameter to the thread pool
+ constructor was SENDER_COUNT. When SENDER_COUNT is much larger than the
+ number of CPUs that is counterproductive. In an ideal world we'd want
+ only as many threads as CPUs here. OTOH the P2P messaging system at the
+ time this comment was written, used blocking I/O, so we were not, as it
+ turns out, living in that ideal world.
+ */
+ final ExecutorService executor = Executors.newFixedThreadPool(SENDER_COUNT);
Review comment:
You might want to consider using `DistributedExecutorServiceRule`. It can be used from within any VM in a dunit test.
```
@Rule
public DistributedExecutorServiceRule executorServiceRule = DistributedExecutorServiceRule();
```
You can also specify a fixed thread count or even declare multiple instances of the rule if you want to use different fixed thread counts in different places:
```
@Rule
public DistributedExecutorServiceRule senderExecutorServiceRule = DistributedExecutorServiceRule(SENDER_COUNT);
```
(Note, I've mostly tested using fixed thread counts in the `ExecutorServiceRule` version).
During rule tearDown it will perform `shutdownNow` which will interrupt any threads that are left running (common problem if the test fails before completing). It will also print stack traces for any threads that were left running.
Most of the submission calls return a `Future` or `CompetableFuture`.
`CyclicBarrier` is also very useful especially if you need to coordinate multiple threads reaching certain points before doing something.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@geode.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org