Allow dynamic update of read and write limits set for Servers

sarankk · vietj · commit 6ddadec7a53e · 2024-02-06T10:08:30.000+01:00
diff --git a/src/main/asciidoc/net.adoc b/src/main/asciidoc/net.adoc
@@ -365,6 +365,18 @@ through {@link io.vertx.core.net.NetServerOptions} and for HttpServer it can be
 {@link examples.NetExamples#configureTrafficShapingForHttpServer}
 ----
 
+These traffic shaping options can also be dynamically updated after server start.
+
+[source,$lang]
+----
+{@link examples.NetExamples#dynamicallyUpdateTrafficShapingForNetServer}
+----
+
+[source,$lang]
+----
+{@link examples.NetExamples#dynamicallyUpdateTrafficShapingForHttpServer}
+----
+
 [[ssl]]
 === Configuring servers and clients to work with SSL/TLS
 
diff --git a/src/main/java/examples/NetExamples.java b/src/main/java/examples/NetExamples.java
@@ -721,6 +721,23 @@ public void configureTrafficShapingForNetServer(Vertx vertx) {
     NetServer server = vertx.createNetServer(options);
   }
 
+  public void dynamicallyUpdateTrafficShapingForNetServer(Vertx vertx) {
+    NetServerOptions options = new NetServerOptions()
+                                 .setHost("localhost")
+                                 .setPort(1234)
+                                 .setTrafficShapingOptions(new TrafficShapingOptions()
+                                                             .setInboundGlobalBandwidth(64 * 1024)
+                                                             .setOutboundGlobalBandwidth(128 * 1024));
+    NetServer server = vertx.createNetServer(options);
+    TrafficShapingOptions update = new TrafficShapingOptions()
+                                     .setInboundGlobalBandwidth(2 * 64 * 1024) // twice
+                                     .setOutboundGlobalBandwidth(128 * 1024); // unchanged
+    server
+      .listen(1234, "localhost")
+      // wait until traffic shaping handler is created for updates
+      .onSuccess(v -> server.updateTrafficShapingOptions(update));
+  }
+
   public void configureTrafficShapingForHttpServer(Vertx vertx) {
     HttpServerOptions options = new HttpServerOptions()
       .setHost("localhost")
@@ -731,4 +748,22 @@ public void configureTrafficShapingForHttpServer(Vertx vertx) {
 
     HttpServer server = vertx.createHttpServer(options);
   }
+
+
+  public void dynamicallyUpdateTrafficShapingForHttpServer(Vertx vertx) {
+    HttpServerOptions options = new HttpServerOptions()
+                                  .setHost("localhost")
+                                  .setPort(1234)
+                                  .setTrafficShapingOptions(new TrafficShapingOptions()
+                                                              .setInboundGlobalBandwidth(64 * 1024)
+                                                              .setOutboundGlobalBandwidth(128 * 1024));
+    HttpServer server = vertx.createHttpServer(options);
+    TrafficShapingOptions update = new TrafficShapingOptions()
+                                     .setInboundGlobalBandwidth(2 * 64 * 1024) // twice
+                                     .setOutboundGlobalBandwidth(128 * 1024); // unchanged
+    server
+      .listen(1234, "localhost")
+      // wait until traffic shaping handler is created for updates
+      .onSuccess(v -> server.updateTrafficShapingOptions(update));
+  }
 }
diff --git a/src/main/java/io/vertx/core/http/HttpServer.java b/src/main/java/io/vertx/core/http/HttpServer.java
@@ -21,6 +21,7 @@
 import io.vertx.core.metrics.Measured;
 import io.vertx.core.net.SSLOptions;
 import io.vertx.core.net.SocketAddress;
+import io.vertx.core.net.TrafficShapingOptions;
 import io.vertx.core.net.impl.SocketAddressImpl;
 import io.vertx.core.streams.ReadStream;
 
@@ -182,6 +183,14 @@ default void updateSSLOptions(SSLOptions options, boolean force, Handler<AsyncRe
     }
   }
 
+  /**
+   * Update traffic shaping options {@code options}, the update happens if valid values are passed for traffic
+   * shaping options. This update happens synchronously and at best effort for rate update to take effect immediately.
+   *
+   * @param options the new traffic shaping options
+   */
+  void updateTrafficShapingOptions(TrafficShapingOptions options);
+
   /**
    * Tell the server to start listening. The server will listen on the port and host specified in the
    * {@link io.vertx.core.http.HttpServerOptions} that was used when creating the server.
diff --git a/src/main/java/io/vertx/core/net/NetServer.java b/src/main/java/io/vertx/core/net/NetServer.java
@@ -256,4 +256,12 @@ default void updateSSLOptions(SSLOptions options, boolean force, Handler<AsyncRe
       fut.onComplete(handler);
     }
   }
+
+  /**
+   * Update traffic shaping options {@code options}, the update happens if valid values are passed for traffic
+   * shaping options. This update happens synchronously and at best effort for rate update to take effect immediately.
+   *
+   * @param options the new traffic shaping options
+   */
+  void updateTrafficShapingOptions(TrafficShapingOptions options);
 }
diff --git a/src/main/java/io/vertx/core/net/TrafficShapingOptions.java b/src/main/java/io/vertx/core/net/TrafficShapingOptions.java
@@ -139,14 +139,14 @@ public TrafficShapingOptions setMaxDelayToWaitUnit(TimeUnit maxDelayToWaitTimeUn
   }
 
   /**
-   * Set the delay between two computations of performances for channels or 0 if no stats are to be computed
+   * Set the delay between two computations of performances for channels
    *
    * @param checkIntervalForStats delay between two computations of performances
    * @return a reference to this, so the API can be used fluently
    */
   public TrafficShapingOptions setCheckIntervalForStats(long checkIntervalForStats) {
     this.checkIntervalForStats = checkIntervalForStats;
-    ObjectUtil.checkPositive(this.checkIntervalForStats, "checkIntervalForStats");
+    ObjectUtil.checkPositiveOrZero(this.checkIntervalForStats, "checkIntervalForStats");
     return this;
   }
 
diff --git a/src/main/java/io/vertx/core/net/impl/TCPServerBase.java b/src/main/java/io/vertx/core/net/impl/TCPServerBase.java
@@ -111,15 +111,13 @@ private GlobalTrafficShapingHandler createTrafficShapingHandler(EventLoopGroup e
       return null;
     }
     GlobalTrafficShapingHandler trafficShapingHandler;
-    if (options.getMaxDelayToWait() != 0 && options.getCheckIntervalForStats() != 0) {
+    if (options.getMaxDelayToWait() != 0) {
       long maxDelayToWaitInMillis = options.getMaxDelayToWaitTimeUnit().toMillis(options.getMaxDelayToWait());
       long checkIntervalForStatsInMillis = options.getCheckIntervalForStatsTimeUnit().toMillis(options.getCheckIntervalForStats());
       trafficShapingHandler = new GlobalTrafficShapingHandler(eventLoopGroup, options.getOutboundGlobalBandwidth(), options.getInboundGlobalBandwidth(), checkIntervalForStatsInMillis, maxDelayToWaitInMillis);
-    } else if (options.getCheckIntervalForStats() != 0) {
+    } else {
       long checkIntervalForStatsInMillis = options.getCheckIntervalForStatsTimeUnit().toMillis(options.getCheckIntervalForStats());
       trafficShapingHandler = new GlobalTrafficShapingHandler(eventLoopGroup, options.getOutboundGlobalBandwidth(), options.getInboundGlobalBandwidth(), checkIntervalForStatsInMillis);
-    } else {
-      trafficShapingHandler = new GlobalTrafficShapingHandler(eventLoopGroup, options.getOutboundGlobalBandwidth(), options.getInboundGlobalBandwidth());
     }
     if (options.getPeakOutboundGlobalBandwidth() != 0) {
       trafficShapingHandler.setMaxGlobalWriteSize(options.getPeakOutboundGlobalBandwidth());
@@ -147,6 +145,31 @@ public Future<Boolean> updateSSLOptions(SSLOptions options, boolean force) {
     }
   }
 
+  public void updateTrafficShapingOptions(TrafficShapingOptions options) {
+    if (options == null) {
+      throw new IllegalArgumentException("Invalid null value passed for traffic shaping options update");
+    }
+    if (trafficShapingHandler == null) {
+      throw new IllegalStateException("Unable to update traffic shaping options because the server was not configured " +
+                                      "to use traffic shaping during startup");
+    }
+    TCPServerBase server = actualServer;
+    if (server != null && server != this) {
+      server.updateTrafficShapingOptions(options);
+    } else {
+      long checkIntervalForStatsInMillis = options.getCheckIntervalForStatsTimeUnit().toMillis(options.getCheckIntervalForStats());
+      trafficShapingHandler.configure(options.getOutboundGlobalBandwidth(), options.getInboundGlobalBandwidth(), checkIntervalForStatsInMillis);
+
+      if (options.getPeakOutboundGlobalBandwidth() != 0) {
+        trafficShapingHandler.setMaxGlobalWriteSize(options.getPeakOutboundGlobalBandwidth());
+      }
+      if (options.getMaxDelayToWait() != 0) {
+        long maxDelayToWaitInMillis = options.getMaxDelayToWaitTimeUnit().toMillis(options.getMaxDelayToWait());
+        trafficShapingHandler.setMaxWriteDelay(maxDelayToWaitInMillis);
+      }
+    }
+  }
+
   public Future<TCPServerBase> bind(SocketAddress address) {
     ContextInternal listenContext = vertx.getOrCreateContext();
     return listen(address, listenContext).map(this);
diff --git a/src/test/java/io/vertx/core/http/HttpBandwidthLimitingTest.java b/src/test/java/io/vertx/core/http/HttpBandwidthLimitingTest.java
@@ -55,22 +55,27 @@ public static Iterable<Object[]> data() {
 
     Function<Vertx, HttpServer> http1ServerFactory = (v) -> Providers.http1Server(v, INBOUND_LIMIT, OUTBOUND_LIMIT);
     Function<Vertx, HttpServer> http2ServerFactory = (v) -> Providers.http2Server(v, INBOUND_LIMIT, OUTBOUND_LIMIT);
+    Function<Vertx, HttpServer> http1NonTrafficShapedServerFactory = (v) -> Providers.http1Server(v, 0, 0);
+    Function<Vertx, HttpServer> http2NonTrafficShapedServerFactory = (v) -> Providers.http1Server(v, 0, 0);
     Function<Vertx, HttpClient> http1ClientFactory = (v) -> v.createHttpClient();
     Function<Vertx, HttpClient> http2ClientFactory = (v) -> v.createHttpClient(createHttp2ClientOptions());
 
     return Arrays.asList(new Object[][] {
-      { 1.1, http1ServerFactory, http1ClientFactory },
-      { 2.0, http2ServerFactory, http2ClientFactory }
+      { 1.1, http1ServerFactory, http1ClientFactory, http1NonTrafficShapedServerFactory },
+      { 2.0, http2ServerFactory, http2ClientFactory, http2NonTrafficShapedServerFactory }
     });
   }
 
   private Function<Vertx, HttpServer> serverFactory;
   private Function<Vertx, HttpClient> clientFactory;
+  private Function<Vertx, HttpServer> nonTrafficShapedServerFactory;
 
   public HttpBandwidthLimitingTest(double protoVersion, Function<Vertx, HttpServer> serverFactory,
-                                   Function<Vertx, HttpClient> clientFactory) {
+                                   Function<Vertx, HttpClient> clientFactory,
+                                   Function<Vertx, HttpServer> nonTrafficShapedServerFactory) {
     this.serverFactory = serverFactory;
     this.clientFactory = clientFactory;
+    this.nonTrafficShapedServerFactory = nonTrafficShapedServerFactory;
   }
 
   @Before
@@ -199,6 +204,63 @@ public void start(Promise<Void> startPromise) {
     Assert.assertTrue(elapsedMillis > expectedTimeMillis(totalReceivedLength.get(), OUTBOUND_LIMIT)); // because there are simultaneous 2 requests
   }
 
+  @Test
+  public void testDynamicOutboundRateUpdate() throws Exception {
+    Buffer expectedBuffer = TestUtils.randomBuffer(TEST_CONTENT_SIZE);
+
+    HttpServer testServer = serverFactory.apply(vertx);
+    testServer.requestHandler(HANDLERS.bufferRead(expectedBuffer));
+    startServer(testServer);
+
+    // update outbound rate to twice the limit
+    TrafficShapingOptions trafficOptions = new TrafficShapingOptions()
+                                             .setInboundGlobalBandwidth(INBOUND_LIMIT) // unchanged
+                                             .setOutboundGlobalBandwidth(2 * OUTBOUND_LIMIT);
+    testServer.updateTrafficShapingOptions(trafficOptions);
+
+    long startTime = System.nanoTime();
+    HttpClient testClient = clientFactory.apply(vertx);
+    read(expectedBuffer, testServer, testClient);
+    await();
+    long elapsedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startTime);
+
+    Assert.assertTrue(elapsedMillis < expectedUpperBoundTimeMillis(TEST_CONTENT_SIZE, OUTBOUND_LIMIT));
+  }
+
+  @Test
+  public void testDynamicInboundRateUpdate() throws Exception {
+    Buffer expectedBuffer = TestUtils.randomBuffer((TEST_CONTENT_SIZE));
+
+    HttpServer testServer = serverFactory.apply(vertx);
+    testServer.requestHandler(HANDLERS.bufferWrite(expectedBuffer));
+    startServer(testServer);
+
+    // update inbound rate to twice the limit
+    TrafficShapingOptions trafficOptions = new TrafficShapingOptions()
+                                             .setOutboundGlobalBandwidth(OUTBOUND_LIMIT) // unchanged
+                                             .setInboundGlobalBandwidth(2 * INBOUND_LIMIT);
+    testServer.updateTrafficShapingOptions(trafficOptions);
+
+    long startTime = System.nanoTime();
+    HttpClient testClient = clientFactory.apply(vertx);
+    write(expectedBuffer, testServer, testClient);
+    await();
+    long elapsedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startTime);
+
+    Assert.assertTrue(elapsedMillis < expectedUpperBoundTimeMillis(TEST_CONTENT_SIZE, INBOUND_LIMIT));
+  }
+
+  @Test(expected = IllegalStateException.class)
+  public void testRateUpdateWhenServerStartedWithoutTrafficShaping() {
+    HttpServer testServer = nonTrafficShapedServerFactory.apply(vertx);
+
+    // update inbound rate to twice the limit
+    TrafficShapingOptions trafficOptions = new TrafficShapingOptions()
+                                             .setOutboundGlobalBandwidth(OUTBOUND_LIMIT)
+                                             .setInboundGlobalBandwidth(2 * INBOUND_LIMIT);
+    testServer.updateTrafficShapingOptions(trafficOptions);
+  }
+
   /**
    * The throttling takes a while to kick in so the expected time cannot be strict especially
    * for small data sizes in these tests.
@@ -211,6 +273,10 @@ private long expectedTimeMillis(long size, int rate) {
     return (long) (TimeUnit.MILLISECONDS.convert(( size / rate), TimeUnit.SECONDS) * 0.5); // multiplied by 0.5 to be more tolerant of time pauses during CI runs
   }
 
+  private long expectedUpperBoundTimeMillis(long size, int rate) {
+    return TimeUnit.MILLISECONDS.convert(( size / rate), TimeUnit.SECONDS); // Since existing rate will be upperbound, runs should complete by this time
+  }
+
   private void read(Buffer expected, HttpServer server, HttpClient client) {
     client.request(HttpMethod.GET, server.actualPort(), DEFAULT_HTTP_HOST,"/buffer-read")
           .compose(req -> req.send()
@@ -280,19 +346,25 @@ static class Providers {
     private static HttpServer http1Server(Vertx vertx, int inboundLimit, int outboundLimit) {
       HttpServerOptions options = new HttpServerOptions()
                                     .setHost(DEFAULT_HTTP_HOST)
-                                    .setPort(DEFAULT_HTTP_PORT)
-                                    .setTrafficShapingOptions(new TrafficShapingOptions()
-                                                                .setInboundGlobalBandwidth(inboundLimit)
-                                                                .setOutboundGlobalBandwidth(outboundLimit));
+                                    .setPort(DEFAULT_HTTP_PORT);
+
+      if (inboundLimit != 0 || outboundLimit != 0) {
+        options.setTrafficShapingOptions(new TrafficShapingOptions()
+                                           .setInboundGlobalBandwidth(inboundLimit)
+                                           .setOutboundGlobalBandwidth(outboundLimit));
+      }
 
       return vertx.createHttpServer(options);
     }
 
     private static HttpServer http2Server(Vertx vertx, int inboundLimit, int outboundLimit) {
-      HttpServerOptions options = createHttp2ServerOptions(DEFAULT_HTTP_PORT, DEFAULT_HTTP_HOST)
-                                    .setTrafficShapingOptions(new TrafficShapingOptions()
-                                                                .setInboundGlobalBandwidth(inboundLimit)
-                                                                .setOutboundGlobalBandwidth(outboundLimit));
+      HttpServerOptions options = createHttp2ServerOptions(DEFAULT_HTTP_PORT, DEFAULT_HTTP_HOST);
+
+      if (inboundLimit != 0 || outboundLimit != 0) {
+        options.setTrafficShapingOptions(new TrafficShapingOptions()
+                                           .setInboundGlobalBandwidth(inboundLimit)
+                                           .setOutboundGlobalBandwidth(outboundLimit));
+      }
 
       return vertx.createHttpServer(options);
     }
diff --git a/src/test/java/io/vertx/core/net/NetBandwidthLimitingTest.java b/src/test/java/io/vertx/core/net/NetBandwidthLimitingTest.java

Original file line number	Diff line number	Diff line change
`@@ -256,4 +256,12 @@ default void updateSSLOptions(SSLOptions options, boolean force, Handler<AsyncRe`
`256`	`256`	`fut.onComplete(handler);`
`257`	`257`	`}`
`258`	`258`	`}`
	`259`	`+`
	`260`	`+ /**`
	`261`	`+ * Update traffic shaping options {@code options}, the update happens if valid values are passed for traffic`
	`262`	`+ * shaping options. This update happens synchronously and at best effort for rate update to take effect immediately.`
	`263`	`+ *`
	`264`	`+ * @param options the new traffic shaping options`
	`265`	`+ */`
	`266`	`+ void updateTrafficShapingOptions(TrafficShapingOptions options);`
`259`	`267`	`}`
Original file line number	Diff line number	Diff line change
`@@ -139,14 +139,14 @@ public TrafficShapingOptions setMaxDelayToWaitUnit(TimeUnit maxDelayToWaitTimeUn`
`139`	`139`	`}`
`140`	`140`
`141`	`141`	`/**`
`142`		`- * Set the delay between two computations of performances for channels or 0 if no stats are to be computed`
	`142`	`+ * Set the delay between two computations of performances for channels`
`143`	`143`	`*`
`144`	`144`	`* @param checkIntervalForStats delay between two computations of performances`
`145`	`145`	`* @return a reference to this, so the API can be used fluently`
`146`	`146`	`*/`
`147`	`147`	`public TrafficShapingOptions setCheckIntervalForStats(long checkIntervalForStats) {`
`148`	`148`	`this.checkIntervalForStats = checkIntervalForStats;`
`149`		`- ObjectUtil.checkPositive(this.checkIntervalForStats, "checkIntervalForStats");`
	`149`	`+ ObjectUtil.checkPositiveOrZero(this.checkIntervalForStats, "checkIntervalForStats");`
`150`	`150`	`return this;`
`151`	`151`	`}`
`152`	`152`