This is an automated email from the ASF dual-hosted git repository.
pjfanning pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko-connectors.git
The following commit(s) were added to refs/heads/main by this push:
new fd74c9506 Unix Domain Socket: fix large-transfer stall from lost
readiness events (#1880)
fd74c9506 is described below
commit fd74c9506992a4dc511fe4dd33c88a806b1e6ec2
Author: PJ Fanning <[email protected]>
AuthorDate: Tue Sep 22 10:58:19 2026 +0100
Unix Domain Socket: fix large-transfer stall from lost readiness events
(#1880)
Motivation:
UnixDomainSocketSpec's "send and receive more ten times the size of a
buffer" test (disabled since 2019) stalls part way through the transfer:
the echoing server never writes a byte, its receive queue then fills,
and both directions deadlock.
jnr-enxio's KQSelector registers kqueue filters edge-triggered
(EV_CLEAR) and its poll() overwrites a key's readyOps for each kevent
entry instead of OR-ing them. When EVFILT_READ and EVFILT_WRITE for the
same channel arrive in one batch, the entry processed last clobbers the
other's readiness, and the consumed edge never fires again. The event
loop gated writes on key.isWritable and reads on key.isReadable, so a
clobbered event stalled that direction forever.
Modification:
Attempt a pending write or read in the NIO event loop whenever the key
is selected, rather than only when the (unreliable) ready ops report it.
A non-blocking operation on a not-ready socket returns 0, which is now
handled: an unfinished write keeps OP_WRITE registered and a 0-byte read
is a no-op instead of being mistaken for EOF. This is self-healing: a
readiness edge can only be lost to another event on the same key, and
that event now triggers the retry. Re-enable the test.
Result:
Sustained transfers complete; the throughput test passes in ~250ms
instead of timing out after a minute.
Tests:
- sbt "unix-domain-socket/testOnly docs.scaladsl.UnixDomainSocketSpec"
(all 7 pass, stable over 5 consecutive runs; the re-enabled test
reproduced the stall before the fix)
References:
Fixes #1858
---
.../impl/UnixDomainSocketImpl.scala | 29 ++++++++++++++++------
.../scala/docs/scaladsl/UnixDomainSocketSpec.scala | 2 +-
2 files changed, 23 insertions(+), 8 deletions(-)
diff --git
a/unix-domain-socket/src/main/scala/org/apache/pekko/stream/connectors/unixdomainsocket/impl/UnixDomainSocketImpl.scala
b/unix-domain-socket/src/main/scala/org/apache/pekko/stream/connectors/unixdomainsocket/impl/UnixDomainSocketImpl.scala
index 3612ed6ee..789eaacde 100644
---
a/unix-domain-socket/src/main/scala/org/apache/pekko/stream/connectors/unixdomainsocket/impl/UnixDomainSocketImpl.scala
+++
b/unix-domain-socket/src/main/scala/org/apache/pekko/stream/connectors/unixdomainsocket/impl/UnixDomainSocketImpl.scala
@@ -200,7 +200,12 @@ private[unixdomainsocket] object UnixDomainSocketImpl {
case null =>
case sendReceiveContext: SendReceiveContext =>
sendReceiveContext.send match {
- case SendRequested(buffer, sent) if keySelectable &&
key.isWritable =>
+ // Attempt the write whenever the key is selected, not only
when it reports
+ // writable. The underlying jnr-enxio selectors register
edge-triggered kqueue
+ // filters and can lose a write-readiness event when a read
event for the same
+ // channel arrives in the same batch - a non-blocking write
simply returns 0
+ // when the socket has no space, so trying is always safe.
+ case SendRequested(buffer, sent) if keySelectable =>
val channel = key.channel().asInstanceOf[UnixSocketChannel]
val written =
@@ -220,10 +225,14 @@ private[unixdomainsocket] object UnixDomainSocketImpl {
log.debug("written: {} remaining: {}", written, remaining)
- if (written >= 0 && remaining == 0) {
- sendReceiveContext.send = SendAvailable(buffer)
- key.interestOps(key.interestOps() &
~SelectionKey.OP_WRITE)
- sent.success(Done)
+ if (written >= 0) {
+ if (remaining == 0) {
+ sendReceiveContext.send = SendAvailable(buffer)
+ key.interestOps(key.interestOps() &
~SelectionKey.OP_WRITE)
+ sent.success(Done)
+ } else {
+ key.interestOps(key.interestOps() |
SelectionKey.OP_WRITE)
+ }
}
case _: SendRequested =>
key.interestOps(key.interestOps() | SelectionKey.OP_WRITE)
@@ -253,7 +262,11 @@ private[unixdomainsocket] object UnixDomainSocketImpl {
} catch { case _: IOException => }
}
sendReceiveContext.receive match {
- case ReceiveAvailable(queue, buffer) if keySelectable &&
key.isReadable =>
+ // As for writes above, attempt the read whenever the key is
selected so that a
+ // read-readiness event lost to the edge-triggered selector
is recovered by the
+ // event that clobbered it. A read of 0 bytes means "no data
yet" and leaves the
+ // registered read interest as it is.
+ case ReceiveAvailable(queue, buffer) if keySelectable =>
buffer.clear()
val channel = key.channel.asInstanceOf[UnixSocketChannel]
@@ -268,12 +281,14 @@ private[unixdomainsocket] object UnixDomainSocketImpl {
log.debug("read: {}", read)
- if (read >= 0) {
+ if (read > 0) {
buffer.flip()
val pendingResult = queue.offer(ByteString(buffer))
pendingResult.onComplete(_ => sel.wakeup())
sendReceiveContext.receive = PendingReceiveAck(queue,
buffer, pendingResult)
key.interestOps(key.interestOps() &
~SelectionKey.OP_READ)
+ } else if (read == 0) {
+ // No data available - the key was selected for another
reason
} else {
queue.complete()
try {
diff --git
a/unix-domain-socket/src/test/scala/docs/scaladsl/UnixDomainSocketSpec.scala
b/unix-domain-socket/src/test/scala/docs/scaladsl/UnixDomainSocketSpec.scala
index 5435afd3c..8d062cde5 100644
--- a/unix-domain-socket/src/test/scala/docs/scaladsl/UnixDomainSocketSpec.scala
+++ b/unix-domain-socket/src/test/scala/docs/scaladsl/UnixDomainSocketSpec.scala
@@ -84,7 +84,7 @@ class UnixDomainSocketSpec
binding.futureValue.unbind().futureValue should be(())
}
- "send and receive more ten times the size of a buffer" ignore {
+ "send and receive more ten times the size of a buffer" in {
val BufferSizeBytes = 64 * 1024
val path = dir.resolve("sock2")
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
