He-Pin commented on code in PR #3740:
URL: https://github.com/apache/pekko/pull/3740#discussion_r4430941678
##########
stream/src/main/scala/org/apache/pekko/stream/impl/io/TLSActor.scala:
##########
@@ -287,6 +289,12 @@ import pekko.util.ByteString
}
}
+ val signallingInboundTruncation = TransferPhase(userOutDemand) { () =>
+ if (tracing) log.debug("signalling SessionTruncated after TransportOut
cancellation")
+ outputBunch.enqueue(UserOut, SessionTruncated)
+ nextPhase(completedPhase)
+ }
Review Comment:
This does not apply to the actor engine. The guard is present, it is just
expressed as the phase precondition instead of an `if`.
`OutputBunch` keeps no per-output element buffer, so there is nothing to
overwrite and no enqueue that can fail:
- `OutputBunch.enqueue` (FanOut.scala) calls
`outputs(id).enqueueOutputElement(elem)` and then refreshes `pending(id)` from
`output.demandAvailable`;
- `SimpleOutputs.enqueueOutputElement` (ActorProcessor.scala) is
`downstreamDemand -= 1; tryOnNext(subscriber, elem)` — it emits immediately
rather than parking an element;
- `OutputBunch.demandAvailableFor(id)` (FanOut.scala) defines `isReady =
pending(id)`, so `pending` mirrors downstream *demand*, not a buffered element.
`signallingInboundTruncation = TransferPhase(userOutDemand)` therefore only
runs while demand is available: `Pump.pump()` executes the action only while
`transferState.isExecutable` (`isReady && !isCompleted`). If `flushToUser()`
consumed the last unit of demand in the same round, `pending(UserOut)` goes
false, the pump parks, and the report is emitted after the next
`SubstreamRequestMore`. That is exactly the ordering the "buffered records
outnumber the plaintext demand" case in `TlsTransportOutputCancelSpec` pins by
granting one demand unit at a time.
`TlsGraphStage` needs the explicit `!userOutput.hasPending` check because
its `OutputSlot` really does hold one element and `enqueue` throws
`IllegalStateException("TLS output slot already has pending data")` on a second
one. The asymmetry is between the two engines' output slots, not a missing
guard here.
Worth noting in passing: `SubstreamCancel` sets `cancelled(id)` without
clearing `pending(id)`, so `userOutDemand.isReady` can stay true after the
plaintext output is cancelled — but `isCompleted` becomes true at the same
time, so `isExecutable` is false and nothing is emitted. That is why the gate
is on the whole `TransferState` rather than on demand alone.
##########
stream/src/main/scala/org/apache/pekko/stream/impl/io/TLSActor.scala:
##########
@@ -352,6 +361,21 @@ import pekko.util.ByteString
}
}
+ /**
+ * Closing the inbound side is what reveals whether the peer's close_notify
is still missing, so
+ * the plaintext output can learn that inbound records buffered from here on
are unreachable. The
+ * read side is cancelled first because delivering that report may have to
wait for plaintext
+ * demand, and a subscribed transport input would overrun its buffer while
the pump is parked.
+ */
+ private def inboundTruncationOwed(): Boolean =
+ if (!corkUser && !engine.isInboundDone && !userOutDemand.isCompleted) {
+ inputBunch.cancel()
+ try {
+ engine.closeInbound()
+ false
+ } catch { case _: SSLException => true }
+ } else false
Review Comment:
Duplicate of the thread on line 296 — same answer: `OutputBunch` has no
element buffer (`enqueueOutputElement` is `downstreamDemand -= 1;
tryOnNext(...)`), `pending(id)` mirrors downstream demand, and
`TransferPhase(userOutDemand)` is the guard, since `Pump.pump()` only runs the
action while `isReady && !isCompleted`. The `!userOutput.hasPending` check in
`TlsGraphStage` exists because that engine's `OutputSlot` buffers one element
and throws on a second enqueue; the actor engine has no such slot.
--
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]
