Merge branch 'fix/52' into 'master'

Improving multi-threadability with Netty

See merge request mangadex/mangadex_at_home!14

dev/settings.json

@@ -3,5 +3,6 @@
   "max_cache_size_mib": 2048,
   "client_port": 8080,
   "max_burst_rate_kib_per_second": 100,
-  "max_bandwidth_mib_per_hour": 1
+  "max_bandwidth_mib_per_hour": 1,
+  "threads_per_cpu": 32
 }

src/main/java/mdnet/base/ClientSettings.java

@@ -16,14 +16,17 @@ public final class ClientSettings {
 	private final int clientPort;
 	@SerializedName("client_secret")
 	private final String clientSecret;
+	@SerializedName("threads_per_cpu")
+	private final int threadsPerCpu;
 
 	public ClientSettings(long maxCacheSizeMib, long maxBandwidthMibPerHour, long maxBurstRateKibPerSecond,
-			int clientPort, String clientSecret) {
+			int clientPort, String clientSecret, int threadsPerCpu) {
 		this.maxCacheSizeMib = maxCacheSizeMib;
 		this.maxBandwidthMibPerHour = maxBandwidthMibPerHour;
 		this.maxBurstRateKibPerSecond = maxBurstRateKibPerSecond;
 		this.clientPort = clientPort;
 		this.clientSecret = Objects.requireNonNull(clientSecret);
+		this.threadsPerCpu = threadsPerCpu;
 	}
 
 	public long getMaxCacheSizeMib() {
@@ -46,11 +49,15 @@ public final class ClientSettings {
 		return clientSecret;
 	}
 
+	public int getThreadsPerCpu() {
+		return threadsPerCpu;
+	}
+
 	@Override
 	public String toString() {
 		return "ClientSettings{" + "maxCacheSizeMib=" + maxCacheSizeMib + ", maxBandwidthMibPerHour="
 				+ maxBandwidthMibPerHour + ", maxBurstRateKibPerSecond=" + maxBurstRateKibPerSecond + ", clientPort="
-				+ clientPort + ", clientSecret='" + "<hidden>" + '\'' + '}';
+				+ clientPort + ", clientSecret='" + clientSecret + '\'' + ", threadsPerCpu=" + threadsPerCpu + '}';
 	}
 
 	public static boolean isSecretValid(String clientSecret) {

src/main/kotlin/mdnet/base/Application.kt

@@ -39,13 +39,13 @@ import javax.crypto.CipherOutputStream
 import javax.crypto.spec.SecretKeySpec
 
 private val LOGGER = LoggerFactory.getLogger("Application")
-private val THREADS_TO_ALLOCATE = Runtime.getRuntime().availableProcessors() * 30 / 2
+private val THREADS_TO_ALLOCATE = 65535 // Have it at the maximum open sockets a user can have in most modern OSes. No reason to limit this, just limit it at the Netty side.
 
 fun getServer(cache: DiskLruCache, serverSettings: ServerSettings, clientSettings: ClientSettings, statistics: AtomicReference<Statistics>): Http4kServer {
     val executor = Executors.newCachedThreadPool()
 
     if (LOGGER.isInfoEnabled) {
-        LOGGER.info("Starting ApacheClient with {} threads", THREADS_TO_ALLOCATE)
+        LOGGER.info("Starting image retriever")
     }
 
     val client = ApacheClient(responseBodyMode = BodyMode.Stream, client = HttpClients.custom()
@@ -193,7 +193,6 @@ fun getServer(cache: DiskLruCache, serverSettings: ServerSettings, clientSetting
                         if (LOGGER.isTraceEnabled) {
                             LOGGER.trace("Request for $sanitizedUri is being served")
                         }
-
                         respondWithImage(mdResponse.body.stream, contentLength, contentType, lastModified)
                     }
                 }

src/main/kotlin/mdnet/base/Netty.kt

@@ -37,9 +37,12 @@ import javax.net.ssl.SSLException
 private val LOGGER = LoggerFactory.getLogger("Application")
 
 class Netty(private val tls: ServerSettings.TlsCert, private val clientSettings: ClientSettings, private val stats: AtomicReference<Statistics>) : ServerConfig {
+    private val threadsToAllocate: Int
+        get() = Runtime.getRuntime().availableProcessors() * clientSettings.threadsPerCpu
+
     override fun toServer(httpHandler: HttpHandler): Http4kServer = object : Http4kServer {
-        private val masterGroup = NioEventLoopGroup()
-        private val workerGroup = NioEventLoopGroup()
+        private val masterGroup = NioEventLoopGroup(threadsToAllocate)
+        private val workerGroup = NioEventLoopGroup(threadsToAllocate)
         private lateinit var closeFuture: ChannelFuture
         private lateinit var address: InetSocketAddress
 
@@ -52,6 +55,10 @@ class Netty(private val tls: ServerSettings.TlsCert, private val clientSettings:
         }
 
         override fun start(): Http4kServer = apply {
+            if (LOGGER.isInfoEnabled) {
+                LOGGER.info("Starting webserver with {} threads", threadsToAllocate)
+            }
+
             val (mainCert, chainCert) = getX509Certs(tls.certificate)
             val sslContext = SslContextBuilder
                 .forServer(getPrivateKey(tls.privateKey), mainCert, chainCert)