/* Mangadex@Home Copyright (c) 2020, MangaDex Network This file is part of MangaDex@Home. MangaDex@Home is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License as published by the Free Software Foundation, either version 3 of the License, or (at your option) any later version. MangaDex@Home is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details. You should have received a copy of the GNU General Public License along with this MangaDex@Home. If not, see . */ package mdnet.netty import io.netty.bootstrap.ServerBootstrap import io.netty.channel.* import io.netty.channel.epoll.Epoll import io.netty.channel.epoll.EpollEventLoopGroup import io.netty.channel.epoll.EpollServerSocketChannel import io.netty.channel.nio.NioEventLoopGroup import io.netty.channel.socket.SocketChannel import io.netty.channel.socket.nio.NioServerSocketChannel import io.netty.handler.codec.DecoderException import io.netty.handler.codec.http.HttpObjectAggregator import io.netty.handler.codec.http.HttpServerCodec import io.netty.handler.codec.http.HttpServerKeepAliveHandler import io.netty.handler.ssl.SslContextBuilder import io.netty.handler.stream.ChunkedWriteHandler import io.netty.handler.timeout.ReadTimeoutException import io.netty.handler.timeout.ReadTimeoutHandler import io.netty.handler.timeout.WriteTimeoutException import io.netty.handler.timeout.WriteTimeoutHandler import io.netty.handler.traffic.GlobalTrafficShapingHandler import io.netty.handler.traffic.TrafficCounter import io.netty.incubator.channel.uring.IOUring import io.netty.incubator.channel.uring.IOUringEventLoopGroup import io.netty.incubator.channel.uring.IOUringServerSocketChannel import io.netty.util.concurrent.DefaultEventExecutorGroup import io.netty.util.internal.SystemPropertyUtil import mdnet.Constants import mdnet.data.Statistics import mdnet.logging.info import mdnet.logging.trace import mdnet.logging.warn import mdnet.settings.ServerSettings import mdnet.settings.TlsCert import org.http4k.core.HttpHandler import org.http4k.server.Http4kChannelHandler import org.http4k.server.Http4kServer import org.http4k.server.ServerConfig import org.slf4j.LoggerFactory import java.io.ByteArrayInputStream import java.io.IOException import java.io.InputStream import java.net.InetSocketAddress import java.net.SocketException import java.security.PrivateKey import java.security.cert.CertificateFactory import java.security.cert.X509Certificate import java.util.Locale import java.util.concurrent.TimeUnit import javax.net.ssl.SSLException sealed class NettyTransport(threads: Int) { abstract val bossGroup: EventLoopGroup abstract val workerGroup: EventLoopGroup abstract val factory: ChannelFactory val executor = DefaultEventExecutorGroup( threads.also { require(threads > 0) { "Threads must be greater than zero" } } ) private class NioTransport(threads: Int) : NettyTransport(threads) { override val bossGroup = NioEventLoopGroup(1) override val workerGroup = NioEventLoopGroup(8) override val factory = ChannelFactory { NioServerSocketChannel() } } private class EpollTransport(threads: Int) : NettyTransport(threads) { override val bossGroup = EpollEventLoopGroup(1) override val workerGroup = EpollEventLoopGroup(8) override val factory = ChannelFactory { EpollServerSocketChannel() } } private class IOUringTransport(threads: Int) : NettyTransport(threads) { override val bossGroup = IOUringEventLoopGroup(1) override val workerGroup = IOUringEventLoopGroup(8) override val factory = ChannelFactory { IOUringServerSocketChannel() } } companion object { private val LOGGER = LoggerFactory.getLogger(NettyTransport::class.java) private fun defaultNumThreads() = Runtime.getRuntime().availableProcessors() * 2 fun bestForPlatform(threads: Int): NettyTransport { val name = SystemPropertyUtil.get("os.name").toLowerCase(Locale.UK).trim { it <= ' ' } val threadsToUse = if (threads == 0) defaultNumThreads() else threads LOGGER.info { "Choosing a transport with $threadsToUse threads" } if (name.startsWith("linux")) { if (!SystemPropertyUtil.get("no-iouring").toBoolean()) { if (IOUring.isAvailable()) { LOGGER.info { "Using IOUring transport" } return IOUringTransport(threadsToUse) } else { LOGGER.info { "IOUring transport not available (this may be normal)" } } } if (!SystemPropertyUtil.get("no-epoll").toBoolean()) { if (Epoll.isAvailable()) { LOGGER.info { "Using Epoll transport" } return EpollTransport(threadsToUse) } else { LOGGER.info { "Epoll transport not available (this may be normal)" } } } } LOGGER.info { "Using Nio transport" } return NioTransport(threadsToUse) } } } class Netty( private val tls: TlsCert, private val serverSettings: ServerSettings, private val statistics: Statistics ) : ServerConfig { override fun toServer(http: HttpHandler): Http4kServer = object : Http4kServer { private val transport = NettyTransport.bestForPlatform(serverSettings.threads) private lateinit var channel: Channel private val burstLimiter = object : GlobalTrafficShapingHandler( transport.workerGroup, serverSettings.maxKilobitsPerSecond * 1000L / 8L, 0, 100 ) { override fun doAccounting(counter: TrafficCounter) { statistics.bytesSent.getAndAccumulate(counter.cumulativeWrittenBytes()) { a, b -> a + b } counter.resetCumulativeTime() } } override fun start(): Http4kServer = apply { LOGGER.info { "Starting Netty!" } val certs = getX509Certs(tls.certificate) val sslContext = SslContextBuilder .forServer(getPrivateKey(tls.privateKey), certs) .protocols("TLSv1.3", "TLSv1.2", "TLSv1.1", "TLSv1") .build() val bootstrap = ServerBootstrap() bootstrap.group(transport.bossGroup, transport.workerGroup) .channelFactory(transport.factory) .childHandler(object : ChannelInitializer() { public override fun initChannel(ch: SocketChannel) { ch.pipeline().addLast("ssl", sslContext.newHandler(ch.alloc())) ch.pipeline().addLast("codec", HttpServerCodec()) ch.pipeline().addLast("keepAlive", HttpServerKeepAliveHandler()) ch.pipeline().addLast("aggregator", HttpObjectAggregator(65536)) ch.pipeline().addLast("burstLimiter", burstLimiter) ch.pipeline().addLast( "readTimeoutHandler", ReadTimeoutHandler(Constants.MAX_READ_TIME_SECONDS) ) ch.pipeline().addLast( "writeTimeoutHandler", WriteTimeoutHandler(Constants.MAX_WRITE_TIME_SECONDS) ) ch.pipeline().addLast("streamer", ChunkedWriteHandler()) ch.pipeline().addLast(transport.executor, "handler", Http4kChannelHandler(http)) ch.pipeline().addLast( "exceptions", object : ChannelInboundHandlerAdapter() { override fun exceptionCaught(ctx: ChannelHandlerContext, cause: Throwable) { if (cause is SSLException || (cause is DecoderException && cause.cause is SSLException)) { LOGGER.trace(cause) { "Ignored invalid SSL connection" } } else if (cause is IOException || cause is SocketException) { LOGGER.trace(cause) { "User (downloader) abruptly closed the connection" } } else if (cause !is ReadTimeoutException && cause !is WriteTimeoutException) { LOGGER.warn(cause) { "Exception in pipeline" } } } } ) } }) .option(ChannelOption.SO_BACKLOG, 1000) .childOption(ChannelOption.SO_KEEPALIVE, true) channel = bootstrap.bind(InetSocketAddress(serverSettings.hostname, serverSettings.port)).sync().channel() } override fun stop() = apply { channel.close().sync() transport.run { bossGroup.shutdownGracefully(0, 500, TimeUnit.MILLISECONDS).sync() workerGroup.shutdownGracefully(0, 500, TimeUnit.MILLISECONDS).sync() executor.shutdownGracefully(0, 500, TimeUnit.MILLISECONDS).sync() } } override fun port(): Int = (channel.localAddress() as InetSocketAddress).port } companion object { private val LOGGER = LoggerFactory.getLogger(Netty::class.java) } } fun getX509Certs(certificates: String): Collection { val targetStream: InputStream = ByteArrayInputStream(certificates.toByteArray()) @Suppress("unchecked_cast") return CertificateFactory.getInstance("X509").generateCertificates(targetStream) as Collection } fun getPrivateKey(privateKey: String): PrivateKey { return loadKey(privateKey)!! }