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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
