He-Pin commented on code in PR #3728:
URL: https://github.com/apache/pekko/pull/3728#discussion_r4407151679
##########
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:
Restored the acquired buffer's capacity check in be1da0e828. An undersized
buffer now falls back to heap before reaching wrap/unwrap.
##########
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:
be1da0e828 returns only direct buffers with exactly the configured pool
capacity. The growth test also checks the entire pool for wrong sizes, heap
buffers, and duplicate references.
##########
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:
Updated the documentation in be1da0e828 to explain per-connection heap
allocation and batching costs. Batching now follows actual transport-buffer
capacity, including after growth.
--
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]
