[EXPERIMENTS] Quota rejection and it's impact on server latencies

clients/da-vinci-client/src/main/java/com/linkedin/davinci/listener/response/ReadResponse.java

@@ -27,6 +27,7 @@ public abstract class ReadResponse {
   private int hadamardProductCount = 0;
   private int countOperatorCount = 0;
   private int rcu = 0;
+  private long responseWriteAndFlushStartTimeNanos = -1;
 
   public void setCompressionStrategy(CompressionStrategy compressionStrategy) {
     this.compressionStrategy = compressionStrategy;
@@ -205,4 +206,16 @@ public int getCountOperatorCount() {
   public abstract ByteBuf getResponseBody();
 
   public abstract int getResponseSchemaIdHeader();
+
+  public void setResponseWriteAndFlushStartTimeNanos(long responseWriteAndFlushStartTimeNanos) {
+    this.responseWriteAndFlushStartTimeNanos = responseWriteAndFlushStartTimeNanos;
+  }
+
+  public void recordResponseWriteAndFlushStartTimeNanos() {
+    this.responseWriteAndFlushStartTimeNanos = System.nanoTime();
+  }
+
+  public long getResponseWriteAndFlushStartTimeNanos() {
+    return responseWriteAndFlushStartTimeNanos;
+  }
 }

internal/venice-client-common/src/main/java/com/linkedin/venice/utils/concurrent/ThreadPoolFactory.java

@@ -23,7 +23,7 @@ public static ThreadPoolExecutor createThreadPool(
     ThreadPoolExecutor executor = new ThreadPoolExecutor(
         threadCount,
         threadCount,
-        0,
+        600,
         TimeUnit.MILLISECONDS,
         getExecutionQueue(capacity, blockingQueueType),
         new DaemonThreadFactory(threadNamePrefix));

internal/venice-common/src/main/java/com/linkedin/venice/throttle/TokenBucket.java

@@ -18,8 +18,8 @@ public class TokenBucket {
   private final Clock clock;
   private final AtomicLong tokens;
   private final AtomicLong tokensRequestedSinceLastRefill;
-  private volatile long previousRefillTime;
-  private volatile long nextUpdateTime;
+  private final AtomicLong previousRefillTime = new AtomicLong();
+  private final AtomicLong nextUpdateTime = new AtomicLong();
 
   /**
    * This constructor should only be used by tests.  Application should not specify it's own instance of Clock
@@ -51,8 +51,9 @@ public TokenBucket(long capacity, long refillAmount, long refillInterval, TimeUn
 
     tokens = new AtomicLong(capacity);
     tokensRequestedSinceLastRefill = new AtomicLong(0);
-    previousRefillTime = clock.millis();
-    nextUpdateTime = previousRefillTime + refillIntervalMs;
+    long timeNow = clock.millis();
+    previousRefillTime.set(timeNow);
+    nextUpdateTime.set(timeNow + refillIntervalMs);
 
     refillPerSecond = refillAmount / (float) refillUnit.toSeconds(refillInterval);
   }
@@ -74,25 +75,18 @@ public TokenBucket(long capacity, long refillAmount, long refillInterval, TimeUn
    * function is short-circuited. Consumers of the token bucket should always retry.
    */
   private void update() {
-    if (clock.millis() > nextUpdateTime) {
-      synchronized (this) {
-        long timeNow = clock.millis();
-        if (timeNow > nextUpdateTime) {
-          long refillCount = (timeNow - nextUpdateTime) / refillIntervalMs + 1;
-          long totalRefillAmount = refillCount * refillAmount;
-          tokens.getAndAccumulate(totalRefillAmount, (existing, toAdd) -> {
-            long newTokens = existing + toAdd;
-            if (newTokens > capacity) {
-              return capacity;
-            } else {
-              return newTokens;
-            }
-          });
-          previousRefillTime = timeNow;
-          tokensRequestedSinceLastRefill.set(0);
-          nextUpdateTime = timeNow + refillIntervalMs;
-        }
-      }
+    long timeNow = clock.millis();
+    long currentNextUpdateTime = nextUpdateTime.get();
+    if (timeNow > currentNextUpdateTime
+        && nextUpdateTime.compareAndSet(currentNextUpdateTime, timeNow + refillIntervalMs)) {
+      long refillCount = (timeNow - currentNextUpdateTime) / refillIntervalMs + 1;
+      long totalRefillAmount = refillCount * refillAmount;
+      tokens.getAndAccumulate(totalRefillAmount, (existing, toAdd) -> {
+        long newTokens = existing + toAdd;
+        return Math.min(newTokens, capacity);
+      });
+      previousRefillTime.set(timeNow);
+      tokensRequestedSinceLastRefill.set(0);
     }
   }
 
@@ -135,7 +129,7 @@ public float getAmortizedRefillPerSecond() {
   }
 
   public double getStaleUsageRatio() {
-    long timeSinceLastRefill = TimeUnit.MILLISECONDS.toSeconds(clock.millis() - previousRefillTime);
+    long timeSinceLastRefill = TimeUnit.MILLISECONDS.toSeconds(clock.millis() - previousRefillTime.get());
     if (timeSinceLastRefill > 0) {
       return ((double) tokensRequestedSinceLastRefill.get() / (double) timeSinceLastRefill) / refillPerSecond;
     } else {

internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/fastclient/AvroStoreClientEndToEndTest.java

@@ -280,6 +280,18 @@ private void runTest(
     }
   }
 
+  @Test
+  public void testFastClientForNettyStats() throws Exception {
+    testFastClientGet(
+        false,
+        false,
+        false,
+        false,
+        2,
+        RequestType.MULTI_GET,
+        StoreMetadataFetchMode.SERVER_BASED_METADATA);
+  }
+
   @Test(dataProvider = "FastClient-Test-Permutations", timeOut = TIME_OUT)
   public void testFastClientGet(
       boolean dualRead,

internal/venice-test-common/src/integrationTest/java/com/linkedin/venice/integration/utils/TestVeniceServer.java

@@ -12,6 +12,7 @@
 import com.linkedin.venice.helix.HelixCustomizedViewOfflinePushRepository;
 import com.linkedin.venice.listener.ListenerService;
 import com.linkedin.venice.listener.StorageReadRequestHandler;
+import com.linkedin.venice.listener.VeniceServerNettyStats;
 import com.linkedin.venice.meta.ReadOnlySchemaRepository;
 import com.linkedin.venice.meta.ReadOnlyStoreRepository;
 import com.linkedin.venice.security.SSLFactory;
@@ -82,7 +83,8 @@ protected StorageReadRequestHandler createRequestHandler(
           boolean parallelBatchGetEnabled,
           int parallelBatchGetChunkSize,
           StorageEngineBackedCompressorFactory compressorFactory,
-          Optional<ResourceReadUsageTracker> resourceReadUsageTracker) {
+          Optional<ResourceReadUsageTracker> resourceReadUsageTracker,
+          VeniceServerNettyStats nettyStats) {
 
         return new StorageReadRequestHandler(
             executor,
@@ -98,7 +100,8 @@ protected StorageReadRequestHandler createRequestHandler(
             parallelBatchGetChunkSize,
             serverConfig,
             compressorFactory,
-            resourceReadUsageTracker) {
+            resourceReadUsageTracker,
+            nettyStats) {
           @Override
           public void channelRead(ChannelHandlerContext context, Object message) throws Exception {
             RequestHandler handler = requestHandler.get();

internal/venice-test-common/src/integrationTest/resources/log4j2.properties

@@ -1,22 +1,55 @@
-status = error
 name = PropertiesConfig
 
 filters = threshold
 
 filter.threshold.type = ThresholdFilter
-filter.threshold.level = debug
+filter.threshold.level = info
 
-appenders = console
+#appenders = console
 
 appender.console.type = Console
 appender.console.name = STDOUT
 appender.console.layout.type = PatternLayout
-appender.console.layout.pattern = %d{yyyy-MM-dd HH:mm:ss} - [%X{component}] %p [%c{1}] [%t] %m%n
+appender.console.layout.pattern = %d{yyyy-MM-dd HH:mm:ss} %p [%c{1}] [%t] %m%n
 
-rootLogger.level = info
-rootLogger.appenderRefs = stdout
-rootLogger.appenderRef.stdout.ref = STDOUT
+appender.rolling.type = RollingFile
+appender.rolling.name = fileLogger
+appender.rolling.fileName= /Users/sumane/logs/venice-test.log
+appender.rolling.filePattern= ${basePath}/app_%d{yyyyMMdd}.log.gz
+appender.rolling.layout.type = PatternLayout
+appender.rolling.layout.pattern = %d{yyyy-MM-dd HH:mm:ss} %p [%c{1.}] [%t] %m%n
+appender.rolling.policies.type = Policies
+
+rootLogger.level = ERROR
+#rootLogger.appenderRefs = stdout
+#rootLogger.appenderRef.stdout.ref = STDOUT
+#rootLogger.appenderRefs = RollingFile
+rootLogger.appenderRef.RollingFile.ref = fileLogger
 
-# Set package org.apache.zookeeper log level to error.
 logger.zk.name = org.apache.zookeeper
-logger.zk.level = error
+logger.zk.level = off
+logger.apache.name = org.apache
+logger.apache.level = error
+logger.kafka.name = kafka
+logger.kafka.level = error
+logger.e.name = org.eclipse
+logger.e.level = off
+logger.d2.name = com.linkedin.d2
+logger.d2.level = error
+logger.venice.name = com.linkedin.venice
+logger.venice.level = info
+logger.dvc.name = com.linkedin.davinci
+logger.dvc.level = info
+logger.vs.name = com.linkedin.venice.serialization.avro
+logger.vs.level = ERROR
+logger.vu.name = com.linkedin.venice.utils
+logger.vu.level = ERROR
+#com.linkedin.venice.listener
+logger.vl.name = com.linkedin.venice.listener
+logger.vl.level = debug
+#c.l.v.helix
+logger.helix.name = com.linkedin.venice.helix
+logger.helix.level = error
+#com.linkedin.venice.listener.ServerConnectionStatsHandler
+logger.scs.name = com.linkedin.venice.listener.ServerConnectionStatsHandler
+logger.scs.level = debug

services/venice-server/src/main/java/com/linkedin/venice/listener/HttpChannelInitializer.java

@@ -34,8 +34,10 @@
 import io.netty.channel.socket.SocketChannel;
 import io.netty.handler.codec.http.HttpObjectAggregator;
 import io.netty.handler.codec.http.HttpServerCodec;
+import io.netty.handler.flush.FlushConsolidationHandler;
 import io.netty.handler.timeout.IdleStateHandler;
 import io.tehuti.metrics.MetricsRepository;
+import java.time.Clock;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Optional;
@@ -61,6 +63,7 @@ public class HttpChannelInitializer extends ChannelInitializer<SocketChannel> {
   private final ReadQuotaEnforcementHandler quotaEnforcer;
   private final VeniceHttp2PipelineInitializerBuilder http2PipelineInitializerBuilder;
   private final ServerConnectionStats serverConnectionStats;
+  private VeniceServerNettyStats nettyStats;
   AggServerQuotaUsageStats quotaUsageStats;
   AggServerQuotaTokenBucketStats quotaTokenBucketStats;
   List<ServerInterceptor> aclInterceptors;
@@ -78,6 +81,31 @@ public HttpChannelInitializer(
       Optional<StaticAccessController> routerAccessController,
       Optional<DynamicAccessController> storeAccessController,
       StorageReadRequestHandler requestHandler) {
+    this(
+        storeMetadataRepository,
+        customizedViewRepository,
+        metricsRepository,
+        sslFactory,
+        sslHandshakeExecutor,
+        serverConfig,
+        routerAccessController,
+        storeAccessController,
+        requestHandler,
+        null);
+  }
+
+  public HttpChannelInitializer(
+      ReadOnlyStoreRepository storeMetadataRepository,
+      CompletableFuture<HelixCustomizedViewOfflinePushRepository> customizedViewRepository,
+      MetricsRepository metricsRepository,
+      Optional<SSLFactory> sslFactory,
+      Executor sslHandshakeExecutor,
+      VeniceServerConfig serverConfig,
+      Optional<StaticAccessController> routerAccessController,
+      Optional<DynamicAccessController> storeAccessController,
+      StorageReadRequestHandler requestHandler,
+      VeniceServerNettyStats nettyStats) {
+    this.nettyStats = nettyStats;
     this.serverConfig = serverConfig;
     this.requestHandler = requestHandler;
     this.isDaVinciClient = serverConfig.isDaVinciClient();
@@ -133,7 +161,9 @@ public HttpChannelInitializer(
           customizedViewRepository,
           nodeId,
           quotaUsageStats,
-          metricsRepository);
+          metricsRepository,
+          nettyStats,
+          Clock.systemUTC());
 
       // Token Bucket Stats for a store must be initialized when that store is created
       this.quotaTokenBucketStats = new AggServerQuotaTokenBucketStats(metricsRepository, quotaEnforcer);
@@ -174,6 +204,7 @@ interface ChannelPipelineConsumer {
 
   @Override
   public void initChannel(SocketChannel ch) {
+    ch.pipeline().addLast(new FlushConsolidationHandler(16, true));
     if (sslFactory.isPresent()) {
       SslInitializer sslInitializer = new SslInitializer(SslUtils.toAlpiniSSLFactory(sslFactory.get()), false);
       if (sslHandshakeExecutor != null) {
@@ -184,9 +215,9 @@ public void initChannel(SocketChannel ch) {
     }
     ChannelPipelineConsumer httpPipelineInitializer = (pipeline, whetherNeedServerCodec) -> {
       ServerConnectionStatsHandler serverConnectionStatsHandler =
-          new ServerConnectionStatsHandler(serverConnectionStats, serverConfig.getRouterPrincipalName());
+          new ServerConnectionStatsHandler(serverConnectionStats, nettyStats, serverConfig.getRouterPrincipalName());
       pipeline.addLast(serverConnectionStatsHandler);
-      StatsHandler statsHandler = new StatsHandler(singleGetStats, multiGetStats, computeStats);
+      StatsHandler statsHandler = new StatsHandler(singleGetStats, multiGetStats, computeStats, nettyStats);
       pipeline.addLast(statsHandler);
       if (whetherNeedServerCodec) {
         pipeline.addLast(new HttpServerCodec());
@@ -245,7 +276,7 @@ public void initChannel(SocketChannel ch) {
   public VeniceServerGrpcRequestProcessor initGrpcRequestProcessor() {
     VeniceServerGrpcRequestProcessor grpcServerRequestProcessor = new VeniceServerGrpcRequestProcessor();
 
-    StatsHandler statsHandler = new StatsHandler(singleGetStats, multiGetStats, computeStats);
+    StatsHandler statsHandler = new StatsHandler(singleGetStats, multiGetStats, computeStats, nettyStats);
     GrpcStatsHandler grpcStatsHandler = new GrpcStatsHandler(statsHandler);
     grpcServerRequestProcessor.addHandler(grpcStatsHandler);
 

services/venice-server/src/main/java/com/linkedin/venice/listener/ListenerService.java

@@ -88,6 +88,8 @@ public ListenerService(
     this.isGrpcEnabled = serverConfig.isGrpcEnabled();
     this.grpcPort = serverConfig.getGrpcPort();
 
+    VeniceServerNettyStats nettyStats = new VeniceServerNettyStats(metricsRepository, "NettyStats");
+
     executor = createThreadPool(
         serverConfig.getRestServiceStorageThreadNum(),
         "StorageExecutionThread",
@@ -121,7 +123,8 @@ public ListenerService(
         serverConfig.isEnableParallelBatchGet(),
         serverConfig.getParallelBatchGetChunkSize(),
         compressorFactory,
-        resourceReadUsageTracker);
+        resourceReadUsageTracker,
+        nettyStats);
 
     storageReadRequestHandler = requestHandler;
 
@@ -134,7 +137,8 @@ public ListenerService(
         serverConfig,
         routerAccessController,
         storeAccessController,
-        requestHandler);
+        requestHandler,
+        nettyStats);
 
     Class<? extends ServerChannel> serverSocketChannelClass = NioServerSocketChannel.class;
     boolean epollEnabled = serverConfig.isRestServiceEpollEnabled();
@@ -186,6 +190,7 @@ public ListenerService(
   public boolean startInner() throws Exception {
     serverFuture = bootstrap.bind(port).sync();
     LOGGER.info("Listener service started on port: {}", port);
+    LOGGER.info("####EXPERIMENT#### -- No flush in quota code path");
 
     if (isGrpcEnabled) {
       grpcServer.start();
@@ -236,7 +241,8 @@ protected StorageReadRequestHandler createRequestHandler(
       boolean parallelBatchGetEnabled,
       int parallelBatchGetChunkSize,
       StorageEngineBackedCompressorFactory compressorFactory,
-      Optional<ResourceReadUsageTracker> resourceReadUsageTracker) {
+      Optional<ResourceReadUsageTracker> resourceReadUsageTracker,
+      VeniceServerNettyStats nettyStats) {
     return new StorageReadRequestHandler(
         executor,
         computeExecutor,
@@ -251,6 +257,7 @@ protected StorageReadRequestHandler createRequestHandler(
         parallelBatchGetChunkSize,
         serverConfig,
         compressorFactory,
-        resourceReadUsageTracker);
+        resourceReadUsageTracker,
+        nettyStats);
   }
 }