Catch only non-fatal exceptions in Source.lines #187
Conversation
Otherwise Control-C doesn't terminate the program.
|
Could you provide an example where this doesn't work? In the stream operators, we generally catch all errors as a rule, and propagate them to the channel. The idea is that only the ultimate consumer will throw, and in effect wind down the whole scope. But that idea might be wrong, of course ;) |
|
Well, to be fair, I have a slight modification of this which does a tail on a file, so it runs continuously and "forever", and in that scenario it's very obvious that (and very annoying that) Control-C doesn't work. However, I would assume that if you have a very very large file then Control-C won't actually interrupt your program until that file has been consumed (ie. the source has been read until the very end), without the change in this PR. I might have time to test that this evening (I need to find/create some very large text file and maybe put it on a slow USB stick). I just started with ox, so I might be doing something incorrectly. I did a quick search for |
|
@wjoel Yes you're right, we don't follow our own advice here ;) See e.g. - it's the same pattern, all exceptions are propagated. (Although, this should probably becase e: Exception, not Throwable)
But I still think that ctrl+c should work just fine - it should interrupt the top-level consuming thread, which should end the entire scope, interrupting all daemon forks (such as the one created by |
|
@adamw tl;dr it's fine, but maybe the I see, ok, first I verified that just a simple "OxCat" can be interrupted with Control-C, works as it should: def run(using Ox, IO): Unit =
supervised:
IO.unsafe:
println("hello")
Source.fromFile(Paths.get("/usr/share/dict/linux.words")).linesUtf8.foreach: line =>
println(line)
ExitCode.SuccessAfter thinking about it for a bit, I think the main issue was that I didn't understand the purpose of def tail(path: Path, chunkSize: Int = 1024)(using Ox, StageCapacity, IO): Source[Chunk[Byte]] =
if Files.isDirectory(path) then throw new IOException(s"Path $path is a directory")
val chunks = StageCapacity.newChannel[Chunk[Byte]]
fork {
forever {
val jFileChannel =
try FileChannel.open(path, StandardOpenOption.READ)
catch
case _: UnsupportedOperationException =>
// Some file systems don't support file channels
Files.newByteChannel(path, StandardOpenOption.READ)
jFileChannel.position(jFileChannel.size())
try {
repeatWhile {
val fileSize = useCloseable(FileChannel.open(path, StandardOpenOption.READ)): fileChannel =>
fileChannel.size()
if jFileChannel.position() > fileSize then
false
else
val buf = ByteBuffer.allocate(chunkSize)
val readBytes = jFileChannel.read(buf)
if readBytes <= 0 then
sleep(100.millis)
else if readBytes > 0 then
chunks.send(
Chunk.fromArray(
if readBytes == chunkSize then buf.array
else buf.array.take(readBytes)
)
)
true
}
} catch case NonFatal(e) => chunks.errorOrClosed(e).discard
finally
try jFileChannel.close()
catch
case NonFatal(closeException) =>
chunks.errorOrClosed(closeException).discard
}
}
chunks... and it works, in the sense that it gets terminated, when I use This modified version works with Control-C: def tail(path: Path, chunkSize: Int = 1024)(using Ox, StageCapacity, IO): Source[Chunk[Byte]] =
if Files.isDirectory(path) then throw new IOException(s"Path $path is a directory")
val chunks = StageCapacity.newChannel[Chunk[Byte]]
fork {
repeatWhile {
println("tail")
val jFileChannel =
try FileChannel.open(path, StandardOpenOption.READ)
catch
case _: UnsupportedOperationException =>
// Some file systems don't support file channels
Files.newByteChannel(path, StandardOpenOption.READ)
jFileChannel.position(jFileChannel.size())
try {
repeatWhile {
val fileSize = useCloseable(FileChannel.open(path, StandardOpenOption.READ)): fileChannel =>
fileChannel.size()
if jFileChannel.position() > fileSize then
false
else
val buf = ByteBuffer.allocate(chunkSize)
val readBytes = jFileChannel.read(buf)
if readBytes <= 0 then
sleep(100.millis)
else if readBytes > 0 then
chunks.send(
Chunk.fromArray(
if readBytes == chunkSize then buf.array
else buf.array.take(readBytes)
)
)
true
}
} catch case e => chunks.errorOrClosed(e).discard
finally
try jFileChannel.close()
catch
case NonFatal(closeException) =>
chunks.errorOrClosed(closeException).discard
!chunks.isClosedForReceive
}
}
chunksAll is well, no need for this PR. Thanks for merging the other one! That |
|
I do wonder if your |
|
This definition of |
Ah of course! There should be The streaming code is different - there, the |
The Ox It seems
It's only this, def run(args: Vector[String])(using Ox, IO): ExitCode =
supervised:
IO.unsafe:
val logLines = tail(warningsLog).myLines(StandardCharsets.UTF_8).map(_.strip)
watchAndPrintStats(logLines)
ExitCode.Success |
Ah! I'm re-reading your code now, and if you have the try-catch inside the But if an exception is thrown inside a stream operator, that's fine as well - the only thing that might happen is that the forks might end for a different reason and in a different order.
The throws-doesn't throw is exactly the difference between Now I'm thinking, maybe we should have some abstraction for a fork-and-propagate scenario, which we can reuse in the stream operators. It might be less error prone then. I'm closing this PR for now, but if something still doesn't work, please let us know :) |
|
See: #191 |
Otherwise Control-C doesn't terminate the program.