Allows decimal values for publishing interval

This commit allows values for --publishing-interval to go
below 1, e.g. 0.5 to publish 2 messages per second. The
lower bound is 0.1, to avoid overloading the scheduling.

This allows scheduled publishers to publish up to 10 messages / second,
which can be useful for some scenarios. These scenarios do not have
to be pure IoT scenarios (--publishing-interval was introduced
initially for IoT scenarios), but rather scenarios with a number
of connections and ingress load in mind.

This commit also changes the calculation of the number of scheduling
threads. It assumes a thread is enough to deal with 100 operations
a second (e.g. 100 publishers publishing at 1 message / second), which
is rather a conservative default.

The number of threads can still be changed with the --producer-scheduler-threads
option.

Fixes #287

Author: acogoluegnes

src/main/java/com/rabbitmq/perf/MulticastParams.java

@@ -1,4 +1,4 @@
-// Copyright (c) 2007-2020 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2007-2022 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // This software, the RabbitMQ Java client library, is triple-licensed under the
 // Mozilla Public License 2.0 ("MPL"), the GNU General Public License version 2
@@ -22,6 +22,7 @@
 
 import com.rabbitmq.perf.PerfTest.EXIT_WHEN;
 import java.io.IOException;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
@@ -92,7 +93,7 @@ public class MulticastParams {
 
     private int routingKeyCacheSize = 0;
     private boolean exclusive = false;
-    private int publishingInterval = -1;
+    private Duration publishingInterval = null;
     private int producerRandomStartDelayInSeconds;
     private int producerSchedulerThreadCount = -1;
     private int consumersThreadPools = -1;
@@ -625,11 +626,11 @@ public boolean isExclusive() {
         return exclusive;
     }
 
-    public void setPublishingInterval(int publishingIntervalInSeconds) {
-        this.publishingInterval = publishingIntervalInSeconds;
+    public void setPublishingInterval(Duration publishingInterval) {
+        this.publishingInterval = publishingInterval;
     }
 
-    public int getPublishingInterval() {
+    public Duration getPublishingInterval() {
         return publishingInterval;
     }
 

src/main/java/com/rabbitmq/perf/MulticastSet.java

@@ -1,4 +1,4 @@
-// Copyright (c) 2007-2020 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2007-2022 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // This software, the RabbitMQ Java client library, is triple-licensed under the
 // Mozilla Public License 2.0 ("MPL"), the GNU General Public License version 2
@@ -29,9 +29,11 @@
 import com.rabbitmq.client.impl.recovery.AutorecoveringConnection;
 import com.rabbitmq.perf.PerfTest.EXIT_WHEN;
 import java.io.IOException;
+import java.math.BigDecimal;
 import java.net.URISyntaxException;
 import java.security.KeyManagementException;
 import java.security.NoSuchAlgorithmException;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
@@ -61,14 +63,6 @@ public class MulticastSet {
     // from Java Client ConsumerWorkService
     public final static int DEFAULT_CONSUMER_WORK_SERVICE_THREAD_POOL_SIZE = Runtime.getRuntime().availableProcessors() * 2;
     private static final Logger LOGGER = LoggerFactory.getLogger(MulticastSet.class);
-    /**
-     * Why 50? This is arbitrary. The fastest rate is 1 message / second when
-     * using a publishing interval, so 1 thread should be able to keep up easily with
-     * up to 50 messages / seconds (ie. 50 producers). Then, a new thread is used
-     * every 50 producers. This is 20 threads for 1000 producers, which seems reasonable.
-     * There's a command line argument to override this anyway.
-     */
-    private static final int PUBLISHING_INTERVAL_NB_PRODUCERS_PER_THREAD = 50;
     private static final String PRODUCER_THREAD_PREFIX = "perf-test-producer-";
     static final String STOP_REASON_REACHED_TIME_LIMIT = "Reached time limit";
     private final Stats stats;
@@ -149,7 +143,30 @@ protected static int nbThreadsForConsumer(MulticastParams params) {
     protected static int nbThreadsForProducerScheduledExecutorService(MulticastParams params) {
         int producerExecutorServiceNbThreads = params.getProducerSchedulerThreadCount();
         if (producerExecutorServiceNbThreads <= 0) {
-            return params.getProducerThreadCount() / PUBLISHING_INTERVAL_NB_PRODUCERS_PER_THREAD + 1;
+            int producerThreadCount = params.getProducerThreadCount();
+            Duration publishingInterval = params.getPublishingInterval() == null ? Duration.ofSeconds(1)
+                : params.getPublishingInterval();
+            long publishingIntervalMs = publishingInterval.toMillis();
+
+            BigDecimal publishingIntervalSeconds = new BigDecimal(publishingIntervalMs)
+                .divide(new BigDecimal(1000));
+            BigDecimal rate = new BigDecimal(producerThreadCount)
+                .divide(publishingIntervalSeconds);
+            /**
+             * Why 100? This is arbitrary. We assume 1 thread is more than enough to handle
+             * the publishing of 100 messages in 1 second, the fastest rate
+             * being 10 messages / second when using --publishing-interval.
+             * Then, a new thread is used
+             * every for every 100 messages / second.
+             * This is 21 threads for 1000 producers publishing 1 message / second,
+             * which seems reasonable.
+             * There's a command line argument to override this anyway.
+             */
+            int threadCount = rate.intValue() / 100 + 1;
+            LOGGER.debug("Using {} thread(s) to schedule {} publisher(s) publishing every {} ms",
+                threadCount, producerThreadCount, publishingInterval.toMillis()
+                );
+            return threadCount;
         } else {
             return producerExecutorServiceNbThreads;
         }
@@ -406,9 +423,9 @@ private void startConsumers(Runnable[] consumerRunnables) throws InterruptedExce
 
     private void startProducers(AgentState[] producerStates) {
         this.messageSizeIndicator.start();
-        if (params.getPublishingInterval() > 0) {
+        if (params.getPublishingInterval() != null) {
             ScheduledExecutorService producersExecutorService = this.threadingHandler.scheduledExecutorService(
-                    PRODUCER_THREAD_PREFIX, nbThreadsForConsumer(params)
+                    PRODUCER_THREAD_PREFIX, nbThreadsForProducerScheduledExecutorService(params)
             );
             Supplier<Integer> startDelaySupplier;
             if (params.getProducerRandomStartDelayInSeconds() > 0) {
@@ -417,13 +434,13 @@ PRODUCER_THREAD_PREFIX, nbThreadsForConsumer(params)
             } else {
                 startDelaySupplier = () -> 0;
             }
-            int publishingInterval = params.getPublishingInterval();
+            Duration publishingInterval = params.getPublishingInterval();
             for (int i = 0; i < producerStates.length; i++) {
                 AgentState producerState = producerStates[i];
                 int delay = startDelaySupplier.get();
                 producerState.task = producersExecutorService.scheduleAtFixedRate(
                         producerState.runnable.createRunnableForScheduling(),
-                        delay, publishingInterval, TimeUnit.SECONDS
+                        delay, publishingInterval.toMillis(), TimeUnit.MILLISECONDS
                 );
             }
         } else {

src/main/java/com/rabbitmq/perf/PerfTest.java

@@ -23,6 +23,8 @@
 import com.rabbitmq.client.impl.DefaultExceptionHandler;
 import com.rabbitmq.client.impl.nio.NioParams;
 import io.micrometer.core.instrument.composite.CompositeMeterRegistry;
+import java.math.BigDecimal;
+import java.time.Duration;
 import java.util.stream.Collectors;
 import org.apache.commons.cli.*;
 import org.slf4j.Logger;
@@ -130,7 +132,16 @@ public static void main(String [] args, PerfTestOptions perfTestOptions) {
             String messageProperties = strArg(cmd, "mp", null);
             int routingKeyCacheSize  = intArg(cmd, "rkcs", 0);
             boolean exclusive = hasOption(cmd, "E");
-            int publishingIntervalInSeconds = intArg(cmd, "P", -1);
+            Duration publishingInterval = null;
+            String publishingIntervalArg = strArg(cmd, "P", null);
+            if (publishingIntervalArg != null) {
+                try {
+                    publishingInterval = parsePublishingInterval(publishingIntervalArg);
+                } catch (IllegalArgumentException e) {
+                    System.out.println("Invalid value for --publishing-interval: " + e.getMessage());
+                    systemExiter.exit(1);
+                }
+            }
             int producerRandomStartDelayInSeconds = intArg(cmd, "prsd", -1);
             int producerSchedulingThreads = intArg(cmd, "pst", -1);
 
@@ -374,7 +385,7 @@ public static void main(String [] args, PerfTestOptions perfTestOptions) {
             p.setMessageProperties(convertKeyValuePairs(messageProperties));
             p.setRoutingKeyCacheSize(routingKeyCacheSize);
             p.setExclusive(exclusive);
-            p.setPublishingInterval(publishingIntervalInSeconds);
+            p.setPublishingInterval(publishingInterval);
             p.setProducerRandomStartDelayInSeconds(producerRandomStartDelayInSeconds);
             p.setProducerSchedulerThreadCount(producerSchedulingThreads);
             p.setConsumersThreadPools(consumersThreadPools);
@@ -811,6 +822,23 @@ private static void versionInformation() {
         System.out.println("\u001B[0m" + info);
     }
 
+    static Duration parsePublishingInterval(String input) {
+        BigDecimal decimalValue;
+        try {
+         decimalValue = new BigDecimal(input);
+        } catch (NumberFormatException e) {
+            throw new IllegalArgumentException("Must be a number");
+        }
+        if (decimalValue.compareTo(BigDecimal.ZERO) <= 0) {
+            throw new IllegalArgumentException("Must be positive");
+        }
+        Duration result = Duration.ofMillis(decimalValue.multiply(BigDecimal.valueOf(1000)).longValue());
+        if (result.toMillis() < 100) {
+            throw new IllegalArgumentException("Cannot be less than 0.1");
+        }
+        return result;
+    }
+
     /**
      * Abstraction to ease testing or PerfTest usage as a library.
      */

src/test/java/com/rabbitmq/perf/MessageCountTimeLimitAndPublishingIntervalRateTest.java

@@ -1,4 +1,4 @@
-// Copyright (c) 2018-2O20 Pivotal Software, Inc.  All rights reserved.
+// Copyright (c) 2018-2O22 Pivotal Software, Inc.  All rights reserved.
 //
 // This software, the RabbitMQ Java client library, is triple-licensed under the
 // Mozilla Public License 2.0 ("MPL"), the GNU General Public License version 2
@@ -19,6 +19,7 @@
 import com.rabbitmq.client.Consumer;
 import com.rabbitmq.client.*;
 import com.rabbitmq.perf.PerfTest.EXIT_WHEN;
+import java.time.Duration;
 import java.util.stream.Collectors;
 import org.assertj.core.api.Assertions;
 import org.assertj.core.api.Condition;
@@ -574,7 +575,7 @@ public void publishingRateLimit() throws InterruptedException {
     @Test
     public void publishingInterval() throws InterruptedException {
         countsAndTimeLimit(0, 0, 6);
-        params.setPublishingInterval(1);
+        params.setPublishingInterval(Duration.ofSeconds(1));
         params.setProducerCount(3);
 
         AtomicInteger publishedMessageCount = new AtomicInteger();

src/test/java/com/rabbitmq/perf/MulticastParamsTest.java

@@ -1,4 +1,4 @@
-// Copyright (c) 2018-2020 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2018-2022 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // This software, the RabbitMQ Java client library, is triple-licensed under the
 // Mozilla Public License 2.0 ("MPL"), the GNU General Public License version 2

src/test/java/com/rabbitmq/perf/MulticastSetTest.java

@@ -18,10 +18,13 @@
 import com.rabbitmq.client.Address;
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.ConnectionFactory;
+import java.time.Duration;
 import org.apache.commons.lang3.time.StopWatch;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
 import org.mockito.ArgumentCaptor;
 import org.mockito.ArgumentMatchers;
 import org.mockito.Mock;
@@ -105,18 +108,35 @@ public void nbThreadsForProducerScheduledExecutorServiceDefaultIsOne() {
     }
 
     @Test
-    public void nbThreadsForProducerScheduledExecutorServiceOneThreadEvery50Producers() {
-        params.setProducerCount(120);
+    public void nbThreadsForProducerScheduledExecutorServiceOneThreadEvery100Producers() {
+        params.setProducerCount(220);
         assertThat(nbThreadsForProducerScheduledExecutorService(params)).isEqualTo(3);
     }
 
     @Test
-    public void nbThreadsForProducerScheduledExecutorServiceOneThreadEvery50ProducersIncludeChannels() {
+    public void nbThreadsForProducerScheduledExecutorServiceOneThreadEvery100ProducersIncludeChannels() {
         params.setProducerCount(30);
-        params.setProducerChannelCount(4);
+        params.setProducerChannelCount(8);
         assertThat(nbThreadsForProducerScheduledExecutorService(params)).isEqualTo(3);
     }
 
+    @ParameterizedTest
+    @CsvSource({
+        "1,1000,1",
+        "200,1000,3",
+        "299,1000,3",
+        "300,1000,4",
+        "200,100,21",
+        "2000,1000,21",
+    })
+    void nbThreadsForProducerScheduledExecutorServiceOK(int producerCount, int publishingIntervalMs,
+            int expectedThreadCount) {
+        params.setProducerCount(producerCount);
+        params.setPublishingInterval(Duration.ofMillis(publishingIntervalMs));
+        assertThat(nbThreadsForProducerScheduledExecutorService(params))
+            .isEqualTo(expectedThreadCount);
+    }
+
     @Test
     public void nbThreadsForProducerScheduledExecutorServiceUseParameterValueWhenSpecified() {
         params.setProducerSchedulerThreadCount(7);

src/test/java/com/rabbitmq/perf/PerfTestTest.java

@@ -19,9 +19,14 @@
 
 import java.util.HashMap;
 import java.util.Map;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.ValueSource;
 
 import static com.rabbitmq.perf.PerfTest.convertKeyValuePairs;
+import static com.rabbitmq.perf.PerfTest.parsePublishingInterval;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 import static org.junit.jupiter.api.Assertions.*;
 
 /**
@@ -113,4 +118,28 @@ void convertPostProcessKeyValuePairs() {
             .containsEntry("x-dead-letter-exchange", "")
             .containsEntry("x-queue-type", "quorum");
     }
+
+    @ParameterizedTest
+    @CsvSource({
+        "1,1000",
+        "0.1,100",
+        "0.5,500",
+        "0.55,550",
+        "2,2000",
+        "15,15000",
+    })
+    void parsePublishingIntervalOK(String input, long expectedMs) {
+        assertThat(parsePublishingInterval(input).toMillis()).isEqualTo(expectedMs);
+    }
+
+    @ParameterizedTest
+    @ValueSource(strings = {
+        "0.09",
+        "-1",
+        "0"
+    })
+    void parsePublishingIntervalKO(String input) {
+        assertThatThrownBy(() -> parsePublishingInterval(input))
+            .isInstanceOf(IllegalArgumentException.class);
+    }
 }

src/test/java/com/rabbitmq/perf/PublisherOnlyStopsCorrectlyTest.java

@@ -1,4 +1,4 @@
-// Copyright (c) 2018-2020 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2018-2022 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // This software, the RabbitMQ Java client library, is triple-licensed under the
 // Mozilla Public License 2.0 ("MPL"), the GNU General Public License version 2
@@ -19,6 +19,7 @@
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.ConnectionFactory;
 import com.rabbitmq.client.impl.AMQImpl;
+import java.time.Duration;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.TestInfo;
@@ -49,12 +50,17 @@ public class PublisherOnlyStopsCorrectlyTest {
     ExecutorService executorService;
 
     static Stream<Arguments> publisherOnlyStopsWhenBrokerCrashesArguments() {
-        return Stream.of(
-                // number of messages before throwing exception, configurator, assertion message
-                Arguments.of(10, (Consumer<MulticastParams>) (params) -> {
-                }, "Sender should have failed and program should stop"),
-                Arguments.of(2, (Consumer<MulticastParams>) (params) -> params.setPublishingInterval(1), "Sender should have failed and program should stop")
-        );
+    return Stream.of(
+        // number of messages before throwing exception, configurator, assertion message
+        Arguments.of(
+            10,
+            (Consumer<MulticastParams>) (params) -> {},
+            "Sender should have failed and program should stop"),
+        Arguments.of(
+            2,
+            (Consumer<MulticastParams>)
+                (params) -> params.setPublishingInterval(Duration.ofSeconds(1)),
+            "Sender should have failed and program should stop"));
     }
 
     @BeforeEach

src/test/java/com/rabbitmq/perf/it/ConnectionRecoveryIT.java

@@ -1,4 +1,4 @@
-// Copyright (c) 2018-2020 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2018-2022 VMware, Inc. or its affiliates.  All rights reserved.
 //
 // This software, the RabbitMQ Java client library, is triple-licensed under the
 // Mozilla Public License 2.0 ("MPL"), the GNU General Public License version 2
@@ -24,6 +24,7 @@
 import com.rabbitmq.perf.NamedThreadFactory;
 import com.rabbitmq.perf.Stats;
 import io.micrometer.core.instrument.composite.CompositeMeterRegistry;
+import java.time.Duration;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -225,7 +226,7 @@ public void shouldStopWhenConnectionRecoveryIsOffAndConnectionsAreKilledAndUsing
         cf.setAutomaticRecoveryEnabled(false);
         configurer.accept(params);
         cfConfigurer.accept(cf);
-        params.setPublishingInterval(1);
+        params.setPublishingInterval(Duration.ofSeconds(1));
         int producerConsumerCount = params.getProducerCount();
 
         MulticastSet set = new MulticastSet(stats, cf, params, "", URIS, latchCompletionHandler(1, info));
@@ -255,7 +256,7 @@ public void shouldRecoverWhenConnectionsAreKilled(Consumer<MulticastParams> conf
     @MethodSource("configurationArguments")
     public void shouldRecoverWhenConnectionsAreKilledAndUsingPublishingInterval(Consumer<MulticastParams> configurer, Consumer<ConnectionFactory> cfConfigurer,
                                                                                 TestInfo info) throws Exception {
-        params.setPublishingInterval(1);
+        params.setPublishingInterval(Duration.ofSeconds(1));
         configurer.accept(params);
         cfConfigurer.accept(cf);
         int producerConsumerCount = params.getProducerCount();