diff --git a/README.md b/README.md index 5c6e6df..24ebaff 100644 --- a/README.md +++ b/README.md @@ -61,6 +61,13 @@ end). - **Result sets are framed in bulk.** One read with the remainder carried forward, never a header read then a payload read: 61 reads over 5000 rows rather than 10,006. See [the benchmark](bench/README.md). +- **The socket NIFs are the default transport, not the inet driver.** A round + trip costs about 97 reductions rather than 323, and lending a connection out + of the pool costs a NIF call rather than two port reassignments. At sixty-four + connections that is about a fifth more queries a second for about a sixth less + of the machine; at sixteen it is nothing at all, because the pool is the + ceiling before the client is. `transport => inet` is one option away. See + [the benchmark](bench/README.md). ## Guides diff --git a/bench/README.md b/bench/README.md index e7f1a18..51e2c91 100644 --- a/bench/README.md +++ b/bench/README.md @@ -48,7 +48,11 @@ clients in the same run, and anything under about ten percent is noise. ## A run 2000 iterations, OTP 29, PostgreSQL 17 in Docker on the same machine, times in -microseconds per operation, mean. +microseconds per operation, mean. The minato column is the `inet` transport, +which is what the default was when the run was made; `minato_bench` takes the +default, so a run made now is the `socket` transport and the whole column moves +together. What that changes is the last two sections, not this one - a percent +on a query against a local server is not where a transport shows up. | workload | minato | epgsql | pgo | node-postgres | postgres.js | asyncpg | psycopg3 | | --- | --- | --- | --- | --- | --- | --- | --- | diff --git a/bench/minato_saturation.erl b/bench/minato_saturation.erl index 1c3050e..1de3a98 100644 --- a/bench/minato_saturation.erl +++ b/bench/minato_saturation.erl @@ -204,10 +204,8 @@ teardown({epgsql, _How}, Connections) -> sql() -> ~"SELECT $1::int4". -connection(inet) -> - base(); -connection(socket) -> - (base())#{transport => socket}. +connection(Transport) -> + (base())#{transport => Transport}. base() -> #{ diff --git a/guides/configuration.md b/guides/configuration.md index 8af4c7d..3fb0d61 100644 --- a/guides/configuration.md +++ b/guides/configuration.md @@ -24,6 +24,7 @@ directly. | `cancel_timeout` | `5000` | how long a cancelled statement has to acknowledge | | `prepared_statements` | `64` | how many statements a connection keeps parsed; `0` disables | | `socket_options` | `binary`, `{active,false}`, `{packet,raw}`, `{nodelay,true}` | each replaces one default | +| `transport` | `socket` | `inet` goes back to the driver every BEAM program uses | | `frame_opts` | `#{max_message_length => 67108864}` | raise it only for a single value near PostgreSQL's 1 GB limit | Two startup parameters are sent unless `parameters` overrides them: @@ -34,6 +35,20 @@ unambiguous under ISO. Set `application_name` here - it is what `pg_stat_activity` shows, and it is the difference between finding the query that is hurting and guessing. +`transport` is `socket`, which is `minato_socket` over the socket NIFs. It does +everything the driver does on every machine minato runs on, costs about a +quarter less per query and about half less per pooled one, and holds more +throughput at the same CPU once a pool is deep enough for per-operation cost to +matter at all - see the [bench](https://github.com/Taure/minato/tree/main/bench) +for where that is and is not true. `inet` is the driver, and is one option away +for a machine or a workload that disagrees. + +The two transports take the same `socket_options`, but the socket transport has +no driver behind it to hold the ones it does not use: `nodelay`, `keepalive`, +`recbuf` and `sndbuf` are set on the socket, `binary`, `{packet, raw}`, +`{active, false}` and `buffer` are accepted and ignored, and anything else is +refused at connect rather than silently dropped. + ## A pool | option | default | | diff --git a/src/minato_conn.erl b/src/minato_conn.erl index 6c8201a..41d66e5 100644 --- a/src/minato_conn.erl +++ b/src/minato_conn.erl @@ -202,10 +202,12 @@ to `prefer` under `ssl => true` and `disable` without it, since there is nothing to bind to on a plain socket. `require` refuses to connect at all unless the server offers `SCRAM-SHA-256-PLUS`. -`transport` is `inet`, the driver every BEAM program uses, or `socket`, which is -`m:minato_socket` over the socket NIFs. `socket` is experimental and faster per -operation, by about a quarter on a query and about a half on a pooled one, and -does everything the driver does on every machine minato runs on. See +`transport` is `socket`, which is `m:minato_socket` over the socket NIFs, or +`inet`, the driver every BEAM program uses. `socket` is the default: it does +everything the driver does on every machine minato runs on, costs about a +quarter less per query and about half less per pooled one, and holds more +throughput at the same CPU once a pool is deep enough for per-operation cost to +show. `inet` is there for a machine or a workload that disagrees. See `m:minato_socket`. `prepared_statements` is how many statements this connection will keep parsed on @@ -316,8 +318,8 @@ opening(Opts) -> end. -spec transport(opts()) -> plain(). -transport(#{transport := socket}) -> minato_socket; -transport(_Opts) -> gen_tcp. +transport(#{transport := inet}) -> gen_tcp; +transport(_Opts) -> minato_socket. -doc """ Close a connection. @@ -534,6 +536,13 @@ The connection is unusable for `next/1` until the messages stop coming, and the two must not be mixed on one connection. """. -spec activate(conn()) -> {ok, conn()} | {error, error()}. +%% Asking twice is asking for the same thing: one message. `{active, once}` +%% twice is harmless on the driver, but a second read on a socket handle is a +%% second operation on the same handle, which a completion system refuses +%% outright. A caller that re-activates on every subscriber - which is what a +%% listener does - would otherwise work on POSIX and not on Windows. +activate(#conn{transport = minato_socket, waiting = Waiting} = Conn) when Waiting =/= undefined -> + {ok, Conn}; activate(#conn{transport = minato_socket} = Conn) -> case minato_socket:activate(handle(Conn)) of {ok, Data} -> {ok, delivered(Conn, Data)}; @@ -651,6 +660,21 @@ handle_message( {'$socket', Socket, completion, {Handle, {error, Reason}}} ) -> {closed, Reason}; +%% The wait can also end with the machine taking it away rather than answering +%% it: a peer that closes while a read is outstanding aborts it on Windows +%% instead of completing it with an error. Ignoring that leaves an owner waiting +%% on a socket that will never speak again, which is a listener that never +%% reconnects. +handle_message( + #conn{transport = minato_socket, socket = Socket, waiting = {completion_info, _Tag, Handle}}, + {'$socket', Socket, abort, {Handle, Reason}} +) -> + {closed, Reason}; +handle_message( + #conn{transport = minato_socket, socket = Socket, waiting = {select_info, _Tag, Handle}}, + {'$socket', Socket, abort, {Handle, Reason}} +) -> + {closed, Reason}; handle_message(#conn{transport = gen_tcp, socket = Socket}, {tcp_closed, Socket}) -> {closed, closed}; handle_message(#conn{transport = gen_tcp, socket = Socket}, {tcp_error, Socket, Reason}) -> diff --git a/src/minato_socket.erl b/src/minato_socket.erl index ed7f5ad..dadc7db 100644 --- a/src/minato_socket.erl +++ b/src/minato_socket.erl @@ -2,10 +2,11 @@ -moduledoc """ The `m:socket` module behind the same calls `m:gen_tcp` answers. -Experimental, and not the default. `m:minato_conn` speaks to its transport -through `connect/4`, `send/2`, `recv/3`, `close/1`, `controlling_process/2` and +The default transport. `m:minato_conn` speaks to its transport through +`connect/4`, `send/2`, `recv/3`, `close/1`, `controlling_process/2` and `setopts/2`, and this module answers all six on top of the socket NIFs so that the transport can be chosen per connection with nothing else changing. +`transport => inet` goes back to `m:gen_tcp`. ## Why @@ -78,10 +79,12 @@ where it hands over the bytes. Open a connection. Takes the option list `m:gen_tcp` would take, so a caller does not have to know -which transport it got. `binary`, `{active, false}` and `{packet, raw}` are what -this module always does and are accepted and ignored; `{nodelay, Boolean}` and -`{buffer, _}` are the two that mean anything here, and anything else is refused -rather than silently dropped. +which transport it got. `binary`, `{mode, binary}`, `{active, false}`, +`{packet, raw}` and `{buffer, _}` are what this module always does or has no +equivalent of, and are accepted and ignored. `{nodelay, _}`, `{keepalive, _}`, +`{recbuf, _}` and `{sndbuf, _}` are set on the socket. Anything else is refused +rather than silently dropped: this is the default transport, and an option that +did something through the driver must not quietly stop doing it here. """. -spec connect( inet:socket_address() | inet:hostname(), inet:port_number(), [term()], timeout() @@ -159,11 +162,16 @@ for the calling process, so whoever wants the message has to make the call. """. -spec activate(socket()) -> {ok, binary()} | {waiting, waiting()} | {error, term()}. activate(Socket) -> - case socket:recv(Socket, 0, nowait) of + try socket:recv(Socket, 0, nowait) of {ok, Data} -> {ok, Data}; {select, Select} -> {waiting, Select}; {completion, Completion} -> {waiting, Completion}; {error, Reason} -> {error, Reason} + catch + %% A socket whose read was aborted from under it has no state left to + %% read in, and Windows raises rather than answering. It is closed, and + %% the caller asked a question that has an answer. + error:{invalid, state} -> {error, closed} end. -doc """ @@ -182,22 +190,36 @@ cancel(Socket, Waiting) -> {error, _Too_late} -> pending end. --doc "Set what can be set. `{active, _}` is refused; see the module note.". +-doc """ +Set what can be set. + +`{active, _}` is refused; see the module note. So is an option this module has +no answer for, because the alternative is a connection that was configured and +silently was not. +""". -spec setopts(socket(), [term()]) -> ok | {error, term()}. -setopts(Socket, [{active, false} | Rest]) -> - setopts(Socket, Rest); -setopts(_Socket, [{active, _Mode} | _Rest]) -> - {error, active_mode_unsupported}; -setopts(Socket, [{nodelay, Value} | Rest]) -> - case socket:setopt(Socket, {tcp, nodelay}, Value) of +setopts(Socket, [Option | Rest]) -> + case setopt(Socket, Option) of ok -> setopts(Socket, Rest); {error, Reason} -> {error, Reason} end; -setopts(Socket, [_Ignored | Rest]) -> - setopts(Socket, Rest); setopts(_Socket, []) -> ok. +-spec setopt(socket(), term()) -> ok | {error, term()}. +setopt(_Socket, {active, false}) -> ok; +setopt(_Socket, {active, _Mode}) -> {error, active_mode_unsupported}; +setopt(Socket, {nodelay, Value}) -> socket:setopt(Socket, {tcp, nodelay}, Value); +setopt(Socket, {keepalive, Value}) -> socket:setopt(Socket, {socket, keepalive}, Value); +setopt(Socket, {recbuf, Value}) -> socket:setopt(Socket, {socket, rcvbuf}, Value); +setopt(Socket, {sndbuf, Value}) -> socket:setopt(Socket, {socket, sndbuf}, Value); +setopt(_Socket, binary) -> ok; +setopt(_Socket, {mode, binary}) -> ok; +setopt(_Socket, {packet, raw}) -> ok; +setopt(_Socket, {packet, 0}) -> ok; +setopt(_Socket, {buffer, _Driver_side}) -> ok; +setopt(_Socket, Option) -> {error, {unsupported_socket_option, Option}}. + %%---------------------------------------------------------------------- %% Opening %%---------------------------------------------------------------------- diff --git a/test/minato_socket_SUITE.erl b/test/minato_transport_SUITE.erl similarity index 51% rename from test/minato_socket_SUITE.erl rename to test/minato_transport_SUITE.erl index 6fade78..198f560 100644 --- a/test/minato_socket_SUITE.erl +++ b/test/minato_transport_SUITE.erl @@ -1,55 +1,75 @@ --module(minato_socket_SUITE). +-module(minato_transport_SUITE). -moduledoc """ -The `socket` transport, against a real PostgreSQL. +Both transports, against a real PostgreSQL. `m:minato_socket` answers the same six calls `m:gen_tcp` answers, so the point of this suite is that everything above the transport cannot tell which one it got: the same authentication, the same query paths, the same errors leaving a -connection usable, and the same pool lending it out and taking it back. +connection usable, and the same pool lending it out and taking it back. Every +case runs twice, once per group, so neither transport rests on being the +default - `socket` is, and `inet` would otherwise be exercised nowhere. The two hard parts are proved rather than described. `m:ssl` takes one of these handles directly, so TLS is the same exchange over a different socket and the certificate hash SCRAM binds to is the same hash. `m:socket` has no active mode, so `LISTEN` is a `select` registered by the owning process and answered in its own mailbox - no reader process between the socket and the connection - and a -notification has to arrive through it. +notification has to arrive through it. The driver gets there by `{active, once}` +instead, and the case asserts which shape it took rather than accepting either. Skips when no server answers. """. -include_lib("stdlib/include/assert.hrl"). --export([all/0, init_per_suite/1, end_per_suite/1]). +-export([all/0, groups/0, init_per_suite/1, end_per_suite/1]). +-export([init_per_group/2, end_per_group/2]). -export([ a_query_runs/1, a_cached_statement_runs_more_than_once/1, a_failing_statement_leaves_the_connection_usable/1, a_transaction_commits/1, a_result_set_spanning_many_reads_arrives_whole/1, + a_socket_option_the_driver_takes_is_taken_here/1, + an_option_this_transport_cannot_honour_is_refused/1, a_pool_lends_a_connection_and_takes_it_back/1, a_pooled_connection_outlives_the_process_that_opened_it/1, a_borrower_that_dies_does_not_take_the_pool_down/1, tls_runs_over_this_transport/1, a_notification_arrives_over_this_transport/1, + a_connection_asked_to_wait_twice_waits_once/1, + a_connection_that_dies_while_waiting_says_so/1, a_connection_goes_back_to_reading_after_waiting/1, a_listener_runs_over_this_transport/1 ]). -define(TIMEOUT, 10000). --spec all() -> [atom()]. +-spec all() -> [{group, atom()}]. all() -> + [{group, socket}, {group, inet}]. + +-spec groups() -> [{atom(), [term()], [atom()]}]. +groups() -> + [{Transport, [], cases()} || Transport <- [socket, inet]]. + +-spec cases() -> [atom()]. +cases() -> [ a_query_runs, a_cached_statement_runs_more_than_once, a_failing_statement_leaves_the_connection_usable, a_transaction_commits, a_result_set_spanning_many_reads_arrives_whole, + a_socket_option_the_driver_takes_is_taken_here, + an_option_this_transport_cannot_honour_is_refused, a_pool_lends_a_connection_and_takes_it_back, a_pooled_connection_outlives_the_process_that_opened_it, a_borrower_that_dies_does_not_take_the_pool_down, tls_runs_over_this_transport, a_notification_arrives_over_this_transport, + a_connection_asked_to_wait_twice_waits_once, + a_connection_that_dies_while_waiting_says_so, a_connection_goes_back_to_reading_after_waiting, a_listener_runs_over_this_transport ]. @@ -60,7 +80,7 @@ all() -> -spec init_per_suite([{atom(), term()}]) -> [{atom(), term()}] | {skip, term()}. init_per_suite(Config) -> {ok, _Started} = application:ensure_all_started(minato), - case minato_conn:connect(maps:remove(transport, opts())) of + case minato_conn:connect((base())#{transport => inet}) of {ok, Conn} -> ok = minato_conn:close(Conn), Config; @@ -72,40 +92,48 @@ init_per_suite(Config) -> end_per_suite(_Config) -> ok. +-spec init_per_group(atom(), [{atom(), term()}]) -> [{atom(), term()}]. +init_per_group(Transport, Config) -> + [{transport, Transport} | Config]. + +-spec end_per_group(atom(), [{atom(), term()}]) -> ok. +end_per_group(_Transport, _Config) -> + ok. + %%---------------------------------------------------------------------- -%% The same client, a different transport +%% The same client, either transport %%---------------------------------------------------------------------- -a_query_runs(_Config) -> - Conn = connected(), +a_query_runs(Config) -> + Conn = connected(Config), {ok, #{rows := [{42}], command := select}, Ready} = minato_query:query(Conn, ~"SELECT $1::int4", [42]), ok = minato_conn:close(Ready). -a_cached_statement_runs_more_than_once(_Config) -> - Conn = connected(), +a_cached_statement_runs_more_than_once(Config) -> + Conn = connected(Config), {ok, #{rows := [{1}]}, First} = minato_query:cached(Conn, ~"SELECT $1::int4", [1]), {ok, #{rows := [{2}]}, Second} = minato_query:cached(First, ~"SELECT $1::int4", [2]), ?assertEqual({1, 64}, minato_conn:statements(Second)), ok = minato_conn:close(Second). -a_failing_statement_leaves_the_connection_usable(_Config) -> - Conn = connected(), +a_failing_statement_leaves_the_connection_usable(Config) -> + Conn = connected(Config), {error, {pgsql_error, #{code := ~"42601"}}, Usable} = minato_query:query(Conn, ~"NOT SQL AT ALL", []), {ok, #{rows := [{1}]}, Ready} = minato_query:query(Usable, ~"SELECT $1::int4", [1]), ok = minato_conn:close(Ready). -a_transaction_commits(_Config) -> - Conn = connected(), +a_transaction_commits(Config) -> + Conn = connected(Config), {ok, done, Ready} = minato_txn:transaction(Conn, fun(Inside) -> {ok, #{rows := [{7}]}, Written} = minato_query:query(Inside, ~"SELECT $1::int4", [7]), {ok, done, Written} end), ok = minato_conn:close(Ready). -a_result_set_spanning_many_reads_arrives_whole(_Config) -> - Conn = connected(), +a_result_set_spanning_many_reads_arrives_whole(Config) -> + Conn = connected(Config), Sql = ~"SELECT i, repeat('x', 32) FROM generate_series(1, 5000) i", {ok, #{rows := Rows, num_rows := 5000}, Ready} = minato_query:cached(Conn, Sql, []), ?assertEqual(5000, length(Rows)), @@ -113,25 +141,47 @@ a_result_set_spanning_many_reads_arrives_whole(_Config) -> ?assertEqual({5000, ~"xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"}, lists:last(Rows)), ok = minato_conn:close(Ready). +%%---------------------------------------------------------------------- +%% Socket options, which the socket transport has to answer for itself +%%---------------------------------------------------------------------- + +a_socket_option_the_driver_takes_is_taken_here(Config) -> + Opts = (opts(Config))#{ + socket_options => [{keepalive, true}, {recbuf, 65536}, {sndbuf, 65536}] + }, + {ok, Conn} = minato_conn:connect(Opts), + {ok, #{rows := [{1}]}, Ready} = minato_query:query(Conn, ~"SELECT $1::int4", [1]), + ok = minato_conn:close(Ready). + +an_option_this_transport_cannot_honour_is_refused(Config) -> + Opts = (opts(Config))#{socket_options => [{send_timeout, 5000}]}, + ok = judged(transport(Config), minato_conn:connect(Opts)). + +%% `send_timeout` is the driver's, and this module has no answer for it. The +%% refusal is the point: an option that did something through the driver has to +%% either keep doing it or say that it stopped, never quietly become nothing. +judged(socket, {error, {socket, {unsupported_socket_option, {send_timeout, 5000}}}}) -> ok; +judged(inet, {ok, Conn}) -> minato_conn:close(Conn). + %%---------------------------------------------------------------------- %% The pool, which is where the ownership question lives %%---------------------------------------------------------------------- -a_pool_lends_a_connection_and_takes_it_back(_Config) -> - Name = pool(lends), +a_pool_lends_a_connection_and_takes_it_back(Config) -> + Name = pool(lends, Config), {ok, #{rows := [{42}]}} = minato:query(Name, ~"SELECT $1::int4", [42]), {ok, #{rows := [{43}]}} = minato:query(Name, ~"SELECT $1::int4", [43]), #{disconnects := 0, checkouts := 2} = minato_pool:stats(Name), _ = minato:stop_pool(Name). -a_pooled_connection_outlives_the_process_that_opened_it(_Config) -> - Name = pool(outlives), +a_pooled_connection_outlives_the_process_that_opened_it(Config) -> + Name = pool(outlives, Config), _ = [minato:query(Name, ~"SELECT $1::int4", [N]) || N <- lists:seq(1, 20)], #{disconnects := 0} = minato_pool:stats(Name), _ = minato:stop_pool(Name). -a_borrower_that_dies_does_not_take_the_pool_down(_Config) -> - Name = pool(borrower), +a_borrower_that_dies_does_not_take_the_pool_down(Config) -> + Name = pool(borrower, Config), {ok, _Result} = minato:query(Name, ~"SELECT $1::int4", [1]), {Pid, Ref} = spawn_monitor(fun() -> {ok, _Conn} = minato_pool:checkout(Name), @@ -145,13 +195,13 @@ a_borrower_that_dies_does_not_take_the_pool_down(_Config) -> _ = minato:stop_pool(Name). %%---------------------------------------------------------------------- -%% TLS, which ssl does over one of these handles unchanged +%% TLS, which ssl does over either handle unchanged %%---------------------------------------------------------------------- -tls_runs_over_this_transport(_Config) -> +tls_runs_over_this_transport(Config) -> case reachable() of false -> {skip, no_tls_server}; - true -> over_tls() + true -> over_tls(Config) end. %% The test certificates are made by the second compose service, so a machine @@ -160,8 +210,8 @@ tls_runs_over_this_transport(_Config) -> reachable() -> filelib:is_regular(certificate("ca.crt")). -over_tls() -> - case minato_conn:connect(secured()) of +over_tls(Config) -> + case minato_conn:connect(secured(Config)) of {ok, Conn} -> {ok, #{rows := [{42}]}, Ready} = minato_query:query(Conn, ~"SELECT $1::int4", [42]), ?assertNotEqual(undefined, minato_conn:channel_binding(Ready)), @@ -178,47 +228,111 @@ over_tls() -> %% Waiting for the server without a read outstanding %%---------------------------------------------------------------------- -a_notification_arrives_over_this_transport(_Config) -> - Conn = connected(), - {ok, _Results, Listening} = minato_query:simple(Conn, ~"LISTEN minato_socket_suite"), +a_notification_arrives_over_this_transport(Config) -> + Channel = channel(~"minato_transport_suite", Config), + Conn = connected(Config), + {ok, _Results, Listening} = minato_query:simple(Conn, <<"LISTEN ", Channel/binary>>), {ok, Waiting} = minato_conn:activate(Listening), - ok = notify(~"minato_socket_suite", ~"without a read outstanding"), + ok = notify(Channel, ~"without a read outstanding", Config), {Kind, Notification} = awaited(Waiting), - ?assertEqual(shape(os:type()), Kind), + ?assertEqual(shape(transport(Config)), Kind), ?assertMatch( - {notification_response, _Pid, ~"minato_socket_suite", ~"without a read outstanding"}, + {notification_response, _Pid, Channel, ~"without a read outstanding"}, Notification ), ok = minato_conn:close(Waiting). -a_connection_goes_back_to_reading_after_waiting(_Config) -> - Conn = connected(), +%% A listener re-activates whenever anything about its subscribers changes, so a +%% connection that is already waiting is asked to wait again. One wait is what +%% it asked for both times. +a_connection_asked_to_wait_twice_waits_once(Config) -> + Channel = channel(~"minato_transport_twice", Config), + Conn = connected(Config), + {ok, _Results, Listening} = minato_query:simple(Conn, <<"LISTEN ", Channel/binary>>), + {ok, Once} = minato_conn:activate(Listening), + {ok, Twice} = minato_conn:activate(Once), + ok = notify(Channel, ~"asked twice", Config), + {_Kind, Notification} = awaited(Twice), + ?assertMatch({notification_response, _Pid, Channel, ~"asked twice"}, Notification), + ok = minato_conn:close(Twice). + +%% The server going away while a wait is outstanding is not the same as a read +%% finding a closed socket, and the machines do not agree on how it is said: a +%% `select` that never becomes readable, a completion answered with an error, or +%% the wait taken away entirely with an `abort`. All three have to reach the +%% owner as a close, because an owner that hears nothing is a listener that +%% never reconnects. +a_connection_that_dies_while_waiting_says_so(Config) -> + Marker = atom_to_binary(named(dies, Config)), + {ok, Conn} = minato_conn:connect((opts(Config))#{ + parameters => #{~"application_name" => Marker} + }), + {ok, Waiting} = minato_conn:activate(Conn), + ok = terminated(Marker, Config), + ?assertMatch({closed, _Reason}, closed(Waiting)). + +%% A wait is for one message, so what arrives before the close - the server's +%% own `FATAL` on its way out - is read and the connection armed again, which is +%% what a listener does between notifications. Arming a socket the server has +%% already gone from is the close said in the other place it can be said, and +%% which of the two happens is a race with the server's own shutdown. +closed(Conn) -> + receive + Message -> + case minato_conn:handle_message(Conn, Message) of + {messages, _Any, Again} -> rearmed(Again); + ignore -> closed(Conn); + {closed, Reason} -> {closed, Reason} + end + after ?TIMEOUT -> ct:fail(never_said_it_closed) + end. + +rearmed(Conn) -> + case minato_conn:activate(Conn) of + {ok, Armed} -> closed(Armed); + {error, Reason} -> {closed, Reason} + end. + +terminated(Marker, Config) -> + {ok, Conn} = minato_conn:connect(opts(Config)), + Sql = ~"SELECT pg_terminate_backend(pid) FROM pg_stat_activity WHERE application_name = $1", + {ok, #{rows := Rows}, Ready} = minato_query:query(Conn, Sql, [Marker]), + ok = minato_conn:close(Ready), + ?assertMatch([_Killed | _Rest], Rows), + ok. + +a_connection_goes_back_to_reading_after_waiting(Config) -> + Conn = connected(Config), {ok, Waiting} = minato_conn:activate(Conn), {ok, Passive} = minato_conn:deactivate(Waiting), {ok, #{rows := [{1}]}, Ready} = minato_query:query(Passive, ~"SELECT $1::int4", [1]), ok = minato_conn:close(Ready). -a_listener_runs_over_this_transport(_Config) -> - Name = minato_socket_suite_listener, - {ok, _Pid} = minato:start_listener(Name, #{connection => opts()}), - ok = minato:listen(Name, ~"minato_socket_listener"), - ok = notify(~"minato_socket_listener", ~"all the way up"), +a_listener_runs_over_this_transport(Config) -> + Name = named(listener, Config), + First = channel(~"minato_transport_listener", Config), + Second = channel(~"minato_transport_listener_again", Config), + {ok, _Pid} = minato:start_listener(Name, #{connection => opts(Config)}), + ok = minato:listen(Name, First), + ok = notify(First, ~"all the way up", Config), receive - {minato_notification, ~"minato_socket_listener", ~"all the way up", _From} -> ok + {minato_notification, First, ~"all the way up", _From} -> ok after ?TIMEOUT -> ct:fail(no_notification) end, - ok = minato:listen(Name, ~"minato_socket_listener_again"), - ok = notify(~"minato_socket_listener_again", ~"and again"), + ok = minato:listen(Name, Second), + ok = notify(Second, ~"and again", Config), receive - {minato_notification, ~"minato_socket_listener_again", ~"and again", _Also} -> ok + {minato_notification, Second, ~"and again", _Also} -> ok after ?TIMEOUT -> ct:fail(no_second_notification) end, ok = minato:stop_listener(Name). %% The kind is asserted rather than assumed, because it is the whole difference -%% between the two machines: POSIX says the socket is readable and leaves the -%% bytes in it, Windows completes the read and puts them in the message. A run -%% that took the other path would otherwise pass without ever touching this one. +%% between the machines and between the transports: POSIX says the socket is +%% readable and leaves the bytes in it, Windows completes the read and puts them +%% in the message, and the driver sends a `tcp` message with no `$socket` in +%% sight. A run that took another path would otherwise pass without ever +%% touching this one. awaited(Conn) -> awaited(Conn, undefined). @@ -237,23 +351,25 @@ awaited(Conn, Kind) -> kind({'$socket', _Socket, Kind, _What}, _Seen) -> Kind; kind(_Sent_from_here, Seen) -> Seen. +shape(inet) -> undefined; +shape(socket) -> shape(os:type()); shape({win32, _Flavour}) -> completion; shape(_Posix) -> select. -notify(Channel, Payload) -> - {ok, Conn} = minato_conn:connect(opts()), +notify(Channel, Payload, Config) -> + {ok, Conn} = minato_conn:connect(opts(Config)), Sql = <<"NOTIFY ", Channel/binary, ", '", Payload/binary, "'">>, {ok, _Results, Ready} = minato_query:simple(Conn, Sql), minato_conn:close(Ready). %%---------------------------------------------------------------------- -connected() -> - {ok, Conn} = minato_conn:connect(opts()), +connected(Config) -> + {ok, Conn} = minato_conn:connect(opts(Config)), Conn. -secured() -> - (opts())#{ +secured(Config) -> + (opts(Config))#{ host => env("MINATO_TLS_HOST", "localhost"), port => list_to_integer(env("MINATO_TLS_PORT", "55433")), ssl => true, @@ -272,21 +388,36 @@ directory({error, Reason}) -> directory(Directory) -> Directory. -pool(Case) -> - Name = list_to_atom("minato_socket_suite_" ++ atom_to_list(Case)), +%% Both groups run every case, so a pool, a listener or a `LISTEN` channel named +%% after the case alone would be named twice. +pool(Case, Config) -> + Name = named(Case, Config), {ok, _Pid} = minato:start_pool(Name, #{ - size => 2, min_size => 2, connection => opts() + size => 2, min_size => 2, connection => opts(Config) }), Name. -opts() -> +named(Case, Config) -> + list_to_atom( + "minato_transport_suite_" ++ atom_to_list(transport(Config)) ++ "_" ++ atom_to_list(Case) + ). + +channel(Prefix, Config) -> + <>. + +transport(Config) -> + proplists:get_value(transport, Config). + +opts(Config) -> + (base())#{transport => transport(Config)}. + +base() -> #{ host => env("MINATO_PG_HOST", "127.0.0.1"), port => list_to_integer(env("MINATO_PG_PORT", "55432")), user => list_to_binary(env("MINATO_PG_USER", "minato")), password => list_to_binary(env("MINATO_PG_PASSWORD", "minato")), database => list_to_binary(env("MINATO_PG_DATABASE", "minato_test")), - transport => socket, timeout => ?TIMEOUT }.