Use 2x CPUs as the number of delegate processes.

3 files changed, 19 insertions, 40 deletions

diff --git a/include/delegate.hrl b/include/delegate.hrl
deleted file mode 100644
index 38f8d42f..00000000
--- a/include/delegate.hrl
+++ /dev/null
@@ -1,32 +0,0 @@
-%%   The contents of this file are subject to the Mozilla Public License
-%%   Version 1.1 (the "License"); you may not use this file except in
-%%   compliance with the License. You may obtain a copy of the License at
-%%   http://www.mozilla.org/MPL/
-%%
-%%   Software distributed under the License is distributed on an "AS IS"
-%%   basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See the
-%%   License for the specific language governing rights and limitations
-%%   under the License.
-%%
-%%   The Original Code is RabbitMQ.
-%%
-%%   The Initial Developers of the Original Code are LShift Ltd,
-%%   Cohesive Financial Technologies LLC, and Rabbit Technologies Ltd.
-%%
-%%   Portions created before 22-Nov-2008 00:00:00 GMT by LShift Ltd,
-%%   Cohesive Financial Technologies LLC, or Rabbit Technologies Ltd
-%%   are Copyright (C) 2007-2008 LShift Ltd, Cohesive Financial
-%%   Technologies LLC, and Rabbit Technologies Ltd.
-%%
-%%   Portions created by LShift Ltd are Copyright (C) 2007-2010 LShift
-%%   Ltd. Portions created by Cohesive Financial Technologies LLC are
-%%   Copyright (C) 2007-2010 Cohesive Financial Technologies
-%%   LLC. Portions created by Rabbit Technologies Ltd are Copyright
-%%   (C) 2007-2010 Rabbit Technologies Ltd.
-%%
-%%   All Rights Reserved.
-%%
-%%   Contributor(s): ______________________________________.
-%%
-
--define(DELEGATE_PROCESSES, 10).
diff --git a/src/delegate.erl b/src/delegate.erl
index 2724736e..03dd06ac 100644
--- a/src/delegate.erl
+++ b/src/delegate.erl
@@ -30,14 +30,14 @@
 %%
 
 -module(delegate).
--include("delegate.hrl").
+-define(DELEGATE_PROCESS_COUNT_MULTIPLIER, 2).
 
 -behaviour(gen_server2).
 
 -export([start_link/1, cast/2, call/2,
          gs2_call/3, gs2_pcall/4,
          gs2_cast/2, gs2_pcast/3,
-         server/1]).
+         server/1, process_count/0]).
 
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2,
          terminate/2, code_change/3]).
@@ -76,7 +76,7 @@ call(Pids, FPid) when is_list(Pids) ->
     call_per_node(split_delegate_per_node(Pids), FPid).
 
 internal_call(Node, Thunk) when is_atom(Node) ->
-    gen_server2:call({server(), Node}, {thunk, Thunk}, infinity).
+    gen_server2:call({server(Node), Node}, {thunk, Thunk}, infinity).
 
 
 cast(Pid, FPid) when is_pid(Pid) ->
@@ -88,7 +88,7 @@ cast(Pids, FPid) when is_list(Pids) ->
     ok.
 
 internal_cast(Node, Thunk) when is_atom(Node) ->
-    gen_server2:cast({server(), Node}, {thunk, Thunk}).
+    gen_server2:cast({server(Node), Node}, {thunk, Thunk}).
 
 %%----------------------------------------------------------------------------
 
@@ -120,8 +120,8 @@ delegate_per_node(NodePids, FPid, DelegateFun) ->
         [DelegateFun(Node, fun() -> [safe_invoke(FPid, Pid) || Pid <- Pids] end)
         || {Node, Pids} <- NodePids]).
 
-server() ->
-    server(erlang:phash2(self(), ?DELEGATE_PROCESSES)).
+server(Node) when is_atom(Node) ->
+    server(erlang:phash2(self(), process_count(Node)));
 
 server(Hash) ->
     list_to_atom("delegate_process_" ++ integer_to_list(Hash)).
@@ -134,6 +134,18 @@ safe_invoke(FPid, Pid) ->
             {ok, Result, Pid}
     end.
 
+process_count(Node) ->
+    case get({process_count, Node}) of
+        undefined ->
+            Count = rpc:call(Node, delegate, process_count, []),
+            put({process_count, Node}, Count),
+            Count;
+        Count -> Count
+    end.
+
+process_count() ->
+    ?DELEGATE_PROCESS_COUNT_MULTIPLIER * erlang:system_info(schedulers).
+
 %%--------------------------------------------------------------------
 
 init([]) ->
diff --git a/src/delegate_sup.erl b/src/delegate_sup.erl
index 99a5e4d3..dd3d0eef 100644
--- a/src/delegate_sup.erl
+++ b/src/delegate_sup.erl
@@ -30,7 +30,6 @@
 %%
 
 -module(delegate_sup).
--include("delegate.hrl").
 
 -behaviour(supervisor).
 
@@ -51,6 +50,6 @@ init(_Args) ->
     {ok, {{one_for_one, 10, 10},
         [{delegate:server(Hash), {delegate, start_link, [Hash]},
           transient, 16#ffffffff, worker, [delegate]} ||
-                Hash <- lists:seq(0, ?DELEGATE_PROCESSES - 1)]}}.
+                Hash <- lists:seq(0, delegate:process_count() - 1)]}}.
 
 %%--------------------------------------------------------------------