tuned GC parameters in Docker images

ensured in-memory-cache is allocated to heap memory
rebuild native image with wider ISA compatibility
2025-03-24 14:42:04 +08:00 · 2025-03-11 12:29:43 +08:00 · 2025-03-10 22:28:55 +08:00 · 2025-03-10 10:09:30 +08:00 · 2025-03-09 22:16:03 +08:00 · 2025-03-09 13:57:52 +08:00
25 changed files with 561 additions and 214 deletions
--- a/benchmark/rbcs-filesystem.yml
+++ b/benchmark/rbcs-filesystem.yml
@@ -0,0 +1,93 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: rbcs-server
+data:
+  rbcs-server.xml: |
+        <?xml version="1.0" encoding="UTF-8" standalone="yes"?>
+        <rbcs:server xmlns:xs="http://www.w3.org/2001/XMLSchema-instance"
+                     xmlns:rbcs="urn:net.woggioni.rbcs.server"
+                     xmlns:rbcs-memcache="urn:net.woggioni.rbcs.server.memcache"
+                     xs:schemaLocation="urn:net.woggioni.rbcs.server.memcache jpms://net.woggioni.rbcs.server.memcache/net/woggioni/rbcs/server/memcache/schema/rbcs-memcache.xsd urn:net.woggioni.rbcs.server jpms://net.woggioni.rbcs.server/net/woggioni/rbcs/server/schema/rbcs-server.xsd"
+        >
+            <bind host="0.0.0.0" port="8080" incoming-connections-backlog-size="128"/>
+            <connection
+                    max-request-size="0xd000000"
+                    idle-timeout="PT15S"
+                    read-idle-timeout="PT30S"
+                    write-idle-timeout="PT30S"/>
+            <event-executor use-virtual-threads="true"/>
+            <cache xs:type="rbcs:fileSystemCacheType" max-age="P7D" enable-compression="false" path="/rbcs/cache"/>
+        </rbcs:server>
+
+---
+apiVersion: v1
+kind: PersistentVolumeClaim
+metadata:
+  name: rbcs-pvc
+  namespace: default
+spec:
+  accessModes:
+    - ReadWriteOnce
+  storageClassName: local-path
+  resources:
+    requests:
+      storage: 16Gi
+---
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: rbcs-deployment
+  labels:
+    app: rbcs
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: rbcs
+  template:
+    metadata:
+      labels:
+        app: rbcs
+    spec:
+      containers:
+        - name: rbcs
+          image: gitea.woggioni.net/woggioni/rbcs:native
+          imagePullPolicy: Always
+          args: ['server', '-c', 'rbcs-server.xml']
+          ports:
+          - containerPort: 8080
+          volumeMounts:
+          - name: config-volume
+            mountPath: /rbcs/rbcs-server.xml
+            subPath: rbcs-server.xml
+          - name: cache-volume
+            mountPath: /rbcs/cache
+          resources:
+            requests:
+              memory: "0.25Gi"
+              cpu: "1"
+            limits:
+              memory: "0.25Gi"
+              cpu: "3.5"      
+      volumes:
+        - name: config-volume
+          configMap:
+            name: rbcs-server
+        - name: cache-volume
+          persistentVolumeClaim:
+            claimName: rbcs-pvc
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: rbcs-service
+spec:
+  type: LoadBalancer
+  ports:
+  - port: 8080
+    targetPort: 8080
+    protocol: TCP
+  selector:
+    app: rbcs
+
--- a/benchmark/rbcs-in-memory.yml
+++ b/benchmark/rbcs-in-memory.yml
@@ -0,0 +1,76 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: rbcs-server
+data:
+  rbcs-server.xml: |
+        <?xml version="1.0" encoding="UTF-8" standalone="yes"?>
+        <rbcs:server xmlns:xs="http://www.w3.org/2001/XMLSchema-instance"
+                     xmlns:rbcs="urn:net.woggioni.rbcs.server"
+                     xmlns:rbcs-memcache="urn:net.woggioni.rbcs.server.memcache"
+                     xs:schemaLocation="urn:net.woggioni.rbcs.server.memcache jpms://net.woggioni.rbcs.server.memcache/net/woggioni/rbcs/server/memcache/schema/rbcs-memcache.xsd urn:net.woggioni.rbcs.server jpms://net.woggioni.rbcs.server/net/woggioni/rbcs/server/schema/rbcs-server.xsd"
+        >
+            <bind host="0.0.0.0" port="8080" incoming-connections-backlog-size="128"/>
+            <connection
+                    max-request-size="0xd000000"
+                    idle-timeout="PT15S"
+                    read-idle-timeout="PT30S"
+                    write-idle-timeout="PT30S"/>
+            <event-executor use-virtual-threads="true"/>
+            <cache xs:type="rbcs:inMemoryCacheType" max-age="P7D" enable-compression="false" max-size="0xb0000000" />
+        </rbcs:server>
+
+---
+
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: rbcs-deployment
+  labels:
+    app: rbcs
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: rbcs
+  template:
+    metadata:
+      labels:
+        app: rbcs
+    spec:
+      containers:
+        - name: rbcs
+          image: gitea.woggioni.net/woggioni/rbcs:native
+          imagePullPolicy: Always
+          args: ['server', '-c', 'rbcs-server.xml']
+          ports:
+          - containerPort: 8080
+          volumeMounts:
+          - name: config-volume
+            mountPath: /rbcs/rbcs-server.xml
+            subPath: rbcs-server.xml
+          resources:
+            requests:
+              memory: "0.5Gi"
+              cpu: "1"
+            limits:
+              memory: "4Gi"
+              cpu: "3.5"      
+      volumes:
+        - name: config-volume
+          configMap:
+            name: rbcs-server
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: rbcs-service
+spec:
+  type: LoadBalancer
+  ports:
+  - port: 8080
+    targetPort: 8080
+    protocol: TCP
+  selector:
+    app: rbcs
+
--- a/benchmark/rbcs-memcache.yml
+++ b/benchmark/rbcs-memcache.yml
@@ -0,0 +1,117 @@
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: rbcs-server
+data:
+  rbcs-server.xml: |
+        <?xml version="1.0" encoding="UTF-8" standalone="yes"?>
+        <rbcs:server xmlns:xs="http://www.w3.org/2001/XMLSchema-instance"
+                     xmlns:rbcs="urn:net.woggioni.rbcs.server"
+                     xmlns:rbcs-memcache="urn:net.woggioni.rbcs.server.memcache"
+                     xs:schemaLocation="urn:net.woggioni.rbcs.server.memcache jpms://net.woggioni.rbcs.server.memcache/net/woggioni/rbcs/server/memcache/schema/rbcs-memcache.xsd urn:net.woggioni.rbcs.server jpms://net.woggioni.rbcs.server/net/woggioni/rbcs/server/schema/rbcs-server.xsd"
+        >
+            <bind host="0.0.0.0" port="8080" incoming-connections-backlog-size="128"/>
+            <connection
+                    max-request-size="0xd000000"
+                    idle-timeout="PT15S"
+                    read-idle-timeout="PT30S"
+                    write-idle-timeout="PT30S"/>
+            <event-executor use-virtual-threads="true"/>
+            <!--cache xs:type="rbcs:inMemoryCacheType" max-age="P7D" enable-compression="false" max-size="0x10000000" /-->
+            <cache xs:type="rbcs-memcache:memcacheCacheType" max-age="P7D" chunk-size="0x1000" digest="MD5">
+                <server host="memcached-service" port="11211" max-connections="256"/>
+            </cache>
+        </rbcs:server>
+
+---
+
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: rbcs-deployment
+  labels:
+    app: rbcs
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: rbcs
+  template:
+    metadata:
+      labels:
+        app: rbcs
+    spec:
+      containers:
+        - name: rbcs
+          image: gitea.woggioni.net/woggioni/rbcs:native
+          imagePullPolicy: Always
+          args: ['server', '-c', 'rbcs-server.xml']
+          ports:
+          - containerPort: 8080
+          volumeMounts:
+          - name: config-volume
+            mountPath: /rbcs/rbcs-server.xml
+            subPath: rbcs-server.xml
+          resources:
+            requests:
+              memory: "0.25Gi"
+              cpu: "1"
+            limits:
+              memory: "0.25Gi"
+              cpu: "1"      
+      volumes:
+        - name: config-volume
+          configMap:
+            name: rbcs-server
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: rbcs-service
+spec:
+  type: LoadBalancer
+  ports:
+  - port: 8080
+    targetPort: 8080
+    protocol: TCP
+  selector:
+    app: rbcs
+---
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: memcached-deployment
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: memcached
+  template:
+    metadata:
+      labels:
+        app: memcached
+    spec:
+      containers:
+      - name: memcached
+        image: memcached
+        args: ["-I", "128m", "-m", "4096"]
+        resources:
+          requests:
+            memory: "1Gi"
+            cpu: "500m"        # 0.5 CPU
+          limits:
+            memory: "5Gi"
+            cpu: "500m"        # 0.5 CP
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: memcached-service
+spec:
+  type: ClusterIP  # ClusterIP makes it accessible only within the cluster
+  ports:
+  - port: 11211     # Default memcached port
+    targetPort: 11211
+    protocol: TCP
+  selector:
+    app: memcached
--- a/docker/Dockerfile
+++ b/docker/Dockerfile
@@ -5,7 +5,7 @@ WORKDIR /home/luser

 FROM base-release AS release-vanilla
 ADD rbcs-cli-envelope-*.jar rbcs.jar
-ENTRYPOINT ["java", "-XX:+UseSerialGC", "-XX:GCTimeRatio=24", "-jar", "/home/luser/rbcs.jar", "server"]
+ENTRYPOINT ["java", "-Dlogback.configurationFile=logback.xml", "-XX:MaxRAMPercentage=70", "-XX:GCTimeRatio=24", "-XX:+UseZGC", "-XX:+ZGenerational", "-jar", "/home/luser/rbcs.jar"]

 FROM base-release AS release-memcache
 ADD --chown=luser:luser rbcs-cli-envelope-*.jar rbcs.jar
@@ -14,10 +14,10 @@ WORKDIR /home/luser/plugins
 RUN --mount=type=bind,source=.,target=/build/distributions tar -xf /build/distributions/rbcs-server-memcache*.tar
 WORKDIR /home/luser
 ADD logback.xml .
-ENTRYPOINT ["java", "-Dlogback.configurationFile=logback.xml", "-XX:+UseSerialGC", "-XX:GCTimeRatio=24", "-jar", "/home/luser/rbcs.jar", "server"]
+ENTRYPOINT ["java", "-Dlogback.configurationFile=logback.xml", "-XX:MaxRAMPercentage=70", "-XX:GCTimeRatio=24", "-XX:+UseZGC", "-XX:+ZGenerational", "-jar", "/home/luser/rbcs.jar"]

 FROM scratch AS release-native
 ADD rbcs-cli.upx /rbcs/rbcs-cli
 ENV RBCS_CONFIGURATION_DIR="/rbcs"
 WORKDIR /rbcs
-ENTRYPOINT ["/rbcs/rbcs-cli"]
+ENTRYPOINT ["/rbcs/rbcs-cli", "-XX:MaximumHeapSizePercent=70"]
--- a/gradle.properties
+++ b/gradle.properties
@@ -2,7 +2,7 @@ org.gradle.configuration-cache=false
 org.gradle.parallel=true
 org.gradle.caching=true

-rbcs.version = 0.2.0
+rbcs.version = 0.2.1

 lys.version = 2025.03.08

--- a/rbcs-api/src/main/java/net/woggioni/rbcs/api/CacheHandler.java
+++ b/rbcs-api/src/main/java/net/woggioni/rbcs/api/CacheHandler.java
@@ -16,7 +16,7 @@ public abstract class CacheHandler extends ChannelInboundHandlerAdapter {
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        if(!requestFinished && msg instanceof CacheMessage) {
-            if(msg instanceof CacheMessage.LastCacheContent || msg instanceof CacheMessage.CacheGetRequest) requestFinished = true;
+            if(msg instanceof CacheMessage.LastCacheContent) requestFinished = true;
            try {
                channelRead0(ctx, (CacheMessage) msg);
            } finally {
--- a/rbcs-api/src/main/java/net/woggioni/rbcs/api/CacheValueMetadata.java
+++ b/rbcs-api/src/main/java/net/woggioni/rbcs/api/CacheValueMetadata.java
@@ -1,10 +1,9 @@
 package net.woggioni.rbcs.api;

+import java.io.Serializable;
 import lombok.Getter;
 import lombok.RequiredArgsConstructor;

-import java.io.Serializable;
-
@Getter
@RequiredArgsConstructor
 public class CacheValueMetadata implements Serializable {
--- a/rbcs-api/src/main/java/net/woggioni/rbcs/api/Configuration.java
+++ b/rbcs-api/src/main/java/net/woggioni/rbcs/api/Configuration.java
@@ -1,16 +1,15 @@
 package net.woggioni.rbcs.api;


-import lombok.EqualsAndHashCode;
-import lombok.NonNull;
-import lombok.Value;
-
 import java.nio.file.Path;
 import java.security.cert.X509Certificate;
 import java.time.Duration;
 import java.util.Map;
 import java.util.Set;
 import java.util.stream.Collectors;
+import lombok.EqualsAndHashCode;
+import lombok.NonNull;
+import lombok.Value;

@Value
 public class Configuration {
--- a/rbcs-cli/build.gradle
+++ b/rbcs-cli/build.gradle
@@ -105,6 +105,7 @@ tasks.named(NativeImagePlugin.CONFIGURE_NATIVE_IMAGE_TASK_NAME, NativeImageConfi
    systemProperty('io.netty.leakDetectionLevel', 'DISABLED')
    modularity.inferModulePath = false
    enabled = true
+    systemProperty('gradle.tmp.dir', temporaryDir.toString())
 }

 nativeImage {
--- a/rbcs-cli/native-image/native-image.properties
+++ b/rbcs-cli/native-image/native-image.properties
@@ -1,2 +1,2 @@
-Args=-O3 -march=skylake --gc=serial --install-exit-handlers --initialize-at-run-time=io.netty --enable-url-protocols=jpms --initialize-at-build-time=net.woggioni.rbcs.common.RbcsUrlStreamHandlerFactory,net.woggioni.rbcs.common.RbcsUrlStreamHandlerFactory$JpmsHandler
+Args=-O3 -march=x86-64-v2 --gc=serial --install-exit-handlers --initialize-at-run-time=io.netty --enable-url-protocols=jpms --initialize-at-build-time=net.woggioni.rbcs.common.RbcsUrlStreamHandlerFactory,net.woggioni.rbcs.common.RbcsUrlStreamHandlerFactory$JpmsHandler
 #-H:TraceClassInitialization=io.netty.handler.ssl.BouncyCastleAlpnSslUtils
--- a/rbcs-cli/native-image/reflect-config.json
+++ b/rbcs-cli/native-image/reflect-config.json
@@ -637,6 +637,10 @@
  "name":"sun.security.provider.DSA$SHA256withDSA",
  "methods":[{"name":"<init>","parameterTypes":[] }]
 },
+{
+  "name":"sun.security.provider.JavaKeyStore$JKS",
+  "methods":[{"name":"<init>","parameterTypes":[] }]
+},
 {
  "name":"sun.security.provider.MD5",
  "methods":[{"name":"<init>","parameterTypes":[] }]
@@ -725,14 +729,6 @@
  "name":"sun.security.x509.CertificatePoliciesExtension",
  "methods":[{"name":"<init>","parameterTypes":["java.lang.Boolean","java.lang.Object"] }]
 },
-{
-  "name":"sun.security.x509.ExtendedKeyUsageExtension",
-  "methods":[{"name":"<init>","parameterTypes":["java.lang.Boolean","java.lang.Object"] }]
-},
-{
-  "name":"sun.security.x509.IssuerAlternativeNameExtension",
-  "methods":[{"name":"<init>","parameterTypes":["java.lang.Boolean","java.lang.Object"] }]
-},
 {
  "name":"sun.security.x509.KeyUsageExtension",
  "methods":[{"name":"<init>","parameterTypes":["java.lang.Boolean","java.lang.Object"] }]
--- a/rbcs-cli/src/configureNativeImage/kotlin/net/woggioni/rbcs/cli/graal/GraalNativeImageConfiguration.kt
+++ b/rbcs-cli/src/configureNativeImage/kotlin/net/woggioni/rbcs/cli/graal/GraalNativeImageConfiguration.kt
@@ -179,6 +179,8 @@ object GraalNativeImageConfiguration {
            } catch (ee : ExecutionException) {
            }
        }
-        RemoteBuildCacheServerCli.main("--help")
+        System.setProperty("net.woggioni.rbcs.conf.dir", System.getProperty("gradle.tmp.dir"))
+        RemoteBuildCacheServerCli.createCommandLine().execute("--version")
+        RemoteBuildCacheServerCli.createCommandLine().execute("server", "-t", "PT10S")
    }
 }
--- a/rbcs-cli/src/main/kotlin/net/woggioni/rbcs/cli/RemoteBuildCacheServerCli.kt
+++ b/rbcs-cli/src/main/kotlin/net/woggioni/rbcs/cli/RemoteBuildCacheServerCli.kt
@@ -26,8 +26,8 @@ class RemoteBuildCacheServerCli : RbcsCommand() {
        private fun setPropertyIfNotPresent(key: String, value: String) {
            System.getProperty(key) ?: System.setProperty(key, value)
        }
-        @JvmStatic
-        fun main(vararg args: String) {
+
+        fun createCommandLine() : CommandLine {
            setPropertyIfNotPresent("logback.configurationFile", "net/woggioni/rbcs/cli/logback.xml")
            setPropertyIfNotPresent("io.netty.leakDetectionLevel", "DISABLED")
            val currentClassLoader = RemoteBuildCacheServerCli::class.java.classLoader
@@ -56,7 +56,12 @@ class RemoteBuildCacheServerCli : RbcsCommand() {
                    addSubcommand(GetCommand())
                    addSubcommand(HealthCheckCommand())
                })
-            System.exit(commandLine.execute(*args))
+            return commandLine
+        }
+
+        @JvmStatic
+        fun main(vararg args: String) {
+            System.exit(createCommandLine().execute(*args))
        }
    }

--- a/rbcs-cli/src/main/kotlin/net/woggioni/rbcs/cli/impl/commands/ClientCommand.kt
+++ b/rbcs-cli/src/main/kotlin/net/woggioni/rbcs/cli/impl/commands/ClientCommand.kt
@@ -6,7 +6,6 @@ import net.woggioni.rbcs.client.Configuration
 import net.woggioni.rbcs.common.createLogger
 import net.woggioni.rbcs.common.debug
 import picocli.CommandLine
-import java.lang.IllegalArgumentException
 import java.nio.file.Path

@CommandLine.Command(
--- a/rbcs-server-memcache/src/main/kotlin/net/woggioni/rbcs/server/memcache/MemcacheCacheConfiguration.kt
+++ b/rbcs-server-memcache/src/main/kotlin/net/woggioni/rbcs/server/memcache/MemcacheCacheConfiguration.kt
@@ -1,7 +1,6 @@
 package net.woggioni.rbcs.server.memcache

 import io.netty.channel.ChannelFactory
-import io.netty.channel.ChannelHandler
 import io.netty.channel.EventLoopGroup
 import io.netty.channel.pool.FixedChannelPool
 import io.netty.channel.socket.DatagramChannel
--- a/rbcs-server-memcache/src/main/kotlin/net/woggioni/rbcs/server/memcache/MemcacheCacheHandler.kt
+++ b/rbcs-server-memcache/src/main/kotlin/net/woggioni/rbcs/server/memcache/MemcacheCacheHandler.kt
@@ -69,10 +69,14 @@ class MemcacheCacheHandler(
        }
    }

+    private interface InProgressRequest {
+
+    }
+
    private inner class InProgressGetRequest(
-        private val key: String,
+        val key: String,
        private val ctx: ChannelHandlerContext
-    ) {
+    ) : InProgressRequest {
        private val acc = ctx.alloc().compositeBuffer()
        private val chunk = ctx.alloc().compositeBuffer()
        private val outputStream = ByteBufOutputStream(chunk).let {
@@ -107,17 +111,17 @@ class MemcacheCacheHandler(
                }
            if (responseSent) {
                acc.readBytes(outputStream, acc.readableBytes())
-                if(acc.readableBytes() >= chunkSize) {
+                if (acc.readableBytes() >= chunkSize) {
                    flush(false)
                }
            }
        }

-        private fun flush(last : Boolean) {
+        private fun flush(last: Boolean) {
            val toSend = extractChunk(chunk, ctx.alloc())
-            val msg = if(last) {
+            val msg = if (last) {
                log.trace(ctx) {
-                    "Sending last chunk to client on channel"
+                    "Sending last chunk to client"
                }
                LastCacheContent(toSend)
            } else {
@@ -144,14 +148,14 @@ class MemcacheCacheHandler(
    }

    private inner class InProgressPutRequest(
-        private val ch : NettyChannel,
-        metadata : CacheValueMetadata,
-        val digest : ByteBuf,
+        private val ch: NettyChannel,
+        metadata: CacheValueMetadata,
+        val digest: ByteBuf,
        val requestController: CompletableFuture<MemcacheRequestController>,
        private val alloc: ByteBufAllocator
-    ) {
+    ) : InProgressRequest {
        private var totalSize = 0
-        private var tmpFile : FileChannel? = null
+        private var tmpFile: FileChannel? = null
        private val accumulator = alloc.compositeBuffer()
        private val stream = ByteBufOutputStream(accumulator).let {
            if (compressionEnabled) {
@@ -178,7 +182,7 @@ class MemcacheCacheHandler(
            tmpFile?.let {
                flushToDisk(it, accumulator)
            }
-            if(accumulator.readableBytes() > 0x100000) {
+            if (accumulator.readableBytes() > 0x100000) {
                log.debug(ch) {
                    "Entry is too big, buffering it into a file"
                }
@@ -195,18 +199,18 @@ class MemcacheCacheHandler(
            }
        }

-        private fun flushToDisk(fc : FileChannel, buf : CompositeByteBuf) {
+        private fun flushToDisk(fc: FileChannel, buf: CompositeByteBuf) {
            val chunk = extractChunk(buf, alloc)
            fc.write(chunk.nioBuffer())
            chunk.release()
        }

-        fun commit() : Pair<Int, ReadableByteChannel> {
+        fun commit(): Pair<Int, ReadableByteChannel> {
            digest.release()
            accumulator.retain()
            stream.close()
            val fileChannel = tmpFile
-            return if(fileChannel != null) {
+            return if (fileChannel != null) {
                flushToDisk(fileChannel, accumulator)
                accumulator.release()
                fileChannel.position(0)
@@ -227,8 +231,7 @@ class MemcacheCacheHandler(
        }
    }

-    private var inProgressPutRequest: InProgressPutRequest? = null
-    private var inProgressGetRequest: InProgressGetRequest? = null
+    private var inProgressRequest: InProgressRequest? = null

    override fun channelRead0(ctx: ChannelHandlerContext, msg: CacheMessage) {
        when (msg) {
@@ -255,7 +258,7 @@ class MemcacheCacheHandler(
                        log.debug(ctx) {
                            "Cache hit for key ${msg.key} on memcache"
                        }
-                        inProgressGetRequest = InProgressGetRequest(msg.key, ctx)
+                        inProgressRequest = InProgressGetRequest(msg.key, ctx)
                    }

                    BinaryMemcacheResponseStatus.KEY_ENOENT -> {
@@ -269,19 +272,26 @@ class MemcacheCacheHandler(

            override fun contentReceived(content: MemcacheContent) {
                log.trace(ctx) {
-                    "${if(content is LastMemcacheContent) "Last chunk" else "Chunk"} of ${content.content().readableBytes()} bytes received from memcache for key ${msg.key}"
+                    "${if (content is LastMemcacheContent) "Last chunk" else "Chunk"} of ${
+                        content.content().readableBytes()
+                    } bytes received from memcache for key ${msg.key}"
                }
-                inProgressGetRequest?.write(content.content())
+                (inProgressRequest as? InProgressGetRequest)?.let { inProgressGetRequest ->
+                    inProgressGetRequest.write(content.content())
                    if (content is LastMemcacheContent) {
-                    inProgressGetRequest?.commit()
+                        inProgressRequest = null
+                        inProgressGetRequest.commit()
+                    }
                }
            }

            override fun exceptionCaught(ex: Throwable) {
+                (inProgressRequest as? InProgressGetRequest).let { inProgressGetRequest ->
                    inProgressGetRequest?.let {
-                    inProgressGetRequest = null
+                        inProgressRequest = null
                        it.rollback()
                    }
+                }
                this@MemcacheCacheHandler.exceptionCaught(ctx, ex)
            }
        }
@@ -311,6 +321,7 @@ class MemcacheCacheHandler(
                        }
                        sendMessageAndFlush(ctx, CachePutResponse(msg.key))
                    }
+
                    else -> this@MemcacheCacheHandler.exceptionCaught(ctx, MemcacheException(status))
                }
            }
@@ -327,21 +338,30 @@ class MemcacheCacheHandler(
                this@MemcacheCacheHandler.exceptionCaught(ctx, ex)
            }
        }
-        inProgressPutRequest = InProgressPutRequest(ctx.channel(), msg.metadata, key, requestController, ctx.alloc())
+        inProgressRequest = InProgressPutRequest(ctx.channel(), msg.metadata, key, requestController, ctx.alloc())
    }

    private fun handleCacheContent(ctx: ChannelHandlerContext, msg: CacheContent) {
-        inProgressPutRequest?.let { request ->
+        val request = inProgressRequest
+        when (request) {
+            is InProgressPutRequest -> {
                log.trace(ctx) {
                    "Received chunk of ${msg.content().readableBytes()} bytes for memcache"
                }
                request.write(msg.content())
            }
+
+            is InProgressGetRequest -> {
+                msg.release()
+            }
+        }
    }

    private fun handleLastCacheContent(ctx: ChannelHandlerContext, msg: LastCacheContent) {
-        inProgressPutRequest?.let { request ->
-            inProgressPutRequest = null
+        val request = inProgressRequest
+        when (request) {
+            is InProgressPutRequest -> {
+                inProgressRequest = null
                log.trace(ctx) {
                    "Received last chunk of ${msg.content().readableBytes()} bytes for memcache"
                }
@@ -356,7 +376,7 @@ class MemcacheCacheHandler(
                    "Trying to send SET request to memcache"
                }
                request.requestController.whenComplete { requestController, ex ->
-                if(ex == null) {
+                    if (ex == null) {
                        log.trace(ctx) {
                            "Sending SET request to memcache"
                        }
@@ -374,7 +394,7 @@ class MemcacheCacheHandler(
                            while (true) {
                                val read = source.read(bb)
                                bb.limit()
-                            if(read >= 0 && bb.position() < chunkSize && bb.hasRemaining()) {
+                                if (read >= 0 && bb.position() < chunkSize && bb.hasRemaining()) {
                                    continue
                                }
                                val chunk = ctx.alloc().buffer(chunkSize)
@@ -384,7 +404,7 @@ class MemcacheCacheHandler(
                                log.trace(ctx) {
                                    "Sending ${chunk.readableBytes()} bytes chunk to memcache"
                                }
-                            if(read < 0) {
+                                if (read < 0) {
                                    requestController.sendContent(DefaultLastMemcacheContent(chunk))
                                    break
                                } else {
@@ -398,18 +418,23 @@ class MemcacheCacheHandler(
                }
            }
        }
+    }

    override fun exceptionCaught(ctx: ChannelHandlerContext, cause: Throwable) {
-        inProgressGetRequest?.let {
-            inProgressGetRequest = null
-            it.rollback()
-        }
-        inProgressPutRequest?.let {
-            inProgressPutRequest = null
-            it.requestController.thenAccept { controller ->
+        val request = inProgressRequest
+        when (request) {
+            is InProgressPutRequest -> {
+                inProgressRequest = null
+                request.requestController.thenAccept { controller ->
                    controller.exceptionCaught(cause)
                }
-            it.rollback()
+                request.rollback()
+            }
+
+            is InProgressGetRequest -> {
+                inProgressRequest = null
+                request.rollback()
+            }
        }
        super.exceptionCaught(ctx, cause)
    }
--- a/rbcs-server/src/main/kotlin/net/woggioni/rbcs/server/RemoteBuildCacheServer.kt
+++ b/rbcs-server/src/main/kotlin/net/woggioni/rbcs/server/RemoteBuildCacheServer.kt
@@ -54,6 +54,7 @@ import net.woggioni.rbcs.server.auth.RoleAuthorizer
 import net.woggioni.rbcs.server.configuration.Parser
 import net.woggioni.rbcs.server.configuration.Serializer
 import net.woggioni.rbcs.server.exception.ExceptionHandler
+import net.woggioni.rbcs.server.handler.BlackHoleRequestHandler
 import net.woggioni.rbcs.server.handler.MaxRequestSizeHandler
 import net.woggioni.rbcs.server.handler.ServerHandler
 import net.woggioni.rbcs.server.throttling.BucketManager
@@ -361,6 +362,7 @@ class RemoteBuildCacheServer(private val cfg: Configuration) {
            }
            pipeline.addLast(eventExecutorGroup, ServerHandler.NAME, serverHandler)
            pipeline.addLast(ExceptionHandler.NAME, ExceptionHandler)
+            pipeline.addLast(BlackHoleRequestHandler.NAME, BlackHoleRequestHandler())
        }

        override fun asyncClose() = cacheHandlerFactory.asyncClose()
--- a/rbcs-server/src/main/kotlin/net/woggioni/rbcs/server/cache/FileSystemCacheHandler.kt
+++ b/rbcs-server/src/main/kotlin/net/woggioni/rbcs/server/cache/FileSystemCacheHandler.kt
@@ -2,7 +2,6 @@ package net.woggioni.rbcs.server.cache

 import io.netty.buffer.ByteBuf
 import io.netty.channel.ChannelHandlerContext
-import io.netty.channel.SimpleChannelInboundHandler
 import io.netty.handler.codec.http.LastHttpContent
 import io.netty.handler.stream.ChunkedNioFile
 import net.woggioni.rbcs.api.CacheHandler
@@ -29,10 +28,16 @@ class FileSystemCacheHandler(
    private val chunkSize: Int
 ) : CacheHandler() {

+    private interface InProgressRequest{
+
+    }
+
+    private class InProgressGetRequest(val request : CacheGetRequest) : InProgressRequest
+
    private inner class InProgressPutRequest(
        val key : String,
        private val fileSink : FileSystemCache.FileSink
-    ) {
+    ) : InProgressRequest {

        private val stream = Channels.newOutputStream(fileSink.channel).let {
            if (compressionEnabled) {
@@ -56,7 +61,7 @@ class FileSystemCacheHandler(
        }
    }

-    private var inProgressPutRequest: InProgressPutRequest? = null
+    private var inProgressRequest: InProgressRequest? = null

    override fun channelRead0(ctx: ChannelHandlerContext, msg: CacheMessage) {
        when (msg) {
@@ -69,9 +74,35 @@ class FileSystemCacheHandler(
    }

    private fun handleGetRequest(ctx: ChannelHandlerContext, msg: CacheGetRequest) {
+        inProgressRequest = InProgressGetRequest(msg)
+
+    }
+
+    private fun handlePutRequest(ctx: ChannelHandlerContext, msg: CachePutRequest) {
        val key = String(Base64.getUrlEncoder().encode(processCacheKey(msg.key, digestAlgorithm)))
+        val sink = cache.put(key, msg.metadata)
+        inProgressRequest = InProgressPutRequest(msg.key, sink)
+    }
+
+    private fun handleCacheContent(ctx: ChannelHandlerContext, msg: CacheContent) {
+        val request = inProgressRequest
+        if(request is InProgressPutRequest) {
+            request.write(msg.content())
+        }
+    }
+
+    private fun handleLastCacheContent(ctx: ChannelHandlerContext, msg: LastCacheContent) {
+        when(val request = inProgressRequest) {
+            is InProgressPutRequest -> {
+                inProgressRequest = null
+                request.write(msg.content())
+                request.commit()
+                sendMessageAndFlush(ctx, CachePutResponse(request.key))
+            }
+            is InProgressGetRequest -> {
+                val key = String(Base64.getUrlEncoder().encode(processCacheKey(request.request.key, digestAlgorithm)))
                cache.get(key)?.also { entryValue ->
-            sendMessageAndFlush(ctx, CacheValueFoundResponse(msg.key, entryValue.metadata))
+                    sendMessageAndFlush(ctx, CacheValueFoundResponse(request.request.key, entryValue.metadata))
                    entryValue.channel.let { channel ->
                        if(compressionEnabled) {
                            InflaterInputStream(Channels.newInputStream(channel)).use { stream ->
@@ -96,28 +127,11 @@ class FileSystemCacheHandler(
                    }
                } ?: sendMessageAndFlush(ctx, CacheValueNotFoundResponse())
            }
-
-    private fun handlePutRequest(ctx: ChannelHandlerContext, msg: CachePutRequest) {
-        val key = String(Base64.getUrlEncoder().encode(processCacheKey(msg.key, digestAlgorithm)))
-        val sink = cache.put(key, msg.metadata)
-        inProgressPutRequest = InProgressPutRequest(msg.key, sink)
-    }
-
-    private fun handleCacheContent(ctx: ChannelHandlerContext, msg: CacheContent) {
-        inProgressPutRequest!!.write(msg.content())
-    }
-
-    private fun handleLastCacheContent(ctx: ChannelHandlerContext, msg: LastCacheContent) {
-        inProgressPutRequest?.let { request ->
-            inProgressPutRequest = null
-            request.write(msg.content())
-            request.commit()
-            sendMessageAndFlush(ctx, CachePutResponse(request.key))
        }
    }

    override fun exceptionCaught(ctx: ChannelHandlerContext, cause: Throwable) {
-        inProgressPutRequest?.rollback()
+        (inProgressRequest as? InProgressPutRequest)?.rollback()
        super.exceptionCaught(ctx, cause)
    }
 }
--- a/rbcs-server/src/main/kotlin/net/woggioni/rbcs/server/cache/InMemoryCache.kt
+++ b/rbcs-server/src/main/kotlin/net/woggioni/rbcs/server/cache/InMemoryCache.kt
@@ -75,6 +75,10 @@ class InMemoryCache(
                                cond.await(interval.toMillis(), TimeUnit.MILLISECONDS)
                            }
                        }
+                        map.forEach {
+                            it.value.content.release()
+                        }
+                        map.clear()
                    }
                    complete(null)
                } catch (ex: Throwable) {
--- a/rbcs-server/src/main/kotlin/net/woggioni/rbcs/server/cache/InMemoryCacheConfiguration.kt
+++ b/rbcs-server/src/main/kotlin/net/woggioni/rbcs/server/cache/InMemoryCacheConfiguration.kt
@@ -4,7 +4,6 @@ import io.netty.channel.ChannelFactory
 import io.netty.channel.EventLoopGroup
 import io.netty.channel.socket.DatagramChannel
 import io.netty.channel.socket.SocketChannel
-import io.netty.util.concurrent.Future
 import net.woggioni.rbcs.api.CacheHandlerFactory
 import net.woggioni.rbcs.api.Configuration
 import net.woggioni.rbcs.common.RBCS
--- a/rbcs-server/src/main/kotlin/net/woggioni/rbcs/server/cache/InMemoryCacheHandler.kt
+++ b/rbcs-server/src/main/kotlin/net/woggioni/rbcs/server/cache/InMemoryCacheHandler.kt
@@ -2,19 +2,11 @@ package net.woggioni.rbcs.server.cache

 import io.netty.buffer.ByteBuf
 import io.netty.channel.ChannelHandlerContext
-import io.netty.channel.SimpleChannelInboundHandler
 import net.woggioni.rbcs.api.CacheHandler
 import net.woggioni.rbcs.api.message.CacheMessage
-import net.woggioni.rbcs.api.message.CacheMessage.CacheContent
-import net.woggioni.rbcs.api.message.CacheMessage.CacheGetRequest
-import net.woggioni.rbcs.api.message.CacheMessage.CachePutRequest
-import net.woggioni.rbcs.api.message.CacheMessage.CachePutResponse
-import net.woggioni.rbcs.api.message.CacheMessage.CacheValueFoundResponse
-import net.woggioni.rbcs.api.message.CacheMessage.CacheValueNotFoundResponse
-import net.woggioni.rbcs.api.message.CacheMessage.LastCacheContent
+import net.woggioni.rbcs.api.message.CacheMessage.*
 import net.woggioni.rbcs.common.ByteBufOutputStream
 import net.woggioni.rbcs.common.RBCS.processCacheKey
-import net.woggioni.rbcs.common.trace
 import java.util.zip.Deflater
 import java.util.zip.DeflaterOutputStream
 import java.util.zip.InflaterOutputStream
@@ -26,7 +18,15 @@ class InMemoryCacheHandler(
    private val compressionLevel: Int
 ) : CacheHandler() {

-    private interface InProgressPutRequest : AutoCloseable {
+    private interface InProgressRequest : AutoCloseable {
+    }
+
+    private class InProgressGetRequest(val request : CacheGetRequest) : InProgressRequest {
+        override fun close() {
+        }
+    }
+
+    private interface InProgressPutRequest : InProgressRequest {
        val request: CachePutRequest
        val buf: ByteBuf

@@ -35,19 +35,15 @@ class InMemoryCacheHandler(

    private inner class InProgressPlainPutRequest(ctx: ChannelHandlerContext, override val request: CachePutRequest) :
        InProgressPutRequest {
-        override val buf = ctx.alloc().compositeBuffer()
-
-        private val stream = ByteBufOutputStream(buf).let {
-            if (compressionEnabled) {
-                DeflaterOutputStream(it, Deflater(compressionLevel))
-            } else {
-                it
-            }
-        }
+        override val buf = ctx.alloc().compositeHeapBuffer()

        override fun append(buf: ByteBuf) {
+            if(buf.isDirect) {
+                this.buf.writeBytes(buf)
+            } else {
                this.buf.addComponent(true, buf.retain())
            }
+        }

        override fun close() {
            buf.release()
@@ -74,7 +70,7 @@ class InMemoryCacheHandler(
        }
    }

-    private var inProgressPutRequest: InProgressPutRequest? = null
+    private var inProgressRequest: InProgressRequest? = null

    override fun channelRead0(ctx: ChannelHandlerContext, msg: CacheMessage) {
        when (msg) {
@@ -87,8 +83,30 @@ class InMemoryCacheHandler(
    }

    private fun handleGetRequest(ctx: ChannelHandlerContext, msg: CacheGetRequest) {
-        cache.get(processCacheKey(msg.key, digestAlgorithm))?.let { value ->
-            sendMessageAndFlush(ctx, CacheValueFoundResponse(msg.key, value.metadata))
+        inProgressRequest = InProgressGetRequest(msg)
+    }
+
+    private fun handlePutRequest(ctx: ChannelHandlerContext, msg: CachePutRequest) {
+        inProgressRequest = if(compressionEnabled) {
+            InProgressCompressedPutRequest(ctx, msg)
+        } else {
+            InProgressPlainPutRequest(ctx, msg)
+        }
+    }
+
+    private fun handleCacheContent(ctx: ChannelHandlerContext, msg: CacheContent) {
+        val req = inProgressRequest
+        if(req is InProgressPutRequest) {
+            req.append(msg.content())
+        }
+    }
+
+    private fun handleLastCacheContent(ctx: ChannelHandlerContext, msg: LastCacheContent) {
+        handleCacheContent(ctx, msg)
+        when(val req = inProgressRequest) {
+            is InProgressGetRequest -> {
+                cache.get(processCacheKey(req.request.key, digestAlgorithm))?.let { value ->
+                    sendMessageAndFlush(ctx, CacheValueFoundResponse(req.request.key, value.metadata))
                    if (compressionEnabled) {
                        val buf = ctx.alloc().heapBuffer()
                        InflaterOutputStream(ByteBufOutputStream(buf)).use {
@@ -102,37 +120,21 @@ class InMemoryCacheHandler(
                    }
                } ?: sendMessage(ctx, CacheValueNotFoundResponse())
            }
-
-    private fun handlePutRequest(ctx: ChannelHandlerContext, msg: CachePutRequest) {
-        inProgressPutRequest = if(compressionEnabled) {
-            InProgressCompressedPutRequest(ctx, msg)
-        } else {
-            InProgressPlainPutRequest(ctx, msg)
-        }
-    }
-
-    private fun handleCacheContent(ctx: ChannelHandlerContext, msg: CacheContent) {
-        inProgressPutRequest?.append(msg.content())
-    }
-
-    private fun handleLastCacheContent(ctx: ChannelHandlerContext, msg: LastCacheContent) {
-        handleCacheContent(ctx, msg)
-        inProgressPutRequest?.let { inProgressRequest ->
-            inProgressPutRequest = null
-            val buf = inProgressRequest.buf
+            is InProgressPutRequest -> {
+                this.inProgressRequest = null
+                val buf = req.buf
                buf.retain()
-            inProgressRequest.close()
-            val cacheKey = processCacheKey(inProgressRequest.request.key, digestAlgorithm)
-            cache.put(cacheKey, CacheEntry(inProgressRequest.request.metadata, buf))
-            sendMessageAndFlush(ctx, CachePutResponse(inProgressRequest.request.key))
+                req.close()
+                val cacheKey = processCacheKey(req.request.key, digestAlgorithm)
+                cache.put(cacheKey, CacheEntry(req.request.metadata, buf))
+                sendMessageAndFlush(ctx, CachePutResponse(req.request.key))
+            }
        }
    }

    override fun exceptionCaught(ctx: ChannelHandlerContext, cause: Throwable) {
-        inProgressPutRequest?.let { req ->
-            req.buf.release()
-            inProgressPutRequest = null
-        }
+        inProgressRequest?.close()
+        inProgressRequest = null
        super.exceptionCaught(ctx, cause)
    }
 }
--- a/rbcs-server/src/main/kotlin/net/woggioni/rbcs/server/handler/BlackHoleRequestHandler.kt
+++ b/rbcs-server/src/main/kotlin/net/woggioni/rbcs/server/handler/BlackHoleRequestHandler.kt
@@ -0,0 +1,13 @@
+package net.woggioni.rbcs.server.handler
+
+import io.netty.channel.ChannelHandlerContext
+import io.netty.channel.SimpleChannelInboundHandler
+import io.netty.handler.codec.http.HttpContent
+
+class BlackHoleRequestHandler : SimpleChannelInboundHandler<HttpContent>() {
+    companion object {
+        val NAME = BlackHoleRequestHandler::class.java.name
+    }
+    override fun channelRead0(ctx: ChannelHandlerContext, msg: HttpContent) {
+    }
+}
--- a/rbcs-server/src/main/kotlin/net/woggioni/rbcs/server/throttling/ThrottlingHandler.kt
+++ b/rbcs-server/src/main/kotlin/net/woggioni/rbcs/server/throttling/ThrottlingHandler.kt
@@ -94,6 +94,9 @@ class ThrottlingHandler(private val bucketManager : BucketManager,
                    handleBuckets(buckets, ctx, msg, false)
                }, waitDuration.toMillis(), TimeUnit.MILLISECONDS)
            } else {
+                queuedContent?.let { qc ->
+                    qc.forEach { it.release() }
+                }
                this.queuedContent = null
                sendThrottledResponse(ctx, waitDuration)
            }
--- a/rbcs-server/src/test/java/net/woggioni/rbcs/server/test/utils/CertificateUtils.java
+++ b/rbcs-server/src/test/java/net/woggioni/rbcs/server/test/utils/CertificateUtils.java
@@ -1,5 +1,14 @@
 package net.woggioni.rbcs.server.test.utils;

+import java.math.BigInteger;
+import java.security.KeyPair;
+import java.security.KeyPairGenerator;
+import java.security.PrivateKey;
+import java.security.SecureRandom;
+import java.security.cert.X509Certificate;
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
+import java.util.Date;
 import org.bouncycastle.asn1.DERSequence;
 import org.bouncycastle.asn1.x500.X500Name;
 import org.bouncycastle.asn1.x509.BasicConstraints;
@@ -15,16 +24,6 @@ import org.bouncycastle.cert.jcajce.JcaX509v3CertificateBuilder;
 import org.bouncycastle.operator.ContentSigner;
 import org.bouncycastle.operator.jcajce.JcaContentSignerBuilder;

-import java.math.BigInteger;
-import java.security.KeyPair;
-import java.security.KeyPairGenerator;
-import java.security.PrivateKey;
-import java.security.SecureRandom;
-import java.security.cert.X509Certificate;
-import java.time.Instant;
-import java.time.temporal.ChronoUnit;
-import java.util.Date;
-
 public class CertificateUtils {

    public record X509Credentials(
--- a/rbcs-server/src/test/kotlin/net/woggioni/rbcs/server/test/BasicAuthServerTest.kt
+++ b/rbcs-server/src/test/kotlin/net/woggioni/rbcs/server/test/BasicAuthServerTest.kt
@@ -154,7 +154,7 @@ class BasicAuthServerTest : AbstractBasicAuthServerTest() {
    }

    @Test
-    @Order(6)
+    @Order(8)
    fun getAsAThrottledUser() {
        val client: HttpClient = HttpClient.newHttpClient()

@@ -172,7 +172,7 @@ class BasicAuthServerTest : AbstractBasicAuthServerTest() {
    }

    @Test
-    @Order(7)
+    @Order(9)
    fun getAsAThrottledUser2() {
        val client: HttpClient = HttpClient.newHttpClient()
Author	SHA1	Message	Date
Walter Oggioni	69f215e68f	tuned GC parameters in Docker images All checks were successful CI / build (push) Successful in 16m35s Details	2025-03-24 14:42:04 +08:00
Walter Oggioni	222b475223	ensured in-memory-cache is allocated to heap memory All checks were successful CI / build (push) Successful in 12m29s Details	2025-03-11 12:29:43 +08:00
Walter Oggioni	ede515e2ca	rebuild native image with wider ISA compatibility All checks were successful CI / build (push) Successful in 14m13s Details	2025-03-10 22:28:55 +08:00
Walter Oggioni	974fdb7a91	added k8s benchmark files All checks were successful CI / build (push) Successful in 15m39s Details	2025-03-10 10:09:30 +08:00
Walter Oggioni	a294229ff0	fixed memory leak in MemcachedCacheHandler	2025-03-09 22:16:03 +08:00
Walter Oggioni	9600dd7e4f	solved issue with ignored `HttpContent` and `HttpCacheContent` messages in the Netty pipeline All checks were successful CI / build (push) Successful in 12m48s Details	2025-03-09 13:57:52 +08:00