Merge branch 'improve_pipeline_balance' of https://github.com/benjaminplee/ibrowse into merge_pull_req_123
diff --git a/Makefile b/Makefile
index b596b64..28dfda8 100644
--- a/Makefile
+++ b/Makefile
@@ -15,9 +15,11 @@
mkdir -p $(DESTDIR)/lib/ibrowse-$(IBROWSE_VSN)/
cp -r ebin $(DESTDIR)/lib/ibrowse-$(IBROWSE_VSN)/
-test: all
+eunit_test: all
./rebar eunit
- erl -noshell -pa .eunit -pa test -s ibrowse -s ibrowse_test unit_tests \
+
+test: all
+ erl -noshell -pa test -pa ebin -s ibrowse_test unit_tests \
-s ibrowse_test verify_chunked_streaming \
-s ibrowse_test test_chunked_streaming_once \
-s erlang halt
diff --git a/include/ibrowse.hrl b/include/ibrowse.hrl
index d6b8f1c..5da1f0d 100644
--- a/include/ibrowse.hrl
+++ b/include/ibrowse.hrl
@@ -12,7 +12,7 @@
host_type % 'hostname', 'ipv4_address' or 'ipv6_address'
}).
--record(lb_pid, {host_port, pid}).
+-record(lb_pid, {host_port, pid, ets_tid}).
-record(client_conn, {key, cur_pipeline_size = 0, reqs_served = 0}).
diff --git a/src/ibrowse.erl b/src/ibrowse.erl
index 951cfe1..51fcb86 100644
--- a/src/ibrowse.erl
+++ b/src/ibrowse.erl
@@ -252,6 +252,11 @@
%% headers. Not quite sure why someone would want this, but one of my
%% users asked for it, so here it is. </li>
%%
+%% <li> The <code>preserve_status_line</code> option is to get the raw status line as a custom header
+%% in the response. The status line is returned as a tuple {ibrowse_status_line, Status_line_binary}
+%% If both the <code>give_raw_headers</code> and <code>preserve_status_line</code> are specified
+%% in a request, only the <code>give_raw_headers</code> is honoured. </li>
+%%
%% <li> The <code>preserve_chunked_encoding</code> option enables the caller
%% to receive the raw data stream when the Transfer-Encoding of the server
%% response is Chunked.
@@ -317,7 +322,7 @@
#url{host = Host,
port = Port,
protocol = Protocol} = Parsed_url ->
- Lb_pid = case ets:lookup(?LOAD_BALANCER_NAMED_TABLE, {Host, Port}) of
+ Lb_pid = case ets:lookup(ibrowse_lb, {Host, Port}) of
[] ->
get_lb_pid(Parsed_url);
[#lb_pid{pid = Lb_pid_1}] ->
@@ -336,7 +341,7 @@
Max_sessions,
Max_pipeline_size,
{SSLOptions, IsSSL},
- Headers, Method, Body, Options_1, Timeout, 0);
+ Headers, Method, Body, Options_1, Timeout, Timeout, os:timestamp(), 0);
Err ->
{error, {url_parsing_failed, Err}}
end.
@@ -345,29 +350,41 @@
Max_sessions,
Max_pipeline_size,
{SSLOptions, IsSSL},
- Headers, Method, Body, Options_1, Timeout, Try_count) when Try_count < 3 ->
+ Headers, Method, Body, Options_1, Timeout,
+ Ori_timeout, Req_start_time, Try_count) when Try_count =< 3 ->
ProcessOptions = get_value(worker_process_options, Options_1, []),
case ibrowse_lb:spawn_connection(Lb_pid, Parsed_url,
Max_sessions,
Max_pipeline_size,
{SSLOptions, IsSSL},
ProcessOptions) of
- {ok, Conn_Pid} ->
+ {ok, {_Pid_cur_spec_size, _, Conn_Pid}} ->
case do_send_req(Conn_Pid, Parsed_url, Headers,
Method, Body, Options_1, Timeout) of
{error, sel_conn_closed} ->
- try_routing_request(Lb_pid, Parsed_url,
- Max_sessions,
- Max_pipeline_size,
- {SSLOptions, IsSSL},
- Headers, Method, Body, Options_1, Timeout, Try_count + 1);
+ Time_now = os:timestamp(),
+ Time_taken_so_far = trunc(round(timer:now_diff(Time_now, Req_start_time)/1000)),
+ Time_remaining = Ori_timeout - Time_taken_so_far,
+ Time_remaining_percent = trunc(round((Time_remaining/Ori_timeout)*100)),
+ %% io:format("~p -- Time_remaining: ~p (~p%)~n", [self(), Time_remaining, Time_remaining_percent]),
+ case (Time_remaining > 0) andalso (Time_remaining_percent >= 5) of
+ true ->
+ try_routing_request(Lb_pid, Parsed_url,
+ Max_sessions,
+ Max_pipeline_size,
+ {SSLOptions, IsSSL},
+ Headers, Method, Body, Options_1,
+ Time_remaining, Ori_timeout, Req_start_time, Try_count + 1);
+ false ->
+ {error, retry_later}
+ end;
Res ->
Res
end;
Err ->
Err
end;
-try_routing_request(_, _, _, _, _, _, _, _, _, _, _) ->
+try_routing_request(_, _, _, _, _, _, _, _, _, _, _, _, _) ->
{error, retry_later}.
merge_options(Host, Port, Options) ->
@@ -441,14 +458,29 @@
Headers, Method, ensure_bin(Body),
Options, Timeout) of
{'EXIT', {timeout, _}} ->
+ P_info = case catch erlang:process_info(Conn_Pid, [messages, message_queue_len, backtrace]) of
+ [_|_] = Conn_Pid_info_list ->
+ Conn_Pid_info_list;
+ _ ->
+ process_info_not_available
+ end,
+ (catch lager:error("{ibrowse_http_client, send_req, ~1000.p} gen_server call timeout.~nProcess info: ~p~n",
+ [[Conn_Pid, Parsed_url, Headers, Method, Body, Options, Timeout], P_info])),
{error, req_timedout};
- {'EXIT', {noproc, {gen_server, call, [Conn_Pid, _, _]}}} ->
+ {'EXIT', {normal, _}} = Ex_rsn ->
+ (catch lager:error("{ibrowse_http_client, send_req, ~1000.p} gen_server call got ~1000.p~n",
+ [[Conn_Pid, Parsed_url, Headers, Method, Body, Options, Timeout], Ex_rsn])),
+ {error, req_timedout};
+ {error, X} when X == connection_closed;
+ X == {send_failed, {error, enotconn}};
+ X == {send_failed,{error,einval}};
+ X == {send_failed,{error,closed}};
+ X == connection_closing;
+ ((X == connection_closed_no_retry) andalso ((Method == get) orelse (Method == head))) ->
{error, sel_conn_closed};
- {'EXIT', {normal, _}} ->
- {error, sel_conn_closed};
- {'EXIT', {connection_closed, _}} ->
- {error, sel_conn_closed};
- {error, connection_closed} ->
+ {error, connection_closed_no_retry} ->
+ {error, connection_closed};
+ {error, {'EXIT', {noproc, _}}} ->
{error, sel_conn_closed};
{'EXIT', Reason} ->
{error, {'EXIT', Reason}};
@@ -558,7 +590,7 @@
%% <code>stream_to</code> option
%% @spec stream_next(Req_id :: req_id()) -> ok | {error, unknown_req_id}
stream_next(Req_id) ->
- case ets:lookup(?STREAM_TABLE, {req_id_pid, Req_id}) of
+ case ets:lookup(ibrowse_stream, {req_id_pid, Req_id}) of
[] ->
{error, unknown_req_id};
[{_, Pid}] ->
@@ -573,7 +605,7 @@
%% error returned.
%% @spec stream_close(Req_id :: req_id()) -> ok | {error, unknown_req_id}
stream_close(Req_id) ->
- case ets:lookup(?STREAM_TABLE, {req_id_pid, Req_id}) of
+ case ets:lookup(ibrowse_stream, {req_id_pid, Req_id}) of
[] ->
{error, unknown_req_id};
[{_, Pid}] ->
@@ -619,7 +651,7 @@
io:format("~80.80.=s~n", [""]),
Metrics = get_metrics(),
lists:foreach(
- fun({Host, Port, Lb_pid, Tid, Size}) ->
+ fun({Host, Port, {Lb_pid, _, Tid, Size, _}}) ->
io:format("~40.40s | ~-5.5s | ~-5.5s | ~p~n",
[Host ++ ":" ++ integer_to_list(Port),
integer_to_list(Tid),
@@ -637,74 +669,68 @@
%% included.
show_dest_status(Host, Port) ->
case get_metrics(Host, Port) of
- {Lb_pid, MsgQueueSize, Tid, Size,
- {{First_p_sz, First_speculative_sz},
- {Last_p_sz, Last_speculative_sz}}} ->
+ {Lb_pid, MsgQueueSize,
+ Tid, Size,
+ {{First_p_sz, First_p_sz},
+ {Last_p_sz, Last_p_sz}}} ->
io:format("Load Balancer Pid : ~p~n"
"LB process msg q size : ~p~n"
"LB ETS table id : ~p~n"
"Num Connections : ~p~n"
- "Smallest pipeline : ~p:~p~n"
- "Largest pipeline : ~p:~p~n",
+ "Smallest pipeline : ~p~n"
+ "Largest pipeline : ~p~n",
[Lb_pid, MsgQueueSize, Tid, Size,
- First_p_sz, First_speculative_sz,
- Last_p_sz, Last_speculative_sz]);
+ First_p_sz, Last_p_sz]);
_Err ->
io:format("Metrics not available~n", [])
end.
get_metrics() ->
- Dests = lists:filter(fun({lb_pid, {Host, Port}, _}) when is_list(Host),
- is_integer(Port) ->
- true;
- (_) ->
- false
- end, ets:tab2list(?LOAD_BALANCER_NAMED_TABLE)),
- All_ets = ets:all(),
- lists:map(fun({lb_pid, {Host, Port}, Lb_pid}) ->
- case lists:dropwhile(
- fun(Tid) ->
- ets:info(Tid, owner) /= Lb_pid
- end, All_ets) of
- [] ->
- {Host, Port, Lb_pid, unknown, 0};
- [Tid | _] ->
- Size = case catch (ets:info(Tid, size)) of
- N when is_integer(N) -> N;
- _ -> 0
- end,
- {Host, Port, Lb_pid, Tid, Size}
- end
- end, Dests).
+ Dests = lists:filter(
+ fun(#lb_pid{host_port = {Host, Port}}) when is_list(Host),
+ is_integer(Port) ->
+ true;
+ (_) ->
+ false
+ end, ets:tab2list(ibrowse_lb)),
+ lists:foldl(
+ fun(#lb_pid{host_port = {X_host, X_port}}, X_acc) ->
+ case get_metrics(X_host, X_port) of
+ {_, _, _, _, _} = X_res ->
+ [{X_host, X_port, X_res} | X_acc];
+ _X_res ->
+ X_acc
+ end
+ end, [], Dests).
get_metrics(Host, Port) ->
- case ets:lookup(?LOAD_BALANCER_NAMED_TABLE, {Host, Port}) of
+ case ets:lookup(ibrowse_lb, {Host, Port}) of
[] ->
no_active_processes;
- [#lb_pid{pid = Lb_pid}] ->
- MsgQueueSize = (catch process_info(Lb_pid, message_queue_len)),
- %% {Lb_pid, MsgQueueSize,
- case lists:dropwhile(
- fun(Tid) ->
- ets:info(Tid, owner) /= Lb_pid
- end, ets:all()) of
- [] ->
- {Lb_pid, MsgQueueSize, unknown, 0, unknown};
- [Tid | _] ->
+ [#lb_pid{pid = Lb_pid, ets_tid = Tid}] ->
+ MsgQueueSize = case (catch process_info(Lb_pid, message_queue_len)) of
+ {message_queue_len, Msg_q_len} ->
+ Msg_q_len;
+ _ ->
+ -1
+ end,
+ case Tid of
+ undefined ->
+ {Lb_pid, MsgQueueSize, undefined, 0, {{0, 0}, {0, 0}}};
+ _ ->
try
Size = ets:info(Tid, size),
case Size of
0 ->
- ok;
+ {Lb_pid, MsgQueueSize, Tid, 0, {{0, 0}, {0, 0}}};
_ ->
- First = ets:first(Tid),
- Last = ets:last(Tid),
- [{_, First_p_sz, First_speculative_sz}] = ets:lookup(Tid, First),
- [{_, Last_p_sz, Last_speculative_sz}] = ets:lookup(Tid, Last),
- {Lb_pid, MsgQueueSize, Tid, Size,
- {{First_p_sz, First_speculative_sz}, {Last_p_sz, Last_speculative_sz}}}
+ {First_p_sz, _, _} = ets:first(Tid),
+ {Last_p_sz, _, _} = ets:last(Tid),
+ {Lb_pid, MsgQueueSize,
+ Tid, Size,
+ {{First_p_sz, First_p_sz}, {Last_p_sz, Last_p_sz}}}
end
- catch _:_ ->
+ catch _:_Err ->
not_available
end
end
@@ -746,9 +772,9 @@
State = #state{},
put(my_trace_flag, State#state.trace),
put(ibrowse_trace_token, "ibrowse"),
- ?LOAD_BALANCER_NAMED_TABLE = ets:new(?LOAD_BALANCER_NAMED_TABLE, [named_table, public, {keypos, 2}]),
- ?CONF_TABLE = ets:new(?CONF_TABLE, [named_table, protected, {keypos, 2}]),
- ?STREAM_TABLE = ets:new(?STREAM_TABLE, [named_table, public]),
+ ibrowse_lb = ets:new(ibrowse_lb, [named_table, public, {keypos, 2}]),
+ ibrowse_conf = ets:new(ibrowse_conf, [named_table, protected, {keypos, 2}]),
+ ibrowse_stream = ets:new(ibrowse_stream, [named_table, public]),
import_config(),
{ok, #state{}}.
@@ -770,7 +796,7 @@
end.
apply_config(Terms) ->
- ets:delete_all_objects(?CONF_TABLE),
+ ets:delete_all_objects(ibrowse_conf),
insert_config(Terms).
insert_config(Terms) ->
@@ -783,12 +809,12 @@
{{options, Host, Port}, Options}],
lists:foreach(
fun({X, Y}) ->
- ets:insert(?CONF_TABLE,
+ ets:insert(ibrowse_conf,
#ibrowse_conf{key = X,
value = Y})
end, I);
({K, V}) ->
- ets:insert(?CONF_TABLE,
+ ets:insert(ibrowse_conf,
#ibrowse_conf{key = K,
value = V});
(X) ->
@@ -799,7 +825,7 @@
%% @doc Internal export
get_config_value(Key) ->
try
- [#ibrowse_conf{value = V}] = ets:lookup(?CONF_TABLE, Key),
+ [#ibrowse_conf{value = V}] = ets:lookup(ibrowse_conf, Key),
V
catch
error:badarg ->
@@ -809,7 +835,7 @@
%% @doc Internal export
get_config_value(Key, DefVal) ->
try
- case ets:lookup(?CONF_TABLE, Key) of
+ case ets:lookup(ibrowse_conf, Key) of
[] ->
DefVal;
[#ibrowse_conf{value = V}] ->
@@ -821,7 +847,7 @@
end.
set_config_value(Key, Val) ->
- ets:insert(?CONF_TABLE, #ibrowse_conf{key = Key, value = Val}).
+ ets:insert(ibrowse_conf, #ibrowse_conf{key = Key, value = Val}).
%%--------------------------------------------------------------------
%% Function: handle_call/3
%% Description: Handling call messages
@@ -833,7 +859,7 @@
%% {stop, Reason, State} (terminate/2 is called)
%%--------------------------------------------------------------------
handle_call({get_lb_pid, #url{host = Host, port = Port} = Url}, _From, State) ->
- Pid = do_get_connection(Url, ets:lookup(?LOAD_BALANCER_NAMED_TABLE, {Host, Port})),
+ Pid = do_get_connection(Url, ets:lookup(ibrowse_lb, {Host, Port})),
{reply, Pid, State};
handle_call(stop, _From, State) ->
@@ -841,7 +867,7 @@
ets:foldl(fun(#lb_pid{pid = Pid}, Acc) ->
ibrowse_lb:stop(Pid),
Acc
- end, [], ?LOAD_BALANCER_NAMED_TABLE),
+ end, [], ibrowse_lb),
{stop, normal, ok, State};
handle_call({set_config_value, Key, Val}, _From, State) ->
@@ -888,7 +914,7 @@
%%--------------------------------------------------------------------
handle_info(all_trace_off, State) ->
Mspec = [{{ibrowse_conf,{trace,'$1','$2'},true},[],[{{'$1','$2'}}]}],
- Trace_on_dests = ets:select(?CONF_TABLE, Mspec),
+ Trace_on_dests = ets:select(ibrowse_conf, Mspec),
Fun = fun(#lb_pid{host_port = {H, P}, pid = Pid}, _) ->
case lists:member({H, P}, Trace_on_dests) of
false ->
@@ -899,8 +925,8 @@
(_, Acc) ->
Acc
end,
- ets:foldl(Fun, undefined, ?LOAD_BALANCER_NAMED_TABLE),
- ets:select_delete(?CONF_TABLE, [{{ibrowse_conf,{trace,'$1','$2'},true},[],['true']}]),
+ ets:foldl(Fun, undefined, ibrowse_lb),
+ ets:select_delete(ibrowse_conf, [{{ibrowse_conf,{trace,'$1','$2'},true},[],['true']}]),
{noreply, State};
handle_info({trace, Bool}, State) ->
@@ -915,8 +941,8 @@
(_, Acc) ->
Acc
end,
- ets:foldl(Fun, undefined, ?LOAD_BALANCER_NAMED_TABLE),
- ets:insert(?CONF_TABLE, #ibrowse_conf{key = {trace, Host, Port},
+ ets:foldl(Fun, undefined, ibrowse_lb),
+ ets:insert(ibrowse_conf, #ibrowse_conf{key = {trace, Host, Port},
value = Bool}),
{noreply, State};
@@ -944,7 +970,6 @@
%%--------------------------------------------------------------------
do_get_connection(#url{host = Host, port = Port}, []) ->
{ok, Pid} = ibrowse_lb:start_link([Host, Port]),
- ets:insert(?LOAD_BALANCER_NAMED_TABLE, #lb_pid{host_port = {Host, Port}, pid = Pid}),
Pid;
do_get_connection(_Url, [#lb_pid{pid = Pid}]) ->
Pid.
diff --git a/src/ibrowse_http_client.erl b/src/ibrowse_http_client.erl
index d92db42..92e4964 100644
--- a/src/ibrowse_http_client.erl
+++ b/src/ibrowse_http_client.erl
@@ -1,9 +1,6 @@
%%%-------------------------------------------------------------------
%%% File : ibrowse_http_client.erl
%%% Author : Chandrashekhar Mullaparthi <chandrashekhar.mullaparthi@t-mobile.co.uk>
-%%% Benjamin Lee <http://github.com/benjaminplee>
-%%% Dan Schwabe <http://github.com/dfschwabe>
-%%% Brian Richards <http://github.com/richbria>
%%% Description : The name says it all
%%%
%%% Created : 11 Oct 2003 by Chandrashekhar Mullaparthi <chandrashekhar.mullaparthi@t-mobile.co.uk>
@@ -22,7 +19,6 @@
start/1,
start/2,
stop/1,
- trace/2,
send_req/7
]).
@@ -57,7 +53,8 @@
deleted_crlf = false, transfer_encoding,
chunk_size, chunk_size_buffer = <<>>,
recvd_chunk_size, interim_reply_sent = false,
- lb_ets_tid, prev_req_id
+ lb_ets_tid, cur_pipeline_size = 0, prev_req_id,
+ proc_state
}).
-record(request, {url, method, options, from,
@@ -77,6 +74,12 @@
-define(DEFAULT_STREAM_CHUNK_SIZE, 1024*1024).
-define(dec2hex(X), erlang:integer_to_list(X, 16)).
+
+%% Macros to prevent spelling mistakes causing bugs
+-define(dont_retry_pipelined_requests, dont_retry_pipelined_requests).
+-define(can_retry_pipelined_requests, can_retry_pipelined_requests).
+-define(dead_proc_walking, dead_proc_walking).
+
%%====================================================================
%% External functions
%%====================================================================
@@ -105,13 +108,16 @@
ok
end.
-trace(Conn_pid, Bool) ->
- catch Conn_pid ! {trace, Bool}.
-
send_req(Conn_Pid, Url, Headers, Method, Body, Options, Timeout) ->
- gen_server:call(
- Conn_Pid,
- {send_req, {Url, Headers, Method, Body, Options, Timeout}}, Timeout).
+ case catch gen_server:call(Conn_Pid,
+ {send_req, {Url, Headers, Method, Body, Options, Timeout}}, Timeout) of
+ {'EXIT', {timeout, _}} ->
+ {error, req_timedout};
+ {'EXIT', {noproc, _}} ->
+ {error, connection_closed};
+ Res ->
+ Res
+ end.
%%====================================================================
%% Server functions
@@ -126,6 +132,7 @@
%% {stop, Reason}
%%--------------------------------------------------------------------
init({Lb_Tid, #url{host = Host, port = Port}, {SSLOptions, Is_ssl}}) ->
+ process_flag(trap_exit, true),
State = #state{host = Host,
port = Port,
ssl_options = SSLOptions,
@@ -135,6 +142,7 @@
put(my_trace_flag, ibrowse_lib:get_trace_status(Host, Port)),
{ok, set_inac_timer(State)};
init(Url) when is_list(Url) ->
+ process_flag(trap_exit, true),
case catch ibrowse_lib:parse_url(Url) of
#url{protocol = Protocol} = Url_rec ->
init({undefined, Url_rec, {[], Protocol == https}});
@@ -142,6 +150,7 @@
{error, invalid_url}
end;
init({Host, Port}) ->
+ process_flag(trap_exit, true),
State = #state{host = Host,
port = Port},
put(ibrowse_trace_token, [Host, $:, integer_to_list(Port)]),
@@ -163,6 +172,10 @@
handle_call({send_req, _}, _From, #state{is_closing = true} = State) ->
{reply, {error, connection_closing}, State};
+handle_call({send_req, _}, _From, #state{proc_state = ?dead_proc_walking} = State) ->
+ shutting_down(State),
+ {reply, {error, connection_closing}, State};
+
handle_call({send_req, {Url, Headers, Method, Body, Options, Timeout}},
From, State) ->
send_req_1(From, Url, Headers, Method, Body, Options, Timeout, State);
@@ -214,30 +227,40 @@
{noreply, State};
handle_info({stream_close, _Req_id}, State) ->
- shutting_down(State),
- do_close(State),
- do_error_reply(State, closing_on_request),
- {stop, normal, State};
+ State_1 = State#state{proc_state = ?dead_proc_walking},
+ shutting_down(State_1),
+ do_close(State_1),
+ do_error_reply(State_1, closing_on_request),
+ delayed_stop_timer(),
+ {noreply, State_1};
handle_info({tcp_closed, _Sock}, State) ->
do_trace("TCP connection closed by peer!~n", []),
- handle_sock_closed(State),
- {stop, normal, State};
+ State_1 = State#state{proc_state = ?dead_proc_walking},
+ handle_sock_closed(State_1, ?can_retry_pipelined_requests),
+ delayed_stop_timer(),
+ {noreply, State_1};
handle_info({ssl_closed, _Sock}, State) ->
do_trace("SSL connection closed by peer!~n", []),
- handle_sock_closed(State),
- {stop, normal, State};
+ State_1 = State#state{proc_state = ?dead_proc_walking},
+ handle_sock_closed(State_1, ?can_retry_pipelined_requests),
+ delayed_stop_timer(),
+ {noreply, State_1};
handle_info({tcp_error, _Sock, Reason}, State) ->
do_trace("Error on connection to ~1000.p:~1000.p -> ~1000.p~n",
[State#state.host, State#state.port, Reason]),
- handle_sock_closed(State),
- {stop, normal, State};
+ State_1 = State#state{proc_state = ?dead_proc_walking},
+ handle_sock_closed(State_1, ?dont_retry_pipelined_requests),
+ delayed_stop_timer(),
+ {noreply, State_1};
handle_info({ssl_error, _Sock, Reason}, State) ->
do_trace("Error on SSL connection to ~1000.p:~1000.p -> ~1000.p~n",
[State#state.host, State#state.port, Reason]),
- handle_sock_closed(State),
- {stop, normal, State};
+ State_1 = State#state{proc_state = ?dead_proc_walking},
+ handle_sock_closed(State_1, ?dont_retry_pipelined_requests),
+ delayed_stop_timer(),
+ {noreply, State_1};
handle_info({req_timedout, From}, State) ->
case lists:keysearch(From, #request.from, queue:to_list(State#state.reqs)) of
@@ -245,21 +268,28 @@
{noreply, State};
{value, #request{stream_to = StreamTo, req_id = ReqId}} ->
catch StreamTo ! {ibrowse_async_response_timeout, ReqId},
- shutting_down(State),
- do_error_reply(State, req_timedout),
- {stop, normal, State}
+ State_1 = State#state{proc_state = ?dead_proc_walking},
+ shutting_down(State_1),
+ do_error_reply(State_1, req_timedout),
+ delayed_stop_timer(),
+ {noreply, State_1}
end;
handle_info(timeout, State) ->
do_trace("Inactivity timeout triggered. Shutting down connection~n", []),
- shutting_down(State),
- do_error_reply(State, req_timedout),
- {stop, normal, State};
+ State_1 = State#state{proc_state = ?dead_proc_walking},
+ shutting_down(State_1),
+ do_error_reply(State_1, req_timedout),
+ delayed_stop_timer(),
+ {noreply, State_1};
handle_info({trace, Bool}, State) ->
put(my_trace_flag, Bool),
{noreply, State};
+handle_info(delayed_stop, State) ->
+ {stop, normal, State};
+
handle_info(Info, State) ->
io:format("Unknown message recvd for ~1000.p:~1000.p -> ~p~n",
[State#state.host, State#state.port, Info]),
@@ -271,9 +301,10 @@
%% Description: Shutdown the server
%% Returns: any (ignored by gen_server)
%%--------------------------------------------------------------------
-terminate(_Reason, State) ->
+terminate(_Reason, #state{lb_ets_tid = Tid} = State) ->
do_close(State),
shutting_down(State),
+ (catch ets:select_delete(Tid, [{{{'_','_','$1'},'_'},[{'==','$1',{const,self()}}],[true]}])),
ok.
%%--------------------------------------------------------------------
@@ -293,16 +324,20 @@
%%--------------------------------------------------------------------
handle_sock_data(Data, #state{status=idle}=State) ->
do_trace("Data recvd on socket in state idle!. ~1000.p~n", [Data]),
- shutting_down(State),
- do_error_reply(State, data_in_status_idle),
- do_close(State),
- {stop, normal, State};
+ State_1 = State#state{proc_state = ?dead_proc_walking},
+ shutting_down(State_1),
+ do_error_reply(State_1, data_in_status_idle),
+ do_close(State_1),
+ delayed_stop_timer(),
+ {noreply, State_1};
handle_sock_data(Data, #state{status = get_header}=State) ->
case parse_response(Data, State) of
{error, _Reason} ->
- shutting_down(State),
- {stop, normal, State};
+ State_1 = State#state{proc_state = ?dead_proc_walking},
+ shutting_down(State_1),
+ delayed_stop_timer(),
+ {noreply, State_1};
#state{socket = Socket, status = Status, cur_req = CurReq} = State_1 ->
_ = case {Status, CurReq} of
{get_header, #request{caller_controls_socket = true}} ->
@@ -323,10 +358,12 @@
true ->
case accumulate_response(Data, State) of
{error, Reason} ->
- shutting_down(State),
- fail_pipelined_requests(State,
+ State_1 = State#state{proc_state = ?dead_proc_walking},
+ shutting_down(State_1),
+ fail_pipelined_requests(State_1,
{error, {Reason, {stat_code, StatCode}, Headers}}),
- {stop, normal, State};
+ delayed_stop_timer(),
+ {noreply, State_1};
State_1 ->
_ = active_once(State_1),
State_2 = set_inac_timer(State_1),
@@ -335,10 +372,12 @@
_ ->
case parse_11_response(Data, State) of
{error, Reason} ->
- shutting_down(State),
- fail_pipelined_requests(State,
+ State_1 = State#state{proc_state = ?dead_proc_walking},
+ shutting_down(State_1),
+ fail_pipelined_requests(State_1,
{error, {Reason, {stat_code, StatCode}, Headers}}),
- {stop, normal, State};
+ delayed_stop_timer(),
+ {noreply, State_1};
#state{cur_req = #request{caller_controls_socket = Ccs},
interim_reply_sent = Irs} = State_1 ->
_ = case Irs of
@@ -459,11 +498,11 @@
%%--------------------------------------------------------------------
%% Handles the case when the server closes the socket
%%--------------------------------------------------------------------
-handle_sock_closed(#state{status=get_header} = State) ->
+handle_sock_closed(#state{status=get_header} = State, _) ->
shutting_down(State),
- do_error_reply(State, connection_closed);
+ do_error_reply(State, connection_closed_no_retry);
-handle_sock_closed(#state{cur_req=undefined} = State) ->
+handle_sock_closed(#state{cur_req=undefined} = State, _) ->
shutting_down(State);
%% We check for IsClosing because this the server could have sent a
@@ -477,7 +516,7 @@
recvd_headers = Headers,
status_line = Status_line,
raw_headers = Raw_headers
- }=State) ->
+ }=State, Retry_state) ->
#request{from=From, stream_to=StreamTo, req_id=ReqId,
response_format = Resp_format,
options = Options,
@@ -505,30 +544,35 @@
{ok, SC, Headers, Buf, Raw_req}
end,
State_1 = do_reply(State, From, StreamTo, ReqId, Resp_format, Reply),
- ok = do_error_reply(State_1#state{reqs = Reqs_1}, connection_closed),
+ case Retry_state of
+ ?dont_retry_pipelined_requests ->
+ ok = do_error_reply(State_1#state{reqs = Reqs_1}, connection_closed_no_retry);
+ ?can_retry_pipelined_requests ->
+ ok = do_error_reply(State_1#state{reqs = Reqs_1}, connection_closed)
+ end,
State_1;
_ ->
- ok = do_error_reply(State, connection_closed),
+ case Retry_state of
+ ?dont_retry_pipelined_requests ->
+ ok = do_error_reply(State, connection_closed_no_retry);
+ ?can_retry_pipelined_requests ->
+ ok = do_error_reply(State, connection_closed)
+ end,
State
end.
-do_connect(Host, Port, Options, State, Timeout) ->
- SockOptions = get_sock_options(Host, Options, State#state.ssl_options),
- case {get_value(socks5_host, Options, undefined), State#state.is_ssl} of
- {undefined, true} ->
- ssl:connect(Host, Port, SockOptions, Timeout);
- {undefined, false} ->
- gen_tcp:connect(Host, Port, SockOptions, Timeout);
- {_, _} ->
- case {ibrowse_socks5:connect(Host, Port, Options, SockOptions, Timeout),
- State#state.is_ssl} of
- {{ok, Socket}, true} ->
- ssl:connect(Socket, SockOptions, Timeout);
- {{ok, Socket}, false} ->
- {ok, Socket};
- {Else, _} ->
- Else
- end
+do_connect(Host, Port, Options, #state{is_ssl = true,
+ use_proxy = false,
+ ssl_options = SSLOptions},
+ Timeout) ->
+ ssl:connect(Host, Port, get_sock_options(Host, Options, SSLOptions), Timeout);
+do_connect(Host, Port, Options, _State, Timeout) ->
+ Socks5Host = get_value(socks5_host, Options, undefined),
+ case Socks5Host of
+ undefined ->
+ gen_tcp:connect(Host, Port, get_sock_options(Host, Options, []), Timeout);
+ _ ->
+ catch ibrowse_socks5:connect(Host, Port, Options)
end.
get_sock_options(Host, Options, SSLOptions) ->
@@ -718,10 +762,12 @@
connect_timeout = Conn_timeout},
send_req_1(From, Url, Headers, Method, Body, Options, Timeout, State_3);
Err ->
- shutting_down(State_2),
+ State_3 = State_2#state{proc_state = ?dead_proc_walking},
+ shutting_down(State_3),
do_trace("Error connecting. Reason: ~1000.p~n", [Err]),
gen_server:reply(From, {error, {conn_failed, Err}}),
- {stop, normal, State_2}
+ delayed_stop_timer(),
+ {noreply, State_3}
end;
%% Send a CONNECT request.
@@ -765,23 +811,28 @@
{ok, _Sent_body} ->
trace_request_body(Body_1),
_ = active_once(State_1),
- State_2 = State_1#state{status = get_header,
- cur_req = NewReq,
- proxy_tunnel_setup = in_progress,
- tunnel_setup_queue = [{From, Url, Headers, Method, Body, Options, Timeout}]},
+ State_1_1 = inc_pipeline_counter(State_1),
+ State_2 = State_1_1#state{status = get_header,
+ cur_req = NewReq,
+ proxy_tunnel_setup = in_progress,
+ tunnel_setup_queue = [{From, Url, Headers, Method, Body, Options, Timeout}]},
State_3 = set_inac_timer(State_2),
{noreply, State_3};
Err ->
- shutting_down(State_1),
+ State_2 = State_1#state{proc_state = ?dead_proc_walking},
+ shutting_down(State_2),
do_trace("Send failed... Reason: ~p~n", [Err]),
gen_server:reply(From, {error, {send_failed, Err}}),
- {stop, normal, State_1}
+ delayed_stop_timer(),
+ {noreply, State_2}
end;
Err ->
- shutting_down(State_1),
+ State_2 = State_1#state{proc_state = ?dead_proc_walking},
+ shutting_down(State_2),
do_trace("Send failed... Reason: ~p~n", [Err]),
gen_server:reply(From, {error, {send_failed, Err}}),
- {stop, normal, State_1}
+ delayed_stop_timer(),
+ {noreply, State_2}
end;
send_req_1(From, Url, Headers, Method, Body, Options, Timeout,
@@ -808,7 +859,7 @@
{Caller, once} when is_pid(Caller) or
is_atom(Caller) ->
Async_pid_rec = {{req_id_pid, ReqId}, self()},
- true = ets:insert(?STREAM_TABLE, Async_pid_rec),
+ true = ets:insert(ibrowse_stream, Async_pid_rec),
{Caller, true};
undefined ->
{undefined, false};
@@ -855,14 +906,15 @@
Raw_req = list_to_binary([Req, Sent_body]),
NewReq_1 = NewReq#request{raw_req = Raw_req},
State_1 = State#state{reqs=queue:in(NewReq_1, State#state.reqs)},
- _ = active_once(State_1),
- State_2 = case Status of
+ State_2 = inc_pipeline_counter(State_1),
+ _ = active_once(State_2),
+ State_3 = case Status of
idle ->
- State_1#state{
+ State_2#state{
status = get_header,
cur_req = NewReq_1};
_ ->
- State_1
+ State_2
end,
case StreamTo of
undefined ->
@@ -876,19 +928,23 @@
catch StreamTo ! {ibrowse_async_raw_req, Raw_req}
end
end,
- State_3 = set_inac_timer(State_2),
- {noreply, State_3};
+ State_4 = set_inac_timer(State_3),
+ {noreply, State_4};
Err ->
- shutting_down(State),
+ State_2 = State#state{proc_state = ?dead_proc_walking},
+ shutting_down(State_2),
do_trace("Send failed... Reason: ~p~n", [Err]),
gen_server:reply(From, {error, {send_failed, Err}}),
- {stop, normal, State}
+ delayed_stop_timer(),
+ {noreply, State_2}
end;
Err ->
- shutting_down(State),
+ State_2 = State#state{proc_state = ?dead_proc_walking},
+ shutting_down(State_2),
do_trace("Send failed... Reason: ~p~n", [Err]),
gen_server:reply(From, {error, {send_failed, Err}}),
- {stop, normal, State}
+ delayed_stop_timer(),
+ {noreply, State_2}
end.
maybe_modify_headers(#url{}, connect, _, Headers, State) ->
@@ -1447,7 +1503,7 @@
_ ->
{file, TmpFilename}
end,
- {Resp_headers_1, Raw_headers_1} = maybe_add_custom_headers(RespHeaders, Raw_headers, Options),
+ {Resp_headers_1, Raw_headers_1} = maybe_add_custom_headers(Status_line, RespHeaders, Raw_headers, Options),
Give_raw_req = get_value(return_raw_request, Options, false),
Reply = case get_value(give_raw_headers, Options, false) of
true when Give_raw_req == false ->
@@ -1474,7 +1530,7 @@
reply_buffer = RepBuf
} = State) ->
Body = RepBuf,
- {Resp_headers_1, Raw_headers_1} = maybe_add_custom_headers(Resp_headers, Raw_headers, Options),
+ {Resp_headers_1, Raw_headers_1} = maybe_add_custom_headers(Status_line, Resp_headers, Raw_headers, Options),
Give_raw_req = get_value(return_raw_request, Options, false),
Reply = case get_value(give_raw_headers, Options, false) of
true when Give_raw_req == false ->
@@ -1779,7 +1835,7 @@
recvd_headers = Headers, http_status_code = StatCode,
cur_req = #request{options = Opts}
}) ->
- {Headers_1, Raw_headers_1} = maybe_add_custom_headers(Headers, Raw_headers, Opts),
+ {Headers_1, Raw_headers_1} = maybe_add_custom_headers(Status_line, Headers, Raw_headers, Opts),
case Give_raw_headers of
false ->
catch StreamTo ! {ibrowse_async_headers, ReqId, StatCode, Headers_1};
@@ -1787,7 +1843,7 @@
catch StreamTo ! {ibrowse_async_headers, ReqId, Status_line, Raw_headers_1}
end.
-maybe_add_custom_headers(Headers, Raw_headers, Opts) ->
+maybe_add_custom_headers(Status_line, Headers, Raw_headers, Opts) ->
Custom_headers = get_value(add_custom_headers, Opts, []),
Headers_1 = Headers ++ Custom_headers,
Raw_headers_1 = case Custom_headers of
@@ -1797,7 +1853,12 @@
_ ->
Raw_headers
end,
- {Headers_1, Raw_headers_1}.
+ case get_value(preserve_status_line, Opts, false) of
+ true ->
+ {[{ibrowse_status_line, Status_line} | Headers_1], Raw_headers_1};
+ false ->
+ {Headers_1, Raw_headers_1}
+ end.
format_response_data(Resp_format, Body) ->
case Resp_format of
@@ -1816,13 +1877,13 @@
do_reply(State, From, undefined, _, Resp_format, {ok, St_code, Headers, Body}) ->
Msg_1 = {ok, St_code, Headers, format_response_data(Resp_format, Body)},
gen_server:reply(From, Msg_1),
- report_request_complete(State);
+ dec_pipeline_counter(State);
do_reply(State, From, undefined, _, _, Msg) ->
gen_server:reply(From, Msg),
- report_request_complete(State);
+ dec_pipeline_counter(State);
do_reply(#state{prev_req_id = Prev_req_id} = State,
_From, StreamTo, ReqId, Resp_format, {ok, _, _, Body}) ->
- State_1 = report_request_complete(State),
+ State_1 = dec_pipeline_counter(State),
case Body of
[] ->
ok;
@@ -1841,10 +1902,10 @@
%% stream_once and sync requests on the same connection, it will
%% take a while for the req_id-pid mapping to get cleared, but it
%% should do no harm.
- ets:delete(?STREAM_TABLE, {req_id_pid, Prev_req_id}),
+ ets:delete(ibrowse_stream, {req_id_pid, Prev_req_id}),
State_1#state{prev_req_id = ReqId};
do_reply(State, _From, StreamTo, ReqId, Resp_format, Msg) ->
- State_1 = report_request_complete(State),
+ State_1 = dec_pipeline_counter(State),
Msg_1 = format_response_data(Resp_format, Msg),
catch StreamTo ! {ibrowse_async_response, ReqId, Msg_1},
State_1.
@@ -1859,7 +1920,7 @@
ReqList = queue:to_list(Reqs),
lists:foreach(fun(#request{from=From, stream_to=StreamTo, req_id=ReqId,
response_format = Resp_format}) ->
- ets:delete(?STREAM_TABLE, {req_id_pid, ReqId}),
+ ets:delete(ibrowse_stream, {req_id_pid, ReqId}),
do_reply(State, From, StreamTo, ReqId, Resp_format, {error, Err})
end, ReqList),
lists:foreach(
@@ -1944,15 +2005,30 @@
shutting_down(#state{lb_ets_tid = undefined}) ->
ok;
-shutting_down(#state{lb_ets_tid = Tid}) ->
- ibrowse_lb:report_connection_down(Tid).
+shutting_down(#state{lb_ets_tid = Tid,
+ cur_pipeline_size = _Sz}) ->
+ (catch ets:select_delete(Tid, [{{{'_', '_', '$1'},'_'},[{'==','$1',{const,self()}}],[true]}])).
-report_request_complete(#state{is_closing = true} = State) ->
+inc_pipeline_counter(#state{is_closing = true} = State) ->
State;
-report_request_complete(#state{lb_ets_tid = undefined} = State) ->
+inc_pipeline_counter(#state{lb_ets_tid = undefined} = State) ->
State;
-report_request_complete(#state{lb_ets_tid = Tid} = State) ->
- ibrowse_lb:report_request_complete(Tid),
+inc_pipeline_counter(#state{cur_pipeline_size = Pipe_sz} = State) ->
+ State#state{cur_pipeline_size = Pipe_sz + 1}.
+
+dec_pipeline_counter(#state{cur_pipeline_size = Pipe_sz,
+ lb_ets_tid = Tid,
+ proc_state = Proc_state} = State) when Tid /= undefined,
+ Proc_state /= ?dead_proc_walking ->
+ Ts = os:timestamp(),
+ catch ets:insert(Tid, {{Pipe_sz - 1, os:timestamp(), self()}, []}),
+ (catch ets:select_delete(Tid, [{{{'_', '$2', '$1'},'_'},
+ [{'==', '$1', {const,self()}},
+ {'<', '$2', {const,Ts}}
+ ],
+ [true]}])),
+ State#state{cur_pipeline_size = Pipe_sz - 1};
+dec_pipeline_counter(State) ->
State.
flatten([H | _] = L) when is_integer(H) ->
@@ -2038,3 +2114,6 @@
{value, {_, Val}} ->
Val
end.
+
+delayed_stop_timer() ->
+ erlang:send_after(500, self(), delayed_stop).
diff --git a/src/ibrowse_lb.erl b/src/ibrowse_lb.erl
index 794ba45..894d8ad 100644
--- a/src/ibrowse_lb.erl
+++ b/src/ibrowse_lb.erl
@@ -1,25 +1,23 @@
%%%-------------------------------------------------------------------
%%% File : ibrowse_lb.erl
%%% Author : chandru <chandrashekhar.mullaparthi@t-mobile.co.uk>
-%%% Benjamin Lee <http://github.com/benjaminplee>
-%%% Dan Schwabe <http://github.com/dfschwabe>
-%%% Brian Richards <http://github.com/richbria>
%%% Description :
%%%
%%% Created : 6 Mar 2008 by chandru <chandrashekhar.mullaparthi@t-mobile.co.uk>
%%%-------------------------------------------------------------------
-
-module(ibrowse_lb).
-author(chandru).
-behaviour(gen_server).
+%%--------------------------------------------------------------------
+%% Include files
+%%--------------------------------------------------------------------
+%%--------------------------------------------------------------------
%% External exports
-export([
start_link/1,
spawn_connection/6,
- stop/1,
- report_connection_down/1,
- report_request_complete/1
+ stop/1
]).
%% gen_server callbacks
@@ -33,18 +31,13 @@
]).
-record(state, {parent_pid,
- ets_tid,
- host,
- port,
- max_sessions,
- max_pipeline_size,
- proc_state}).
-
--define(PIPELINE_MAX, 99999).
--define(MAX_RETRIES, 3).
--define(KEY_MATCHSPEC_BY_PID(Pid), [{{{'_', '_', Pid}, '_'}, [], ['$_']}]).
--define(KEY_MATCHSPEC_BY_PID_FOR_DELETE(Pid), [{{{'_', '_', Pid}, '_'}, [], [true]}]).
--define(KEY_MATCHSPEC_FOR_DELETE(Key), [{{Key, '_'}, [], [true]}]).
+ ets_tid,
+ host,
+ port,
+ max_sessions,
+ max_pipeline_size,
+ proc_state
+ }).
-include("ibrowse.hrl").
@@ -58,37 +51,10 @@
start_link(Args) ->
gen_server:start_link(?MODULE, Args, []).
-spawn_connection(Lb_pid,
- Url,
- Max_sessions,
- Max_pipeline_size,
- SSL_options,
- Process_options)
- when is_pid(Lb_pid),
- is_record(Url, url),
- is_integer(Max_pipeline_size),
- is_integer(Max_sessions) ->
- gen_server:call(Lb_pid,
- {spawn_connection, Url, Max_sessions, Max_pipeline_size, SSL_options, Process_options}).
-
-stop(Lb_pid) ->
- case catch gen_server:call(Lb_pid, stop) of
- {'EXIT', {timeout, _}} ->
- exit(Lb_pid, kill);
- ok ->
- ok
- end.
-
-report_connection_down(Tid) ->
- %% Don't cascade errors since Tid is really managed by other process
- catch ets:select_delete(Tid, ?KEY_MATCHSPEC_BY_PID_FOR_DELETE(self())).
-
-report_request_complete(Tid) ->
- report_request_complete(Tid, ?MAX_RETRIES).
-
%%====================================================================
%% Server functions
%%====================================================================
+
%%--------------------------------------------------------------------
%% Function: init/1
%% Description: Initiates the server
@@ -99,19 +65,37 @@
%%--------------------------------------------------------------------
init([Host, Port]) ->
process_flag(trap_exit, true),
-
Max_sessions = ibrowse:get_config_value({max_sessions, Host, Port}, 10),
Max_pipe_sz = ibrowse:get_config_value({max_pipeline_size, Host, Port}, 10),
-
put(my_trace_flag, ibrowse_lib:get_trace_status(Host, Port)),
put(ibrowse_trace_token, ["LB: ", Host, $:, integer_to_list(Port)]),
+ State = #state{parent_pid = whereis(ibrowse),
+ host = Host,
+ port = Port,
+ max_pipeline_size = Max_pipe_sz,
+ max_sessions = Max_sessions},
+ State_1 = maybe_create_ets(State),
+ {ok, State_1}.
- {ok, #state{parent_pid = whereis(ibrowse),
- host = Host,
- port = Port,
- max_pipeline_size = Max_pipe_sz,
- max_sessions = Max_sessions}}.
+spawn_connection(Lb_pid, Url,
+ Max_sessions,
+ Max_pipeline_size,
+ SSL_options,
+ Process_options)
+ when is_pid(Lb_pid),
+ is_record(Url, url),
+ is_integer(Max_pipeline_size),
+ is_integer(Max_sessions) ->
+ gen_server:call(Lb_pid,
+ {spawn_connection, Url, Max_sessions, Max_pipeline_size, SSL_options, Process_options}).
+stop(Lb_pid) ->
+ case catch gen_server:call(Lb_pid, stop) of
+ {'EXIT', {timeout, _}} ->
+ exit(Lb_pid, kill);
+ ok ->
+ ok
+ end.
%%--------------------------------------------------------------------
%% Function: handle_call/3
%% Description: Handling call messages
@@ -122,29 +106,36 @@
%% {stop, Reason, Reply, State} | (terminate/2 is called)
%% {stop, Reason, State} (terminate/2 is called)
%%--------------------------------------------------------------------
+
handle_call(stop, _From, #state{ets_tid = undefined} = State) ->
gen_server:reply(_From, ok),
{stop, normal, State};
+
handle_call(stop, _From, #state{ets_tid = Tid} = State) ->
- for_each_connection_pid(Tid, fun(Pid) -> ibrowse_http_client:stop(Pid) end),
+ stop_all_conn_procs(Tid),
gen_server:reply(_From, ok),
{stop, normal, State};
handle_call(_, _From, #state{proc_state = shutting_down} = State) ->
{reply, {error, shutting_down}, State};
-handle_call({spawn_connection, Url, Max_sess, Max_pipe, SSL_options, Process_options}, _From, State) ->
- State_1 = maybe_create_ets(State),
- Tid = State_1#state.ets_tid,
- Reply = case num_current_connections(Tid) of
- X when X >= Max_sess ->
- find_best_connection(Tid, Max_pipe);
- _ ->
- Result = {ok, Pid} = ibrowse_http_client:start_link({Tid, Url, SSL_options}, Process_options),
- record_new_connection(Tid, Pid),
- Result
- end,
- {reply, Reply, State_1#state{max_sessions = Max_sess, max_pipeline_size = Max_pipe}};
+handle_call({spawn_connection, Url, Max_sess, Max_pipe, SSL_options, Process_options}, _From,
+ State) ->
+ State_1 = maybe_create_ets(State),
+ Tid = State_1#state.ets_tid,
+ Tid_size = ets:info(Tid, size),
+ case Tid_size >= Max_sess of
+ true ->
+ Reply = find_best_connection(Tid, Max_pipe),
+ {reply, Reply, State_1#state{max_sessions = Max_sess,
+ max_pipeline_size = Max_pipe}};
+ false ->
+ {ok, Pid} = ibrowse_http_client:start({Tid, Url, SSL_options}, Process_options),
+ Ts = os:timestamp(),
+ ets:insert(Tid, {{1, Ts, Pid}, []}),
+ {reply, {ok, {1, Ts, Pid}}, State_1#state{max_sessions = Max_sess,
+ max_pipeline_size = Max_pipe}}
+ end;
handle_call(Request, _From, State) ->
Reply = {unknown_request, Request},
@@ -167,22 +158,27 @@
%% {noreply, State, Timeout} |
%% {stop, Reason, State} (terminate/2 is called)
%%--------------------------------------------------------------------
-handle_info({'EXIT', Parent, _Reason}, #state{parent_pid = Parent} = State) ->
- {stop, normal, State};
+
handle_info({trace, Bool}, #state{ets_tid = undefined} = State) ->
put(my_trace_flag, Bool),
{noreply, State};
+
handle_info({trace, Bool}, #state{ets_tid = Tid} = State) ->
- for_each_connection_pid(Tid, fun(Pid) -> ibrowse_http_client:trace(Pid, Bool) end),
+ ets:foldl(fun({{_, Pid}, _}, Acc) when is_pid(Pid) ->
+ catch Pid ! {trace, Bool},
+ Acc;
+ (_, Acc) ->
+ Acc
+ end, undefined, Tid),
put(my_trace_flag, Bool),
{noreply, State};
handle_info(timeout, State) ->
%% We can't shutdown the process immediately because a request
%% might be in flight. So we first remove the entry from the
- %% load balancer named ets table, and then shutdown a couple
- %% of seconds later
- ets:delete(?LOAD_BALANCER_NAMED_TABLE, {State#state.host, State#state.port}),
+ %% ibrowse_lb ets table, and then shutdown a couple of seconds
+ %% later
+ ets:delete(ibrowse_lb, {State#state.host, State#state.port}),
erlang:send_after(2000, self(), shutdown),
{noreply, State#state{proc_state = shutting_down}};
@@ -197,12 +193,17 @@
%% Description: Shutdown the server
%% Returns: any (ignored by gen_server)
%%--------------------------------------------------------------------
-terminate(_Reason, #state{host = Host, port = Port}) ->
- % Use delete_object instead of delete in case another process for this host/port
- % has been spawned, in which case will be deleting the wrong record because pid won't match.
- ets:delete_object(?LOAD_BALANCER_NAMED_TABLE, #lb_pid{host_port = {Host, Port}, pid = self()}),
+terminate(_Reason, #state{host = Host, port = Port, ets_tid = Tid} = _State) ->
+ catch ets:delete(ibrowse_lb, {Host, Port}),
+ stop_all_conn_procs(Tid),
ok.
+stop_all_conn_procs(Tid) ->
+ ets:foldl(fun({{_, _, Pid}, _}, Acc) ->
+ ibrowse_http_client:stop(Pid),
+ Acc
+ end, [], Tid).
+
%%--------------------------------------------------------------------
%% Func: code_change/3
%% Purpose: Convert process state when code is changed
@@ -214,73 +215,19 @@
%%--------------------------------------------------------------------
%%% Internal functions
%%--------------------------------------------------------------------
-find_best_connection(Tid, Max_pipeline_size) ->
- find_best_connection(Tid, Max_pipeline_size, ?MAX_RETRIES).
-
-find_best_connection(_Tid, _Max_pipeline_size, 0) ->
- {error, retry_later};
-find_best_connection(Tid, Max_pipeline_size, RemainingRetries) ->
+find_best_connection(Tid, Max_pipe) ->
case ets:first(Tid) of
- {Size, _Timestamp, Pid} = Key when Size < Max_pipeline_size ->
- case record_request_for_connection(Tid, Key) of
- true ->
- {ok, Pid};
- false ->
- find_best_connection(Tid, Max_pipeline_size, RemainingRetries - 1)
- end;
- _ ->
+ {Spec_size, Ts, Pid} = First when Spec_size < Max_pipe ->
+ ets:delete(Tid, First),
+ ets:insert(Tid, {{Spec_size + 1, Ts, Pid}, []}),
+ {ok, First};
+ _ ->
{error, retry_later}
end.
-maybe_create_ets(#state{ets_tid = undefined} = State) ->
- Tid = ets:new(?CONNECTIONS_LOCAL_TABLE, [public, ordered_set]),
+maybe_create_ets(#state{ets_tid = undefined, host = Host, port = Port} = State) ->
+ Tid = ets:new(ibrowse_lb, [public, ordered_set]),
+ ets:insert(ibrowse_lb, #lb_pid{host_port = {Host, Port}, pid = self(), ets_tid = Tid}),
State#state{ets_tid = Tid};
maybe_create_ets(State) ->
State.
-
-%% Ets connection table utility methods
-num_current_connections(Tid) ->
- catch ets:info(Tid, size).
-
-record_new_connection(Tid, Pid) ->
- catch ets:insert(Tid, {new_key(Pid), undefined}).
-
-record_request_for_connection(Tid, Key) ->
- case ets:select_delete(Tid, ?KEY_MATCHSPEC_FOR_DELETE(Key)) of
- 1 ->
- ets:insert(Tid, {incremented(Key), undefined}),
- true;
- _ ->
- false
- end.
-
-report_request_complete(_Tid, 0) ->
- false;
-report_request_complete(Tid, RemainingRetries) ->
- %% Don't cascade errors since Tid is really managed by other process
- catch case ets:select(Tid, ?KEY_MATCHSPEC_BY_PID(self())) of
- [{MatchKey, _}] ->
- case ets:select_delete(Tid, ?KEY_MATCHSPEC_FOR_DELETE(MatchKey)) of
- 1 ->
- ets:insert(Tid, {decremented(MatchKey), undefined}),
- true;
- _ ->
- report_request_complete(Tid, RemainingRetries - 1)
- end;
- _ ->
- false
- end.
-
-
-new_key(Pid) ->
- {1, os:timestamp(), Pid}.
-
-incremented({Size, Timestamp, Pid}) ->
- {Size + 1, Timestamp, Pid}.
-
-decremented({Size, _Timestamp, Pid}) ->
- {Size - 1, os:timestamp(), Pid}.
-
-for_each_connection_pid(Tid, Fun) ->
- ets:foldl(fun({{_, _, Pid}, _}, _) -> Fun(Pid) end, undefined, Tid),
- ok.
diff --git a/src/ibrowse_lib.erl b/src/ibrowse_lib.erl
index 3362b39..1f0a61a 100644
--- a/src/ibrowse_lib.erl
+++ b/src/ibrowse_lib.erl
@@ -28,7 +28,8 @@
get_value/2,
get_value/3,
parse_url/1,
- printable_date/0
+ printable_date/0,
+ printable_date/1
]).
get_trace_status(Host, Port) ->
@@ -368,8 +369,11 @@
default_port(ftp) -> 21.
printable_date() ->
- {{Y,Mo,D},{H, M, S}} = calendar:local_time(),
- {_,_,MicroSecs} = now(),
+ printable_date(os:timestamp()).
+
+printable_date(Now) ->
+ {{Y,Mo,D},{H, M, S}} = calendar:now_to_local_time(Now),
+ {_,_,MicroSecs} = Now,
[integer_to_list(Y),
$-,
integer_to_list(Mo),
diff --git a/test/ibrowse_functional_tests.erl b/test/ibrowse_functional_tests.erl
index e55c5b2..3517011 100644
--- a/test/ibrowse_functional_tests.erl
+++ b/test/ibrowse_functional_tests.erl
@@ -17,6 +17,8 @@
-define(LONG_TIMEOUT_MS, 30000).
-define(PAUSE_FOR_CONNECTIONS_MS, 2000).
+-compile(export_all).
+
setup() ->
application:start(crypto),
application:start(public_key),
@@ -141,6 +143,7 @@
timer:sleep(?PAUSE_FOR_CONNECTIONS_MS), %% Wait for everyone to get in line
+ ibrowse:show_dest_status("localhost", 8181),
Counts = [Count || {_Pid, Count} <- ibrowse_test_server:get_conn_pipeline_depth()],
?assertEqual(MaxSessions, length(Counts)),
diff --git a/test/ibrowse_load_test.erl b/test/ibrowse_load_test.erl
new file mode 100644
index 0000000..5ff308e
--- /dev/null
+++ b/test/ibrowse_load_test.erl
@@ -0,0 +1,181 @@
+-module(ibrowse_load_test).
+-compile(export_all).
+
+-define(ibrowse_load_test_counters, ibrowse_load_test_counters).
+
+start(Num_workers, Num_requests, Max_sess) ->
+ proc_lib:spawn(fun() ->
+ start_1(Num_workers, Num_requests, Max_sess)
+ end).
+
+query_state() ->
+ ibrowse_load_test ! query_state.
+
+shutdown() ->
+ ibrowse_load_test ! shutdown.
+
+start_1(Num_workers, Num_requests, Max_sess) ->
+ register(ibrowse_load_test, self()),
+ application:start(ibrowse),
+ application:set_env(ibrowse, inactivity_timeout, 5000),
+ Ulimit = os:cmd("ulimit -n"),
+ case catch list_to_integer(string:strip(Ulimit, right, $\n)) of
+ X when is_integer(X), X > 3000 ->
+ ok;
+ X ->
+ io:format("Load test not starting. {insufficient_value_for_ulimit, ~p}~n", [X]),
+ exit({insufficient_value_for_ulimit, X})
+ end,
+ ets:new(?ibrowse_load_test_counters, [named_table, public]),
+ ets:new(ibrowse_load_timings, [named_table, public]),
+ try
+ ets:insert(?ibrowse_load_test_counters, [{success, 0},
+ {failed, 0},
+ {timeout, 0},
+ {retry_later, 0},
+ {one_request_only, 0}
+ ]),
+ ibrowse:set_max_sessions("localhost", 8081, Max_sess),
+ Start_time = now(),
+ Workers = spawn_workers(Num_workers, Num_requests),
+ erlang:send_after(1000, self(), print_diagnostics),
+ ok = wait_for_workers(Workers),
+ End_time = now(),
+ Time_in_secs = trunc(round(timer:now_diff(End_time, Start_time) / 1000000)),
+ Req_count = Num_workers * Num_requests,
+ [{_, Success_count}] = ets:lookup(?ibrowse_load_test_counters, success),
+ case Success_count == Req_count of
+ true ->
+ io:format("Test success. All requests succeeded~n", []);
+ false when Success_count > 0 ->
+ io:format("Test failed. Some successes~n", []);
+ false ->
+ io:format("Test failed. ALL requests FAILED~n", [])
+ end,
+ case Time_in_secs > 0 of
+ true ->
+ io:format("Reqs/sec achieved : ~p~n", [trunc(round(Success_count / Time_in_secs))]);
+ false ->
+ ok
+ end,
+ io:format("Load test results:~n~p~n", [ets:tab2list(?ibrowse_load_test_counters)]),
+ io:format("Timings: ~p~n", [calculate_timings()])
+ catch Err ->
+ io:format("Err: ~p~n", [Err])
+ after
+ ets:delete(?ibrowse_load_test_counters),
+ ets:delete(ibrowse_load_timings),
+ unregister(ibrowse_load_test)
+ end.
+
+calculate_timings() ->
+ {Max, Min, Mean} = get_mmv(ets:first(ibrowse_load_timings), {0, 9999999, 0}),
+ Variance = trunc(round(ets:foldl(fun({_, X}, X_acc) ->
+ (X - Mean)*(X-Mean) + X_acc
+ end, 0, ibrowse_load_timings) / ets:info(ibrowse_load_timings, size))),
+ Std_dev = trunc(round(math:sqrt(Variance))),
+ {ok, [{max, Max},
+ {min, Min},
+ {mean, Mean},
+ {variance, Variance},
+ {standard_deviation, Std_dev}]}.
+
+get_mmv('$end_of_table', {Max, Min, Total}) ->
+ Mean = trunc(round(Total / ets:info(ibrowse_load_timings, size))),
+ {Max, Min, Mean};
+get_mmv(Key, {Max, Min, Total}) ->
+ [{_, V}] = ets:lookup(ibrowse_load_timings, Key),
+ get_mmv(ets:next(ibrowse_load_timings, Key), {max(Max, V), min(Min, V), Total + V}).
+
+
+spawn_workers(Num_w, Num_r) ->
+ spawn_workers(Num_w, Num_r, self(), []).
+
+spawn_workers(0, _Num_requests, _Parent, Acc) ->
+ lists:reverse(Acc);
+spawn_workers(Num_workers, Num_requests, Parent, Acc) ->
+ Pid_ref = spawn_monitor(fun() ->
+ random:seed(now()),
+ case catch worker_loop(Parent, Num_requests) of
+ {'EXIT', Rsn} ->
+ io:format("Worker crashed with reason: ~p~n", [Rsn]);
+ _ ->
+ ok
+ end
+ end),
+ spawn_workers(Num_workers - 1, Num_requests, Parent, [Pid_ref | Acc]).
+
+wait_for_workers([]) ->
+ ok;
+wait_for_workers([{Pid, Pid_ref} | T] = Pids) ->
+ receive
+ {done, Pid} ->
+ wait_for_workers(T);
+ {done, Some_pid} ->
+ wait_for_workers([{Pid, Pid_ref} | lists:keydelete(Some_pid, 1, T)]);
+ print_diagnostics ->
+ io:format("~1000.p~n", [ibrowse:get_metrics()]),
+ erlang:send_after(1000, self(), print_diagnostics),
+ wait_for_workers(Pids);
+ query_state ->
+ io:format("Waiting for ~p~n", [Pids]),
+ wait_for_workers(Pids);
+ shutdown ->
+ io:format("Shutting down on command. Still waiting for ~p workers~n", [length(Pids)]);
+ {'DOWN', _, process, _, normal} ->
+ wait_for_workers(Pids);
+ {'DOWN', _, process, Down_pid, Rsn} ->
+ io:format("Worker ~p died. Reason: ~p~n", [Down_pid, Rsn]),
+ wait_for_workers(lists:keydelete(Down_pid, 1, Pids));
+ X ->
+ io:format("Recvd unknown msg: ~p~n", [X]),
+ wait_for_workers(Pids)
+ end.
+
+worker_loop(Parent, 0) ->
+ Parent ! {done, self()};
+worker_loop(Parent, N) ->
+ Delay = random:uniform(100),
+ Url = case Delay rem 10 of
+ %% Change 10 to some number between 0-9 depending on how
+ %% much chaos you want to introduce into the server
+ %% side. The higher the number, the more often the
+ %% server will close a connection after serving the
+ %% first request, thereby forcing the client to
+ %% retry. Any number of 10 or higher will disable this
+ %% chaos mechanism
+ 10 ->
+ ets:update_counter(?ibrowse_load_test_counters, one_request_only, 1),
+ "http://localhost:8081/ibrowse_handle_one_request_only";
+ _ ->
+ "http://localhost:8081/blah"
+ end,
+ Start_time = now(),
+ Res = ibrowse:send_req(Url, [], get),
+ End_time = now(),
+ Time_taken = trunc(round(timer:now_diff(End_time, Start_time) / 1000)),
+ ets:insert(ibrowse_load_timings, {now(), Time_taken}),
+ case Res of
+ {ok, "200", _, _} ->
+ ets:update_counter(?ibrowse_load_test_counters, success, 1);
+ {error, req_timedout} ->
+ ets:update_counter(?ibrowse_load_test_counters, timeout, 1);
+ {error, retry_later} ->
+ ets:update_counter(?ibrowse_load_test_counters, retry_later, 1);
+ {error, Reason} ->
+ update_unknown_counter(Reason, 1);
+ _ ->
+ io:format("~p -- Res: ~p~n", [self(), Res]),
+ ets:update_counter(?ibrowse_load_test_counters, failed, 1)
+ end,
+ timer:sleep(Delay),
+ worker_loop(Parent, N - 1).
+
+update_unknown_counter(Counter, Inc_val) ->
+ case catch ets:update_counter(?ibrowse_load_test_counters, Counter, Inc_val) of
+ {'EXIT', _} ->
+ ets:insert_new(?ibrowse_load_test_counters, {Counter, 0}),
+ update_unknown_counter(Counter, Inc_val);
+ _ ->
+ ok
+ end.
diff --git a/test/ibrowse_test.erl b/test/ibrowse_test.erl
index 4ddb9c1..0787493 100644
--- a/test/ibrowse_test.erl
+++ b/test/ibrowse_test.erl
@@ -8,9 +8,10 @@
load_test/3,
send_reqs_1/3,
do_send_req/2,
+ local_unit_tests/0,
unit_tests/0,
- unit_tests/1,
- unit_tests_1/2,
+ unit_tests/2,
+ unit_tests_1/3,
ue_test/0,
ue_test/1,
verify_chunked_streaming/0,
@@ -34,9 +35,13 @@
test_303_response_with_a_body/1,
test_binary_headers/0,
test_binary_headers/1,
- test_generate_body_0/0
+ test_generate_body_0/0,
+ test_retry_of_requests/0,
+ test_retry_of_requests/1
]).
+-include_lib("ibrowse/include/ibrowse.hrl").
+
test_stream_once(Url, Method, Options) ->
test_stream_once(Url, Method, Options, 5000).
@@ -90,6 +95,8 @@
ets:new(pid_table, [named_table, public]),
ets:new(ibrowse_test_results, [named_table, public]),
ets:new(ibrowse_errors, [named_table, public, ordered_set]),
+ ets:new(ibrowse_counter, [named_table, public, ordered_set]),
+ ets:insert(ibrowse_counter, {req_id, 1}),
init_results(),
process_flag(trap_exit, true),
log_msg("Starting spawning of workers...~n", []),
@@ -207,56 +214,65 @@
%%------------------------------------------------------------------------------
%% Unit Tests
%%------------------------------------------------------------------------------
+-define(LOCAL_TESTS, [
+ {local_test_fun, test_20122010, []},
+ {local_test_fun, test_pipeline_head_timeout, []},
+ {local_test_fun, test_head_transfer_encoding, []},
+ {local_test_fun, test_head_response_with_body, []},
+ {local_test_fun, test_303_response_with_a_body, []},
+ {local_test_fun, test_binary_headers, []},
+ {local_test_fun, test_retry_of_requests, []}
+ ]).
+
-define(TEST_LIST, [{"http://intranet/messenger", get},
- {"http://www.google.co.uk", get},
- {"http://www.google.com", get},
- {"http://www.google.com", options},
- {"https://mail.google.com", get},
- {"http://www.sun.com", get},
- {"http://www.oracle.com", get},
- {"http://www.bbc.co.uk", get},
- {"http://www.bbc.co.uk", trace},
- {"http://www.bbc.co.uk", options},
- {"http://yaws.hyber.org", get},
- {"http://jigsaw.w3.org/HTTP/ChunkedScript", get},
- {"http://jigsaw.w3.org/HTTP/TE/foo.txt", get},
- {"http://jigsaw.w3.org/HTTP/TE/bar.txt", get},
- {"http://jigsaw.w3.org/HTTP/connection.html", get},
- {"http://jigsaw.w3.org/HTTP/cc.html", get},
- {"http://jigsaw.w3.org/HTTP/cc-private.html", get},
- {"http://jigsaw.w3.org/HTTP/cc-proxy-revalidate.html", get},
- {"http://jigsaw.w3.org/HTTP/cc-nocache.html", get},
- {"http://jigsaw.w3.org/HTTP/h-content-md5.html", get},
- {"http://jigsaw.w3.org/HTTP/h-retry-after.html", get},
- {"http://jigsaw.w3.org/HTTP/h-retry-after-date.html", get},
- {"http://jigsaw.w3.org/HTTP/neg", get},
- {"http://jigsaw.w3.org/HTTP/negbad", get},
- {"http://jigsaw.w3.org/HTTP/400/toolong/", get},
- {"http://jigsaw.w3.org/HTTP/300/", get},
- {"http://jigsaw.w3.org/HTTP/Basic/", get, [{basic_auth, {"guest", "guest"}}]},
- {"http://jigsaw.w3.org/HTTP/CL/", get},
- {"http://www.httpwatch.com/httpgallery/chunked/", get},
- {"https://github.com", get, [{ssl_options, [{depth, 2}]}]},
- {local_test_fun, test_20122010, []},
- {local_test_fun, test_pipeline_head_timeout, []},
- {local_test_fun, test_head_transfer_encoding, []},
- {local_test_fun, test_head_response_with_body, []},
- {local_test_fun, test_303_response_with_a_body, []},
- {local_test_fun, test_binary_headers, []}
- ]).
+ {"http://www.google.co.uk", get},
+ {"http://www.google.com", get},
+ {"http://www.google.com", options},
+ {"https://mail.google.com", get},
+ {"http://www.sun.com", get},
+ {"http://www.oracle.com", get},
+ {"http://www.bbc.co.uk", get},
+ {"http://www.bbc.co.uk", trace},
+ {"http://www.bbc.co.uk", options},
+ {"http://yaws.hyber.org", get},
+ {"http://jigsaw.w3.org/HTTP/ChunkedScript", get},
+ {"http://jigsaw.w3.org/HTTP/TE/foo.txt", get},
+ {"http://jigsaw.w3.org/HTTP/TE/bar.txt", get},
+ {"http://jigsaw.w3.org/HTTP/connection.html", get},
+ {"http://jigsaw.w3.org/HTTP/cc.html", get},
+ {"http://jigsaw.w3.org/HTTP/cc-private.html", get},
+ {"http://jigsaw.w3.org/HTTP/cc-proxy-revalidate.html", get},
+ {"http://jigsaw.w3.org/HTTP/cc-nocache.html", get},
+ {"http://jigsaw.w3.org/HTTP/h-content-md5.html", get},
+ {"http://jigsaw.w3.org/HTTP/h-retry-after.html", get},
+ {"http://jigsaw.w3.org/HTTP/h-retry-after-date.html", get},
+ {"http://jigsaw.w3.org/HTTP/neg", get},
+ {"http://jigsaw.w3.org/HTTP/negbad", get},
+ {"http://jigsaw.w3.org/HTTP/400/toolong/", get},
+ {"http://jigsaw.w3.org/HTTP/300/", get},
+ {"http://jigsaw.w3.org/HTTP/Basic/", get, [{basic_auth, {"guest", "guest"}}]},
+ {"http://jigsaw.w3.org/HTTP/CL/", get},
+ {"http://www.httpwatch.com/httpgallery/chunked/", get},
+ {"https://github.com", get, [{ssl_options, [{depth, 2}]}]}
+ ] ++ ?LOCAL_TESTS).
+
+local_unit_tests() ->
+ unit_tests([], ?LOCAL_TESTS).
unit_tests() ->
- unit_tests([]).
+ error_logger:tty(false),
+ unit_tests([], ?TEST_LIST),
+ error_logger:tty(true).
-unit_tests(Options) ->
+unit_tests(Options, Test_list) ->
application:start(crypto),
application:start(public_key),
- application:start(ssl),
+ application:ensure_all_started(ssl),
(catch ibrowse_test_server:start_server(8181, tcp)),
- ibrowse:start(),
+ application:start(ibrowse),
Options_1 = Options ++ [{connect_timeout, 5000}],
Test_timeout = proplists:get_value(test_timeout, Options, 60000),
- {Pid, Ref} = erlang:spawn_monitor(?MODULE, unit_tests_1, [self(), Options_1]),
+ {Pid, Ref} = erlang:spawn_monitor(?MODULE, unit_tests_1, [self(), Options_1, Test_list]),
receive
{done, Pid} ->
ok;
@@ -269,14 +285,14 @@
catch ibrowse_test_server:stop_server(8181),
ok.
-unit_tests_1(Parent, Options) ->
+unit_tests_1(Parent, Options, Test_list) ->
lists:foreach(fun({local_test_fun, Fun_name, Args}) ->
execute_req(local_test_fun, Fun_name, Args);
({Url, Method}) ->
execute_req(Url, Method, Options);
({Url, Method, X_Opts}) ->
execute_req(Url, Method, X_Opts ++ Options)
- end, ?TEST_LIST),
+ end, Test_list),
Parent ! {done, self()}.
verify_chunked_streaming() ->
@@ -371,6 +387,8 @@
{'EXIT', Reason};
{'DOWN', _, _, _, _} ->
wait_for_resp(Pid);
+ {'EXIT', _, normal} ->
+ wait_for_resp(Pid);
Msg ->
io:format("Recvd unknown message: ~p~n", [Msg]),
wait_for_resp(Pid)
@@ -425,6 +443,7 @@
end.
execute_req(local_test_fun, Method, Args) ->
+ reset_ibrowse(),
io:format(" ~-54.54w: ", [Method]),
Result = (catch apply(?MODULE, Method, Args)),
io:format("~p~n", [Result]);
@@ -539,6 +558,74 @@
end.
%%------------------------------------------------------------------------------
+%% Test that retry of requests happens correctly, and that ibrowse doesn't retry
+%% if there is not enough time left
+%%------------------------------------------------------------------------------
+test_retry_of_requests() ->
+ clear_msg_q(),
+ test_retry_of_requests("http://localhost:8181/ibrowse_handle_one_request_only_with_delay").
+
+test_retry_of_requests(Url) ->
+ reset_ibrowse(),
+ Timeout_1 = 2050,
+ Res_1 = test_retry_of_requests(Url, Timeout_1),
+ case lists:filter(fun({_Pid, {ok, "200", _, _}}) ->
+ true;
+ (_) -> false
+ end, Res_1) of
+ [_|_] = X ->
+ Res_1_1 = Res_1 -- X,
+ case lists:all(
+ fun({_Pid, {error, retry_later}}) ->
+ true;
+ (_) ->
+ false
+ end, Res_1_1) of
+ true ->
+ ok;
+ false ->
+ exit({failed, Timeout_1, Res_1})
+ end;
+ _ ->
+ exit({failed, Timeout_1, Res_1})
+ end,
+ Timeout_2 = 2200,
+ Res_2 = test_retry_of_requests(Url, Timeout_2),
+ case lists:filter(fun({_Pid, {ok, "200", _, _}}) ->
+ true;
+ (_) -> false
+ end, Res_2) of
+ [_|_] = Res_2_X ->
+ Res_2_1 = Res_2 -- Res_2_X,
+ case lists:all(
+ fun({_Pid, {error, X_err_2}}) ->
+ (X_err_2 == retry_later) orelse (X_err_2 == req_timedout);
+ (_) ->
+ false
+ end, Res_2_1) of
+ true ->
+ ok;
+ false ->
+ exit({failed, {?MODULE, ?LINE}, Timeout_2, Res_2})
+ end;
+ _ ->
+ exit({failed, {?MODULE, ?LINE}, Timeout_2, Res_2})
+ end,
+ success.
+
+test_retry_of_requests(Url, Timeout) ->
+ #url{host = Host, port = Port} = ibrowse_lib:parse_url(Url),
+ ibrowse:set_max_sessions(Host, Port, 1),
+ Parent = self(),
+ Pids = lists:map(fun(_) ->
+ spawn(fun() ->
+ Res = (catch ibrowse:send_req(Url, [], get, [], [], Timeout)),
+ Parent ! {self(), Res}
+ end)
+ end, lists:seq(1,10)),
+ accumulate_worker_resp(Pids).
+
+%%------------------------------------------------------------------------------
%% Test what happens when the request at the head of a pipeline times out
%%------------------------------------------------------------------------------
test_pipeline_head_timeout() ->
@@ -547,22 +634,27 @@
test_pipeline_head_timeout(Url) ->
{ok, Pid} = ibrowse:spawn_worker_process(Url),
+ Fixed_timeout = 2000,
Test_parent = self(),
Fun = fun({fixed, Timeout}) ->
- spawn(fun() ->
- do_test_pipeline_head_timeout(Url, Pid, Test_parent, Timeout)
- end);
- (Timeout_mult) ->
- spawn(fun() ->
- Timeout = 1000 + Timeout_mult*1000,
- do_test_pipeline_head_timeout(Url, Pid, Test_parent, Timeout)
- end)
- end,
- Pids = [Fun(X) || X <- [{fixed, 32000} | lists:seq(1,10)]],
+ X_pid = spawn(fun() ->
+ do_test_pipeline_head_timeout(Url, Pid, Test_parent, Timeout)
+ end),
+ %% io:format("Pid ~p with a fixed timeout~n", [X_pid]),
+ X_pid;
+ (Timeout_mult) ->
+ Timeout = Fixed_timeout + Timeout_mult*1000,
+ X_pid = spawn(fun() ->
+ do_test_pipeline_head_timeout(Url, Pid, Test_parent, Timeout)
+ end),
+ %% io:format("Pid ~p with a timeout of ~p~n", [X_pid, Timeout]),
+ X_pid
+ end,
+ Pids = [Fun(X) || X <- [{fixed, Fixed_timeout} | lists:seq(1,10)]],
Result = accumulate_worker_resp(Pids),
case lists:all(fun({_, X_res}) ->
- X_res == {error,req_timedout}
- end, Result) of
+ (X_res == {error,req_timedout}) orelse (X_res == {error, connection_closed})
+ end, Result) of
true ->
success;
false ->
@@ -725,3 +817,7 @@
io:format("~s -- " ++ Fmt, [ibrowse_lib:printable_date() | Args]);
do_trace(_, _, _) ->
ok.
+
+reset_ibrowse() ->
+ application:stop(ibrowse),
+ application:start(ibrowse).
diff --git a/test/ibrowse_test_server.erl b/test/ibrowse_test_server.erl
index dc0d7e2..7025286 100644
--- a/test/ibrowse_test_server.erl
+++ b/test/ibrowse_test_server.erl
@@ -21,26 +21,30 @@
start_server(Port, Sock_type) ->
Fun = fun() ->
- Name = server_proc_name(Port),
- register(Name, self()),
- ets:new(?CONN_PIPELINE_DEPTH, [named_table, public, set]),
- case do_listen(Sock_type, Port, [{active, false},
- {reuseaddr, true},
- {nodelay, true},
- {packet, http}]) of
- {ok, Sock} ->
- do_trace("Server listening on port: ~p~n", [Port]),
- accept_loop(Sock, Sock_type);
- Err ->
- erlang:error(
- lists:flatten(
- io_lib:format(
- "Failed to start server on port ~p. ~p~n",
- [Port, Err]))),
- exit({listen_error, Err})
- end,
- unregister(Name)
- end,
+ Proc_name = server_proc_name(Port),
+ case whereis(Proc_name) of
+ undefined ->
+ register(Proc_name, self()),
+ ets:new(?CONN_PIPELINE_DEPTH, [named_table, public, set]),
+ case do_listen(Sock_type, Port, [{active, false},
+ {reuseaddr, true},
+ {nodelay, true},
+ {packet, http}]) of
+ {ok, Sock} ->
+ do_trace("Server listening on port: ~p~n", [Port]),
+ accept_loop(Sock, Sock_type);
+ Err ->
+ erlang:error(
+ lists:flatten(
+ io_lib:format(
+ "Failed to start server on port ~p. ~p~n",
+ [Port, Err]))),
+ exit({listen_error, Err})
+ end;
+ _X ->
+ ok
+ end
+ end,
spawn_link(Fun).
stop_server(Port) ->
@@ -105,7 +109,7 @@
server_loop(Sock, Sock_type, #request{headers = Headers} = Req) ->
receive
{http, Sock, {http_request, HttpMethod, HttpUri, HttpVersion}} ->
- ets:update_counter(?CONN_PIPELINE_DEPTH, self(), 1),
+ catch ets:update_counter(?CONN_PIPELINE_DEPTH, self(), 1),
server_loop(Sock, Sock_type, Req#request{method = HttpMethod,
uri = HttpUri,
version = HttpVersion});
@@ -113,16 +117,18 @@
server_loop(Sock, Sock_type, Req#request{headers = [H | Headers]});
{http, Sock, http_eoh} ->
case process_request(Sock, Sock_type, Req) of
+ close_connection ->
+ gen_tcp:shutdown(Sock, read_write);
not_done ->
ok;
_ ->
- ets:update_counter(?CONN_PIPELINE_DEPTH, self(), -1)
+ catch ets:update_counter(?CONN_PIPELINE_DEPTH, self(), -1)
end,
server_loop(Sock, Sock_type, #request{});
{http, Sock, {http_error, Err}} ->
- do_trace("Error parsing HTTP request:~n"
- "Req so far : ~p~n"
- "Err : ", [Req, Err]),
+ io:format("Error parsing HTTP request:~n"
+ "Req so far : ~p~n"
+ "Err : ~p", [Req, Err]),
exit({http_error, Err});
{setopts, Opts} ->
setopts(Sock, Sock_type, Opts),
@@ -131,9 +137,9 @@
do_trace("Client closed connection~n", []),
ok;
Other ->
- do_trace("Recvd unknown msg: ~p~n", [Other]),
+ io:format("Recvd unknown msg: ~p~n", [Other]),
exit({unknown_msg, Other})
- after 5000 ->
+ after 120000 ->
do_trace("Timing out client connection~n", []),
ok
end.
@@ -172,7 +178,7 @@
headers = _Headers,
uri = {abs_path, "/ibrowse_inac_timeout_test"}} = Req) ->
do_trace("Recvd req: ~p. Sleeping for 30 secs...~n", [Req]),
- timer:sleep(30000),
+ timer:sleep(3000),
do_trace("...Sending response now.~n", []),
Resp = <<"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n">>,
do_send(Sock, Sock_type, Resp);
@@ -204,7 +210,7 @@
#request{method='HEAD',
headers = _Headers,
uri = {abs_path, "/ibrowse_head_test"}}) ->
- Resp = <<"HTTP/1.1 200 OK\r\nServer: Apache-Coyote/1.1\r\nTransfer-Encoding: chunked\r\nDate: Wed, 04 Apr 2012 16:53:49 GMT\r\nConnection: close\r\n\r\n">>,
+ Resp = <<"HTTP/1.1 200 OK\r\nServer: Apache-Coyote/1.1\r\Date: Wed, 04 Apr 2012 16:53:49 GMT\r\nConnection: close\r\n\r\n">>,
do_send(Sock, Sock_type, Resp);
process_request(Sock, Sock_type,
#request{method='POST',
@@ -218,12 +224,28 @@
uri = {abs_path, "/ibrowse_303_with_body_test"}}) ->
Resp = <<"HTTP/1.1 303 See Other\r\nLocation: http://example.org\r\nContent-Length: 5\r\n\r\nabcde">>,
do_send(Sock, Sock_type, Resp);
+process_request(Sock, Sock_type,
+ #request{method='GET',
+ headers = _Headers,
+ uri = {abs_path, "/ibrowse_handle_one_request_only_with_delay"}}) ->
+ timer:sleep(2000),
+ Resp = <<"HTTP/1.1 200 OK\r\nServer: Apache-Coyote/1.1\r\nDate: Wed, 04 Apr 2012 16:53:49 GMT\r\nConnection: close\r\n\r\n">>,
+ do_send(Sock, Sock_type, Resp),
+ close_connection;
+process_request(Sock, Sock_type,
+ #request{method='GET',
+ headers = _Headers,
+ uri = {abs_path, "/ibrowse_handle_one_request_only"}}) ->
+ Resp = <<"HTTP/1.1 200 OK\r\nServer: Apache-Coyote/1.1\r\nDate: Wed, 04 Apr 2012 16:53:49 GMT\r\nConnection: close\r\n\r\n">>,
+ do_send(Sock, Sock_type, Resp),
+ close_connection;
process_request(_Sock, _Sock_type, #request{uri = {abs_path, "/never_respond"} } ) ->
not_done;
process_request(Sock, Sock_type, Req) ->
do_trace("Recvd req: ~p~n", [Req]),
Resp = <<"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n">>,
- do_send(Sock, Sock_type, Resp).
+ do_send(Sock, Sock_type, Resp),
+ timer:sleep(random:uniform(100)).
do_send(Sock, tcp, Resp) ->
gen_tcp:send(Sock, Resp);
