He-Pin commented on code in PR #3728:
URL: https://github.com/apache/pekko/pull/3728#discussion_r4408560062
##########
stream/src/main/scala/org/apache/pekko/stream/impl/io/TlsGraphStage.scala:
##########
@@ -473,16 +473,43 @@ import pekko.util.ByteString
case BUFFER_UNDERFLOW =>
flushToUser()
+ ensureInboundBufferSizes()
Review Comment:
`BUFFER_UNDERFLOW` is the ordinary "the record is not complete yet" path, so
this now runs on every partial TLS record - `engine.getSession()` plus two
accessors plus two capacity comparisons - and the same again in the
`BUFFER_OVERFLOW` branch at line 480.
Measured on JDK 25, `getPacketBufferSize()`/`getApplicationBufferSize()`
return `16709`/`16704` both before and after `beginHandshake()`, for TLSv1.2
and TLSv1.3 alike, so with the default provider this call is a guaranteed no-op
and pure overhead on a hot path.
`SSLEngineImpl.getSession()` is not `synchronized` (checked the bytecode),
so this is not a lock cost, but it is still worth guarding. A cheap
precondition that keeps SunJSSE on the fast path: on underflow the unwrap
consumed nothing, so growth can only be needed when the input buffer is
completely full -
```scala
case BUFFER_UNDERFLOW =>
flushToUser()
if (transportInBuffer.remaining() == transportInBuffer.capacity())
ensureInboundBufferSizes()
```
(For the `BUFFER_OVERFLOW` branch the equivalent signal is `userOutBuffer`
being full, which has to be sampled before `flushToUser()` clears it.)
##########
stream/src/main/scala/org/apache/pekko/stream/impl/io/TlsGraphStage.scala:
##########
@@ -473,16 +473,43 @@ import pekko.util.ByteString
case BUFFER_UNDERFLOW =>
flushToUser()
+ ensureInboundBufferSizes()
case BUFFER_OVERFLOW =>
flushToUser()
+ ensureInboundBufferSizes()
transportInput.putBackUnreadBuffer(transportInBuffer)
case null =>
failTls(new IllegalStateException("unexpected status null in
doUnwrap()"))
}
}
+ private def ensureInboundBufferSizes(): Unit = {
+ // Session requirements can increase while an incoming record is being
inspected.
+ val session = engine.getSession
+ val packetSize = math.max(1, session.getPacketBufferSize)
+ transportInBuffer = ensureReadBufferCapacity(transportInBuffer,
packetSize)
+ transportUnreadBuffer =
ensureReadBufferCapacity(transportUnreadBuffer, packetSize)
+ transportPacketSize = math.max(transportPacketSize, packetSize)
Review Comment:
`transportPacketSize` is raised here but `transportOutBuffer` is never grown
to match - it only grows reactively on `BUFFER_OVERFLOW` in `doWrap`.
Two outbound decisions key off `transportPacketSize`:
- `canContinueWrapping` requires `transportOutBuffer.remaining >=
transportPacketSize` (line 427);
- `shouldFlushTransportNow` requires `transportOutBuffer.position() >=
transportPacketSize` (line 562).
Once `transportPacketSize` exceeds `transportOutBuffer.capacity()`, both are
permanently false, so wrap batching and size-triggered flushing are silently
disabled for the lifetime of that connection and it degrades to one record per
`ByteString`. That is stronger than what the new `stream-io.md` paragraph
promises ("can reduce batching, which is bounded by the actual transport buffer
capacity").
Consider growing `transportOutBuffer` here as well - flushing first if it
currently holds deferred ciphertext - or capping `transportPacketSize` at
`transportOutBuffer.capacity()` so the two stay in step.
##########
stream/src/main/scala/org/apache/pekko/stream/impl/io/TlsGraphStage.scala:
##########
@@ -679,18 +709,19 @@ 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 =
Review Comment:
The inner `else` looks unreachable with the shipped pool:
`DirectByteBufferPool.takeBufferFromPool` only ever returns buffers of
`defaultBufferSize` (`DirectByteBufferPool.scala:51-66`), and
`pooledBufferSize` is `tcp.Settings.DirectBufferSize` (`Tcp.scala:667`), so the
outer `requiredCapacity <= pooledBufferSize` guard already implies
`buffer.capacity() >= requiredCapacity`.
Fine to keep as defence - just noting it cannot be covered by a test unless
someone swaps in a different `BufferPool`.
The important part of this change is the companion guard in
`releaseTransportBuffer` (line 722), and it deserves more credit than it
currently gets: `DirectByteBufferPool.release` performs **no** validation
(`DirectByteBufferPool.scala:68-82`) - it stores whatever it is handed and
`acquire()` returns it after a `clear()`. Without the `isDirect && capacity()
== pooledBufferSize` check, every heap fallback buffer allocated by this method
would be released into the shared TCP pool and then handed out to unrelated TCP
connections for the lifetime of the ActorSystem. Worth stating that explicitly
in the PR description, because as written it reads like a cosmetic tidy-up.
##########
stream/src/main/scala/org/apache/pekko/stream/impl/io/TlsGraphStage.scala:
##########
@@ -473,16 +473,43 @@ import pekko.util.ByteString
case BUFFER_UNDERFLOW =>
flushToUser()
+ ensureInboundBufferSizes()
case BUFFER_OVERFLOW =>
flushToUser()
+ ensureInboundBufferSizes()
transportInput.putBackUnreadBuffer(transportInBuffer)
case null =>
failTls(new IllegalStateException("unexpected status null in
doUnwrap()"))
}
}
+ private def ensureInboundBufferSizes(): Unit = {
+ // Session requirements can increase while an incoming record is being
inspected.
+ val session = engine.getSession
+ val packetSize = math.max(1, session.getPacketBufferSize)
+ transportInBuffer = ensureReadBufferCapacity(transportInBuffer,
packetSize)
+ transportUnreadBuffer =
ensureReadBufferCapacity(transportUnreadBuffer, packetSize)
+ transportPacketSize = math.max(transportPacketSize, packetSize)
+
+ val applicationSize = math.max(1, session.getApplicationBufferSize)
+ applicationRecordSize = math.max(applicationRecordSize,
applicationSize)
+ maxUserInBufferSize = math.max(maxUserInBufferSize, applicationSize)
Review Comment:
These use `math.max(current, applicationSize)` - a single record - whereas
`allocateBuffers` (lines 141-143) and `growTransportOutBuffer` (lines 616-618)
derive the same two fields from `applicationBufferSize(applicationSize,
packetSize)`, i.e. `applicationSize * recordsPerEngineCall`.
It is harmless today because this is a `max` and the batch value is normally
larger, but having two formulas for one quantity is a trap for the next reader:
if a session's application size ever grew past the batch value,
`maxUserInBufferSize` would silently stop being a batch size and become a
single-record cap. Consider deriving both from `applicationBufferSize` here too.
##########
stream-tests/src/test/scala/org/apache/pekko/stream/io/TlsGraphStageIsolatedSpec.scala:
##########
@@ -300,13 +372,31 @@ class TlsGraphStageDeferredCloseSpec extends
StreamSpec(TlsGraphStageDeferredClo
dsts: Array[ByteBuffer],
offset: Int,
length: Int): SSLEngineResult = {
+ if (unwrapCount > 0 && negotiatedPacketSize > 0) {
+ reportedPacketSize = negotiatedPacketSize
+ if (negotiatedApplicationSize > 0) reportedApplicationSize =
negotiatedApplicationSize
+ bufferRetries += 1
+ if (bufferRetries > 8) throw new SSLException("TLS buffers were not
resized after repeated retries")
Review Comment:
`bufferRetries` is incremented on every post-handshake scatter `unwrap` and
is never reset, so it really counts "unwraps since the handshake", not retries.
A future case that legitimately needs more than 8 unwrap rounds will fail with
the misleading `"TLS buffers were not resized after repeated retries"`.
Suggest renaming to something like `postHandshakeUnwraps`, and/or resetting
it once an unwrap returns neither `BUFFER_OVERFLOW` nor `BUFFER_UNDERFLOW`.
That keeps the intent (fail fast if the stage never resizes) without coupling
the guard to the total round count.
##########
stream-tests/src/test/scala/org/apache/pekko/stream/io/TlsGraphStageIsolatedSpec.scala:
##########
@@ -450,6 +629,9 @@ object TlsGraphStageDeferredCloseSpec {
val config: Config =
ConfigFactory
.parseString(
- s"""pekko.scheduler.implementation =
"${classOf[DeferredZeroTimerScheduler].getName}"""")
+ s"""pekko.scheduler.implementation =
"${classOf[DeferredZeroTimerScheduler].getName}"
+ |pekko.io.tcp.direct-buffer-size = 18432
Review Comment:
This changes the environment of the three pre-existing tests in this spec,
not just the new ones.
`applicationBufferSize` divides the transport buffer capacity by
`packetSize`, so with the default `direct-buffer-size = 128 KiB` the
pre-existing tests ran with `131072 / 16709 = 7` -> `min(4, 7) = 4` -> `16704 *
4 = 66816` for `maxUserInBufferSize`/`maxUserOutBufferSize`. At `18432` that
becomes `18432 / 16709 = 1` -> `16704`, a 4x reduction in the batching limit
for tests that have nothing to do with buffer sizing.
I can see why the value is needed: the two "grow inbound buffers after
session sizes increase" cases require the initial buffer to be smaller than
`negotiatedPacketSize = 32768`, otherwise `ensureReadBufferCapacity` is a no-op
and `underflowUnwraps`/`overflowUnwraps` stay at 0. But
`TlsGraphStageSmallBufferSpec` (8192) already exists for exactly that purpose.
Suggest moving those two cases there - or into a third dedicated spec - and
leaving this config with only `direct-buffer-pool-limit = 8`, which is the part
the pool-uniqueness assertion genuinely needs (at the default limit of 1000 it
would acquire 128 MiB of direct memory).
If you would rather keep it here, please add a comment explaining the
coupling; right now a future reader tuning this value has no way to know it
changes batching semantics for unrelated tests.
##########
stream-tests/src/test/scala/org/apache/pekko/stream/io/TlsGraphStageIsolatedSpec.scala:
##########
@@ -41,9 +39,12 @@ import pekko.stream.scaladsl._
import pekko.stream.scaladsl.GraphDSL.Implicits._
import pekko.stream.testkit.{ StreamSpec, TestPublisher, TestSubscriber }
import pekko.stream.testkit.scaladsl.{ TestSink, TestSource }
+import pekko.stream.testkit.scaladsl.StreamTestKit.assertAllStagesStopped
import pekko.testkit.{ ExplicitlyTriggeredScheduler, WithLogCapturing }
import pekko.util.ByteString
+import com.typesafe.config.{ Config, ConfigFactory }
Review Comment:
Unrelated churn, although it does move the file onto the repo convention -
`TlsSpec.scala`, `TcpSpec.scala` and `TlsGraphStageEdgeCasesSpec.scala` all put
`com.typesafe.config` after the pekko group, and this file was the outlier.
Fine either way; flagging it only against the AGENTS.md "do not mix formatting
churn" rule in case a maintainer would rather it stayed out of this PR.
##########
docs/src/main/paradox/stream/stream-io.md:
##########
@@ -161,6 +161,14 @@ 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
Review Comment:
#3727 inserts a different paragraph at this exact position, so whichever PR
merges second conflicts here and both paragraphs need to be kept.
On the content: it is accurate, and this is the only place in either PR that
names `pekko.stream.materializer.tls.engine`, which is helpful. One nit - "can
reduce batching, which is bounded by the actual transport buffer capacity"
understates the effect: once the session's packet size exceeds the transport
buffer capacity, `canContinueWrapping` and the size-based flush in
`shouldFlushTransportNow` are both permanently false, so batching is not merely
reduced but switched off. See the inline note on `ensureInboundBufferSizes`.
--
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]
