Skip to content

Commit

Permalink
Do a blocking flush every 100 calls to put_copy_data
Browse files Browse the repository at this point in the history
We had a blocking flush in pg-1.3.x at every call to put_copy_data.
This made sure, that all data is sent until the next put_copy_data.
In ged#462 (and pg-1.4.0 to .2) the behaviour was changed to rely on the non-blocking flushs libpq is doing internally.
This makes a decent performance improvement especially on Windows.
Unfortunately ged#473 proved that memory bloat can happen, when sending the data is slower than calls to put_copy_data happen.

As a trade-off this proposes to do a blocking flush only every 100 calls.

If libpq is running in blocking mode (PG::Connection.async_api = false) put_copy_data does a blocking flush every time new memory is allocated.
Unfortunately we don't have this kind of information, since we don't have access to libpq's PGconn struct and the return codes don't give us an indication when this happens.
So doing a flush at every fixed number of calls is a very simple heuristic.

Fixes ged#473
  • Loading branch information
larskanis committed Aug 9, 2022
1 parent 28b73d2 commit 54e22bc
Show file tree
Hide file tree
Showing 2 changed files with 12 additions and 0 deletions.
1 change: 1 addition & 0 deletions ext/pg_connection.c
Original file line number Diff line number Diff line change
Expand Up @@ -266,6 +266,7 @@ pgconn_s_allocate( VALUE klass )
this->encoder_for_put_copy_data = Qnil;
this->decoder_for_get_copy_data = Qnil;
this->trace_stream = Qnil;
rb_ivar_set(self, rb_intern("@calls_to_put_copy_data"), INT2FIX(0));

return self;
}
Expand Down
11 changes: 11 additions & 0 deletions lib/pg/connection.rb
Original file line number Diff line number Diff line change
Expand Up @@ -408,7 +408,17 @@ def isnonblocking
# See also #copy_data.
#
def put_copy_data(buffer, encoder=nil)
# sync_put_copy_data does a non-blocking attept to flush data.
until res=sync_put_copy_data(buffer, encoder)
# It didn't flush immediately and allocation of more buffering memory failed.
# Wait for all data sent by doing a blocking flush.
res = flush
end

# And do a blocking flush every 100 calls.
# This is to avoid memory bloat, when sending the data is slower than calls to put_copy_data happen.
if (@calls_to_put_copy_data += 1) > 100
@calls_to_put_copy_data = 0
res = flush
end
res
Expand All @@ -431,6 +441,7 @@ def put_copy_end(*args)
until sync_put_copy_end(*args)
flush
end
@calls_to_put_copy_data = 0
flush
end
alias async_put_copy_end put_copy_end
Expand Down

0 comments on commit 54e22bc

Please sign in to comment.