Now that you posted the entire code, here’s exactly what’s going on.
You are spawning 18 parallel processes and each of them is trying to insert 500 records one by one which means 9000 DB transactions. Unless you configure your pool with 9000 connections then an incomplete ingestion is to be expected.
The idea of balancing out your Repo’s pool size and the streaming processing’s max_concurrency parameter is to avoid exactly what your code is doing: never go above the Repo’s pool size.
But your code is doing it, in fact exceeding Repo’s capacity by 500x.
Change your code to do this:
defmodule IngestCSV do
alias NimbleCSV.RFC4180, as: YourCSV
require Logger
NimbleCSV.define(YourCSV, separator: ";")
def load(path) do
path
|> Path.expand()
|> File.stream!(read_ahead: 524_288)
|> YourCSV.parse_stream()
# Remove `Stream.chunk_every`
|> Task.async_stream(fn [name, bar_code, price, currency] ->
# Remove `Enum.map`
FileWatchExample.Product.create_product(%{name: name, bar_code: bar_code, price: price, currency: currency})
end, max_concurrency: 18, on_timeout: :kill_task, ordered: false)
|> Stream.run()
end
end
Now your DB pool’s size is 20 and you will never have more than 18 records being inserted in parallel, thus you should never have dropped ingestions.
If that really works then you can proceed to bump up the Repo’s pool size to e.g. 100 and put max_concurrency at 95 - 98 and that should work fine and accelerate your workload.
I believe you are being tripped up by Task.async_stream’s semantics. The function that is passed to it is going to be working in 18 separate parallel independent processes. BUT, having Enum.map inside of it is still serial i.e. not parallelized. So you do have 18 parallel independent tasks each trying to open 500 DB transactions (inserts) one after another. As you have found that that mostly works but not always – because it’s a very wrongly written parallel code.
