Current section
18 Versions
Jump to
Current section
18 Versions
Compare versions
6
files changed
+170
additions
-30
deletions
| @@ -73,6 +73,7 @@ connect(Opts) -> {ok, Connection :: epgsql:connection()} | {error, Reason :: epg | |
| 73 73 | port => inet:port_number(), |
| 74 74 | ssl => boolean() | required, |
| 75 75 | ssl_opts => [ssl:tls_client_option()], % @see OTP ssl documentation |
| 76 | + socket_active => true | integer(), % @see "Active socket" section below |
| 76 77 | tcp_opts => [gen_tcp:option()], % @see OTP gen_tcp module documentation |
| 77 78 | timeout => timeout(), % socket connect timeout, default: 5000 ms |
| 78 79 | async => pid() | atom(), % process to receive LISTEN/NOTIFY msgs |
| @@ -108,7 +109,7 @@ Only `host` and `username` are mandatory, but most likely you would need `databa | |
| 108 109 | - `ssl` if set to `true`, perform an attempt to connect in ssl mode, but continue unencrypted |
| 109 110 | if encryption isn't supported by server. if set to `required` connection will fail if encryption |
| 110 111 | is not available. |
| 111 | - - `ssl_opts` will be passed as is to `ssl:connect/3` |
| 112 | + - `ssl_opts` will be passed as is to `ssl:connect/3`. |
| 112 113 | - `tcp_opts` will be passed as is to `gen_tcp:connect/3`. Some options are forbidden, such as |
| 113 114 | `mode`, `packet`, `header`, `active`. When `tcp_opts` is not provided, epgsql does some tuning |
| 114 115 | (eg, sets TCP `keepalive` and auto-tunes `buffer`), but when `tcp_opts` is provided, no |
| @@ -125,6 +126,12 @@ Only `host` and `username` are mandatory, but most likely you would need `databa | |
| 125 126 | - `application_name` is an optional string parameter. It is usually set by an application upon |
| 126 127 | connection to the server. The name will be displayed in the `pg_stat_activity` |
| 127 128 | view and included in CSV log entries. |
| 129 | + - `socket_active` is an optional parameter, which can be `true` or an integer in the range -32768 |
| 130 | + to 32767 (inclusive, however only positive value make sense right now). |
| 131 | + This option is used to control the flow of incoming messages from the network socket to make |
| 132 | + sure huge query results won't result in `epgsql` process mailbox overflow. It affects the |
| 133 | + behaviour of some of the commands and interfaces (`epgsqli` and replication), so, use with |
| 134 | + caution! See [Active socket](#active-socket) for more details. |
| 128 135 | |
| 129 136 | Options may be passed as proplist or as map with the same key names. |
| 130 137 | |
| @@ -673,6 +680,48 @@ Retrieve actual value of server-side parameters, such as character endoding, | |
| 673 680 | date/time format and timezone, server version and so on. See [libpq PQparameterStatus](https://www.postgresql.org/docs/current/static/libpq-status.html#LIBPQ-PQPARAMETERSTATUS). |
| 674 681 | Parameter's value may change during connection's lifetime. |
| 675 682 | |
| 683 | + ## Active socket |
| 684 | + |
| 685 | + By default `epgsql` sets its underlying `gen_tcp` or `ssl` socket into `{active, true}` mode |
| 686 | + (make sure you understand the [OTP inet:setopts/2 documentation](https://www.erlang.org/doc/man/inet.html#setopts-2) |
| 687 | + about `active` option). |
| 688 | + That means if PostgreSQL decides to quickly send a huge amount of data to the client (for example, |
| 689 | + client made a SELECT that returns large amount of results or when we are connected in streaming |
| 690 | + replication mode and receiving a lot of updates), underlying network socket might quickly send |
| 691 | + large number of messages to the `epgsql` connection process leading to the growing mailbox and high |
| 692 | + RAM consumption (or even OOM situation in case of really large query result or massive replication |
| 693 | + update). |
| 694 | + |
| 695 | + To avoid such scenarios, `epgsql` can may rely on "TCP backpressure" to prevent socket from sending |
| 696 | + unlimited number of messages - implement a "flow control". To do so, `socket_active => 1..32767` |
| 697 | + could be added at connection time. This option would set `{active, N}` option on the underlying |
| 698 | + socket and would tell the network to send no more than `N` messages to `epgsql` connection and then |
| 699 | + pause to let `epgsql` and the client process the already received network data and then decide how |
| 700 | + to proceed. |
| 701 | + |
| 702 | + The way this pause is signalled to the client and how the socket can be activated again depends on |
| 703 | + the interface client is using: |
| 704 | + |
| 705 | + - when `epgsqli` interface is used, `epgsql` would send all the normal low level messages and then |
| 706 | + at any point it may send `{epgsql, C, socket_passive}` message to signal that socket have been |
| 707 | + paused. `epgsql:activate(C)` must be called to re-activate the socket. |
| 708 | + - when `epgsql` is connected in [Streaming replication](doc/streaming.md) mode and `pid()` is used |
| 709 | + as the receiver of the X-Log Data messages, it would behave in the same way: |
| 710 | + `{epgsql, C, socket_passive}` might be sent along with |
| 711 | + `{epgsql, self(), {x_log_data, _, _, _}}` messages and `epgsql:activate/1` can be used to |
| 712 | + re-activate. |
| 713 | + - in all the other cases (`epgsql` / `epgsqla` command, while `COPY FROM STDIN` mode is active, |
| 714 | + when Streaming replication with Erlang module callback as receiver of X-Log Data or |
| 715 | + while connection is idle) `epgsql` would transparently re-activate the socket automatically: it |
| 716 | + won't prevent high RAM usage from large SELECT result, but it would make sure `epgsql` process |
| 717 | + has no more than `N` messages from the network in its mailbox. |
| 718 | + |
| 719 | + It is a good idea to combine `socket_active => N` with some specific value of |
| 720 | + `tcp_opts => [{buffer, X}]` since each of the `N` messages sent from the network to `epgsql` |
| 721 | + process would contain no more than `X` bytes. So the MAXIMUM amount of data seating at the `epgsql` |
| 722 | + mailbox could be roughly estimated as `N * X`. So if `N = 256` and `X = 512*1024` (512kb) then |
| 723 | + there will be no more than `N * X = 256 * 524288 = 134_217_728` or 128MB of data in the mailbox |
| 724 | + at the same time. |
| 676 725 | |
| 677 726 | ## Streaming replication protocol |
| 678 727 | |
| @@ -698,7 +747,7 @@ Here's how to create a patch that's easy to integrate: | |
| 698 747 | - Create a new branch for the proposed fix. |
| 699 748 | - Make sure it includes a test and documentation, if appropriate. |
| 700 749 | - Open a pull request against the `devel` branch of epgsql. |
| 701 | - - Passing build in travis |
| 750 | + - Passing CI build |
| 702 751 | |
| 703 752 | ## Test Setup |
| 704 753 | |
| @@ -708,11 +757,11 @@ Postgres database. | |
| 708 757 | NOTE: you will need the postgis and hstore extensions to run these |
| 709 758 | tests! On Ubuntu, you can install them with a command like this: |
| 710 759 | |
| 711 | - 1. `apt-get install postgresql-9.3-postgis-2.1 postgresql-contrib` |
| 760 | + 1. `apt-get install postgresql-12-postgis-3 postgresql-contrib` |
| 712 761 | 1. `make test` # Runs the tests |
| 713 762 | |
| 714 763 | NOTE 2: It's possible to run tests on exact postgres version by changing $PATH like |
| 715 764 | |
| 716 | - `PATH=$PATH:/usr/lib/postgresql/9.5/bin/ make test` |
| 765 | + `PATH=$PATH:/usr/lib/postgresql/12/bin/ make test` |
| 717 766 | |
| 718 767 | [](https://github.com/epgsql/epgsql/actions/workflows/ci.yml) |
| @@ -47,4 +47,4 @@ | |
| 47 47 | {<<"links">>,[{<<"Github">>,<<"https://github.com/epgsql/epgsql">>}]}. |
| 48 48 | {<<"name">>,<<"epgsql">>}. |
| 49 49 | {<<"requirements">>,[]}. |
| 50 | - {<<"version">>,<<"4.6.1">>}. |
| 50 | + {<<"version">>,<<"4.7.0">>}. |
| @@ -85,7 +85,7 @@ execute(PgSock, #connect{opts = #{username := Username} = Opts, stage = connect} | |
| 85 85 | execute(PgSock, #connect{stage = auth, auth_send = {PacketType, Data}} = St) -> |
| 86 86 | {send, PacketType, Data, PgSock, St#connect{auth_send = undefined}}. |
| 87 87 | |
| 88 | - -spec open_socket([{atom(), any()}], epgsql:connect_opts()) -> |
| 88 | + -spec open_socket([{atom(), any()}], epgsql:connect_opts_map()) -> |
| 89 89 | {ok , gen_tcp | ssl, gen_tcp:socket() | ssl:sslsocket()} | {error, any()}. |
| 90 90 | open_socket(SockOpts, #{host := Host} = ConnectOpts) -> |
| 91 91 | Timeout = maps:get(timeout, ConnectOpts, 5000), |
| @@ -1,6 +1,6 @@ | |
| 1 1 | {application,epgsql, |
| 2 2 | [{description,"PostgreSQL Client"}, |
| 3 | - {vsn,"4.6.1"}, |
| 3 | + {vsn,"4.7.0"}, |
| 4 4 | {modules,[]}, |
| 5 5 | {registered,[]}, |
| 6 6 | {applications,[kernel,stdlib,ssl]}, |
| @@ -37,13 +37,15 @@ | |
| 37 37 | start_replication/5, |
| 38 38 | start_replication/6, |
| 39 39 | start_replication/7, |
| 40 | - to_map/1]). |
| 41 | - -export([handle_x_log_data/5]). % private |
| 40 | + to_map/1, |
| 41 | + activate/1]). |
| 42 | + %% private |
| 43 | + -export([handle_x_log_data/5]). |
| 42 44 | |
| 43 | - -export_type([connection/0, connect_option/0, connect_opts/0, |
| 45 | + -export_type([connection/0, connect_option/0, connect_opts/0, connect_opts_map/0, |
| 44 46 | connect_error/0, query_error/0, sql_query/0, column/0, |
| 45 47 | type_name/0, epgsql_type/0, statement/0, |
| 46 | - transaction_option/0, transaction_opts/0]). |
| 48 | + transaction_option/0, transaction_opts/0, socket_active/0]). |
| 47 49 | |
| 48 50 | %% Deprecated types |
| 49 51 | -export_type([bind_param/0, typed_param/0, |
| @@ -62,6 +64,7 @@ | |
| 62 64 | -type host() :: inet:ip_address() | inet:hostname(). |
| 63 65 | -type password() :: string() | iodata() | fun( () -> iodata() ). |
| 64 66 | -type connection() :: pid(). |
| 67 | + -type socket_active() :: true | -32768..32767. |
| 65 68 | -type connect_option() :: |
| 66 69 | {host, host()} | |
| 67 70 | {username, string()} | |
| @@ -76,11 +79,10 @@ | |
| 76 79 | {codecs, Codecs :: [{epgsql_codec:codec_mod(), any()}]} | |
| 77 80 | {nulls, Nulls :: [any(), ...]} | % terms to be used as NULL |
| 78 81 | {replication, Replication :: string()} | % Pass "database" to connect in replication mode |
| 79 | - {application_name, ApplicationName :: string()}. |
| 80 | - |
| 81 | - -type connect_opts() :: |
| 82 | - [connect_option()] |
| 83 | - | #{host => host(), |
| 82 | + {application_name, ApplicationName :: string()} | |
| 83 | + {socket_active, Active :: socket_active()}. |
| 84 | + -type connect_opts_map() :: |
| 85 | + #{host => host(), |
| 84 86 | username => string(), |
| 85 87 | password => password(), |
| 86 88 | database => string(), |
| @@ -93,9 +95,12 @@ | |
| 93 95 | codecs => [{epgsql_codec:codec_mod(), any()}], |
| 94 96 | nulls => [any(), ...], |
| 95 97 | replication => string(), |
| 96 | - application_name => string() |
| 98 | + application_name => string(), |
| 99 | + socket_active => socket_active() |
| 97 100 | }. |
| 98 101 | |
| 102 | + -type connect_opts() :: connect_opts_map() | [connect_option()]. |
| 103 | + |
| 99 104 | -type transaction_option() :: |
| 100 105 | {reraise, boolean()} | |
| 101 106 | {ensure_committed, boolean()} | |
| @@ -571,3 +576,16 @@ to_map(Map) when is_map(Map) -> | |
| 571 576 | Map; |
| 572 577 | to_map(List) when is_list(List) -> |
| 573 578 | maps:from_list(List). |
| 579 | + |
| 580 | + %% @doc Activates TCP or SSL socket of a connection. |
| 581 | + %% |
| 582 | + %% If the `socket_active` connection option is supplied the function sets |
| 583 | + %% `{active, X}' the connection's SSL or TCP socket. It sets `{active, true}' otherwise. |
| 584 | + %% |
| 585 | + %% @param Connection connection |
| 586 | + %% @returns `ok' or `{error, Reason}' |
| 587 | + %% |
| 588 | + %% Note: The ssl:reason() type is not exported so that we use `any()' on the spec. |
| 589 | + -spec activate(connection()) -> ok | {error, inet:posix() | any()}. |
| 590 | + activate(Connection) -> |
| 591 | + epgsql_sock:activate(Connection). |
Loading more files…