Skip to content

Commit 188ff2e

Browse files
kyri-petroukyri-petrou
authored
Optimize netty integration and default config (#3114)
1 parent b900d12 commit 188ff2e

File tree

12 files changed

+223
-66
lines changed

12 files changed

+223
-66
lines changed

‎.devcontainer/Dockerfile

+13-5
Original file line numberDiff line numberDiff line change
@@ -1,21 +1,20 @@
11
# See here for image contents: https://github.com/microsoft/vscode-dev-containers/tree/v0.238.1/containers/java/.devcontainer/base.Dockerfile
22

33
# [Choice] Java version (use -bullseye variants on local arm64/Apple Silicon): 11, 17, 11-bullseye, 17-bullseye, 11-buster, 17-buster
4-
ARG VARIANT="11"
5-
FROM mcr.microsoft.com/vscode/devcontainers/java:0-${VARIANT}
4+
ARG VARIANT="17"
5+
FROM mcr.microsoft.com/vscode/devcontainers/java:${VARIANT}
66

77

88
RUN curl -s "https://get.sdkman.io" | bash
99

1010
# Install Scala Lang
11-
ARG SBT_VERSION="1.7.1"
11+
ARG SBT_VERSION="1.10.1"
1212
RUN \
1313
curl -L "https://github.com/sbt/sbt/releases/download/v$SBT_VERSION/sbt-$SBT_VERSION.tgz" | tar zxf - -C /usr/share && \
1414
cd /usr/share/sbt/bin && \
15-
rm sbt.bat sbtn-x86_64-apple-darwin sbtn-x86_64-pc-linux sbtn-x86_64-pc-win32.exe && \
1615
ln -s /usr/share/sbt/bin/sbt /usr/local/bin/sbt
1716

18-
ARG SCALA_VERSION="3.1.3"
17+
ARG SCALA_VERSION="3.3.3"
1918
RUN \
2019
mkdir /setup-project && \
2120
cd /setup-project && \
@@ -24,4 +23,13 @@ RUN \
2423
sbt compile && \
2524
rm -rf /setup-project
2625

26+
RUN \
27+
mkdir /setup-wrk && \
28+
sudo apt-get update -y && sudo apt-get install build-essential libssl-dev git -y && \
29+
git clone https://github.com/wg/wrk.git wrk && \
30+
cd wrk && \
31+
make && \
32+
cp wrk /usr/local/bin && \
33+
rm -rf /setup-wrk
34+
2735
CMD ["sbt"]

‎.devcontainer/devcontainer.json

+2-2
Original file line numberDiff line numberDiff line change
@@ -8,8 +8,8 @@
88
// Update the VARIANT arg to pick a Java version: 11, 17
99
// Append -bullseye or -buster to pin to an OS version.
1010
// Use the -bullseye variants on local arm64/Apple Silicon.
11-
"VARIANT": "11",
12-
"SCALA_VERSION": "3.1.3"
11+
"VARIANT": "17",
12+
"SCALA_VERSION": "3.3.3"
1313
}
1414
},
1515

‎build.sbt

+3-1
Original file line numberDiff line numberDiff line change
@@ -281,6 +281,8 @@ lazy val zioHttpExample = (project in file("zio-http-example"))
281281
.settings(runSettings(Debug.Main))
282282
.settings(libraryDependencies ++= Seq(`jwt-core`, `zio-schema-json`))
283283
.settings(
284+
run / fork := true,
285+
run / javaOptions ++= Seq("-Xms4G", "-Xmx4G", "-XX:+UseG1GC"),
284286
libraryDependencies ++= Seq(
285287
`zio-config`,
286288
`zio-config-magnolia`,
@@ -404,7 +406,7 @@ lazy val docs = project
404406
testFrameworks += new TestFramework("zio.test.sbt.ZTestFramework"),
405407
libraryDependencies ++= Seq(
406408
`jwt-core`,
407-
"dev.zio" %% "zio-test" % ZioVersion,
409+
"dev.zio" %% "zio-test" % ZioVersion,
408410
`zio-config`,
409411
`zio-config-magnolia`,
410412
`zio-config-typesafe`,

‎project/Dependencies.scala

+9-9
Original file line numberDiff line numberDiff line change
@@ -21,15 +21,15 @@ object Dependencies {
2121

2222
val netty =
2323
Seq(
24-
"io.netty" % "netty-codec-http" % NettyVersion,
25-
"io.netty" % "netty-handler-proxy" % NettyVersion,
26-
"io.netty" % "netty-transport-native-epoll" % NettyVersion,
27-
"io.netty" % "netty-transport-native-epoll" % NettyVersion % Runtime classifier "linux-x86_64",
28-
"io.netty" % "netty-transport-native-epoll" % NettyVersion % Runtime classifier "linux-aarch_64",
29-
"io.netty" % "netty-transport-native-kqueue" % NettyVersion,
30-
"io.netty" % "netty-transport-native-kqueue" % NettyVersion % Runtime classifier "osx-x86_64",
31-
"io.netty" % "netty-transport-native-kqueue" % NettyVersion % Runtime classifier "osx-aarch_64",
32-
"com.aayushatharva.brotli4j" % "brotli4j" % "1.16.0" % "provided",
24+
"io.netty" % "netty-codec-http" % NettyVersion,
25+
"io.netty" % "netty-handler-proxy" % NettyVersion,
26+
"io.netty" % "netty-transport-native-epoll" % NettyVersion,
27+
"io.netty" % "netty-transport-native-epoll" % NettyVersion classifier "linux-x86_64",
28+
"io.netty" % "netty-transport-native-epoll" % NettyVersion classifier "linux-aarch_64",
29+
"io.netty" % "netty-transport-native-kqueue" % NettyVersion,
30+
"io.netty" % "netty-transport-native-kqueue" % NettyVersion classifier "osx-x86_64",
31+
"io.netty" % "netty-transport-native-kqueue" % NettyVersion classifier "osx-aarch_64",
32+
"com.aayushatharva.brotli4j" % "brotli4j" % "1.16.0" % "provided",
3333
)
3434

3535
val `netty-incubator` =

‎zio-http-example/src/main/scala/example/PlainTextBenchmarkServer.scala

-2
Original file line numberDiff line numberDiff line change
@@ -36,11 +36,9 @@ object PlainTextBenchmarkServer extends ZIOAppDefault {
3636

3737
private val config = Server.Config.default
3838
.port(8080)
39-
.enableRequestStreaming
4039

4140
private val nettyConfig = NettyConfig.default
4241
.leakDetection(LeakDetectionLevel.DISABLED)
43-
.maxThreads(8)
4442

4543
private val configLayer = ZLayer.succeed(config)
4644
private val nettyConfigLayer = ZLayer.succeed(nettyConfig)

‎zio-http-example/src/main/scala/example/SimpleEffectBenchmarkServer.scala

-2
Original file line numberDiff line numberDiff line change
@@ -33,11 +33,9 @@ object SimpleEffectBenchmarkServer extends ZIOAppDefault {
3333

3434
private val config = Server.Config.default
3535
.port(8080)
36-
.enableRequestStreaming
3736

3837
private val nettyConfig = NettyConfig.default
3938
.leakDetection(LeakDetectionLevel.DISABLED)
40-
.maxThreads(8)
4139

4240
private val configLayer = ZLayer.succeed(config)
4341
private val nettyConfigLayer = ZLayer.succeed(nettyConfig)

‎zio-http/jvm/src/main/scala/zio/http/netty/NettyConfig.scala

+95-34
Original file line numberDiff line numberDiff line change
@@ -30,12 +30,19 @@ final case class NettyConfig(
3030
nThreads: Int,
3131
shutdownQuietPeriodDuration: Duration,
3232
shutdownTimeoutDuration: Duration,
33+
bossGroup: NettyConfig.BossGroup,
3334
) extends EventLoopGroups.Config { self =>
3435

36+
/**
37+
* Configure Netty's boss event-loop group. This only applies to server
38+
* applications and is ignored for the Client
39+
*/
40+
def bossGroup(cfg: NettyConfig.BossGroup): NettyConfig = self.copy(bossGroup = cfg)
41+
3542
def channelType(channelType: ChannelType): NettyConfig = self.copy(channelType = channelType)
3643

3744
/**
38-
* Configure the server to use the leak detection level provided.
45+
* Configure Netty to use the leak detection level provided.
3946
*
4047
* @see
4148
* <a
@@ -44,50 +51,104 @@ final case class NettyConfig(
4451
def leakDetection(level: LeakDetectionLevel): NettyConfig = self.copy(leakDetectionLevel = level)
4552

4653
/**
47-
* Configure the server to use a maximum of nThreads to process requests.
54+
* Configure Netty to use a maximum of `nThreads` for the worker event-loop
55+
* group.
4856
*/
4957
def maxThreads(nThreads: Int): NettyConfig = self.copy(nThreads = nThreads)
5058

51-
val shutdownTimeUnit: TimeUnit = TimeUnit.MILLISECONDS
52-
53-
val shutdownQuietPeriod: Long = shutdownQuietPeriodDuration.toMillis
54-
val shutdownTimeOut: Long = shutdownTimeoutDuration.toMillis
59+
def shutdownTimeUnit: TimeUnit = TimeUnit.MILLISECONDS
60+
def shutdownQuietPeriod: Long = shutdownQuietPeriodDuration.toMillis
61+
def shutdownTimeOut: Long = shutdownTimeoutDuration.toMillis
5562
}
5663

5764
object NettyConfig {
58-
def config: Config[NettyConfig] =
59-
(LeakDetectionLevel.config.nested("leak-detection-level").withDefault(NettyConfig.default.leakDetectionLevel) ++
60-
Config
61-
.string("channel-type")
62-
.mapOrFail {
63-
case "auto" => Right(ChannelType.AUTO)
64-
case "nio" => Right(ChannelType.NIO)
65-
case "epoll" => Right(ChannelType.EPOLL)
66-
case "kqueue" => Right(ChannelType.KQUEUE)
67-
case "uring" => Right(ChannelType.URING)
68-
case other => Left(Config.Error.InvalidData(message = s"Invalid channel type: $other"))
69-
}
70-
.withDefault(NettyConfig.default.channelType) ++
65+
final case class BossGroup(
66+
channelType: ChannelType,
67+
nThreads: Int,
68+
shutdownQuietPeriodDuration: Duration,
69+
shutdownTimeOutDuration: Duration,
70+
) extends EventLoopGroups.Config {
71+
def shutdownTimeUnit: TimeUnit = TimeUnit.MILLISECONDS
72+
def shutdownQuietPeriod: Long = shutdownQuietPeriodDuration.toMillis
73+
def shutdownTimeOut: Long = shutdownTimeOutDuration.toMillis
74+
}
75+
76+
private def baseConfig: Config[EventLoopGroups.Config] =
77+
(Config
78+
.string("channel-type")
79+
.mapOrFail {
80+
case "auto" => Right(ChannelType.AUTO)
81+
case "nio" => Right(ChannelType.NIO)
82+
case "epoll" => Right(ChannelType.EPOLL)
83+
case "kqueue" => Right(ChannelType.KQUEUE)
84+
case "uring" => Right(ChannelType.URING)
85+
case other => Left(Config.Error.InvalidData(message = s"Invalid channel type: $other"))
86+
}
87+
.withDefault(NettyConfig.default.channelType) ++
7188
Config.int("max-threads").withDefault(NettyConfig.default.nThreads) ++
7289
Config.duration("shutdown-quiet-period").withDefault(NettyConfig.default.shutdownQuietPeriodDuration) ++
7390
Config.duration("shutdown-timeout").withDefault(NettyConfig.default.shutdownTimeoutDuration)).map {
74-
case (leakDetectionLevel, channelType, maxThreads, quietPeriod, timeout) =>
75-
NettyConfig(leakDetectionLevel, channelType, maxThreads, quietPeriod, timeout)
91+
case (channelT, maxThreads, quietPeriod, timeout) =>
92+
new EventLoopGroups.Config {
93+
override val channelType: ChannelType = channelT
94+
override val nThreads: Int = maxThreads
95+
override val shutdownQuietPeriod: Long = quietPeriod.toMillis
96+
override val shutdownTimeOut: Long = timeout.toMillis
97+
override val shutdownTimeUnit: TimeUnit = TimeUnit.MILLISECONDS
98+
}
7699
}
77100

78-
val default: NettyConfig = NettyConfig(
79-
LeakDetectionLevel.SIMPLE,
80-
ChannelType.AUTO,
81-
0,
82-
// Defaults taken from io.netty.util.concurrent.AbstractEventExecutor
83-
Duration.fromSeconds(2),
84-
Duration.fromSeconds(15),
85-
)
86-
87-
val defaultWithFastShutdown: NettyConfig = default.copy(
88-
shutdownQuietPeriodDuration = Duration.fromMillis(50),
89-
shutdownTimeoutDuration = Duration.fromMillis(250),
90-
)
101+
def config: Config[NettyConfig] =
102+
(LeakDetectionLevel.config.nested("leak-detection-level").withDefault(NettyConfig.default.leakDetectionLevel) ++
103+
baseConfig.nested("worker-group").orElse(baseConfig) ++
104+
baseConfig.nested("boss-group")).map { case (leakDetectionLevel, worker, boss) =>
105+
def toDuration(n: Long, timeUnit: TimeUnit) = Duration.fromJava(java.time.Duration.of(n, timeUnit.toChronoUnit))
106+
NettyConfig(
107+
leakDetectionLevel,
108+
worker.channelType,
109+
worker.nThreads,
110+
shutdownQuietPeriodDuration = toDuration(worker.shutdownQuietPeriod, worker.shutdownTimeUnit),
111+
shutdownTimeoutDuration = toDuration(worker.shutdownTimeOut, worker.shutdownTimeUnit),
112+
NettyConfig.BossGroup(
113+
boss.channelType,
114+
boss.nThreads,
115+
shutdownQuietPeriodDuration = toDuration(boss.shutdownQuietPeriod, boss.shutdownTimeUnit),
116+
shutdownTimeOutDuration = toDuration(boss.shutdownTimeOut, boss.shutdownTimeUnit),
117+
),
118+
)
119+
}
120+
121+
val default: NettyConfig = {
122+
val quietPeriod = Duration.fromSeconds(2)
123+
val timeout = Duration.fromSeconds(15)
124+
NettyConfig(
125+
LeakDetectionLevel.SIMPLE,
126+
ChannelType.AUTO,
127+
java.lang.Runtime.getRuntime.availableProcessors(),
128+
// Defaults taken from io.netty.util.concurrent.AbstractEventExecutor
129+
shutdownQuietPeriodDuration = quietPeriod,
130+
shutdownTimeoutDuration = timeout,
131+
NettyConfig.BossGroup(
132+
ChannelType.AUTO,
133+
1,
134+
shutdownQuietPeriodDuration = quietPeriod,
135+
shutdownTimeOutDuration = timeout,
136+
),
137+
)
138+
}
139+
140+
val defaultWithFastShutdown: NettyConfig = {
141+
val quietPeriod = Duration.fromMillis(50)
142+
val timeout = Duration.fromMillis(250)
143+
default.copy(
144+
shutdownQuietPeriodDuration = quietPeriod,
145+
shutdownTimeoutDuration = timeout,
146+
bossGroup = default.bossGroup.copy(
147+
shutdownQuietPeriodDuration = quietPeriod,
148+
shutdownTimeOutDuration = timeout,
149+
),
150+
)
151+
}
91152

92153
sealed trait LeakDetectionLevel {
93154
self =>

‎zio-http/jvm/src/main/scala/zio/http/netty/server/NettyDriver.scala

+10-9
Original file line numberDiff line numberDiff line change
@@ -35,7 +35,7 @@ private[zio] final case class NettyDriver(
3535
channelFactory: ChannelFactory[ServerChannel],
3636
channelInitializer: ChannelInitializer[Channel],
3737
serverInboundHandler: ServerInboundHandler,
38-
eventLoopGroup: EventLoopGroup,
38+
eventLoopGroups: ServerEventLoopGroups,
3939
serverConfig: Server.Config,
4040
nettyConfig: NettyConfig,
4141
) extends Driver { self =>
@@ -44,7 +44,7 @@ private[zio] final case class NettyDriver(
4444
for {
4545
chf <- ZIO.attempt {
4646
new ServerBootstrap()
47-
.group(eventLoopGroup)
47+
.group(eventLoopGroups.boss, eventLoopGroups.worker)
4848
.channelFactory(channelFactory)
4949
.childHandler(channelInitializer)
5050
.option[Integer](ChannelOption.SO_BACKLOG, serverConfig.soBacklog)
@@ -84,7 +84,7 @@ private[zio] final case class NettyDriver(
8484
channelFactory <- ChannelFactories.Client.live.build
8585
.provideSomeEnvironment[Scope](_ ++ ZEnvironment[ChannelType.Config](nettyConfig))
8686
nettyRuntime <- NettyRuntime.live.build
87-
} yield NettyClientDriver(channelFactory.get, eventLoopGroup, nettyRuntime.get)
87+
} yield NettyClientDriver(channelFactory.get, eventLoopGroups.worker, nettyRuntime.get)
8888

8989
override def toString: String = s"NettyDriver($serverConfig)"
9090
}
@@ -97,7 +97,7 @@ object NettyDriver {
9797
RoutesRef
9898
& ChannelFactory[ServerChannel]
9999
& ChannelInitializer[Channel]
100-
& EventLoopGroup
100+
& ServerEventLoopGroups
101101
& Server.Config
102102
& NettyConfig
103103
& ServerInboundHandler,
@@ -108,7 +108,7 @@ object NettyDriver {
108108
app <- ZIO.service[RoutesRef]
109109
cf <- ZIO.service[ChannelFactory[ServerChannel]]
110110
cInit <- ZIO.service[ChannelInitializer[Channel]]
111-
elg <- ZIO.service[EventLoopGroup]
111+
elg <- ZIO.service[ServerEventLoopGroups]
112112
sc <- ZIO.service[Server.Config]
113113
nsc <- ZIO.service[NettyConfig]
114114
sih <- ZIO.service[ServerInboundHandler]
@@ -117,14 +117,15 @@ object NettyDriver {
117117
channelFactory = cf,
118118
channelInitializer = cInit,
119119
serverInboundHandler = sih,
120-
eventLoopGroup = elg,
120+
eventLoopGroups = elg,
121121
serverConfig = sc,
122122
nettyConfig = nsc,
123123
)
124124

125-
val manual: ZLayer[EventLoopGroup & ChannelFactory[ServerChannel] & Server.Config & NettyConfig, Nothing, Driver] = {
125+
val manual
126+
: ZLayer[ServerEventLoopGroups & ChannelFactory[ServerChannel] & Server.Config & NettyConfig, Nothing, Driver] = {
126127
implicit val trace: Trace = Trace.empty
127-
ZLayer.makeSome[EventLoopGroup & ChannelFactory[ServerChannel] & Server.Config & NettyConfig, Driver](
128+
ZLayer.makeSome[ServerEventLoopGroups & ChannelFactory[ServerChannel] & Server.Config & NettyConfig, Driver](
128129
ZLayer(AppRef.empty),
129130
ServerChannelInitializer.layer,
130131
ServerInboundHandler.live,
@@ -135,7 +136,7 @@ object NettyDriver {
135136
val customized: ZLayer[Server.Config & NettyConfig, Throwable, Driver] = {
136137
val serverChannelFactory: ZLayer[NettyConfig, Nothing, ChannelFactory[ServerChannel]] =
137138
ChannelFactories.Server.fromConfig
138-
val eventLoopGroup: ZLayer[NettyConfig, Nothing, EventLoopGroup] = EventLoopGroups.live
139+
val eventLoopGroup: ZLayer[NettyConfig, Nothing, ServerEventLoopGroups] = ServerEventLoopGroups.live
139140

140141
ZLayer.makeSome[Server.Config & NettyConfig, Driver](
141142
eventLoopGroup,

‎zio-http/jvm/src/main/scala/zio/http/netty/server/ServerEventLoopGroups.scala

+30
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,30 @@
1+
package zio.http.netty.server
2+
3+
import zio._
4+
import zio.stacktracer.TracingImplicits.disableAutoTrace
5+
6+
import zio.http.netty.{EventLoopGroups, NettyConfig}
7+
8+
import io.netty.channel.EventLoopGroup
9+
10+
final case class ServerEventLoopGroups(
11+
boss: EventLoopGroup,
12+
worker: EventLoopGroup,
13+
)
14+
15+
object ServerEventLoopGroups {
16+
private implicit val trace: Trace = Trace.empty
17+
18+
private def groupLayer(cfg: EventLoopGroups.Config): ULayer[EventLoopGroup] =
19+
(ZLayer.succeed(cfg) >>> EventLoopGroups.live).fresh
20+
21+
val live: ZLayer[NettyConfig, Nothing, ServerEventLoopGroups] = ZLayer.fromZIO {
22+
ZIO.serviceWith[NettyConfig] { cfg =>
23+
val boss = groupLayer(cfg.bossGroup)
24+
val worker = groupLayer(cfg)
25+
boss.zipWithPar(worker) { (boss, worker) =>
26+
ZEnvironment(ServerEventLoopGroups(boss.get, worker.get))
27+
}
28+
}
29+
}.flatten
30+
}

‎zio-http/jvm/src/main/scala/zio/http/netty/server/ServerInboundHandler.scala

+7-1
Original file line numberDiff line numberDiff line change
@@ -148,7 +148,13 @@ private[zio] final case class ServerInboundHandler(
148148
def fastEncode(response: Response, bytes: Array[Byte]) = {
149149
val jResponse = NettyResponseEncoder.fastEncode(method, response, bytes)
150150
val djResponse = jResponse.retainedDuplicate()
151-
ctx.writeAndFlush(djResponse, ctx.voidPromise())
151+
152+
// This handler sits at the tail of the pipeline, so using ctx.channel.writeAndFlush won't add any
153+
// overhead of passing through the pipeline. It's also better to use ctx.channel.writeAndFlush in
154+
// cases that we're writing to the channel from a different thread (which is most of the time as we're
155+
// creating responses in ZIO's executor).
156+
val ch = ctx.channel()
157+
ch.writeAndFlush(djResponse, ch.voidPromise())
152158
true
153159
}
154160

0 commit comments

Comments
 (0)