Ah, this is a great trick! I just tried it, and it reduces the running time to 0.8 sec:
defp filter_line(<<c::utf8, ?,::utf8, _::binary>>)
when c in [?0, ?2, ?4, ?5, ?6, ?8],
do: true
defp filter_line(<<_::utf8, ?,::utf8, _::binary>>),
do: false
defp filter_line(<<_other::utf8, rest::binary>>),
do: filter_line(rest)
I don’t see why does it have to look as close to ruby. Isn’t the point of this exercise to do the work as fast as possible?
If you want to limit the memory usage, then yes. To make it work faster, you can use a read-ahead buffer. Similarly, you could use delayed write on the writing side. Here’s a version which takes less than 2 secs on my machine (with the optimized filter_line):
def main([ filename, "stream" ]) do
File.stream!(filename, read_ahead: 100_000)
|> Stream.filter(&filter_line/1)
|> Stream.into(File.stream!(filename <> ".out", [:delayed_write]))
|> Stream.run
end
It’s not as efficient as the eager one, but much faster than the original. Perhaps it can be further optimized, but I didn’t spend any time investigating it.
Well, we have to read a lot of data, and write a lot of data, so we’re partially I/O bound in both versions. In the streaming one we’re doing I/O less efficiently (but reducing the memory usage), so that’s a trade-off.
I consider I/O bound to mean we’re doing I/O operations, i.e. talking to external devices. Message passing (unless to a different machine) is not such operation so it’s not I/O bound in my opinion. Regardless, message passing overhead might be significant if the processes are doing little work on each message.
In any case I’d say it’s first worth optimizing the sequential algorithm (perhaps by manually recursing, and making the decision with less processing 
) before considering splitting the work over multiple processes. Concurrency is not a remedy for a suboptimal sequential algorithm
