:poolboy doesn't play nice with Task.async. Help

Questions / Help
,
1

I’m fairly new to elixir. Here’s the code:

defmodule MovAvi.Pool.Executor do
  defmodule Worker do
    use GenServer

    def start_link(args) do
      GenServer.start_link(__MODULE__, args, [])
    end

    def init(args \\ []) do
      uniq_id = :rand.uniform(500)
      IO.puts("Starting worker with uniq_id: #{uniq_id}")
      {:ok, {uniq_id}}
    end

    def worker_called(server, args) do
      GenServer.call(server, {:worker_called, args})
    end

    def handle_call({:worker_called, args}, _from, state) do
      IO.puts("worker_called #{args} #{inspect(state)}")
      :timer.sleep(500)
      {:reply, nil, state}
    end
  end

  use GenServer

  def start_link(args) do
    GenServer.start_link(__MODULE__, args, name: __MODULE__)
  end

  def init(args \\ [pool_size: 2, pool_overflow: 0]) do
    {:ok, pool} =
      :poolboy.start_link(
        worker_module: args[:worker_module],
        size: args[:pool_size],
        max_overflow: args[:pool_overflow]
      )

    {:ok, pool}
  end

  def wait_for_tasks(tasks) do
    GenServer.call(__MODULE__, {:wait_for_tasks, tasks})
  end

  def run_with_task(id) do
    GenServer.call(__MODULE__, {:run_with_task, id})
  end

  def run(id) do
    GenServer.call(__MODULE__, {:run, id})
  end

  def handle_call({:wait_for_tasks, tasks}, _from, state) do
    tasks |> Enum.each(&Task.await(&1))
    {:reply, nil, state}
  end

  def handle_call({:run_with_task, id}, _from, pool) do
    task =
      Task.async(fn ->
        :poolboy.transaction(pool, fn worker ->
          Worker.worker_called(worker, id)
        end)
      end)

    {:reply, task, pool}
  end

  def handle_call({:run, id}, _from, pool) do
    :poolboy.transaction(pool, fn worker ->
      Worker.worker_called(worker, id)
    end)

    {:reply, nil, pool}
  end
end

{:ok, executer} =
  MovAvi.Pool.Executor.start_link(worker_module: MovAvi.Pool.Executor.Worker, pool_size: 2)

Enum.map(0..5, fn id ->
  MovAvi.Pool.Executor.run(id)
end)

IO.puts("=== running run_with_task ===")

Enum.map(6..10, fn id ->
  MovAvi.Pool.Executor.run_with_task(id)
end)
|> MovAvi.Pool.Executor.wait_for_tasks()

I’m playing around with :poolboy and ran into an issue. With normal run the GenServer is blocked till the worker :poolboy.transaction is completed this is expected. Also only 2 workers are created as mentioned.

But when I use run_with_task, :poolboy creates new worker for every id. I’m confused why this is happening.

I’m expecting run_with_task to create all Task and the Task should be blocked for :poolboy to allocate workers.

Any insights and help is appreciated. Also If you think there’s a better way to write the code do let me know.

Thanks

2

Executor.init receives [worker_module: MovAvi.Pool.Executor.Worker, pool_size: 2] as args, so args[:pool_overflow] returns nil, not 0. Consider making the default arg an empty list and using a default per item instead like args[:pool_overflow] || 0.

(In fact the default arg for init will never be used, since the callback used by OTP is init/1, not init/0)

3

Thanks :smile:

For some reason I was thinking defaults will fill missing keys. Sort of merge args with defaults.

Everything works as expected.