Question regarding improving fizzbuzz IO performance

Should this use ordered: true on Task.async_stream? I’d expect the result should still be printed in order and not just randomly.

Two related gotchas with IO.stream:

• it’s represented by a “file server” process, so “chunking” into bigger pieces helps by reducing the number of messages sent to that process (chunks vs individual lines)

• it can still only print one message at a time, no matter how many workers are sending messages

You could likely boost performance by building bigger iolists and sending those to IO in one go.

@LostKobrakai ordered: true is the default for Task.async_stream. From the documentation:

:ordered - whether the results should be returned in the same order as the input stream. When the output is ordered, Elixir may need to buffer results to emit them in the original order. Setting this option to false disables the need to buffer at the cost of removing ordering. This is also useful when you’re using the tasks only for the side effects. Note that regardless of what :ordered is set to, the tasks will process asynchronously. (NOTE: This is what I’m doing in the original fizzbuzz_with_io approach, that’s why the results are unordered) If you need to process elements in order, consider using Enum.map/2 or Enum.each/2 instead. Defaults to true.

If I try to print the results in order, I would have to go with fizzbuzz_no_io approach and turn my driver module into the following:

defmodule Fizzbuzz.Cli do
  def main([lower, upper]) do
    {lower, upper} = {String.to_integer(lower), String.to_integer(upper)}
    chunk_size = min(div(upper - lower, System.schedulers_online), 5000)
    input_enumerable = get_input_ranges(lower, upper, chunk_size)
    IO.inspect(input_enumerable)
    stream = Task.async_stream(input_enumerable, fn range -> Fizzbuzz.fizzbuzz_no_io(range) end, timeout: :infinity)
    Enum.reduce(stream, :ok, fn {:ok, res}, _acc -> res |> Enum.into(IO.stream) end)
  end

  def main(_), do: IO.puts("Usage: fizzbuzz 1 10000")

  defp get_input_ranges(lower, upper, chunk_size) do
    if chunk_size >= 10 do
      if lower >= upper, do: [], else: [lower..min(lower+chunk_size, upper) | get_input_ranges(min(lower+chunk_size, upper) + 1, upper, chunk_size)]
    else
      [lower..upper]
    end
  end
end

Fizzbuzz code:

def fizzbuzz_no_io(enumerable) do
  Stream.map(enumerable, &reply/1)
  |> Stream.chunk_every(6000)
  |> Enum.to_list()
end

Which brings me to my next order of business…

Doing so prints the results in order, but drops the IO speed. @al2o3cr 's answer is a great start for me, but I feel I’ll need to dive deeper into IO internals to see if there’s any scope for improvement.

I’ve been playing with chunk_size in my driver program and the chunk_every values in fizzbuzz_no_io which yield different speed results for different values. Currently, the maximum IO throughput I’ve managed so far is about 40 MiB/s on my machine, purely with hit-and-trial, using a chunk_size of 5000 and chunk_every of 6000.

One last thing that I still don’t understand is that, which seems weird if I think about it - I’m dividing the problem into chunks of size 5000 each, processing each chunk in a Task which will return me a list of size 5000. Why then does removing chunk_every line in this case drop my IO speed to 2.5-3 MiB/s, when my original input coming from Stream.map is of size 5000?

To put it simply, my remaining question is this: Why does chunking a stream of size 5000 into a chunk of size 6000 speed up the IO drastically from 2.5-3 MiB/s to 40 MiB/s?

Thank you for the insightful answers once again. Apologies if I missed something.

I am on my phone, but my blind bet is IOlist concatenation. For smaller lists these are kept as is while for larger ones these are concatenated into large binary blob that is then sent to the OS.

Thanks to some learning on this thread and some findings regarding cost of passing data between different processes, I’ve made some further improvements:

• Using GenServer instead of Task - divide the input into chunks and spin up a GenServer for every chunk, which computes the results for that chunk in parallel.
• Printing the results by sending a print call to every GenServer in order. No need to move data back to main process unlike a Task, which returns the output to the main process, which is then again passed to IO.puts. Instead we directly call IO.puts from our GenServer, which saves 1 data move and I think that’s the reason of the speedup.

Fizzbuzz.Worker module:

defmodule Fizzbuzz.Worker do
  use GenServer

  def init([range]) do
    send(self(), {:calculate, range})
    {:ok, []}
  end

  def handle_info({:calculate, range}, _state) do
    res = Fizzbuzz.fizzbuzz_no_io(range)
    {:noreply, res}
  end

  def handle_call(:print, _from, results) do
    IO.binwrite(results)
    {:reply, :ok, []}
  end
end

Main module:

defmodule Fizzbuzz.Cli do
  def main([lower, upper]) do
    {lower, upper} = {String.to_integer(lower), String.to_integer(upper)}
    chunk_size = min(div(upper - lower, System.schedulers_online), 6000)
    input_enumerable = get_input_ranges(lower, upper, chunk_size)
    IO.inspect(input_enumerable)
    Task.async_stream(input_enumerable, fn input -> elem(GenServer.start_link(Fizzbuzz.Worker, [input]), 1) end, timeout: :infinity)
    |> Stream.map(fn {:ok, res} -> res end)
    |> Stream.each(
      fn pid ->
        GenServer.call(pid, :print)
        Process.exit(pid, :kill)
      end)
    |> Stream.run()
  end

  def main(_), do: IO.puts("Usage: fizzbuzz 1 10000")

  defp get_input_ranges(lower, upper, chunk_size) do
    if chunk_size >= 10 do
      if lower >= upper, do: [], else: [lower..min(lower+chunk_size, upper) | get_input_ranges(min(lower+chunk_size, upper) + 1, upper, chunk_size)]
    else
      [lower..upper]
    end
  end
end

As a result, I’m getting speeds of 80+ MiB/s on my system, and everything is getting printed in order. It’s not in the order of GiB/s, but I’m happy with my gains for now.

Thanks for all the help once again. Any suggestions/improvements to the above code are welcome.

You can likely optimize it further if :print is not a call but a cast (does not require a response).

@dimitarvp I tried that, but I face two problems: 1) I have to wait for all GenServers to finish (I don’t know how to do that properly, so I just added a Process.sleep at end of my driver module for the timebeing). 2) it breaks the order in which numbers are printed.

Oh I must have missed that: is the right printing order a hard requirement?

No worries @dimitarvp, I missed that the first time too. But the problem does mention that the output should be a valid Fizzbuzz output, so I believe the answer is yes. Otherwise I’d happily go all out with the handle_cast approach, since I’m easily getting 110 MiB/s throughput with that.

I’m happy to report that I’m now able to achieve 150+ MiB/s throughput on my system by making a small change to Fizzbuzz.reply function:

defp reply(n) do
  case {rem(n, 3), rem(n, 5)} do
    {0, 0} -> "FizzBuzz\n"
    {0, _} -> "Fizz\n"
    {_, 0} -> "Buzz\n"
    {_, _} -> [Integer.to_string(n), "\n"]
  end
end