-include("ejabberd_stacktrace.hrl").
-record(state,
- {db_ref = self() :: pid(),
- db_type = odbc :: pgsql | mysql | sqlite | odbc | mssql,
- db_version = undefined :: undefined | non_neg_integer(),
- host = <<"">> :: binary(),
- pending_requests :: p1_queue:queue()}).
+ {db_ref :: undefined | pid(),
+ db_type = odbc :: pgsql | mysql | sqlite | odbc | mssql,
+ db_version :: undefined | non_neg_integer(),
+ host :: binary(),
+ pending_requests :: p1_queue:queue(),
+ overload_reported :: undefined | integer()}).
-define(STATE_KEY, ejabberd_sql_state).
-
-define(NESTING_KEY, ejabberd_sql_nesting_level).
-
-define(TOP_LEVEL_TXN, 0).
-
-define(MAX_TRANSACTION_RESTARTS, 10).
-
-define(KEEPALIVE_QUERY, [<<"SELECT 1;">>]).
-
-define(PREPARE_KEY, ejabberd_sql_prepare).
-
%%-define(DBGFSM, true).
-
-ifdef(DBGFSM).
-
-define(FSMOPTS, [{debug, [trace]}]).
-
-else.
-
-define(FSMOPTS, []).
-
-endif.
-%%%----------------------------------------------------------------------
-%%% API
-%%%----------------------------------------------------------------------
--spec start_link(binary(), pos_integer()) -> {ok, pid()} | {error, term()}.
-start_link(Host, I) ->
- Proc = binary_to_atom(get_worker_name(Host, I), utf8),
- p1_fsm:start_link({local, Proc}, ?MODULE, [Host],
- fsm_limit_opts() ++ ?FSMOPTS).
-
+-type state() :: #state{}.
-type sql_query_simple() :: [sql_query() | binary()] | #sql_query{} |
fun(() -> any()) | fun((atom(), _) -> any()).
-type sql_query() :: sql_query_simple() |
{selected, [any()]} |
ok.
--spec sql_query(binary(), sql_query()) -> sql_query_result().
+%%%----------------------------------------------------------------------
+%%% API
+%%%----------------------------------------------------------------------
+-spec start_link(binary(), pos_integer()) -> {ok, pid()} | {error, term()}.
+start_link(Host, I) ->
+ Proc = binary_to_atom(get_worker_name(Host, I), utf8),
+ p1_fsm:start_link({local, Proc}, ?MODULE, [Host],
+ fsm_limit_opts() ++ ?FSMOPTS).
+-spec sql_query(binary(), sql_query()) -> sql_query_result().
sql_query(Host, Query) ->
sql_call(Host, {sql_query, Query}).
-spec sql_transaction(binary(), [sql_query()] | fun(() -> any())) ->
{atomic, any()} |
{aborted, any()}.
-
sql_transaction(Host, Queries)
when is_list(Queries) ->
F = fun () ->
sql_bloc(Host, F) -> sql_call(Host, {sql_bloc, F}).
sql_call(Host, Msg) ->
+ Timeout = query_timeout(Host),
case get(?STATE_KEY) of
undefined ->
Proc = get_worker(Host),
- sync_send_event(Proc, {sql_cmd, Msg,
- erlang:monotonic_time(millisecond)},
- query_timeout(Host));
+ sync_send_event(Proc, {sql_cmd, Msg, current_time() + Timeout},
+ Timeout);
_State ->
nested_op(Msg)
end.
keep_alive(Host, Proc) ->
- case sync_send_event(Proc,
- {sql_cmd, {sql_query, ?KEEPALIVE_QUERY},
- erlang:monotonic_time(millisecond)},
- query_timeout(Host)) of
+ Timeout = query_timeout(Host),
+ case sync_send_event(
+ Proc,
+ {sql_cmd, {sql_query, ?KEEPALIVE_QUERY}, current_time() + Timeout},
+ Timeout) of
{selected,_,[[<<"1">>]]} ->
ok;
_Err ->
?ERROR_MSG("Keep alive query failed, closing connection: ~p", [_Err]),
- sync_send_event(Proc, force_timeout, query_timeout(Host))
+ sync_send_event(Proc, force_timeout, Timeout)
end.
sync_send_event(Proc, Msg, Timeout) ->
end.
-spec sql_query_t(sql_query()) -> sql_query_result().
-
%% This function is intended to be used from inside an sql_transaction:
sql_query_t(Query) ->
QRes = sql_query_internal(Query),
case QRes of
- {error, Reason} -> throw({aborted, Reason});
+ {error, Reason} -> restart(Reason);
Rs when is_list(Rs) ->
case lists:keysearch(error, 1, Rs) of
- {value, {error, Reason}} -> throw({aborted, Reason});
+ {value, {error, Reason}} -> restart(Reason);
_ -> QRes
end;
_ -> QRes
[Event]),
{next_state, connecting, State}.
-connecting({sql_cmd, {sql_query, ?KEEPALIVE_QUERY},
- _Timestamp},
+connecting({sql_cmd, {sql_query, ?KEEPALIVE_QUERY}, Timestamp},
From, State) ->
- p1_fsm:reply(From,
- {error, <<"SQL connection failed">>}),
+ reply(From, {error, <<"SQL connection failed">>}, Timestamp),
{next_state, connecting, State};
connecting({sql_cmd, Command, Timestamp} = Req, From,
State) ->
try p1_queue:in({sql_cmd, Command, From, Timestamp},
State#state.pending_requests)
catch error:full ->
+ Err = <<"SQL request queue is overfilled">>,
+ ?ERROR_MSG("~s, bouncing all pending requests", [Err]),
Q = p1_queue:dropwhile(
- fun({sql_cmd, _, To, _Timestamp}) ->
- p1_fsm:reply(
- To, {error, <<"SQL connection failed">>}),
+ fun({sql_cmd, _, To, TS}) ->
+ reply(To, {error, Err}, TS),
true
end, State#state.pending_requests),
p1_queue:in({sql_cmd, Command, From, Timestamp}, Q)
connecting(Request, {Who, _Ref}, State) ->
?WARNING_MSG("Unexpected call ~p from ~p in 'connecting'",
[Request, Who]),
- {reply, {error, badarg}, connecting, State}.
+ {next_state, connecting, State}.
session_established({sql_cmd, Command, Timestamp}, From,
State) ->
run_sql_cmd(Command, From, State, Timestamp);
session_established(Request, {Who, _Ref}, State) ->
- ?WARNING_MSG("Unexpected call ~p from ~p in 'session_establ"
- "ished'",
+ ?WARNING_MSG("Unexpected call ~p from ~p in 'session_established'",
[Request, Who]),
- {reply, {error, badarg}, session_established, State}.
+ {next_state, session_established, State}.
session_established({sql_cmd, Command, From, Timestamp},
State) ->
{next_state, connecting, State}.
run_sql_cmd(Command, From, State, Timestamp) ->
- QueryTimeout = query_timeout(State#state.host),
- case erlang:monotonic_time(millisecond) - Timestamp of
- Age when Age < QueryTimeout ->
- put(?NESTING_KEY, ?TOP_LEVEL_TXN),
- put(?STATE_KEY, State),
- abort_on_driver_error(outer_op(Command), From);
- Age ->
- ?ERROR_MSG("Database was not available or too slow, "
- "discarding ~p milliseconds old request~n~p~n",
- [Age, Command]),
- {next_state, session_established, State}
+ case current_time() >= Timestamp of
+ true ->
+ State1 = report_overload(State),
+ {next_state, session_established, State1};
+ false ->
+ put(?NESTING_KEY, ?TOP_LEVEL_TXN),
+ put(?STATE_KEY, State),
+ abort_on_driver_error(outer_op(Command), From, Timestamp)
end.
%% Only called by handle_call, only handles top level operations.
{error, <<"killed">>};
exit:{normal, _} ->
{error, <<"terminated unexpectedly">>};
+ exit:{shutdown, _} ->
+ {error, <<"shutdown">>};
?EX_RULE(Class, Reason, Stack) ->
StackTrace = ?EX_STACK(Stack),
?ERROR_MSG("Internal error while processing SQL query:~n** ~s",
generic_sql_query_format(SQLQuery).
%% Generate the OTP callback return tuple depending on the driver result.
-abort_on_driver_error({error,
- <<"query timed out">>} = Reply,
- From) ->
- p1_fsm:reply(From, Reply),
+abort_on_driver_error({error, <<"query timed out">>} = Reply, From, Timestamp) ->
+ reply(From, Reply, Timestamp),
{stop, timeout, get(?STATE_KEY)};
-abort_on_driver_error({error,
- <<"Failed sending data on socket", _/binary>>} = Reply,
- From) ->
- p1_fsm:reply(From, Reply),
+abort_on_driver_error({error, <<"Failed sending data on socket", _/binary>>} = Reply,
+ From, Timestamp) ->
+ reply(From, Reply, Timestamp),
{stop, closed, get(?STATE_KEY)};
-abort_on_driver_error({error,
- <<"SQL connection failed">>} = Reply,
- From) ->
- p1_fsm:reply(From, Reply),
+abort_on_driver_error({error, <<"SQL connection failed">>} = Reply, From, Timestamp) ->
+ reply(From, Reply, Timestamp),
{stop, timeout, get(?STATE_KEY)};
-abort_on_driver_error({error,
- <<"Communication link failure">>} = Reply,
- From) ->
- p1_fsm:reply(From, Reply),
+abort_on_driver_error({error, <<"Communication link failure">>} = Reply, From, Timestamp) ->
+ reply(From, Reply, Timestamp),
{stop, closed, get(?STATE_KEY)};
-abort_on_driver_error(Reply, From) ->
- p1_fsm:reply(From, Reply),
+abort_on_driver_error(Reply, From, Timestamp) ->
+ reply(From, Reply, Timestamp),
{next_state, session_established, get(?STATE_KEY)}.
+-spec report_overload(state()) -> state().
+report_overload(#state{overload_reported = PrevTime} = State) ->
+ CurrTime = current_time(),
+ case PrevTime == undefined orelse (CurrTime - PrevTime) > timer:seconds(30) of
+ true ->
+ ?ERROR_MSG("SQL connection pool is overloaded, "
+ "discarding stale requests", []),
+ State#state{overload_reported = current_time()};
+ false ->
+ State
+ end.
+
+-spec reply({pid(), term()}, term(), integer()) -> term().
+reply(From, Reply, Timestamp) ->
+ case current_time() >= Timestamp of
+ true -> ok;
+ false -> p1_fsm:reply(From, Reply)
+ end.
+
%% == pure ODBC code
%% part of init/1
query_timeout(LServer) ->
ejabberd_option:sql_query_timeout(LServer).
+current_time() ->
+ erlang:monotonic_time(millisecond).
+
%% ***IMPORTANT*** This error format requires extended_errors turned on.
extended_error({"08S01", _, Reason}) ->
% TCP Provider: The specified network name is no longer available
projects / ejabberd / commitdiff