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.git
The following commit(s) were added to refs/heads/main by this push:
new 14dc8a676c fix: close IO resources left open on abrupt or failing
paths (#3532)
14dc8a676c is described below
commit 14dc8a676ccf127fc3aea5099ab3a0c258c9a815
Author: PJ Fanning <[email protected]>
AuthorDate: Tue Sep 8 09:43:25 2026 +0100
fix: close IO resources left open on abrupt or failing paths (#3532)
Motivation:
Three places under src/main open an IO resource and leave it open on at
least one reachable path:
- ResolvConfParser.parseFile never closes the Stream returned by
Files.lines, so the fd on /etc/resolv.conf stays open until GC. This is
the only use of Files.lines in the tree.
- InputStreamSource.postStop fails the materialized value without closing
the user-supplied InputStream. Every handler path closes it, but abrupt
termination (materializer or actor system shutdown) skips the handlers,
so the stream leaks. The sibling stages FileSource and
OutputStreamGraphStage already close in postStop.
- ArteryAeronUdpTransport.autoSelectPort leaks the DatagramChannel when
bind throws.
Modification:
Close the Files.lines stream in a finally; parseLines consumes the
iterator eagerly, so an eager close is safe.
Close the input stream in InputStreamSource.postStop before failing the
promise. The close is inlined rather than delegated to closeInputStream,
because that helper routes failures through failStage, which is pointless
on an already torn-down stage. A close failure is logged at debug and the
promise still fails with AbruptStageTerminationException.
Wrap the bind in autoSelectPort in try/finally.
Result:
None of the three paths leaks a file descriptor.
Tests:
- sbt "stream-tests/testOnly
org.apache.pekko.stream.io.InputStreamSourceSpec" - 9 succeeded, 0 failed. The
new case
"close the input stream on actor materializer shutdown" fails without the
InputStreamSource change (8 succeeded,
1 failed) and passes with it.
- sbt "actor/compile" "stream/compile" "remote/compile" - success
- sbt "actor/mimaReportBinaryIssues" "stream/mimaReportBinaryIssues"
"remote/mimaReportBinaryIssues" - success
- Native scalafmt run on the four changed files.
- No test for the ResolvConfParser and autoSelectPort fixes: neither
changes observable behaviour, and asserting an
fd was released is not possible portably from Scala.
References:
None - found by a resource-leak audit of src/main
---
.../pekko/io/dns/internal/ResolvConfParser.scala | 5 ++++-
.../artery/aeron/ArteryAeronUdpTransport.scala | 8 ++++----
.../pekko/stream/io/InputStreamSourceSpec.scala | 20 +++++++++++++++++++-
.../pekko/stream/impl/io/InputStreamSource.scala | 7 +++++++
4 files changed, 34 insertions(+), 6 deletions(-)
diff --git
a/actor/src/main/scala/org/apache/pekko/io/dns/internal/ResolvConfParser.scala
b/actor/src/main/scala/org/apache/pekko/io/dns/internal/ResolvConfParser.scala
index 7678af8537..07f07b7521 100644
---
a/actor/src/main/scala/org/apache/pekko/io/dns/internal/ResolvConfParser.scala
+++
b/actor/src/main/scala/org/apache/pekko/io/dns/internal/ResolvConfParser.scala
@@ -33,7 +33,10 @@ private[dns] object ResolvConfParser {
*/
def parseFile(file: File): Try[ResolvConf] = {
Try {
- parseLines(Files.lines(file.toPath).iterator().asScala)
+ // the stream holds the file open until closed, and parseLines consumes
it eagerly
+ val lines = Files.lines(file.toPath)
+ try parseLines(lines.iterator().asScala)
+ finally lines.close()
}
}
diff --git
a/remote/src/main/scala/org/apache/pekko/remote/artery/aeron/ArteryAeronUdpTransport.scala
b/remote/src/main/scala/org/apache/pekko/remote/artery/aeron/ArteryAeronUdpTransport.scala
index 93160b85ba..d8be809b91 100644
---
a/remote/src/main/scala/org/apache/pekko/remote/artery/aeron/ArteryAeronUdpTransport.scala
+++
b/remote/src/main/scala/org/apache/pekko/remote/artery/aeron/ArteryAeronUdpTransport.scala
@@ -481,9 +481,9 @@ private[remote] class ArteryAeronUdpTransport(_system:
ExtendedActorSystem, _pro
import java.nio.channels.DatagramChannel
val socket = DatagramChannel.open().socket()
- socket.bind(new InetSocketAddress(hostname, 0))
- val port = socket.getLocalPort
- socket.close()
- port
+ try {
+ socket.bind(new InetSocketAddress(hostname, 0))
+ socket.getLocalPort
+ } finally socket.close()
}
}
diff --git
a/stream-tests/src/test/scala/org/apache/pekko/stream/io/InputStreamSourceSpec.scala
b/stream-tests/src/test/scala/org/apache/pekko/stream/io/InputStreamSourceSpec.scala
index fa6de2b7ed..b2fa0c9fca 100644
---
a/stream-tests/src/test/scala/org/apache/pekko/stream/io/InputStreamSourceSpec.scala
+++
b/stream-tests/src/test/scala/org/apache/pekko/stream/io/InputStreamSourceSpec.scala
@@ -14,7 +14,7 @@
package org.apache.pekko.stream.io
import java.io.{ ByteArrayInputStream, InputStream }
-import java.util.concurrent.CountDownLatch
+import java.util.concurrent.{ CountDownLatch, TimeUnit }
import scala.annotation.nowarn
import scala.util.Success
@@ -123,6 +123,24 @@ class InputStreamSourceSpec extends
StreamSpec(UnboundedMailboxConfig) {
f.failed.futureValue shouldBe an[AbruptStageTerminationException]
}
+ "close the input stream on actor materializer shutdown" in {
+ val mat = ActorMaterializer()
+ val closed = new CountDownLatch(1)
+ val source = StreamConverters.fromInputStream(() =>
+ new InputStream {
+ override def read(): Int = -1
+ override def close(): Unit = closed.countDown()
+ })
+ val pubSink = Sink.asPublisher[ByteString](false)
+ val (f, neverPub) = source.toMat(pubSink)(Keep.both).run()(mat)
+ val c = TestSubscriber.manualProbe[ByteString]()
+ neverPub.subscribe(c)
+ c.expectSubscription()
+ mat.shutdown()
+ f.failed.futureValue shouldBe an[AbruptStageTerminationException]
+ closed.await(3, TimeUnit.SECONDS) shouldBe true
+ }
+
"emit as soon as read" in {
val latch = new CountDownLatch(1)
val probe = StreamConverters
diff --git
a/stream/src/main/scala/org/apache/pekko/stream/impl/io/InputStreamSource.scala
b/stream/src/main/scala/org/apache/pekko/stream/impl/io/InputStreamSource.scala
index 5a483700ad..cc43be35d7 100644
---
a/stream/src/main/scala/org/apache/pekko/stream/impl/io/InputStreamSource.scala
+++
b/stream/src/main/scala/org/apache/pekko/stream/impl/io/InputStreamSource.scala
@@ -101,6 +101,13 @@ private[pekko] final class InputStreamSource(factory: ()
=> InputStream, chunkSi
override def postStop(): Unit = {
if (!isClosed) {
+ // abrupt termination skips the handlers, so this is the last chance
to release the stream.
+ // The stage is already torn down here, so a close failure is not
routed through failStage.
+ try {
+ if (inputStream ne null) inputStream.close()
+ } catch {
+ case NonFatal(ex) => log.debug("Failed to close input stream on
abrupt termination: {}", ex.getMessage)
+ }
mat.tryFailure(new AbruptStageTerminationException(this))
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
