summaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/priority_queue.erl40
-rw-r--r--src/rabbit.erl25
-rw-r--r--src/rabbit_alarm.erl52
-rw-r--r--src/rabbit_amqqueue.erl4
-rw-r--r--src/rabbit_channel.erl4
-rw-r--r--src/rabbit_control.erl48
-rw-r--r--src/rabbit_guid.erl26
-rw-r--r--src/rabbit_heartbeat.erl4
-rw-r--r--src/rabbit_memsup.erl142
-rw-r--r--src/rabbit_memsup_darwin.erl88
-rw-r--r--src/rabbit_memsup_linux.erl115
-rw-r--r--src/rabbit_misc.erl43
-rw-r--r--src/rabbit_mnesia.erl97
-rw-r--r--src/rabbit_net.erl132
-rw-r--r--src/rabbit_networking.erl61
-rw-r--r--src/rabbit_reader.erl20
-rw-r--r--src/rabbit_tests.erl114
-rw-r--r--src/rabbit_writer.erl4
-rw-r--r--src/tcp_listener.erl29
-rw-r--r--src/tcp_listener_sup.erl14
20 files changed, 778 insertions, 284 deletions
diff --git a/src/priority_queue.erl b/src/priority_queue.erl
index 732757c4..c74b39a9 100644
--- a/src/priority_queue.erl
+++ b/src/priority_queue.erl
@@ -55,7 +55,8 @@
-module(priority_queue).
--export([new/0, is_queue/1, is_empty/1, len/1, to_list/1, in/2, in/3, out/1]).
+-export([new/0, is_queue/1, is_empty/1, len/1, to_list/1, in/2, in/3,
+ out/1, join/2]).
%%----------------------------------------------------------------------------
@@ -73,6 +74,7 @@
-spec(in/2 :: (any(), pqueue()) -> pqueue()).
-spec(in/3 :: (any(), priority(), pqueue()) -> pqueue()).
-spec(out/1 :: (pqueue()) -> {empty | {value, any()}, pqueue()}).
+-spec(join/2 :: (pqueue(), pqueue()) -> pqueue()).
-endif.
@@ -147,6 +149,42 @@ out({pqueue, [{P, Q} | Queues]}) ->
end,
{R, NewQ}.
+join(A, {queue, [], []}) ->
+ A;
+join({queue, [], []}, B) ->
+ B;
+join({queue, AIn, AOut}, {queue, BIn, BOut}) ->
+ {queue, BIn, AOut ++ lists:reverse(AIn, BOut)};
+join(A = {queue, _, _}, {pqueue, BPQ}) ->
+ {Pre, Post} = lists:splitwith(fun ({P, _}) -> P < 0 end, BPQ),
+ Post1 = case Post of
+ [] -> [ {0, A} ];
+ [ {0, ZeroQueue} | Rest ] -> [ {0, join(A, ZeroQueue)} | Rest ];
+ _ -> [ {0, A} | Post ]
+ end,
+ {pqueue, Pre ++ Post1};
+join({pqueue, APQ}, B = {queue, _, _}) ->
+ {Pre, Post} = lists:splitwith(fun ({P, _}) -> P < 0 end, APQ),
+ Post1 = case Post of
+ [] -> [ {0, B} ];
+ [ {0, ZeroQueue} | Rest ] -> [ {0, join(ZeroQueue, B)} | Rest ];
+ _ -> [ {0, B} | Post ]
+ end,
+ {pqueue, Pre ++ Post1};
+join({pqueue, APQ}, {pqueue, BPQ}) ->
+ {pqueue, merge(APQ, BPQ, [])}.
+
+merge([], BPQ, Acc) ->
+ lists:reverse(Acc, BPQ);
+merge(APQ, [], Acc) ->
+ lists:reverse(Acc, APQ);
+merge([{P, A}|As], [{P, B}|Bs], Acc) ->
+ merge(As, Bs, [ {P, join(A, B)} | Acc ]);
+merge([{PA, A}|As], Bs = [{PB, _}|_], Acc) when PA < PB ->
+ merge(As, Bs, [ {PA, A} | Acc ]);
+merge(As = [{_, _}|_], [{PB, B}|Bs], Acc) ->
+ merge(As, Bs, [ {PB, B} | Acc ]).
+
r2f([]) -> {queue, [], []};
r2f([_] = R) -> {queue, [], R};
r2f([X,Y]) -> {queue, [X], [Y]};
diff --git a/src/rabbit.erl b/src/rabbit.erl
index 6ce90d93..ef1e0049 100644
--- a/src/rabbit.erl
+++ b/src/rabbit.erl
@@ -133,13 +133,13 @@ start(normal, []) ->
ok = start_child(rabbit_log),
ok = rabbit_hooks:start(),
- ok = rabbit_amqqueue:start(),
+ ok = rabbit_binary_generator:
+ check_empty_content_body_frame_size(),
{ok, MemoryAlarms} = application:get_env(memory_alarms),
ok = rabbit_alarm:start(MemoryAlarms),
- ok = rabbit_binary_generator:
- check_empty_content_body_frame_size(),
+ ok = rabbit_amqqueue:start(),
ok = start_child(rabbit_router),
ok = start_child(rabbit_node_monitor)
@@ -168,12 +168,27 @@ start(normal, []) ->
{"TCP listeners",
fun () ->
ok = rabbit_networking:start(),
- {ok, TCPListeners} = application:get_env(tcp_listeners),
+ {ok, TcpListeners} = application:get_env(tcp_listeners),
lists:foreach(
fun ({Host, Port}) ->
ok = rabbit_networking:start_tcp_listener(Host, Port)
end,
- TCPListeners)
+ TcpListeners)
+ end},
+ {"SSL listeners",
+ fun () ->
+ case application:get_env(ssl_listeners) of
+ {ok, []} ->
+ ok;
+ {ok, SslListeners} ->
+ ok = rabbit_misc:start_applications([crypto, ssl]),
+
+ {ok, SslOpts} = application:get_env(ssl_options),
+
+ [rabbit_networking:start_ssl_listener
+ (Host, Port, SslOpts) || {Host, Port} <- SslListeners],
+ ok
+ end
end}]),
io:format("~nbroker running~n"),
diff --git a/src/rabbit_alarm.erl b/src/rabbit_alarm.erl
index 21999f16..309c9a0e 100644
--- a/src/rabbit_alarm.erl
+++ b/src/rabbit_alarm.erl
@@ -41,7 +41,7 @@
-define(MEMSUP_CHECK_INTERVAL, 1000).
%% OSes on which we know memory alarms to be trustworthy
--define(SUPPORTED_OS, [{unix, linux}]).
+-define(SUPPORTED_OS, [{unix, linux}, {unix, darwin}]).
-record(alarms, {alertees, system_memory_high_watermark = false}).
@@ -136,33 +136,35 @@ code_change(_OldVsn, State, _Extra) ->
%%----------------------------------------------------------------------------
start_memsup() ->
- Mod = case os:type() of
- %% memsup doesn't take account of buffers or cache when
- %% considering "free" memory - therefore on Linux we can
- %% get memory alarms very easily without any pressure
- %% existing on memory at all. Therefore we need to use
- %% our own simple memory monitor.
- %%
- {unix, linux} -> rabbit_memsup_linux;
-
- %% Start memsup programmatically rather than via the
- %% rabbitmq-server script. This is not quite the right
- %% thing to do as os_mon checks to see if memsup is
- %% available before starting it, but as memsup is
- %% available everywhere (even on VXWorks) it should be
- %% ok.
- %%
- %% One benefit of the programmatic startup is that we
- %% can add our alarm_handler before memsup is running,
- %% thus ensuring that we notice memory alarms that go
- %% off on startup.
- %%
- _ -> memsup
- end,
+ {Mod, Args} =
+ case os:type() of
+ %% memsup doesn't take account of buffers or cache when
+ %% considering "free" memory - therefore on Linux we can
+ %% get memory alarms very easily without any pressure
+ %% existing on memory at all. Therefore we need to use
+ %% our own simple memory monitor.
+ %%
+ {unix, linux} -> {rabbit_memsup, [rabbit_memsup_linux]};
+ {unix, darwin} -> {rabbit_memsup, [rabbit_memsup_darwin]};
+
+ %% Start memsup programmatically rather than via the
+ %% rabbitmq-server script. This is not quite the right
+ %% thing to do as os_mon checks to see if memsup is
+ %% available before starting it, but as memsup is
+ %% available everywhere (even on VXWorks) it should be
+ %% ok.
+ %%
+ %% One benefit of the programmatic startup is that we
+ %% can add our alarm_handler before memsup is running,
+ %% thus ensuring that we notice memory alarms that go
+ %% off on startup.
+ %%
+ _ -> {memsup, []}
+ end,
%% This is based on os_mon:childspec(memsup, true)
{ok, _} = supervisor:start_child(
os_mon_sup,
- {memsup, {Mod, start_link, []},
+ {memsup, {Mod, start_link, Args},
permanent, 2000, worker, [Mod]}),
ok.
diff --git a/src/rabbit_amqqueue.erl b/src/rabbit_amqqueue.erl
index 4903c2c5..f05f7880 100644
--- a/src/rabbit_amqqueue.erl
+++ b/src/rabbit_amqqueue.erl
@@ -303,10 +303,10 @@ basic_cancel(#amqqueue{pid = QPid}, ChPid, ConsumerTag, OkMsg) ->
infinity).
notify_sent(QPid, ChPid) ->
- gen_server2:cast(QPid, {notify_sent, ChPid}).
+ gen_server2:pcast(QPid, 8, {notify_sent, ChPid}).
unblock(QPid, ChPid) ->
- gen_server2:cast(QPid, {unblock, ChPid}).
+ gen_server2:pcast(QPid, 8, {unblock, ChPid}).
internal_delete(QueueName) ->
rabbit_misc:execute_mnesia_transaction(
diff --git a/src/rabbit_channel.erl b/src/rabbit_channel.erl
index 58b94234..16b7c938 100644
--- a/src/rabbit_channel.erl
+++ b/src/rabbit_channel.erl
@@ -157,6 +157,10 @@ handle_cast({conserve_memory, Conserve}, State) ->
State#ch.writer_pid, #'channel.flow'{active = not(Conserve)}),
noreply(State).
+handle_info({'EXIT', WriterPid, Reason = {writer, send_failed, _Error}},
+ State = #ch{writer_pid = WriterPid}) ->
+ State#ch.reader_pid ! {channel_exit, State#ch.channel, Reason},
+ {stop, normal, State};
handle_info({'EXIT', _Pid, Reason}, State) ->
{stop, Reason, State};
diff --git a/src/rabbit_control.erl b/src/rabbit_control.erl
index 37e4d189..cf20520e 100644
--- a/src/rabbit_control.erl
+++ b/src/rabbit_control.erl
@@ -329,22 +329,23 @@ format_info_item(Items, Key) ->
{value, Info = {Key, Value}} = lists:keysearch(Key, 1, Items),
case Info of
{_, #resource{name = Name}} ->
- url_encode(Name);
+ escape(Name);
_ when Key =:= address; Key =:= peer_address andalso is_tuple(Value) ->
inet_parse:ntoa(Value);
_ when is_pid(Value) ->
atom_to_list(node(Value));
_ when is_binary(Value) ->
- url_encode(Value);
+ escape(Value);
_ ->
io_lib:format("~w", [Value])
end.
display_list(L) when is_list(L) ->
lists:foreach(fun (I) when is_binary(I) ->
- io:format("~s~n", [url_encode(I)]);
+ io:format("~s~n", [escape(I)]);
(I) when is_tuple(I) ->
- display_row([url_encode(V) || V <- tuple_to_list(I)])
+ display_row([escape(V)
+ || V <- tuple_to_list(I)])
end,
lists:sort(L)),
ok;
@@ -356,32 +357,23 @@ call(Node, {Mod, Fun, Args}) ->
rpc_call(Node, Mod, Fun, Args) ->
rpc:call(Node, Mod, Fun, Args, ?RPC_TIMEOUT).
-%% url_encode is lifted from ibrowse, modified to preserve some characters
-url_encode(Bin) when binary(Bin) ->
- url_encode_char(lists:reverse(binary_to_list(Bin)), []).
-
-url_encode_char([X | T], Acc) when X >= $a, X =< $z ->
- url_encode_char(T, [X | Acc]);
-url_encode_char([X | T], Acc) when X >= $A, X =< $Z ->
- url_encode_char(T, [X | Acc]);
-url_encode_char([X | T], Acc) when X >= $0, X =< $9 ->
- url_encode_char(T, [X | Acc]);
-url_encode_char([X | T], Acc)
- when X == $-; X == $_; X == $.; X == $~;
- X == $!; X == $*; X == $'; X == $(;
- X == $); X == $;; X == $:; X == $@;
- X == $&; X == $=; X == $+; X == $$;
- X == $,; X == $/; X == $?; X == $%;
- X == $#; X == $[; X == $] ->
- url_encode_char(T, [X | Acc]);
-url_encode_char([X | T], Acc) ->
- url_encode_char(T, [$%, d2h(X bsr 4), d2h(X band 16#0f) | Acc]);
-url_encode_char([], Acc) ->
+%% escape does C-style backslash escaping of non-printable ASCII
+%% characters. We don't escape characters above 127, since they may
+%% form part of UTF-8 strings.
+
+escape(Bin) when binary(Bin) ->
+ escape_char(lists:reverse(binary_to_list(Bin)), []).
+
+escape_char([$\\ | T], Acc) ->
+ escape_char(T, [$\\, $\\ | Acc]);
+escape_char([X | T], Acc) when X > 32, X /= 127 ->
+ escape_char(T, [X | Acc]);
+escape_char([X | T], Acc) ->
+ escape_char(T, [$\\, $0 + (X bsr 6), $0 + (X band 8#070 bsr 3),
+ $0 + (X band 7) | Acc]);
+escape_char([], Acc) ->
Acc.
-d2h(N) when N<10 -> N+$0;
-d2h(N) -> N+$a-10.
-
list_replace(Find, Replace, List) ->
[case X of Find -> Replace; _ -> X end || X <- List].
diff --git a/src/rabbit_guid.erl b/src/rabbit_guid.erl
index 2be00503..b789fbd1 100644
--- a/src/rabbit_guid.erl
+++ b/src/rabbit_guid.erl
@@ -42,6 +42,7 @@
terminate/2, code_change/3]).
-define(SERVER, ?MODULE).
+-define(SERIAL_FILENAME, "rabbit_serial").
-record(state, {serial}).
@@ -59,17 +60,28 @@
%%----------------------------------------------------------------------------
start_link() ->
- %% The persister can get heavily loaded, and we don't want that to
- %% impact guid generation. We therefore keep the serial in a
- %% separate process rather than calling rabbit_persister:serial/0
- %% directly in the functions below.
gen_server:start_link({local, ?SERVER}, ?MODULE,
- [rabbit_persister:serial()], []).
+ [update_disk_serial()], []).
+
+update_disk_serial() ->
+ Filename = filename:join(rabbit_mnesia:dir(), ?SERIAL_FILENAME),
+ Serial = case rabbit_misc:read_term_file(Filename) of
+ {ok, [Num]} -> Num;
+ {error, enoent} -> rabbit_persister:serial();
+ {error, Reason} ->
+ throw({error, {cannot_read_serial_file, Filename, Reason}})
+ end,
+ case rabbit_misc:write_term_file(Filename, [Serial + 1]) of
+ ok -> ok;
+ {error, Reason1} ->
+ throw({error, {cannot_write_serial_file, Filename, Reason1}})
+ end,
+ Serial.
%% generate a guid that is monotonically increasing per process.
%%
%% The id is only unique within a single cluster and as long as the
-%% persistent message store hasn't been deleted.
+%% serial store hasn't been deleted.
guid() ->
%% We don't use erlang:now() here because a) it may return
%% duplicates when the system clock has been rewound prior to a
@@ -77,7 +89,7 @@ guid() ->
%% now() to move ahead of the system time), and b) it is really
%% slow since it takes a global lock and makes a system call.
%%
- %% rabbit_persister:serial/0, in combination with self/0 (which
+ %% A persisted serial number, in combination with self/0 (which
%% includes the node name) uniquely identifies a process in space
%% and time. We combine that with a process-local counter to give
%% us a GUID that is monotonically increasing per process.
diff --git a/src/rabbit_heartbeat.erl b/src/rabbit_heartbeat.erl
index 0a68c9ad..ed0066fe 100644
--- a/src/rabbit_heartbeat.erl
+++ b/src/rabbit_heartbeat.erl
@@ -53,7 +53,7 @@ start_heartbeat(Sock, TimeoutSec) ->
spawn_link(fun () -> heartbeater(Sock, TimeoutSec * 1000 div 2,
send_oct, 0,
fun () ->
- catch gen_tcp:send(Sock, rabbit_binary_generator:build_heartbeat_frame()),
+ catch rabbit_net:send(Sock, rabbit_binary_generator:build_heartbeat_frame()),
continue
end,
erlang:monitor(process, Parent)) end),
@@ -73,7 +73,7 @@ heartbeater(Sock, TimeoutMillisec, StatName, Threshold, Handler, MonitorRef) ->
{'DOWN', MonitorRef, process, _Object, _Info} -> ok;
Other -> exit({unexpected_message, Other})
after TimeoutMillisec ->
- case inet:getstat(Sock, [StatName]) of
+ case rabbit_net:getstat(Sock, [StatName]) of
{ok, [{StatName, NewStatVal}]} ->
if NewStatVal =/= StatVal ->
F({NewStatVal, 0});
diff --git a/src/rabbit_memsup.erl b/src/rabbit_memsup.erl
new file mode 100644
index 00000000..b0d57cb2
--- /dev/null
+++ b/src/rabbit_memsup.erl
@@ -0,0 +1,142 @@
+%% 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-2009 LShift
+%% Ltd. Portions created by Cohesive Financial Technologies LLC are
+%% Copyright (C) 2007-2009 Cohesive Financial Technologies
+%% LLC. Portions created by Rabbit Technologies Ltd are Copyright
+%% (C) 2007-2009 Rabbit Technologies Ltd.
+%%
+%% All Rights Reserved.
+%%
+%% Contributor(s): ______________________________________.
+%%
+
+-module(rabbit_memsup).
+
+-behaviour(gen_server).
+
+-export([start_link/1]).
+
+-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
+ terminate/2, code_change/3]).
+
+-export([update/0]).
+
+-record(state, {memory_fraction,
+ timeout,
+ timer,
+ mod,
+ mod_state,
+ alarmed
+ }).
+
+-define(SERVER, memsup). %% must be the same as the standard memsup
+
+-define(DEFAULT_MEMORY_CHECK_INTERVAL, 1000).
+
+%%----------------------------------------------------------------------------
+
+-ifdef(use_specs).
+
+-spec(start_link/1 :: (atom()) -> {'ok', pid()} | 'ignore' | {'error', any()}).
+-spec(update/0 :: () -> 'ok').
+
+-endif.
+
+%%----------------------------------------------------------------------------
+
+start_link(Args) ->
+ gen_server:start_link({local, ?SERVER}, ?MODULE, [Args], []).
+
+update() ->
+ gen_server:cast(?SERVER, update).
+
+%%----------------------------------------------------------------------------
+
+init([Mod]) ->
+ Fraction = os_mon:get_env(memsup, system_memory_high_watermark),
+ TRef = start_timer(?DEFAULT_MEMORY_CHECK_INTERVAL),
+ InitState = Mod:init(),
+ State = #state { memory_fraction = Fraction,
+ timeout = ?DEFAULT_MEMORY_CHECK_INTERVAL,
+ timer = TRef,
+ mod = Mod,
+ mod_state = InitState,
+ alarmed = false },
+ {ok, internal_update(State)}.
+
+start_timer(Timeout) ->
+ {ok, TRef} = timer:apply_interval(Timeout, ?MODULE, update, []),
+ TRef.
+
+%% Export the same API as the real memsup. Note that
+%% get_sysmem_high_watermark gives an int in the range 0 - 100, while
+%% set_sysmem_high_watermark takes a float in the range 0.0 - 1.0.
+handle_call(get_sysmem_high_watermark, _From, State) ->
+ {reply, trunc(100 * State#state.memory_fraction), State};
+
+handle_call({set_sysmem_high_watermark, Float}, _From, State) ->
+ {reply, ok, State#state{memory_fraction = Float}};
+
+handle_call(get_check_interval, _From, State) ->
+ {reply, State#state.timeout, State};
+
+handle_call({set_check_interval, Timeout}, _From, State) ->
+ {ok, cancel} = timer:cancel(State#state.timer),
+ {reply, ok, State#state{timeout = Timeout, timer = start_timer(Timeout)}};
+
+handle_call(get_memory_data, _From,
+ State = #state { mod = Mod, mod_state = ModState }) ->
+ {reply, Mod:get_memory_data(ModState), State};
+
+handle_call(_Request, _From, State) ->
+ {noreply, State}.
+
+handle_cast(update, State) ->
+ {noreply, internal_update(State)};
+
+handle_cast(_Request, State) ->
+ {noreply, State}.
+
+handle_info(_Info, State) ->
+ {noreply, State}.
+
+terminate(_Reason, _State) ->
+ ok.
+
+code_change(_OldVsn, State, _Extra) ->
+ {ok, State}.
+
+internal_update(State = #state { memory_fraction = MemoryFraction,
+ alarmed = Alarmed,
+ mod = Mod, mod_state = ModState }) ->
+ ModState1 = Mod:update(ModState),
+ {MemTotal, MemUsed, _BigProc} = Mod:get_memory_data(ModState1),
+ NewAlarmed = MemUsed / MemTotal > MemoryFraction,
+ case {Alarmed, NewAlarmed} of
+ {false, true} ->
+ alarm_handler:set_alarm({system_memory_high_watermark, []});
+ {true, false} ->
+ alarm_handler:clear_alarm(system_memory_high_watermark);
+ _ ->
+ ok
+ end,
+ State #state { mod_state = ModState1, alarmed = NewAlarmed }.
diff --git a/src/rabbit_memsup_darwin.erl b/src/rabbit_memsup_darwin.erl
new file mode 100644
index 00000000..3de2d843
--- /dev/null
+++ b/src/rabbit_memsup_darwin.erl
@@ -0,0 +1,88 @@
+%% 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-2009 LShift
+%% Ltd. Portions created by Cohesive Financial Technologies LLC are
+%% Copyright (C) 2007-2009 Cohesive Financial Technologies
+%% LLC. Portions created by Rabbit Technologies Ltd are Copyright
+%% (C) 2007-2009 Rabbit Technologies Ltd.
+%%
+%% All Rights Reserved.
+%%
+%% Contributor(s): ______________________________________.
+%%
+
+-module(rabbit_memsup_darwin).
+
+-export([init/0, update/1, get_memory_data/1]).
+
+-record(state, {total_memory,
+ allocated_memory}).
+
+%%----------------------------------------------------------------------------
+
+-ifdef(use_specs).
+
+-type(state() :: #state { total_memory :: ('undefined' | non_neg_integer()),
+ allocated_memory :: ('undefined' | non_neg_integer())
+ }).
+
+-spec(init/0 :: () -> state()).
+-spec(update/1 :: (state()) -> state()).
+-spec(get_memory_data/1 :: (state()) -> {non_neg_integer(), non_neg_integer(),
+ ('undefined' | pid())}).
+
+-endif.
+
+%%----------------------------------------------------------------------------
+
+init() ->
+ #state{total_memory = undefined,
+ allocated_memory = undefined}.
+
+update(State) ->
+ File = os:cmd("/usr/bin/vm_stat"),
+ Lines = string:tokens(File, "\n"),
+ Dict = dict:from_list(lists:map(fun parse_line/1, Lines)),
+ [PageSize, Inactive, Active, Free, Wired] =
+ [dict:fetch(Key, Dict) ||
+ Key <- [page_size, 'Pages inactive', 'Pages active', 'Pages free',
+ 'Pages wired down']],
+ MemTotal = PageSize * (Inactive + Active + Free + Wired),
+ MemUsed = PageSize * (Active + Wired),
+ State#state{total_memory = MemTotal, allocated_memory = MemUsed}.
+
+get_memory_data(State) ->
+ {State#state.total_memory, State#state.allocated_memory, undefined}.
+
+%%----------------------------------------------------------------------------
+
+%% A line looks like "Foo bar: 123456."
+parse_line(Line) ->
+ [Name, RHS | _Rest] = string:tokens(Line, ":"),
+ case Name of
+ "Mach Virtual Memory Statistics" ->
+ ["(page", "size", "of", PageSize, "bytes)"] =
+ string:tokens(RHS, " "),
+ {page_size, list_to_integer(PageSize)};
+ _ ->
+ [Value | _Rest1] = string:tokens(RHS, " ."),
+ {list_to_atom(Name), list_to_integer(Value)}
+ end.
diff --git a/src/rabbit_memsup_linux.erl b/src/rabbit_memsup_linux.erl
index ffdc7e99..ca942d7c 100644
--- a/src/rabbit_memsup_linux.erl
+++ b/src/rabbit_memsup_linux.erl
@@ -31,104 +31,44 @@
-module(rabbit_memsup_linux).
--behaviour(gen_server).
+-export([init/0, update/1, get_memory_data/1]).
--export([start_link/0]).
-
--export([init/1, handle_call/3, handle_cast/2, handle_info/2,
- terminate/2, code_change/3]).
-
--export([update/0]).
-
--define(SERVER, memsup). %% must be the same as the standard memsup
-
--define(DEFAULT_MEMORY_CHECK_INTERVAL, 1000).
-
--record(state, {memory_fraction, alarmed, timeout, timer}).
+-record(state, {total_memory,
+ allocated_memory}).
%%----------------------------------------------------------------------------
-ifdef(use_specs).
--spec(start_link/0 :: () -> {'ok', pid()} | 'ignore' | {'error', any()}).
--spec(update/0 :: () -> 'ok').
-
--endif.
-
-%%----------------------------------------------------------------------------
-
-start_link() ->
- gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).
+-type(state() :: #state { total_memory :: ('undefined' | non_neg_integer()),
+ allocated_memory :: ('undefined' | non_neg_integer())
+ }).
+-spec(init/0 :: () -> state()).
+-spec(update/1 :: (state()) -> state()).
+-spec(get_memory_data/1 :: (state()) -> {non_neg_integer(), non_neg_integer(),
+ ('undefined' | pid())}).
-update() ->
- gen_server:cast(?SERVER, update).
+-endif.
%%----------------------------------------------------------------------------
-init(_Args) ->
- Fraction = os_mon:get_env(memsup, system_memory_high_watermark),
- TRef = start_timer(?DEFAULT_MEMORY_CHECK_INTERVAL),
- {ok, #state{alarmed = false,
- memory_fraction = Fraction,
- timeout = ?DEFAULT_MEMORY_CHECK_INTERVAL,
- timer = TRef}}.
-
-start_timer(Timeout) ->
- {ok, TRef} = timer:apply_interval(Timeout, ?MODULE, update, []),
- TRef.
-
-%% Export the same API as the real memsup. Note that
-%% get_sysmem_high_watermark gives an int in the range 0 - 100, while
-%% set_sysmem_high_watermark takes a float in the range 0.0 - 1.0.
-handle_call(get_sysmem_high_watermark, _From, State) ->
- {reply, trunc(100 * State#state.memory_fraction), State};
-
-handle_call({set_sysmem_high_watermark, Float}, _From, State) ->
- {reply, ok, State#state{memory_fraction = Float}};
+init() ->
+ #state{total_memory = undefined,
+ allocated_memory = undefined}.
-handle_call(get_check_interval, _From, State) ->
- {reply, State#state.timeout, State};
-
-handle_call({set_check_interval, Timeout}, _From, State) ->
- {ok, cancel} = timer:cancel(State#state.timer),
- {reply, ok, State#state{timeout = Timeout, timer = start_timer(Timeout)}};
-
-handle_call(_Request, _From, State) ->
- {noreply, State}.
-
-handle_cast(update, State = #state{alarmed = Alarmed,
- memory_fraction = MemoryFraction}) ->
+update(State) ->
File = read_proc_file("/proc/meminfo"),
Lines = string:tokens(File, "\n"),
Dict = dict:from_list(lists:map(fun parse_line/1, Lines)),
- MemTotal = dict:fetch('MemTotal', Dict),
- MemUsed = MemTotal
- - dict:fetch('MemFree', Dict)
- - dict:fetch('Buffers', Dict)
- - dict:fetch('Cached', Dict),
- NewAlarmed = MemUsed / MemTotal > MemoryFraction,
- case {Alarmed, NewAlarmed} of
- {false, true} ->
- alarm_handler:set_alarm({system_memory_high_watermark, []});
- {true, false} ->
- alarm_handler:clear_alarm(system_memory_high_watermark);
- _ ->
- ok
- end,
- {noreply, State#state{alarmed = NewAlarmed}};
-
-handle_cast(_Request, State) ->
- {noreply, State}.
-
-handle_info(_Info, State) ->
- {noreply, State}.
-
-terminate(_Reason, _State) ->
- ok.
-
-code_change(_OldVsn, State, _Extra) ->
- {ok, State}.
+ [MemTotal, MemFree, Buffers, Cached] =
+ [dict:fetch(Key, Dict) ||
+ Key <- ['MemTotal', 'MemFree', 'Buffers', 'Cached']],
+ MemUsed = MemTotal - MemFree - Buffers - Cached,
+ State#state{total_memory = MemTotal, allocated_memory = MemUsed}.
+
+get_memory_data(State) ->
+ {State#state.total_memory, State#state.allocated_memory, undefined}.
%%----------------------------------------------------------------------------
@@ -152,5 +92,10 @@ read_proc_file(IoDevice, Acc) ->
%% A line looks like "FooBar: 123456 kB"
parse_line(Line) ->
- [Name, Value | _] = string:tokens(Line, ": "),
- {list_to_atom(Name), list_to_integer(Value)}.
+ [Name, RHS | _Rest] = string:tokens(Line, ":"),
+ [Value | UnitsRest] = string:tokens(RHS, " "),
+ Value1 = case UnitsRest of
+ [] -> list_to_integer(Value); %% no units
+ ["kB"] -> list_to_integer(Value) * 1024
+ end,
+ {list_to_atom(Name), Value1}.
diff --git a/src/rabbit_misc.erl b/src/rabbit_misc.erl
index abf4c7cc..95a274e3 100644
--- a/src/rabbit_misc.erl
+++ b/src/rabbit_misc.erl
@@ -50,9 +50,11 @@
-export([intersperse/2, upmap/2, map_in_order/2]).
-export([table_foreach/2]).
-export([dirty_read_all/1, dirty_foreach_key/2, dirty_dump_log/1]).
+-export([read_term_file/1, write_term_file/2]).
-export([append_file/2, ensure_parent_dirs_exist/1]).
-export([format_stderr/2]).
-export([start_applications/1, stop_applications/1]).
+-export([unfold/2, ceil/1]).
-import(mnesia).
-import(lists).
@@ -65,6 +67,8 @@
-include_lib("kernel/include/inet.hrl").
+-type(ok_or_error() :: 'ok' | {'error', any()}).
+
-spec(method_record_type/1 :: (tuple()) -> atom()).
-spec(polite_pause/0 :: () -> 'done').
-spec(polite_pause/1 :: (non_neg_integer()) -> 'done').
@@ -88,9 +92,9 @@
-spec(r_arg/4 :: (vhost() | r(atom()), K, amqp_table(), binary()) ->
undefined | r(K) when is_subtype(K, atom())).
-spec(rs/1 :: (r(atom())) -> string()).
--spec(enable_cover/0 :: () -> 'ok' | {'error', any()}).
+-spec(enable_cover/0 :: () -> ok_or_error()).
-spec(report_cover/0 :: () -> 'ok').
--spec(enable_cover/1 :: (string()) -> 'ok' | {'error', any()}).
+-spec(enable_cover/1 :: (string()) -> ok_or_error()).
-spec(report_cover/1 :: (string()) -> 'ok').
-spec(throw_on_error/2 ::
(atom(), thunk({error, any()} | {ok, A} | A)) -> A).
@@ -100,7 +104,7 @@
-spec(with_vhost/2 :: (vhost(), thunk(A)) -> A).
-spec(with_user_and_vhost/3 :: (username(), vhost(), thunk(A)) -> A).
-spec(execute_mnesia_transaction/1 :: (thunk(A)) -> A).
--spec(ensure_ok/2 :: ('ok' | {'error', any()}, atom()) -> 'ok').
+-spec(ensure_ok/2 :: (ok_or_error(), atom()) -> 'ok').
-spec(localnode/1 :: (atom()) -> erlang_node()).
-spec(tcp_name/3 :: (atom(), ip_address(), ip_port()) -> atom()).
-spec(intersperse/2 :: (A, [A]) -> [A]).
@@ -110,12 +114,16 @@
-spec(dirty_read_all/1 :: (atom()) -> [any()]).
-spec(dirty_foreach_key/2 :: (fun ((any()) -> any()), atom()) ->
'ok' | 'aborted').
--spec(dirty_dump_log/1 :: (string()) -> 'ok' | {'error', any()}).
--spec(append_file/2 :: (string(), string()) -> 'ok' | {'error', any()}).
+-spec(dirty_dump_log/1 :: (string()) -> ok_or_error()).
+-spec(read_term_file/1 :: (string()) -> {'ok', [any()]} | {'error', any()}).
+-spec(write_term_file/2 :: (string(), [any()]) -> ok_or_error()).
+-spec(append_file/2 :: (string(), string()) -> ok_or_error()).
-spec(ensure_parent_dirs_exist/1 :: (string()) -> 'ok').
-spec(format_stderr/2 :: (string(), [any()]) -> 'ok').
-spec(start_applications/1 :: ([atom()]) -> 'ok').
-spec(stop_applications/1 :: ([atom()]) -> 'ok').
+-spec(unfold/2 :: (fun ((A) -> ({'true', B, A} | 'false')), A) -> {[B], A}).
+-spec(ceil/1 :: (number()) -> number()).
-endif.
@@ -360,7 +368,9 @@ dirty_foreach_key1(F, TableName, K) ->
end.
dirty_dump_log(FileName) ->
- {ok, LH} = disk_log:open([{name, dirty_dump_log}, {mode, read_only}, {file, FileName}]),
+ {ok, LH} = disk_log:open([{name, dirty_dump_log},
+ {mode, read_only},
+ {file, FileName}]),
dirty_dump_log1(LH, disk_log:chunk(LH, start)),
disk_log:close(LH).
@@ -374,6 +384,12 @@ dirty_dump_log1(LH, {K, Terms, BadBytes}) ->
dirty_dump_log1(LH, disk_log:chunk(LH, K)).
+read_term_file(File) -> file:consult(File).
+
+write_term_file(File, Terms) ->
+ file:write_file(File, list_to_binary([io_lib:format("~w.~n", [Term]) ||
+ Term <- Terms])).
+
append_file(File, Suffix) ->
case file:read_file_info(File) of
{ok, FInfo} -> append_file(File, FInfo#file_info.size, Suffix);
@@ -444,3 +460,18 @@ stop_applications(Apps) ->
cannot_stop_application,
Apps).
+unfold(Fun, Init) ->
+ unfold(Fun, [], Init).
+
+unfold(Fun, Acc, Init) ->
+ case Fun(Init) of
+ {true, E, I} -> unfold(Fun, [E|Acc], I);
+ false -> {Acc, Init}
+ end.
+
+ceil(N) ->
+ T = trunc(N),
+ case N - T of
+ 0 -> N;
+ _ -> 1 + T
+ end.
diff --git a/src/rabbit_mnesia.erl b/src/rabbit_mnesia.erl
index 575ecb0a..37e20335 100644
--- a/src/rabbit_mnesia.erl
+++ b/src/rabbit_mnesia.erl
@@ -149,6 +149,11 @@ table_definitions() ->
table_names() ->
[Tab || {Tab, _} <- table_definitions()].
+replicated_table_names() ->
+ [Tab || {Tab, Attrs} <- table_definitions(),
+ not lists:member({local_content, true}, Attrs)
+ ].
+
dir() -> mnesia:system_info(directory).
ensure_mnesia_dir() ->
@@ -192,28 +197,16 @@ cluster_nodes_config_filename() ->
create_cluster_nodes_config(ClusterNodes) ->
FileName = cluster_nodes_config_filename(),
- Handle = case file:open(FileName, [write]) of
- {ok, Device} -> Device;
- {error, Reason} ->
- throw({error, {cannot_create_cluster_nodes_config,
- FileName, Reason}})
- end,
- try
- ok = io:write(Handle, ClusterNodes),
- ok = io:put_chars(Handle, [$.])
- after
- case file:close(Handle) of
- ok -> ok;
- {error, Reason1} ->
- throw({error, {cannot_close_cluster_nodes_config,
- FileName, Reason1}})
- end
- end,
- ok.
+ case rabbit_misc:write_term_file(FileName, [ClusterNodes]) of
+ ok -> ok;
+ {error, Reason} ->
+ throw({error, {cannot_create_cluster_nodes_config,
+ FileName, Reason}})
+ end.
read_cluster_nodes_config() ->
FileName = cluster_nodes_config_filename(),
- case file:consult(FileName) of
+ case rabbit_misc:read_term_file(FileName) of
{ok, [ClusterNodes]} -> ClusterNodes;
{error, enoent} ->
case application:get_env(cluster_config) of
@@ -250,12 +243,10 @@ delete_cluster_nodes_config() ->
%% standalone disk node, or disk or ram node connected to the
%% specified cluster nodes.
init_db(ClusterNodes) ->
- WasDiskNode = mnesia:system_info(use_dir),
- IsDiskNode = ClusterNodes == [] orelse
- lists:member(node(), ClusterNodes),
case mnesia:change_config(extra_db_nodes, ClusterNodes -- [node()]) of
{ok, []} ->
- if WasDiskNode and IsDiskNode ->
+ case mnesia:system_info(use_dir) of
+ true ->
case check_schema_integrity() of
ok ->
ok;
@@ -270,22 +261,18 @@ init_db(ClusterNodes) ->
ok = move_db(),
ok = create_schema()
end;
- WasDiskNode ->
- throw({error, {cannot_convert_disk_node_to_ram_node,
- ClusterNodes}});
- IsDiskNode ->
- ok = create_schema();
- true ->
- throw({error, {unable_to_contact_cluster_nodes,
- ClusterNodes}})
+ false ->
+ ok = create_schema()
end;
{ok, [_|_]} ->
- ok = wait_for_tables(),
- ok = create_local_table_copies(
- case IsDiskNode of
- true -> disc;
- false -> ram
- end);
+ IsDiskNode = ClusterNodes == [] orelse
+ lists:member(node(), ClusterNodes),
+ ok = wait_for_replicated_tables(),
+ ok = create_local_table_copy(schema, disc_copies),
+ ok = create_local_table_copies(case IsDiskNode of
+ true -> disc;
+ false -> ram
+ end);
{error, Reason} ->
%% one reason we may end up here is if we try to join
%% nodes together that are currently running standalone or
@@ -336,40 +323,36 @@ create_tables() ->
table_definitions()),
ok.
+table_has_copy_type(TabDef, DiscType) ->
+ lists:member(node(), proplists:get_value(DiscType, TabDef, [])).
+
create_local_table_copies(Type) ->
- ok = if Type /= ram -> create_local_table_copy(schema, disc_copies);
- true -> ok
- end,
lists:foreach(
fun({Tab, TabDef}) ->
- HasDiscCopies =
- lists:keymember(disc_copies, 1, TabDef),
- HasDiscOnlyCopies =
- lists:keymember(disc_only_copies, 1, TabDef),
+ HasDiscCopies = table_has_copy_type(TabDef, disc_copies),
+ HasDiscOnlyCopies = table_has_copy_type(TabDef, disc_only_copies),
+ LocalTab = proplists:get_bool(local_content, TabDef),
StorageType =
- case Type of
- disc ->
+ if
+ Type =:= disc orelse LocalTab ->
if
- HasDiscCopies -> disc_copies;
+ HasDiscCopies -> disc_copies;
HasDiscOnlyCopies -> disc_only_copies;
- true -> ram_copies
+ true -> ram_copies
end;
%% unused code - commented out to keep dialyzer happy
-%% disc_only ->
+%% Type =:= disc_only ->
%% if
%% HasDiscCopies or HasDiscOnlyCopies ->
%% disc_only_copies;
%% true -> ram_copies
%% end;
- ram ->
+ Type =:= ram ->
ram_copies
end,
ok = create_local_table_copy(Tab, StorageType)
end,
table_definitions()),
- ok = if Type == ram -> create_local_table_copy(schema, ram_copies);
- true -> ok
- end,
ok.
create_local_table_copy(Tab, Type) ->
@@ -384,10 +367,14 @@ create_local_table_copy(Tab, Type) ->
end,
ok.
-wait_for_tables() ->
+wait_for_replicated_tables() -> wait_for_tables(replicated_table_names()).
+
+wait_for_tables() -> wait_for_tables(table_names()).
+
+wait_for_tables(TableNames) ->
case check_schema_integrity() of
ok ->
- case mnesia:wait_for_tables(table_names(), 30000) of
+ case mnesia:wait_for_tables(TableNames, 30000) of
ok -> ok;
{timeout, BadTabs} ->
throw({error, {timeout_waiting_for_tables, BadTabs}});
diff --git a/src/rabbit_net.erl b/src/rabbit_net.erl
new file mode 100644
index 00000000..a5ccc8e9
--- /dev/null
+++ b/src/rabbit_net.erl
@@ -0,0 +1,132 @@
+%% 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-2009 LShift
+%% Ltd. Portions created by Cohesive Financial Technologies LLC are
+%% Copyright (C) 2007-2009 Cohesive Financial Technologies
+%% LLC. Portions created by Rabbit Technologies Ltd are Copyright
+%% (C) 2007-2009 Rabbit Technologies Ltd.
+%%
+%% All Rights Reserved.
+%%
+%% Contributor(s): ______________________________________.
+%%
+
+-module(rabbit_net).
+-include("rabbit.hrl").
+-include_lib("kernel/include/inet.hrl").
+
+-export([async_recv/3, close/1, controlling_process/2,
+ getstat/2, peername/1, port_command/2,
+ send/2, sockname/1]).
+%%---------------------------------------------------------------------------
+
+-ifdef(use_specs).
+
+-type(stat_option() ::
+ 'recv_cnt' | 'recv_max' | 'recv_avg' | 'recv_oct' | 'recv_dvi' |
+ 'send_cnt' | 'send_max' | 'send_avg' | 'send_oct' | 'send_pend').
+-type(error() :: {'error', any()}).
+
+-spec(async_recv/3 :: (socket(), integer(), timeout()) -> {'ok', any()}).
+-spec(close/1 :: (socket()) -> 'ok' | error()).
+-spec(controlling_process/2 :: (socket(), pid()) -> 'ok' | error()).
+-spec(port_command/2 :: (socket(), iolist()) -> 'true').
+-spec(send/2 :: (socket(), binary() | iolist()) -> 'ok' | error()).
+-spec(peername/1 :: (socket()) ->
+ {'ok', {ip_address(), non_neg_integer()}} | error()).
+-spec(sockname/1 :: (socket()) ->
+ {'ok', {ip_address(), non_neg_integer()}} | error()).
+-spec(getstat/2 :: (socket(), [stat_option()]) ->
+ {'ok', [{stat_option(), integer()}]} | error()).
+
+-endif.
+
+%%---------------------------------------------------------------------------
+
+
+async_recv(Sock, Length, Timeout) when is_record(Sock, ssl_socket) ->
+ Pid = self(),
+ Ref = make_ref(),
+
+ spawn(fun() -> Pid ! {inet_async, Sock, Ref,
+ ssl:recv(Sock#ssl_socket.ssl, Length, Timeout)}
+ end),
+
+ {ok, Ref};
+
+async_recv(Sock, Length, infinity) when is_port(Sock) ->
+ prim_inet:async_recv(Sock, Length, -1);
+
+async_recv(Sock, Length, Timeout) when is_port(Sock) ->
+ prim_inet:async_recv(Sock, Length, Timeout).
+
+close(Sock) when is_record(Sock, ssl_socket) ->
+ ssl:close(Sock#ssl_socket.ssl);
+
+close(Sock) when is_port(Sock) ->
+ gen_tcp:close(Sock).
+
+
+controlling_process(Sock, Pid) when is_record(Sock, ssl_socket) ->
+ ssl:controlling_process(Sock#ssl_socket.ssl, Pid);
+
+controlling_process(Sock, Pid) when is_port(Sock) ->
+ gen_tcp:controlling_process(Sock, Pid).
+
+
+getstat(Sock, Stats) when is_record(Sock, ssl_socket) ->
+ inet:getstat(Sock#ssl_socket.tcp, Stats);
+
+getstat(Sock, Stats) when is_port(Sock) ->
+ inet:getstat(Sock, Stats).
+
+
+peername(Sock) when is_record(Sock, ssl_socket) ->
+ ssl:peername(Sock#ssl_socket.ssl);
+
+peername(Sock) when is_port(Sock) ->
+ inet:peername(Sock).
+
+
+port_command(Sock, Data) when is_record(Sock, ssl_socket) ->
+ case ssl:send(Sock#ssl_socket.ssl, Data) of
+ ok ->
+ self() ! {inet_reply, Sock, ok},
+ true;
+ {error, Reason} ->
+ erlang:error(Reason)
+ end;
+
+port_command(Sock, Data) when is_port(Sock) ->
+ erlang:port_command(Sock, Data).
+
+send(Sock, Data) when is_record(Sock, ssl_socket) ->
+ ssl:send(Sock#ssl_socket.ssl, Data);
+
+send(Sock, Data) when is_port(Sock) ->
+ gen_tcp:send(Sock, Data).
+
+
+sockname(Sock) when is_record(Sock, ssl_socket) ->
+ ssl:sockname(Sock#ssl_socket.ssl);
+
+sockname(Sock) when is_port(Sock) ->
+ inet:sockname(Sock).
diff --git a/src/rabbit_networking.erl b/src/rabbit_networking.erl
index 2dbd5a5a..eed21a01 100644
--- a/src/rabbit_networking.erl
+++ b/src/rabbit_networking.erl
@@ -31,18 +31,28 @@
-module(rabbit_networking).
--export([start/0, start_tcp_listener/2, stop_tcp_listener/2,
- on_node_down/1, active_listeners/0, node_listeners/1,
- connections/0, connection_info/1, connection_info/2,
- connection_info_all/0, connection_info_all/1]).
+-export([start/0, start_tcp_listener/2, start_ssl_listener/3,
+ stop_tcp_listener/2, on_node_down/1, active_listeners/0,
+ node_listeners/1, connections/0, connection_info/1,
+ connection_info/2, connection_info_all/0,
+ connection_info_all/1]).
%%used by TCP-based transports, e.g. STOMP adapter
-export([check_tcp_listener_address/3]).
--export([tcp_listener_started/2, tcp_listener_stopped/2, start_client/1]).
+-export([tcp_listener_started/2, ssl_connection_upgrade/2,
+ tcp_listener_stopped/2, start_client/1]).
-include("rabbit.hrl").
-include_lib("kernel/include/inet.hrl").
+-define(RABBIT_TCP_OPTS, [
+ binary,
+ {packet, raw}, % no packaging
+ {reuseaddr, true}, % allow rebind without waiting
+ %% {nodelay, true}, % TCP_NODELAY - disable Nagle's alg.
+ %% {delay_send, true},
+ {exit_on_close, false}
+ ]).
%%----------------------------------------------------------------------------
-ifdef(use_specs).
@@ -52,6 +62,7 @@
-spec(start/0 :: () -> 'ok').
-spec(start_tcp_listener/2 :: (host(), ip_port()) -> 'ok').
+-spec(start_ssl_listener/3 :: (host(), ip_port(), [info()]) -> 'ok').
-spec(stop_tcp_listener/2 :: (host(), ip_port()) -> 'ok').
-spec(active_listeners/0 :: () -> [listener()]).
-spec(node_listeners/1 :: (erlang_node()) -> [listener()]).
@@ -96,21 +107,24 @@ check_tcp_listener_address(NamePrefix, Host, Port) ->
{IPAddress, Name}.
start_tcp_listener(Host, Port) ->
- {IPAddress, Name} = check_tcp_listener_address(rabbit_tcp_listener_sup, Host, Port),
+ start_listener(Host, Port, "TCP Listener",
+ {?MODULE, start_client, []}).
+
+start_ssl_listener(Host, Port, SslOpts) ->
+ start_listener(Host, Port, "SSL Listener",
+ {?MODULE, ssl_connection_upgrade, [SslOpts]}).
+
+start_listener(Host, Port, Label, OnConnect) ->
+ {IPAddress, Name} =
+ check_tcp_listener_address(rabbit_tcp_listener_sup, Host, Port),
{ok,_} = supervisor:start_child(
rabbit_sup,
{Name,
{tcp_listener_sup, start_link,
- [IPAddress, Port,
- [binary,
- {packet, raw}, % no packaging
- {reuseaddr, true}, % allow rebind without waiting
- %% {nodelay, true}, % TCP_NODELAY - disable Nagle's alg.
- %% {delay_send, true},
- {exit_on_close, false}],
+ [IPAddress, Port, ?RABBIT_TCP_OPTS ,
{?MODULE, tcp_listener_started, []},
{?MODULE, tcp_listener_stopped, []},
- {?MODULE, start_client, []}]},
+ OnConnect, Label]},
transient, infinity, supervisor, [tcp_listener_sup]}),
ok.
@@ -148,10 +162,27 @@ on_node_down(Node) ->
start_client(Sock) ->
{ok, Child} = supervisor:start_child(rabbit_tcp_client_sup, []),
- ok = gen_tcp:controlling_process(Sock, Child),
+ ok = rabbit_net:controlling_process(Sock, Child),
Child ! {go, Sock},
Child.
+ssl_connection_upgrade(SslOpts, Sock) ->
+ {ok, {PeerAddress, PeerPort}} = rabbit_net:peername(Sock),
+ PeerIp = inet_parse:ntoa(PeerAddress),
+
+ case ssl:ssl_accept(Sock, SslOpts) of
+ {ok, SslSock} ->
+ rabbit_log:info("upgraded TCP connection from ~s:~p to SSL~n",
+ [PeerIp, PeerPort]),
+ RabbitSslSock = #ssl_socket{tcp = Sock, ssl = SslSock},
+ start_client(RabbitSslSock);
+ {error, Reason} ->
+ gen_tcp:close(Sock),
+ rabbit_log:error("failed to upgrade TCP connection from ~s:~p "
+ "to SSL: ~n~p~n", [PeerIp, PeerPort, Reason]),
+ {error, Reason}
+ end.
+
connections() ->
[Pid || {_, Pid, _, _} <- supervisor:which_children(
rabbit_tcp_client_sup)].
diff --git a/src/rabbit_reader.erl b/src/rabbit_reader.erl
index 426b99eb..69dbc008 100644
--- a/src/rabbit_reader.erl
+++ b/src/rabbit_reader.erl
@@ -200,7 +200,7 @@ inet_op(F) -> rabbit_misc:throw_on_error(inet_error, F).
peername(Sock) ->
try
- {Address, Port} = inet_op(fun () -> inet:peername(Sock) end),
+ {Address, Port} = inet_op(fun () -> rabbit_net:peername(Sock) end),
AddressS = inet_parse:ntoa(Address),
{AddressS, Port}
catch
@@ -286,7 +286,7 @@ mainloop(Parent, Deb, State = #v1{sock= Sock, recv_ref = Ref}) ->
%% since this termination is initiated by our parent it is
%% probably more important to exit quickly.
exit(Reason);
- {'EXIT', _Pid, E = {writer, send_failed, _Error}} ->
+ {channel_exit, _Chan, E = {writer, send_failed, _Error}} ->
throw(E);
{channel_exit, Channel, Reason} ->
mainloop(Parent, Deb, handle_channel_exit(Channel, Reason, State));
@@ -323,8 +323,8 @@ mainloop(Parent, Deb, State = #v1{sock= Sock, recv_ref = Ref}) ->
end.
switch_callback(OldState, NewCallback, Length) ->
- Ref = inet_op(fun () -> prim_inet:async_recv(
- OldState#v1.sock, Length, -1) end),
+ Ref = inet_op(fun () -> rabbit_net:async_recv(
+ OldState#v1.sock, Length, infinity) end),
OldState#v1{callback = NewCallback,
recv_ref = Ref}.
@@ -539,7 +539,7 @@ handle_input(handshake, <<"AMQP",1,1,ProtocolMajor,ProtocolMinor>>,
end;
handle_input(handshake, Other, #v1{sock = Sock}) ->
- ok = inet_op(fun () -> gen_tcp:send(
+ ok = inet_op(fun () -> rabbit_net:send(
Sock, <<"AMQP",1,1,
?PROTOCOL_VERSION_MAJOR,
?PROTOCOL_VERSION_MINOR>>) end),
@@ -675,23 +675,23 @@ infos(Items, State) -> [{Item, i(Item, State)} || Item <- Items].
i(pid, #v1{}) ->
self();
i(address, #v1{sock = Sock}) ->
- {ok, {A, _}} = inet:sockname(Sock),
+ {ok, {A, _}} = rabbit_net:sockname(Sock),
A;
i(port, #v1{sock = Sock}) ->
- {ok, {_, P}} = inet:sockname(Sock),
+ {ok, {_, P}} = rabbit_net:sockname(Sock),
P;
i(peer_address, #v1{sock = Sock}) ->
- {ok, {A, _}} = inet:peername(Sock),
+ {ok, {A, _}} = rabbit_net:peername(Sock),
A;
i(peer_port, #v1{sock = Sock}) ->
- {ok, {_, P}} = inet:peername(Sock),
+ {ok, {_, P}} = rabbit_net:peername(Sock),
P;
i(SockStat, #v1{sock = Sock}) when SockStat =:= recv_oct;
SockStat =:= recv_cnt;
SockStat =:= send_oct;
SockStat =:= send_cnt;
SockStat =:= send_pend ->
- case inet:getstat(Sock, [SockStat]) of
+ case rabbit_net:getstat(Sock, [SockStat]) of
{ok, [{SockStat, StatVal}]} -> StatVal;
{error, einval} -> undefined;
{error, Error} -> throw({cannot_get_socket_stats, Error})
diff --git a/src/rabbit_tests.erl b/src/rabbit_tests.erl
index e5100ccd..b4cd30bc 100644
--- a/src/rabbit_tests.erl
+++ b/src/rabbit_tests.erl
@@ -49,6 +49,7 @@ test_content_prop_roundtrip(Datum, Binary) ->
all_tests() ->
passed = test_priority_queue(),
+ passed = test_unfold(),
passed = test_parsing(),
passed = test_topic_matching(),
passed = test_log_management(),
@@ -75,7 +76,8 @@ test_priority_queue() ->
%% 1-element priority Q
Q1 = priority_queue:in(foo, 1, priority_queue:new()),
- {true, false, 1, [{1, foo}], [foo]} = test_priority_queue(Q1),
+ {true, false, 1, [{1, foo}], [foo]} =
+ test_priority_queue(Q1),
%% 2-element same-priority Q
Q2 = priority_queue:in(bar, 1, Q1),
@@ -91,6 +93,71 @@ test_priority_queue() ->
Q4 = priority_queue:in(foo, -1, priority_queue:new()),
{true, false, 1, [{-1, foo}], [foo]} = test_priority_queue(Q4),
+ %% merge 2 * 1-element no-priority Qs
+ Q5 = priority_queue:join(priority_queue:in(foo, Q),
+ priority_queue:in(bar, Q)),
+ {true, false, 2, [{0, foo}, {0, bar}], [foo, bar]} =
+ test_priority_queue(Q5),
+
+ %% merge 1-element no-priority Q with 1-element priority Q
+ Q6 = priority_queue:join(priority_queue:in(foo, Q),
+ priority_queue:in(bar, 1, Q)),
+ {true, false, 2, [{1, bar}, {0, foo}], [bar, foo]} =
+ test_priority_queue(Q6),
+
+ %% merge 1-element priority Q with 1-element no-priority Q
+ Q7 = priority_queue:join(priority_queue:in(foo, 1, Q),
+ priority_queue:in(bar, Q)),
+ {true, false, 2, [{1, foo}, {0, bar}], [foo, bar]} =
+ test_priority_queue(Q7),
+
+ %% merge 2 * 1-element same-priority Qs
+ Q8 = priority_queue:join(priority_queue:in(foo, 1, Q),
+ priority_queue:in(bar, 1, Q)),
+ {true, false, 2, [{1, foo}, {1, bar}], [foo, bar]} =
+ test_priority_queue(Q8),
+
+ %% merge 2 * 1-element different-priority Qs
+ Q9 = priority_queue:join(priority_queue:in(foo, 1, Q),
+ priority_queue:in(bar, 2, Q)),
+ {true, false, 2, [{2, bar}, {1, foo}], [bar, foo]} =
+ test_priority_queue(Q9),
+
+ %% merge 2 * 1-element different-priority Qs (other way around)
+ Q10 = priority_queue:join(priority_queue:in(bar, 2, Q),
+ priority_queue:in(foo, 1, Q)),
+ {true, false, 2, [{2, bar}, {1, foo}], [bar, foo]} =
+ test_priority_queue(Q10),
+
+ %% merge 2 * 2-element multi-different-priority Qs
+ Q11 = priority_queue:join(Q6, Q5),
+ {true, false, 4, [{1, bar}, {0, foo}, {0, foo}, {0, bar}],
+ [bar, foo, foo, bar]} = test_priority_queue(Q11),
+
+ %% and the other way around
+ Q12 = priority_queue:join(Q5, Q6),
+ {true, false, 4, [{1, bar}, {0, foo}, {0, bar}, {0, foo}],
+ [bar, foo, bar, foo]} = test_priority_queue(Q12),
+
+ %% merge with negative priorities
+ Q13 = priority_queue:join(Q4, Q5),
+ {true, false, 3, [{0, foo}, {0, bar}, {-1, foo}], [foo, bar, foo]} =
+ test_priority_queue(Q13),
+
+ %% and the other way around
+ Q14 = priority_queue:join(Q5, Q4),
+ {true, false, 3, [{0, foo}, {0, bar}, {-1, foo}], [foo, bar, foo]} =
+ test_priority_queue(Q14),
+
+ %% joins with empty queues:
+ Q1 = priority_queue:join(Q, Q1),
+ Q1 = priority_queue:join(Q1, Q),
+
+ %% insert with priority into non-empty zero-priority queue
+ Q15 = priority_queue:in(baz, 1, Q5),
+ {true, false, 3, [{1, baz}, {0, foo}, {0, bar}], [baz, foo, bar]} =
+ test_priority_queue(Q15),
+
passed.
priority_queue_in_all(Q, L) ->
@@ -116,6 +183,14 @@ test_simple_n_element_queue(N) ->
{true, false, N, ToListRes, Items} = test_priority_queue(Q),
passed.
+test_unfold() ->
+ {[], test} = rabbit_misc:unfold(fun (_V) -> false end, test),
+ List = lists:seq(2,20,2),
+ {List, 0} = rabbit_misc:unfold(fun (0) -> false;
+ (N) -> {true, N*2, N-1}
+ end, 10),
+ passed.
+
test_parsing() ->
passed = test_content_properties(),
passed.
@@ -408,19 +483,17 @@ test_cluster_management() ->
end,
ClusteringSequence),
- %% attempt to convert a disk node into a ram node
+ %% convert a disk node into a ram node
ok = control_action(reset, []),
ok = control_action(start_app, []),
ok = control_action(stop_app, []),
- {error, {cannot_convert_disk_node_to_ram_node, _}} =
- control_action(cluster, ["invalid1@invalid",
- "invalid2@invalid"]),
+ ok = control_action(cluster, ["invalid1@invalid",
+ "invalid2@invalid"]),
- %% attempt to join a non-existing cluster as a ram node
+ %% join a non-existing cluster as a ram node
ok = control_action(reset, []),
- {error, {unable_to_contact_cluster_nodes, _}} =
- control_action(cluster, ["invalid1@invalid",
- "invalid2@invalid"]),
+ ok = control_action(cluster, ["invalid1@invalid",
+ "invalid2@invalid"]),
SecondaryNode = rabbit_misc:localnode(hare),
case net_adm:ping(SecondaryNode) of
@@ -436,11 +509,12 @@ test_cluster_management2(SecondaryNode) ->
NodeS = atom_to_list(node()),
SecondaryNodeS = atom_to_list(SecondaryNode),
- %% attempt to convert a disk node into a ram node
+ %% make a disk node
ok = control_action(reset, []),
ok = control_action(cluster, [NodeS]),
- {error, {unable_to_join_cluster, _, _}} =
- control_action(cluster, [SecondaryNodeS]),
+ %% make a ram node
+ ok = control_action(reset, []),
+ ok = control_action(cluster, [SecondaryNodeS]),
%% join cluster as a ram node
ok = control_action(reset, []),
@@ -453,21 +527,21 @@ test_cluster_management2(SecondaryNode) ->
ok = control_action(start_app, []),
ok = control_action(stop_app, []),
- %% attempt to join non-existing cluster as a ram node
- {error, _} = control_action(cluster, ["invalid1@invalid",
- "invalid2@invalid"]),
-
+ %% join non-existing cluster as a ram node
+ ok = control_action(cluster, ["invalid1@invalid",
+ "invalid2@invalid"]),
%% turn ram node into disk node
+ ok = control_action(reset, []),
ok = control_action(cluster, [SecondaryNodeS, NodeS]),
ok = control_action(start_app, []),
ok = control_action(stop_app, []),
- %% attempt to convert a disk node into a ram node
- {error, {cannot_convert_disk_node_to_ram_node, _}} =
- control_action(cluster, ["invalid1@invalid",
- "invalid2@invalid"]),
+ %% convert a disk node into a ram node
+ ok = control_action(cluster, ["invalid1@invalid",
+ "invalid2@invalid"]),
%% turn a disk node into a ram node
+ ok = control_action(reset, []),
ok = control_action(cluster, [SecondaryNodeS]),
ok = control_action(start_app, []),
ok = control_action(stop_app, []),
diff --git a/src/rabbit_writer.erl b/src/rabbit_writer.erl
index e338ddfe..1679ce7c 100644
--- a/src/rabbit_writer.erl
+++ b/src/rabbit_writer.erl
@@ -169,7 +169,7 @@ assemble_frames(Channel, MethodRecord, Content, FrameMax) ->
tcp_send(Sock, Data) ->
rabbit_misc:throw_on_error(inet_error,
- fun () -> gen_tcp:send(Sock, Data) end).
+ fun () -> rabbit_net:send(Sock, Data) end).
internal_send_command(Sock, Channel, MethodRecord) ->
ok = tcp_send(Sock, assemble_frames(Channel, MethodRecord)).
@@ -206,6 +206,6 @@ internal_send_command_async(Sock, Channel, MethodRecord, Content, FrameMax) ->
ok.
port_cmd(Sock, Data) ->
- try erlang:port_command(Sock, Data)
+ try rabbit_net:port_command(Sock, Data)
catch error:Error -> exit({writer, send_failed, Error})
end.
diff --git a/src/tcp_listener.erl b/src/tcp_listener.erl
index 92a47cf1..4a2e149b 100644
--- a/src/tcp_listener.erl
+++ b/src/tcp_listener.erl
@@ -33,28 +33,28 @@
-behaviour(gen_server).
--export([start_link/7]).
+-export([start_link/8]).
-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
terminate/2, code_change/3]).
--record(state, {sock, on_startup, on_shutdown}).
+-record(state, {sock, on_startup, on_shutdown, label}).
%%--------------------------------------------------------------------
start_link(IPAddress, Port, SocketOpts,
ConcurrentAcceptorCount, AcceptorSup,
- OnStartup, OnShutdown) ->
+ OnStartup, OnShutdown, Label) ->
gen_server:start_link(
?MODULE, {IPAddress, Port, SocketOpts,
ConcurrentAcceptorCount, AcceptorSup,
- OnStartup, OnShutdown}, []).
+ OnStartup, OnShutdown, Label}, []).
%%--------------------------------------------------------------------
init({IPAddress, Port, SocketOpts,
ConcurrentAcceptorCount, AcceptorSup,
- {M,F,A} = OnStartup, OnShutdown}) ->
+ {M,F,A} = OnStartup, OnShutdown, Label}) ->
process_flag(trap_exit, true),
case gen_tcp:listen(Port, SocketOpts ++ [{ip, IPAddress},
{active, false}]) of
@@ -65,15 +65,16 @@ init({IPAddress, Port, SocketOpts,
end,
lists:duplicate(ConcurrentAcceptorCount, dummy)),
{ok, {LIPAddress, LPort}} = inet:sockname(LSock),
- error_logger:info_msg("started TCP listener on ~s:~p~n",
- [inet_parse:ntoa(LIPAddress), LPort]),
+ error_logger:info_msg("started ~s on ~s:~p~n",
+ [Label, inet_parse:ntoa(LIPAddress), LPort]),
apply(M, F, A ++ [IPAddress, Port]),
- {ok, #state{sock=LSock,
- on_startup = OnStartup, on_shutdown = OnShutdown}};
+ {ok, #state{sock = LSock,
+ on_startup = OnStartup, on_shutdown = OnShutdown,
+ label = Label}};
{error, Reason} ->
error_logger:error_msg(
- "failed to start TCP listener on ~s:~p - ~p~n",
- [inet_parse:ntoa(IPAddress), Port, Reason]),
+ "failed to start ~s on ~s:~p - ~p~n",
+ [Label, inet_parse:ntoa(IPAddress), Port, Reason]),
{stop, {cannot_listen, IPAddress, Port, Reason}}
end.
@@ -86,11 +87,11 @@ handle_cast(_Msg, State) ->
handle_info(_Info, State) ->
{noreply, State}.
-terminate(_Reason, #state{sock=LSock, on_shutdown = {M,F,A}}) ->
+terminate(_Reason, #state{sock=LSock, on_shutdown = {M,F,A}, label=Label}) ->
{ok, {IPAddress, Port}} = inet:sockname(LSock),
gen_tcp:close(LSock),
- error_logger:info_msg("stopped TCP listener on ~s:~p~n",
- [inet_parse:ntoa(IPAddress), Port]),
+ error_logger:info_msg("stopped ~s on ~s:~p~n",
+ [Label, inet_parse:ntoa(IPAddress), Port]),
apply(M, F, A ++ [IPAddress, Port]).
code_change(_OldVsn, State, _Extra) ->
diff --git a/src/tcp_listener_sup.erl b/src/tcp_listener_sup.erl
index 901a0da3..d6bbac08 100644
--- a/src/tcp_listener_sup.erl
+++ b/src/tcp_listener_sup.erl
@@ -33,23 +33,23 @@
-behaviour(supervisor).
--export([start_link/6, start_link/7]).
+-export([start_link/7, start_link/8]).
-export([init/1]).
start_link(IPAddress, Port, SocketOpts, OnStartup, OnShutdown,
- AcceptCallback) ->
+ AcceptCallback, Label) ->
start_link(IPAddress, Port, SocketOpts, OnStartup, OnShutdown,
- AcceptCallback, 1).
+ AcceptCallback, 1, Label).
start_link(IPAddress, Port, SocketOpts, OnStartup, OnShutdown,
- AcceptCallback, ConcurrentAcceptorCount) ->
+ AcceptCallback, ConcurrentAcceptorCount, Label) ->
supervisor:start_link(
?MODULE, {IPAddress, Port, SocketOpts, OnStartup, OnShutdown,
- AcceptCallback, ConcurrentAcceptorCount}).
+ AcceptCallback, ConcurrentAcceptorCount, Label}).
init({IPAddress, Port, SocketOpts, OnStartup, OnShutdown,
- AcceptCallback, ConcurrentAcceptorCount}) ->
+ AcceptCallback, ConcurrentAcceptorCount, Label}) ->
%% This is gross. The tcp_listener needs to know about the
%% tcp_acceptor_sup, and the only way I can think of accomplishing
%% that without jumping through hoops is to register the
@@ -62,5 +62,5 @@ init({IPAddress, Port, SocketOpts, OnStartup, OnShutdown,
{tcp_listener, {tcp_listener, start_link,
[IPAddress, Port, SocketOpts,
ConcurrentAcceptorCount, Name,
- OnStartup, OnShutdown]},
+ OnStartup, OnShutdown, Label]},
transient, 100, worker, [tcp_listener]}]}}.
View Lorry Controller Status