Filled in more functional tests

Added additional functional tests ensuring that
the pipelines empty and added some robustness
around the test server.

test/ibrowse_functional_tests.erl

@@ -12,6 +12,9 @@
 
 -define(SERVER_PORT, 8181).
 -define(BASE_URL, "http://localhost:" ++ integer_to_list(?SERVER_PORT)).
+-define(SHORT_TIMEOUT_MS, 5000).
+-define(LONG_TIMEOUT_MS, 30000).
+-define(PAUSE_FOR_CONNECTIONS_MS, 2000).
 
 setup() ->
     application:start(crypto),
@@ -33,7 +36,10 @@ running_server_fixture_test_() ->
      [
         ?TIMEDTEST("Simple request can be honored", simple_request),
         ?TIMEDTEST("Slow server causes timeout", slow_server_timeout),
-        ?TIMEDTEST("Requests are balanced over connections", balanced_connections)
+        ?TIMEDTEST("Pipeline depth goes down with responses", pipeline_depth),
+        ?TIMEDTEST("Timeout closes pipe", closing_pipes),
+        ?TIMEDTEST("Requests are balanced over connections", balanced_connections),
+        ?TIMEDTEST("Pipeline too small signals retries", small_pipeline)
      ]
     }.
 
@@ -43,6 +49,44 @@ simple_request() ->
 slow_server_timeout() ->
     ?assertMatch({error, req_timedout}, ibrowse:send_req(?BASE_URL ++ "/never_respond", [], get, [], [], 5000)).
 
+pipeline_depth() ->
+    MaxSessions = 2,
+    MaxPipeline = 2,
+    RequestsSent = 2,
+    EmptyPipelineDepth = 0,
+
+    ?assertEqual([], ibrowse_test_server:get_conn_pipeline_depth()),
+
+    Fun = fun() -> ibrowse:send_req(?BASE_URL, [], get, [], [{max_sessions, MaxSessions}, {max_pipeline_size, MaxPipeline}], ?SHORT_TIMEOUT_MS) end,
+    times(RequestsSent, fun() -> spawn_link(Fun) end),
+
+    timer:sleep(?PAUSE_FOR_CONNECTIONS_MS),
+
+    Counts = [Count || {_Pid, Count} <- ibrowse_test_server:get_conn_pipeline_depth()],
+    ?assertEqual(MaxSessions, length(Counts)),
+    ?assertEqual(lists:duplicate(MaxSessions, EmptyPipelineDepth), Counts).
+
+closing_pipes() ->
+    MaxSessions = 2,
+    MaxPipeline = 2,
+    RequestsSent = 2,
+    BalancedNumberOfRequestsPerConnection = 1,
+
+    ?assertEqual([], ibrowse_test_server:get_conn_pipeline_depth()),
+
+    Fun = fun() -> ibrowse:send_req(?BASE_URL ++ "/never_respond", [], get, [], [{max_sessions, MaxSessions}, {max_pipeline_size, MaxPipeline}], ?SHORT_TIMEOUT_MS) end,
+    times(RequestsSent, fun() -> spawn_link(Fun) end),
+
+    timer:sleep(?PAUSE_FOR_CONNECTIONS_MS),
+
+    Counts = [Count || {_Pid, Count} <- ibrowse_test_server:get_conn_pipeline_depth()],
+    ?assertEqual(MaxSessions, length(Counts)),
+    ?assertEqual(lists:duplicate(MaxSessions, BalancedNumberOfRequestsPerConnection), Counts),
+
+    timer:sleep(?SHORT_TIMEOUT_MS),
+
+    ?assertEqual([], ibrowse_test_server:get_conn_pipeline_depth()).
+
 balanced_connections() ->
     MaxSessions = 4,
     MaxPipeline = 100,
@@ -51,16 +95,38 @@ balanced_connections() ->
 
     ?assertEqual([], ibrowse_test_server:get_conn_pipeline_depth()),
 
-    Fun = fun() -> ibrowse:send_req(?BASE_URL ++ "/never_respond", [], get, [], [{max_sessions, MaxSessions}, {max_pipeline_size, MaxPipeline}], 30000) end,
+    Fun = fun() -> ibrowse:send_req(?BASE_URL ++ "/never_respond", [], get, [], [{max_sessions, MaxSessions}, {max_pipeline_size, MaxPipeline}], ?LONG_TIMEOUT_MS) end,
     times(RequestsSent, fun() -> spawn_link(Fun) end),
 
-    timer:sleep(1000),
+    timer:sleep(?PAUSE_FOR_CONNECTIONS_MS),
 
     Counts = [Count || {_Pid, Count} <- ibrowse_test_server:get_conn_pipeline_depth()],
     ?assertEqual(MaxSessions, length(Counts)),
 
     ?assertEqual(lists:duplicate(MaxSessions, BalancedNumberOfRequestsPerConnection), Counts).
 
+small_pipeline() ->
+    MaxSessions = 10,
+    MaxPipeline = 10,
+    RequestsSent = 100,
+    FullRequestsPerConnection = 10,
+
+    ?assertEqual([], ibrowse_test_server:get_conn_pipeline_depth()),
+
+    Fun = fun() -> ibrowse:send_req(?BASE_URL ++ "/never_respond", [], get, [], [{max_sessions, MaxSessions}, {max_pipeline_size, MaxPipeline}], ?SHORT_TIMEOUT_MS) end,
+    times(RequestsSent, fun() -> spawn(Fun) end),
+
+    timer:sleep(?PAUSE_FOR_CONNECTIONS_MS),  %% Wait for everyone to get in line
+
+    Counts = [Count || {_Pid, Count} <- ibrowse_test_server:get_conn_pipeline_depth()],
+    ?assertEqual(MaxSessions, length(Counts)),
+
+    ?assertEqual(lists:duplicate(MaxSessions, FullRequestsPerConnection), Counts),
+
+    Response = ibrowse:send_req(?BASE_URL ++ "/never_respond", [], get, [], [{max_sessions, MaxSessions}, {max_pipeline_size, MaxPipeline}], ?SHORT_TIMEOUT_MS),
+
+    ?assertEqual({error, retry_later}, Response).
+
 times(0, _) ->
     ok;
 times(X, Fun) ->

test/ibrowse_test_server.erl

@@ -68,11 +68,7 @@ do_accept(ssl, Listen_sock) ->
 accept_loop(Sock, Sock_type) ->
     case do_accept(Sock_type, Sock) of
         {ok, Conn} ->
-            Pid = spawn_link(
-              fun() ->
-                      server_loop(Conn, Sock_type, #request{})
-              end),
-            ets:insert(?CONN_PIPELINE_DEPTH, {Pid, 0}),
+            Pid = spawn_link(fun() -> connection(Conn, Sock_type) end),
             set_controlling_process(Conn, Sock_type, Pid),
             Pid ! {setopts, [{active, true}]},
             accept_loop(Sock, Sock_type);
@@ -87,6 +83,14 @@ accept_loop(Sock, Sock_type) ->
             Err
     end.
 
+connection(Conn, Sock_type) ->
+    ets:insert(?CONN_PIPELINE_DEPTH, {self(), 0}),
+    try
+        server_loop(Conn, Sock_type, #request{})
+    after
+        ets:delete(?CONN_PIPELINE_DEPTH, self())
+    end.
+
 set_controlling_process(Sock, tcp, Pid) ->
     gen_tcp:controlling_process(Sock, Pid);
 set_controlling_process(Sock, ssl, Pid) ->
@@ -100,14 +104,19 @@ setopts(Sock, ssl, Opts) ->
 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),
             server_loop(Sock, Sock_type, Req#request{method = HttpMethod,
                                                      uri = HttpUri,
                                                      version = HttpVersion});
         {http, Sock, {http_header, _, _, _, _} = H} ->
             server_loop(Sock, Sock_type, Req#request{headers = [H | Headers]});
         {http, Sock, http_eoh} ->
-            ets:update_counter(?CONN_PIPELINE_DEPTH, self(), 1),
-            process_request(Sock, Sock_type, Req),
+            case process_request(Sock, Sock_type, Req) of
+                not_done ->
+                    ok;
+                _ ->
+                    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"
@@ -172,7 +181,6 @@ process_request(Sock, Sock_type,
                          uri = {abs_path, "/ibrowse_head_transfer_enc"}}) ->
     Resp = <<"HTTP/1.1 400 Bad Request\r\nServer: Apache-Coyote/1.1\r\nContent-Length:5\r\nDate: Wed, 04 Apr 2012 16:53:49 GMT\r\n\r\nabcde">>,
     do_send(Sock, Sock_type, Resp);
-
 process_request(Sock, Sock_type,
                 #request{method='GET',
                          headers = Headers,
@@ -210,7 +218,7 @@ process_request(Sock, Sock_type,
     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{uri = {abs_path, "/never_respond"} } ) ->
-    noop;
+    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">>,
@@ -221,7 +229,6 @@ do_send(Sock, tcp, Resp) ->
 do_send(Sock, ssl, Resp) ->
     ssl:send(Sock, Resp).
 
-
 %%------------------------------------------------------------------------------
 %% Utility functions
 %%------------------------------------------------------------------------------