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