Timeout errors with long running transactions

cjk

Hi there,

I have a problem with ecto timeouts (so it seems). In a Elixir application I wrote a module for importing data. It gets a CSV file, goes through that file line by line (with a Stream.map()), looks if this dataset already exists in the database, updates it or inserts it as a new dataset. Pretty basic.

To make that operation restartable I wrap this whole process in a database transaction. The amount of datasets is pretty large and it can take up to an hour.

I know of Ecto timeouts, and thus I start that transaction with timeout: :infinity to avoid timeout problems. It works in development, but in prod I get strange errors:

16:02:34.435 [error] Postgrex.Protocol (#PID<0.423.0>) disconnected: ** (DBConnection.ConnectionError) ssl send: closed

 ** (exit) an exception was raised:
     ** (DBConnection.ConnectionError) ssl send: closed
         (ecto_sql) lib/ecto/adapters/sql.ex:624: Ecto.Adapters.SQL.raise_sql_call_error/1
         (ecto_sql) lib/ecto/adapters/sql.ex:557: Ecto.Adapters.SQL.execute/5
         (ecto) lib/ecto/repo/queryable.ex:147: Ecto.Repo.Queryable.execute/4
         (ecto) lib/ecto/repo/queryable.ex:18: Ecto.Repo.Queryable.all/3
         (ecto) lib/ecto/repo/queryable.ex:66: Ecto.Repo.Queryable.one/3
         (termitool) lib/termitool/meta/meta.ex:332: Termitool.Meta.get_user_by/1
         (termitool) lib/termitool/meta/meta.ex:439: Termitool.Meta.get_by_username/1
         (termitool) lib/termitool/meta/meta.ex:465: Termitool.Meta.username_password_auth/2

Basically connection errors at random places in the application. I can avoid this problem by setting timeout: :infinity in the repo configuration, but this seems wrong and dangerous to me.

The code is basically:

Repo.transaction(fn ->
  stream
  |> Stream.map(fn {row, idx} -> update_or_create_row(row) end)
end,
timeout: :infinity)

I am using a stream because that’s what I get from the CSV parsing library.

What am I doing wrong?

Best regards,
CK

18 comments

#ecto

0 5347 18

2019-03-15 12:25:34 UTC

Showing Posts 1 to 10

Show Best Posts
Show All (oldest first)
Show All (newest first)

dimitarvp

This looks strange to me: your last statement in such a processing pipeline should always be using Enum; Stream just returns a function.

Are you sure any actual work is done by this code in production?

Post #1

cjk

Sorry - it‘s an Enum.map, not a Stream.map

Edit: ok, I’m at the PC again.

Sorry, I simplified the code a bit, in fact it is more like this:

Repo.transaction(fn ->
  stream
  |> Stream.map(&create_or_replace/1)
  |> Enum.reduce(…)
end,
timeout: :infinity)

The reduce counts and throws away the success results and collects the error results with their changesets.

Post #2

cjk

Nothing, nobody? Am I missing some important information?

Post #3

NobbZ

Is it the transaction that times out or the individual query?

As far as I can remember DB lessons during university, individual queries during a transaction might take longer than without the transaction due to the doublebookkeeping of indexes or linear scan of things that are only visible in the transaction but not yet commited to the database. Can you check if the problem persists if you increase the timeout of the individual query?

Post #4

kokolegorille

There are some tools to leverage high import, for example GenStage, which will provide back pressure support.

I would also reach for insert_all, because it’s more efficient to send one query with 10’000 insert than 10’000 single insert.

Enum.chunk_by can be useful to treat smaller pieces.

But all of this does not fit well within a huge transaction.

Post #5

cjk

The timeouts appear all over the place in the application during the import job, in parts of the application which have nothing to do with the import and are not wrapped in the transaction. But the queries in the transaction seem not to time out, I did not see the connection errors there.

Post #6

cjk

insert_all is not a viable solution in this case, I either update existing rows or insert new rows, but not all in one table. For example one CSV row can result in 4 rows in 4 different tables.

Post #7

kokolegorille

Instead of inserting new rows, I keep them in a list, and proceed all in one go.

If You have different tables, You still might keep multiple lists (one per table), and proceed with multiple insert_all.

I had similar constraint (huge textfile, multiple tables, over 1’000’000 records) with update or create. I had to be careful to race condition when processing the file concurrently.

After using tasks, poolboy, I finally setup my import pipeline with GenStage, and a couple of insert_all.

Post #8

cjk

Hm. Race conditions on the data are a non-issue in this case. But I will overhaul my import pipeline, thanks for your input…

That said, I guess I found the underlying cause for this problem. Your hint with concurrently importing the data gave me the idea: the CSV library uses workers to parallelize the CSV reading and parsing. That means that my Stream.map() executions are parallelized as well.

Parallelized execution means: different ecto processes are used. And since the server has traffic and more cores than my workstation this means: more Stream workers, more user connections and thus the pool (I was using the default size of 15) could be exhausted pretty fast.

And indeed, if I reduce the pool size on my workstation to 2 and the timeout to 1 second, I get the same errors all over the place.

To avoid that I now use the :caller option on all repo calls, and now it works like a charm on my dev machine. Have still to test it in production, though

This also means: my import did not run in the transaction anyways…

Post #9

cjk

Dammit.

The errors appeared again, random connection drops by Postgrex:

11:38:50.415 [error] Postgrex.Protocol (#PID<0.425.0>) disconnected: ** (DBConnection.ConnectionError) ssl send: closed
11:38:52.415 [error] Postgrex.Protocol (#PID<0.431.0>) disconnected: ** (DBConnection.ConnectionError) ssl send: closed
11:38:54.415 [error] Postgrex.Protocol (#PID<0.418.0>) disconnected: ** (DBConnection.ConnectionError) ssl send: closed
11:38:56.415 [error] Postgrex.Protocol (#PID<0.424.0>) disconnected: ** (DBConnection.ConnectionError) ssl send: closed
11:38:58.415 [error] Postgrex.Protocol (#PID<0.433.0>) disconnected: ** (DBConnection.ConnectionError) ssl send: closed
11:39:00.415 [error] Postgrex.Protocol (#PID<0.426.0>) disconnected: ** (DBConnection.ConnectionError) ssl send: closed

And this time there was not even an import job running, I just removed the timeout: :infinity from the repo configuration.

What’s going on?!