pjfanning commented on code in PR #3728:
URL: https://github.com/apache/pekko/pull/3728#discussion_r4406882497
##########
stream/src/main/scala/org/apache/pekko/stream/impl/io/TlsGraphStage.scala:
##########
@@ -679,18 +679,13 @@ import pekko.util.ByteString
userInBuffer = null
}
- private def acquireTransportBuffer(requiredCapacity: Int): ByteBuffer = {
- val buffer = bufferPool.acquire()
- if (buffer.capacity >= requiredCapacity) buffer
- else {
- releaseTransportBuffer(buffer)
- throw new IllegalStateException(
- s"TLS packet buffer requires $requiredCapacity bytes but
pekko.io.tcp.direct-buffer-size is $pooledBufferSize")
- }
- }
+ private def acquireTransportBuffer(requiredCapacity: Int): ByteBuffer =
+ if (requiredCapacity <= pooledBufferSize) bufferPool.acquire()
+ else ByteBuffer.allocate(requiredCapacity)
private def releaseTransportBuffer(buffer: ByteBuffer): Unit =
- if (buffer ne null) {
+ // Only direct buffers came from the TCP pool; larger heap buffers are
stage-owned.
+ if ((buffer ne null) && buffer.isDirect) {
Review Comment:
Using `isDirect` to decide whether a buffer belongs to the pool is fragile.
If the fallback or growth path later switches to `allocateDirect` (a natural
optimisation), oversized stage-owned buffers would be released into the shared
fixed-size TCP pool, and TCP connections would get buffers of unexpected size.
Suggest tracking ownership explicitly, e.g. a flag per buffer or at least
`buffer.capacity == pooledBufferSize`.
##########
stream/src/main/scala/org/apache/pekko/stream/impl/io/TlsGraphStage.scala:
##########
@@ -679,18 +679,13 @@ import pekko.util.ByteString
userInBuffer = null
}
- private def acquireTransportBuffer(requiredCapacity: Int): ByteBuffer = {
- val buffer = bufferPool.acquire()
- if (buffer.capacity >= requiredCapacity) buffer
- else {
- releaseTransportBuffer(buffer)
- throw new IllegalStateException(
- s"TLS packet buffer requires $requiredCapacity bytes but
pekko.io.tcp.direct-buffer-size is $pooledBufferSize")
- }
- }
+ private def acquireTransportBuffer(requiredCapacity: Int): ByteBuffer =
+ if (requiredCapacity <= pooledBufferSize) bufferPool.acquire()
Review Comment:
This drops the check on the acquired buffer's real capacity. It now assumes
every buffer from `bufferPool` has exactly `pooledBufferSize` (`Tcp`
`DirectBufferSize`) capacity. If those ever diverge, an undersized buffer comes
back silently and fails later as unexplained overflow/underflow loops, where
the old code failed fast here. Suggest keeping the check: acquire, and if
`buffer.capacity < requiredCapacity`, release it and allocate on the heap
instead.
##########
docs/src/main/paradox/stream/stream-io.md:
##########
@@ -161,6 +161,11 @@ Java
The `SSLEngine` instance can then be used with the binding or outgoing
connection factory methods.
+When using `pekko.stream.materializer.tls.engine = graph-stage`, TLS packet
buffers are sized to
+the requirements of the `SSLEngine`. The `pekko.io.tcp.direct-buffer-size`
setting controls the
Review Comment:
"It does not limit the TLS record size" is accurate, but it undersells the
effect. `direct-buffer-size` still decides whether TLS uses pooled direct
buffers or heap buffers allocated per connection (3 × ~16.7 KiB), and it still
sets batching (`pooledBufferSize / packetSize`, so 1 record per engine call
below ~16.7 KiB). Worth saying that lowering it below the TLS packet size has a
performance cost.
##########
stream-tests/src/test/scala/org/apache/pekko/stream/io/TlsGraphStageIsolatedSpec.scala:
##########
@@ -247,19 +248,68 @@ class TlsGraphStageIsolatedSpec extends
StreamSpec(TlsSpec.configOverrides) with
}
}
+class TlsGraphStageSmallBufferSpec
+ extends StreamSpec(
+ ConfigFactory.parseString("pekko.io.tcp.direct-buffer-size = 8192")
+ .withFallback(ConfigFactory.parseString(TlsSpec.configOverrides))) {
+
+ "TlsGraphStage" must {
+ for (protocol <- Seq("TLSv1.2", "TLSv1.3")) {
+ s"round-trip $protocol records larger than the TCP pool buffers" in {
+ assertAllStagesStopped {
+ val context = TlsSpec.initSslContext(protocol)
+ def tls(client: Boolean) =
+ BidiFlow.fromGraph(new TlsGraphStage(
+ () => {
+ val engine = context.createSSLEngine()
+ engine.setUseClientMode(client)
+ engine.setEnabledProtocols(Array(protocol))
+ engine
+ },
+ _ => Success(()),
+
IgnoreComplete)).withAttributes(TlsGraphStage.StreamTlsAttributes)
+
+ val payload =
ByteString.fromArray(Array.tabulate[Byte](32768)(_.toByte))
+ val echo = Flow[SslTlsInbound].collect { case SessionBytes(_, bytes)
=> SendBytes(bytes) }
+ val received = Source.single[SslTlsOutbound](SendBytes(payload))
+ .via(tls(client = true).atop(tls(client =
false).reversed).join(echo))
+ .collect { case SessionBytes(_, bytes) => bytes }
+ .scan(ByteString.empty)(_ ++ _)
+ .filter(_.size >= payload.size)
+ .runWith(Sink.head)
+
+ Await.result(received, remainingOrDefault) shouldBe payload
+ }
+
+ // TLS fallback buffers must not enter the shared, fixed-size TCP pool.
+ val pool = pekko.io.Tcp(system).bufferPool
Review Comment:
This check acquires one buffer from a pool that TLS never used in this run,
because every buffer took the heap fallback. It never exercises the case where
a pooled direct buffer is released and replaced by a heap buffer in
`growTransportOutBuffer`. A regression that releases a grown heap buffer into
the pool, or releases a pooled buffer twice, would pass. Consider asserting the
pool contents after the growth test in `TlsGraphStageDeferredCloseSpec` as
well, e.g. drain the pool and check every buffer is direct with the expected
capacity and that none are duplicates.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
