Chapril/xmpp.chapril.org-ejabberd
25
1
Fork 0
mirror of https://github.com/processone/ejabberd.git synced 2024-11-22 16:20:52 +01:00
Code Releases Activity

Discard queued requests that are too old (the caller has already got a timeout).

Browse Source 
(cherry picked from commit 0e5b930b22)
This commit is contained in:
Jerome Sautret 2010-06-11 16:35:45 +02:00
parent 2028698336
commit 1142cdad1b
1 changed files with 27 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

41
src/odbc/ejabberd_odbc.erl
View File

@ -119,14 +119,14 @@ sql_call(Host, Msg) ->
case get(?STATE_KEY) of case get(?STATE_KEY) of
undefined -> undefined ->
?GEN_FSM:sync_send_event(ejabberd_odbc_sup:get_random_pid(Host), ?GEN_FSM:sync_send_event(ejabberd_odbc_sup:get_random_pid(Host),
{sql_cmd, Msg}, ?TRANSACTION_TIMEOUT); {sql_cmd, Msg, now()}, ?TRANSACTION_TIMEOUT);
_State -> _State ->
nested_op(Msg) nested_op(Msg)
end. end.
% perform a harmless query on all opened connexions to avoid connexion close. % perform a harmless query on all opened connexions to avoid connexion close.
keep_alive(PID) -> keep_alive(PID) ->
?GEN_FSM:sync_send_event(PID, {sql_cmd, {sql_query, ?KEEPALIVE_QUERY}}, ?GEN_FSM:sync_send_event(PID, {sql_cmd, {sql_query, ?KEEPALIVE_QUERY}, now()},
?KEEPALIVE_TIMEOUT). ?KEEPALIVE_TIMEOUT).
%% This function is intended to be used from inside an sql_transaction: %% This function is intended to be used from inside an sql_transaction:
@ -219,22 +219,22 @@ connecting(Event, State) ->
?WARNING_MSG("unexpected event in 'connecting': ~p", [Event]), ?WARNING_MSG("unexpected event in 'connecting': ~p", [Event]),
{next_state, connecting, State}. {next_state, connecting, State}.
connecting({sql_cmd, {sql_query, ?KEEPALIVE_QUERY}}, From, State) -> connecting({sql_cmd, {sql_query, ?KEEPALIVE_QUERY}, _Timestamp}, From, State) ->
?GEN_FSM:reply(From, {error, "SQL connection failed"}), ?GEN_FSM:reply(From, {error, "SQL connection failed"}),
{next_state, connecting, State}; {next_state, connecting, State};
connecting({sql_cmd, Command} = Req, From, State) -> connecting({sql_cmd, Command, Timestamp} = Req, From, State) ->
?DEBUG("queueing pending request while connecting:~n\t~p", [Req]), ?DEBUG("queuing pending request while connecting:~n\t~p", [Req]),
{Len, PendingRequests} = State#state.pending_requests, {Len, PendingRequests} = State#state.pending_requests,
NewPendingRequests = NewPendingRequests =
if Len < State#state.max_pending_requests_len -> if Len < State#state.max_pending_requests_len ->
{Len + 1, queue:in({sql_cmd, Command, From}, PendingRequests)}; {Len + 1, queue:in({sql_cmd, Command, From, Timestamp}, PendingRequests)};
true -> true ->
lists:foreach( lists:foreach(
fun({sql_cmd, _, To}) -> fun({sql_cmd, _, To, _Timestamp}) ->
?GEN_FSM:reply( ?GEN_FSM:reply(
To, {error, "SQL connection failed"}) To, {error, "SQL connection failed"})
end, queue:to_list(PendingRequests)), end, queue:to_list(PendingRequests)),
{1, queue:from_list([{sql_cmd, Command, From}])} {1, queue:from_list([{sql_cmd, Command, From, Timestamp}])}
end, end,
{next_state, connecting, {next_state, connecting,
State#state{pending_requests = NewPendingRequests}}; State#state{pending_requests = NewPendingRequests}};
@ -243,19 +243,15 @@ connecting(Request, {Who, _Ref}, State) ->
[Request, Who]), [Request, Who]),
{reply, {error, badarg}, connecting, State}. {reply, {error, badarg}, connecting, State}.
session_established({sql_cmd, Command}, From, State) -> session_established({sql_cmd, Command, Timestamp}, From, State) ->
put(?NESTING_KEY, ?TOP_LEVEL_TXN), run_sql_cmd(Command, From, State, Timestamp);
put(?STATE_KEY, State),
abort_on_driver_error(outer_op(Command), From);
session_established(Request, {Who, _Ref}, State) -> session_established(Request, {Who, _Ref}, State) ->
?WARNING_MSG("unexpected call ~p from ~p in 'session_established'", ?WARNING_MSG("unexpected call ~p from ~p in 'session_established'",
[Request, Who]), [Request, Who]),
{reply, {error, badarg}, session_established, State}. {reply, {error, badarg}, session_established, State}.
session_established({sql_cmd, Command, From}, State) -> session_established({sql_cmd, Command, From, Timestamp}, State) ->
put(?NESTING_KEY, ?TOP_LEVEL_TXN), run_sql_cmd(Command, From, State, Timestamp);
put(?STATE_KEY, State),
abort_on_driver_error(outer_op(Command), From);
session_established(Event, State) -> session_established(Event, State) ->
?WARNING_MSG("unexpected event in 'session_established': ~p", [Event]), ?WARNING_MSG("unexpected event in 'session_established': ~p", [Event]),
{next_state, session_established, State}. {next_state, session_established, State}.
@ -301,6 +297,19 @@ print_state(State) ->
%%% Internal functions %%% Internal functions
%%%---------------------------------------------------------------------- %%%----------------------------------------------------------------------
run_sql_cmd(Command, From, State, Timestamp) ->
case timer:now_diff(now(), Timestamp) div 1000 of
Age when Age < ?TRANSACTION_TIMEOUT ->
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}
end.
%% Only called by handle_call, only handles top level operations. %% Only called by handle_call, only handles top level operations.
%% @spec outer_op(Op) -> {error, Reason} | {aborted, Reason} | {atomic, Result} %% @spec outer_op(Op) -> {error, Reason} | {aborted, Reason} | {atomic, Result}
outer_op({sql_query, Query}) -> outer_op({sql_query, Query}) ->