replicator, shared/dedicated httpc pool configuration

gistfile1.diff

diff --git a/src/couchdb/couch_api_wrap.erl b/src/couchdb/couch_api_wrap.erl
index 63e607a..8fcd08f 100644
--- a/src/couchdb/couch_api_wrap.erl
+++ b/src/couchdb/couch_api_wrap.erl
@@ -101,7 +101,6 @@ db_open(DbName, Options, Create) ->
     end.
 
 db_close(#httpdb{httpc_pool = Pool}) ->
-    unlink(Pool),
     ok = couch_httpc_pool:stop(Pool);
 db_close(DbName) ->
     catch couch_db:close(DbName).
diff --git a/src/couchdb/couch_api_wrap.hrl b/src/couchdb/couch_api_wrap.hrl
index bcd1b70..d31ea1f 100644
--- a/src/couchdb/couch_api_wrap.hrl
+++ b/src/couchdb/couch_api_wrap.hrl
@@ -11,6 +11,10 @@
 % the License.
 
 
+-record(httpc_pool, {
+    pid = nil,
+    dedicated = false
+}).
 
 -record(httpdb, {
     url,
@@ -23,7 +27,7 @@
     ibrowse_options = [],
     retries = 10,
     wait = 250,         % milliseconds
-    httpc_pool = nil,
+    httpc_pool = #httpc_pool{},
     http_connections,
     http_pipeline_size
 }).
diff --git a/src/couchdb/couch_api_wrap_httpc.erl b/src/couchdb/couch_api_wrap_httpc.erl
index bc9413d..6665309 100644
--- a/src/couchdb/couch_api_wrap_httpc.erl
+++ b/src/couchdb/couch_api_wrap_httpc.erl
@@ -29,12 +29,14 @@
 -define(replace(L, K, V), lists:keystore(K, 1, L, {K, V})).
 
 
-setup(#httpdb{httpc_pool = nil, url = Url, ibrowse_options = IbrowseOptions,
+setup(#httpdb{httpc_pool = #httpc_pool{pid = nil, dedicated = Dedicated},
+    url = Url, ibrowse_options = IbrowseOptions,
     http_connections = MaxConns, http_pipeline_size = PipeSize} = Db) ->
     HttpcPoolOptions = [
         {ssl_options, get_value(ssl_options, IbrowseOptions, [])},
         {max_piped_connections, MaxConns},
-        {pipeline_size, PipeSize}
+        {pipeline_size, PipeSize},
+        {dedicated, Dedicated}
     ],
     {ok, Pid} = couch_httpc_pool:start_link(
         ibrowse_lib:parse_url(Url), HttpcPoolOptions),
diff --git a/src/couchdb/couch_httpc_pool.erl b/src/couchdb/couch_httpc_pool.erl
index 979b3d0..4ee5793 100644
--- a/src/couchdb/couch_httpc_pool.erl
+++ b/src/couchdb/couch_httpc_pool.erl
@@ -34,6 +34,7 @@
 -export([code_change/3, terminate/2]).
 
 -include("couch_db.hrl").
+-include("couch_api_wrap.hrl").
 -include("../ibrowse/ibrowse.hrl").
 
 -import(couch_util, [
@@ -53,24 +54,42 @@
 }).
 
 
-start_link(BaseUrl, Options) ->
-    gen_server:start_link(?MODULE, {BaseUrl, Options}, []).
+start_link(#url{host = Host, port = Port} = BaseUrl, Options) ->
+    Dedicated = get_value(dedicated, Options),
+    Res = case Dedicated of
+    true ->
+        gen_server:start_link(?MODULE, {BaseUrl, Options}, []);
+    false ->
+        Name = list_to_atom(Host ++ ":" ++ integer_to_list(Port)),
+        gen_server:start_link({local, Name}, ?MODULE, {BaseUrl, Options}, [])
+    end,
+    case Res of
+    {ok, Pid} ->
+        {ok, #httpc_pool{pid = Pid, dedicated = Dedicated}};
+    {error, {already_started, Pid}} ->
+        link(Pid),
+        {ok, #httpc_pool{pid = Pid, dedicated = Dedicated}}
+    end.
 
 
-stop(Pool) ->
-    ok = gen_server:call(Pool, stop, infinity).
+stop(#httpc_pool{pid = Pool, dedicated = true}) ->
+    unlink(Pool),
+    ok = gen_server:call(Pool, stop, infinity);
+stop(#httpc_pool{pid = Pool, dedicated = false}) ->
+    unlink(Pool),
+    ok.
 
 
-get_piped_worker(Pool) ->
+get_piped_worker(#httpc_pool{pid = Pool}) ->
     gen_server:call(Pool, get_piped_worker, infinity).
 
 
-get_worker(Pool) ->
+get_worker(#httpc_pool{pid = Pool}) ->
     gen_server:call(Pool, get_worker, infinity).
 
 
 % Only workers without a pipeline need to be released.
-release_worker(Pool, Worker) ->
+release_worker(#httpc_pool{pid = Pool}, Worker) ->
     ok = gen_server:call(Pool, {release_worker, Worker}, infinity).
 
 
diff --git a/src/couchdb/couch_replicator.erl b/src/couchdb/couch_replicator.erl
index 00618b7..b2b1495 100644
--- a/src/couchdb/couch_replicator.erl
+++ b/src/couchdb/couch_replicator.erl
@@ -274,9 +274,13 @@ do_init(#rep{options = Options, id = {BaseId, Ext}} = Rep) ->
         "~ca worker batch size of ~p~n"
         "~c~p HTTP connections, each with a pipeline size of ~p~n"
         "~ca connection timeout of ~p milliseconds~n"
+        "~c~s connections~n"
         "~csocket options are: ~s",
         [BaseId ++ Ext, $\t, CopiersCount, $\t, BatchSize, $\t, MaxHttpConns,
-            HttpPipeSize, $\t, get_value(connection_timeout, Options),
+            HttpPipeSize, $\t, get_value(connection_timeout, Options), $\t,
+            case get_value(dedicated_pool, Options) of
+            true -> "dedicated";
+            false -> "shared" end,
             $\t, io_lib:format("~p", [get_value(socket_options, Options)])]),
 
     ?LOG_DEBUG("Missing rev finder pids are: ~p", [MissingRevFinders]),
diff --git a/src/couchdb/couch_replicator_utils.erl b/src/couchdb/couch_replicator_utils.erl
index 8716f92..bb72ccd 100644
--- a/src/couchdb/couch_replicator_utils.erl
+++ b/src/couchdb/couch_replicator_utils.erl
@@ -156,7 +156,8 @@ parse_rep_db({Props}, ProxyParams, Options) ->
                 ProxyParams ++ ssl_params(Url)]),
         timeout = get_value(connection_timeout, Options),
         http_connections = get_value(http_connections, Options),
-        http_pipeline_size = get_value(http_pipeline_size, Options)
+        http_pipeline_size = get_value(http_pipeline_size, Options),
+        httpc_pool = #httpc_pool{dedicated = get_value(dedicated_pool, Options)}
     };
 parse_rep_db(<<"http://", _/binary>> = Url, ProxyParams, Options) ->
     parse_rep_db({[{<<"url">>, Url}]}, ProxyParams, Options);
@@ -189,6 +190,7 @@ make_options(Props) ->
             "[{keepalive, true}, {nodelay, false}]")),
     lists:ukeymerge(1, Options, [
         {connection_timeout, list_to_integer(DefTimeout)},
+        {dedicated_pool, false},
         {http_connections, list_to_integer(DefConns)},
         {http_pipeline_size, list_to_integer(DefPipeSize)},
         {socket_options, DefSocketOptions},
@@ -224,6 +226,8 @@ convert_options([{<<"http_pipeline_size">>, V} | R]) ->
     [{http_pipeline_size, couch_util:to_integer(V)} | convert_options(R)];
 convert_options([{<<"connection_timeout">>, V} | R]) ->
     [{connection_timeout, couch_util:to_integer(V)} | convert_options(R)];
+convert_options([{<<"dedicated_connections">>, V} | R]) ->
+    [{dedicated_pool, V} | convert_options(R)];
 convert_options([{<<"socket_options">>, V} | R]) ->
     {ok, SocketOptions} = couch_util:parse_term(V),
     [{socket_options, SocketOptions} | convert_options(R)];