diff options
author | Emile Joubert <emile@rabbitmq.com> | 2013-01-25 16:48:16 +0000 |
---|---|---|
committer | Emile Joubert <emile@rabbitmq.com> | 2013-01-25 16:48:16 +0000 |
commit | 58d814681fdc33df35e161c9580c2c924fcffccf (patch) | |
tree | 297ea79866f3b5219dfdf40fd3264f1a119763cc | |
parent | 45f7849be58f78a23e28061fcace8bfdbdaa1e9a (diff) | |
parent | ccd04a74a4780152d42cc0dce4abe9916fd7fdd7 (diff) | |
download | rabbitmq-server-58d814681fdc33df35e161c9580c2c924fcffccf.tar.gz |
Merge default into bug23749 (causing no changes)
-rw-r--r-- | src/rabbit_amqqueue.erl | 6 | ||||
-rw-r--r-- | src/rabbit_amqqueue_process.erl | 101 | ||||
-rw-r--r-- | src/rabbit_channel.erl | 64 | ||||
-rw-r--r-- | src/rabbit_limiter.erl | 107 | ||||
-rw-r--r-- | src/rabbit_misc.erl | 51 | ||||
-rw-r--r-- | src/rabbit_tests.erl | 24 |
6 files changed, 309 insertions, 44 deletions
diff --git a/src/rabbit_amqqueue.erl b/src/rabbit_amqqueue.erl index 9228755e..cb7e961d 100644 --- a/src/rabbit_amqqueue.erl +++ b/src/rabbit_amqqueue.erl @@ -28,7 +28,7 @@ -export([consumers/1, consumers_all/1, consumer_info_keys/0]). -export([basic_get/3, basic_consume/7, basic_cancel/4]). -export([notify_sent/2, notify_sent_queue_down/1, unblock/2, flush_all/2]). --export([notify_down_all/2, limit_all/3]). +-export([notify_down_all/2, limit_all/3, inform_limiter/3]). -export([on_node_down/1]). -export([update/2, store_queue/1, policy_changed/2]). -export([start_mirroring/1, stop_mirroring/1, sync_mirrors/1, @@ -145,6 +145,7 @@ -spec(notify_down_all/2 :: (qpids(), pid()) -> ok_or_errors()). -spec(limit_all/3 :: (qpids(), pid(), rabbit_limiter:token()) -> ok_or_errors()). +-spec(inform_limiter/3 :: (rabbit_types:amqqueue(), pid(), any()) -> 'ok'). -spec(basic_get/3 :: (rabbit_types:amqqueue(), pid(), boolean()) -> {'ok', non_neg_integer(), qmsg()} | 'empty'). -spec(basic_consume/7 :: @@ -532,6 +533,9 @@ notify_down_all(QPids, ChPid) -> limit_all(QPids, ChPid, Limiter) -> delegate:cast(QPids, {limit, ChPid, Limiter}). +inform_limiter(#amqqueue{pid = QPid}, ChPid, Msg) -> + delegate:cast(QPid, {inform_limiter, ChPid, Msg}). + basic_get(#amqqueue{pid = QPid}, ChPid, NoAck) -> delegate:call(QPid, {basic_get, ChPid, NoAck}). diff --git a/src/rabbit_amqqueue_process.erl b/src/rabbit_amqqueue_process.erl index fe3a6099..4e249365 100644 --- a/src/rabbit_amqqueue_process.erl +++ b/src/rabbit_amqqueue_process.erl @@ -65,9 +65,18 @@ monitor_ref, acktags, consumer_count, + %% Queue of {ChPid, #consumer{}} for consumers which have + %% been blocked for any reason blocked_consumers, + %% List of consumer tags which have individually been + %% blocked by the limiter. + blocked_ctags, + %% The limiter itself limiter, + %% Has the limiter imposed a channel-wide block, either + %% because of qos or channel flow? is_limit_active, + %% Internal flow control for queue -> writer unsent_message_count}). %%---------------------------------------------------------------------------- @@ -366,6 +375,7 @@ ch_record(ChPid) -> acktags = queue:new(), consumer_count = 0, blocked_consumers = queue:new(), + blocked_ctags = [], is_limit_active = false, limiter = rabbit_limiter:make_token(), unsent_message_count = 0}, @@ -413,13 +423,6 @@ block_consumer(C = #cr{blocked_consumers = Blocked}, QEntry) -> is_ch_blocked(#cr{unsent_message_count = Count, is_limit_active = Limited}) -> Limited orelse Count >= ?UNSENT_MESSAGE_LIMIT. -ch_record_state_transition(OldCR, NewCR) -> - case {is_ch_blocked(OldCR), is_ch_blocked(NewCR)} of - {true, false} -> unblock; - {false, true} -> block; - {_, _} -> ok - end. - deliver_msgs_to_consumers(_DeliverFun, true, State) -> {true, State}; deliver_msgs_to_consumers(DeliverFun, false, @@ -437,17 +440,29 @@ deliver_msgs_to_consumers(DeliverFun, false, deliver_msg_to_consumer(DeliverFun, E = {ChPid, Consumer}, State) -> C = ch_record(ChPid), case is_ch_blocked(C) of - true -> block_consumer(C, E), - {false, State}; - false -> case rabbit_limiter:can_send(C#cr.limiter, self(), - Consumer#consumer.ack_required) of - false -> block_consumer(C#cr{is_limit_active = true}, E), - {false, State}; - true -> AC1 = queue:in(E, State#q.active_consumers), - deliver_msg_to_consumer( - DeliverFun, Consumer, C, - State#q{active_consumers = AC1}) - end + true -> + block_consumer(C, E), + {false, State}; + false -> + #cr{limiter = Limiter, ch_pid = ChPid, blocked_ctags = BCTags} = C, + #consumer{tag = CTag} = Consumer, + case rabbit_limiter:can_cons_send(Limiter, CTag) of + false -> + block_consumer(C#cr{blocked_ctags = [CTag | BCTags]}, E), + {false, State}; + true -> + case rabbit_limiter:can_ch_send( + Limiter, self(), Consumer#consumer.ack_required) of + false -> + block_consumer(C#cr{is_limit_active = true}, E), + {false, State}; + true -> + AC1 = queue:in(E, State#q.active_consumers), + deliver_msg_to_consumer( + DeliverFun, Consumer, C, + State#q{active_consumers = AC1}) + end + end end. deliver_msg_to_consumer(DeliverFun, @@ -455,8 +470,12 @@ deliver_msg_to_consumer(DeliverFun, ack_required = AckRequired}, C = #cr{ch_pid = ChPid, acktags = ChAckTags, + limiter = Limiter, unsent_message_count = Count}, - State = #q{q = #amqqueue{name = QName}}) -> + State = #q{q = #amqqueue{name = QName}, + backing_queue = BQ, + backing_queue_state = BQS}) -> + rabbit_limiter:record_cons_send(Limiter, ChPid, ConsumerTag, BQ:len(BQS)), {{Message, IsDelivered, AckTag}, Stop, State1} = DeliverFun(AckRequired, State), rabbit_channel:deliver(ChPid, ConsumerTag, AckRequired, @@ -605,16 +624,20 @@ possibly_unblock(State, ChPid, Update) -> not_found -> State; C -> - C1 = Update(C), - case ch_record_state_transition(C, C1) of - ok -> update_ch_record(C1), - State; - unblock -> #cr{blocked_consumers = Consumers} = C1, - update_ch_record( - C1#cr{blocked_consumers = queue:new()}), - AC1 = queue:join(State#q.active_consumers, - Consumers), - run_message_queue(State#q{active_consumers = AC1}) + C1 = #cr{blocked_ctags = BCTags1} = Update(C), + {Blocked, Unblocked} = + lists:partition( + fun({_ChPid, #consumer{tag = CTag}}) -> + is_ch_blocked(C1) orelse lists:member(CTag, BCTags1) + end, queue:to_list(C1#cr.blocked_consumers)), + case Unblocked of + [] -> update_ch_record(C1), + State; + _ -> update_ch_record( + C1#cr{blocked_consumers = queue:from_list(Blocked)}), + AC1 = queue:join(State#q.active_consumers, + queue:from_list(Unblocked)), + run_message_queue(State#q{active_consumers = AC1}) end end. @@ -1112,10 +1135,12 @@ handle_call({basic_cancel, ChPid, ConsumerTag, OkMsg}, From, case lookup_ch(ChPid) of not_found -> reply(ok, State); - C = #cr{blocked_consumers = Blocked} -> + C = #cr{limiter = Limiter, blocked_consumers = Blocked} -> emit_consumer_deleted(ChPid, ConsumerTag, qname(State)), + Limiter1 = rabbit_limiter:forget_consumer(Limiter, ConsumerTag), Blocked1 = remove_consumer(ChPid, ConsumerTag, Blocked), - update_consumer_count(C#cr{blocked_consumers = Blocked1}, -1), + update_consumer_count(C#cr{limiter = Limiter1, + blocked_consumers = Blocked1}, -1), State1 = State#q{ exclusive_consumer = case Holder of {ChPid, ConsumerTag} -> none; @@ -1269,7 +1294,9 @@ handle_cast({limit, ChPid, Limiter}, State) -> false -> ok end, Limited = OldLimited andalso rabbit_limiter:is_enabled(Limiter), - C#cr{limiter = Limiter, is_limit_active = Limited} + C#cr{limiter = rabbit_limiter:copy_queue_state( + OldLimiter, Limiter), + is_limit_active = Limited} end)); handle_cast({flush, ChPid}, State) -> @@ -1302,6 +1329,16 @@ handle_cast(stop_mirroring, State = #q{backing_queue = BQ, noreply(State#q{backing_queue = BQ1, backing_queue_state = BQS1}); +handle_cast({inform_limiter, ChPid, Msg}, + State = #q{backing_queue = BQ, + backing_queue_state = BQS}) -> + #cr{limiter = Lim, + blocked_ctags = BCTags} = ch_record(ChPid), + {Unblock, Lim2} = rabbit_limiter:inform(Lim, ChPid, BQ:len(BQS), Msg), + noreply(possibly_unblock( + State, ChPid, fun(C) -> C#cr{blocked_ctags = BCTags -- Unblock, + limiter = Lim2} end)); + handle_cast(wake_up, State) -> noreply(State). diff --git a/src/rabbit_channel.erl b/src/rabbit_channel.erl index 160512a2..56a10676 100644 --- a/src/rabbit_channel.erl +++ b/src/rabbit_channel.erl @@ -21,7 +21,8 @@ -behaviour(gen_server2). -export([start_link/11, do/2, do/3, do_flow/3, flush/1, shutdown/1]). --export([send_command/2, deliver/4, flushed/2]). +-export([send_command/2, deliver/4, send_credit_reply/2, send_drained/3, + flushed/2]). -export([list/0, info_keys/0, info/1, info/2, info_all/0, info_all/1]). -export([refresh_config_local/0, ready_for_close/1]). -export([force_event_refresh/0]). @@ -38,7 +39,7 @@ queue_names, queue_monitors, consumer_mapping, blocking, queue_consumers, delivering_queues, queue_collector_pid, stats_timer, confirm_enabled, publish_seqno, - unconfirmed, confirmed, capabilities, trace_state}). + unconfirmed, confirmed, capabilities, trace_state, credit_map}). -define(MAX_PERMISSION_CACHE_SIZE, 12). @@ -94,6 +95,9 @@ -spec(deliver/4 :: (pid(), rabbit_types:ctag(), boolean(), rabbit_amqqueue:qmsg()) -> 'ok'). +-spec(send_credit_reply/2 :: (pid(), non_neg_integer()) -> 'ok'). +-spec(send_drained/3 :: (pid(), rabbit_types:ctag(), non_neg_integer()) + -> 'ok'). -spec(flushed/2 :: (pid(), pid()) -> 'ok'). -spec(list/0 :: () -> [pid()]). -spec(list_local/0 :: () -> [pid()]). @@ -138,6 +142,12 @@ send_command(Pid, Msg) -> deliver(Pid, ConsumerTag, AckRequired, Msg) -> gen_server2:cast(Pid, {deliver, ConsumerTag, AckRequired, Msg}). +send_credit_reply(Pid, Len) -> + gen_server2:cast(Pid, {send_credit_reply, Len}). + +send_drained(Pid, ConsumerTag, Count) -> + gen_server2:cast(Pid, {send_drained, ConsumerTag, Count}). + flushed(Pid, QPid) -> gen_server2:cast(Pid, {flushed, QPid}). @@ -209,7 +219,8 @@ init([Channel, ReaderPid, WriterPid, ConnPid, ConnName, Protocol, User, VHost, unconfirmed = dtree:empty(), confirmed = [], capabilities = Capabilities, - trace_state = rabbit_trace:init(VHost)}, + trace_state = rabbit_trace:init(VHost), + credit_map = dict:new()}, State1 = rabbit_event:init_stats_timer(State, #ch.stats_timer), rabbit_event:notify(channel_created, infos(?CREATION_EVENT_KEYS, State1)), rabbit_event:if_enabled(State1, #ch.stats_timer, @@ -315,6 +326,21 @@ handle_cast({deliver, ConsumerTag, AckRequired, Content), noreply(record_sent(ConsumerTag, AckRequired, Msg, State)); +handle_cast({send_credit_reply, Len}, State = #ch{writer_pid = WriterPid}) -> + ok = rabbit_writer:send_command( + WriterPid, #'basic.credit_ok'{available = Len}), + noreply(State); + +handle_cast({send_drained, ConsumerTag, Count}, + State = #ch{writer_pid = WriterPid}) -> + ok = rabbit_writer:send_command( + WriterPid, #'basic.credit_state'{consumer_tag = ConsumerTag, + credit = 0, + count = Count, + available = 0, + drain = true}), + noreply(State); + handle_cast(force_event_refresh, State) -> rabbit_event:notify(channel_created, infos(?CREATION_EVENT_KEYS, State)), noreply(State); @@ -697,7 +723,8 @@ handle_method(#'basic.consume'{queue = QueueNameBin, nowait = NoWait}, _, State = #ch{conn_pid = ConnPid, limiter = Limiter, - consumer_mapping = ConsumerMapping}) -> + consumer_mapping = ConsumerMapping, + credit_map = CreditMap}) -> case dict:find(ConsumerTag, ConsumerMapping) of error -> QueueName = expand_queue_name_shortcut(QueueNameBin, State), @@ -715,6 +742,15 @@ handle_method(#'basic.consume'{queue = QueueNameBin, case rabbit_amqqueue:with_exclusive_access_or_die( QueueName, ConnPid, fun (Q) -> + case dict:find(ActualConsumerTag, CreditMap) of + {ok, {Credit, Count, Drain}} -> + ok = rabbit_amqqueue:inform_limiter( + Q, self(), + {basic_credit, ActualConsumerTag, + Credit, Count, Drain, false}); + error -> + ok + end, {rabbit_amqqueue:basic_consume( Q, NoAck, self(), Limiter, ActualConsumerTag, ExclusiveConsume, @@ -724,9 +760,11 @@ handle_method(#'basic.consume'{queue = QueueNameBin, end) of {ok, Q = #amqqueue{pid = QPid, name = QName}} -> CM1 = dict:store(ActualConsumerTag, Q, ConsumerMapping), + CrM1 = dict:erase(ActualConsumerTag, CreditMap), State1 = monitor_delivering_queue( NoAck, QPid, QName, - State#ch{consumer_mapping = CM1}), + State#ch{consumer_mapping = CM1, + credit_map = CrM1}), {noreply, case NoWait of true -> consumer_monitor(ActualConsumerTag, State1); @@ -1089,6 +1127,22 @@ handle_method(#'channel.flow'{active = false}, _, {noreply, State2} end; +handle_method(#'basic.credit'{consumer_tag = CTag, + credit = Credit, + count = Count, + drain = Drain}, _, + State = #ch{consumer_mapping = Consumers, + credit_map = CMap}) -> + case dict:find(CTag, Consumers) of + {ok, Q} -> ok = rabbit_amqqueue:inform_limiter( + Q, self(), + {basic_credit, CTag, Credit, Count, Drain, true}), + {noreply, State}; + error -> CMap2 = dict:store(CTag, {Credit, Count, Drain}, CMap), + {reply, #'basic.credit_ok'{available = 0}, + State#ch{credit_map = CMap2}} + end; + handle_method(_MethodRecord, _Content, _State) -> rabbit_misc:protocol_error( command_invalid, "unimplemented method", []). diff --git a/src/rabbit_limiter.erl b/src/rabbit_limiter.erl index 8a7d14fe..b97d1073 100644 --- a/src/rabbit_limiter.erl +++ b/src/rabbit_limiter.erl @@ -15,19 +15,25 @@ %% -module(rabbit_limiter). +-include("rabbit_framing.hrl"). -behaviour(gen_server2). -export([init/1, terminate/2, code_change/3, handle_call/3, handle_cast/2, handle_info/2, prioritise_call/3]). + -export([start_link/0, make_token/0, make_token/1, is_enabled/1, enable/2, disable/1]). --export([limit/2, can_send/3, ack/2, register/2, unregister/2]). +-export([limit/2, can_ch_send/3, can_cons_send/2, record_cons_send/4, + ack/2, register/2, unregister/2]). -export([get_limit/1, block/1, unblock/1, is_blocked/1]). +-export([inform/4, forget_consumer/2, copy_queue_state/2]). + +-import(rabbit_misc, [serial_add/2, serial_diff/2]). %%---------------------------------------------------------------------------- --record(token, {pid, enabled}). +-record(token, {pid, enabled, q_state}). -ifdef(use_specs). @@ -42,7 +48,8 @@ -spec(enable/2 :: (token(), non_neg_integer()) -> token()). -spec(disable/1 :: (token()) -> token()). -spec(limit/2 :: (token(), non_neg_integer()) -> 'ok' | {'disabled', token()}). --spec(can_send/3 :: (token(), pid(), boolean()) -> boolean()). +-spec(can_ch_send/3 :: (token(), pid(), boolean()) -> boolean()). +-spec(can_cons_send/2 :: (token(), rabbit_types:ctag()) -> boolean()). -spec(ack/2 :: (token(), non_neg_integer()) -> 'ok'). -spec(register/2 :: (token(), pid()) -> 'ok'). -spec(unregister/2 :: (token(), pid()) -> 'ok'). @@ -50,6 +57,10 @@ -spec(block/1 :: (token()) -> 'ok'). -spec(unblock/1 :: (token()) -> 'ok' | {'disabled', token()}). -spec(is_blocked/1 :: (token()) -> boolean()). +-spec(inform/4 :: (token(), pid(), non_neg_integer(), any()) -> + {[rabbit_types:ctag()], token()}). +-spec(forget_consumer/2 :: (token(), rabbit_types:ctag()) -> token()). +-spec(copy_queue_state/2 :: (token(), token()) -> token()). -endif. @@ -64,6 +75,8 @@ %% notified of a change in the limit or volume that may allow it to %% deliver more messages via the limiter's channel. +-record(credit, {count = 0, credit = 0, drain = false}). + %%---------------------------------------------------------------------------- %% API %%---------------------------------------------------------------------------- @@ -71,7 +84,8 @@ start_link() -> gen_server2:start_link(?MODULE, [], []). make_token() -> make_token(undefined). -make_token(Pid) -> #token{pid = Pid, enabled = false}. +make_token(Pid) -> #token{pid = Pid, enabled = false, + q_state = dict:new()}. is_enabled(#token{enabled = Enabled}) -> Enabled. @@ -88,15 +102,25 @@ limit(Limiter, PrefetchCount) -> %% breaching a limit. Note that we don't use maybe_call here in order %% to avoid always going through with_exit_handler/2, even when the %% limiter is disabled. -can_send(#token{pid = Pid, enabled = true}, QPid, AckRequired) -> +can_ch_send(#token{pid = Pid, enabled = true}, QPid, AckRequired) -> rabbit_misc:with_exit_handler( fun () -> true end, fun () -> gen_server2:call(Pid, {can_send, QPid, AckRequired}, infinity) end); -can_send(_, _, _) -> +can_ch_send(_, _, _) -> true. +can_cons_send(#token{q_state = Credits}, CTag) -> + case dict:find(CTag, Credits) of + {ok, #credit{credit = C}} when C > 0 -> true; + {ok, #credit{}} -> false; + error -> true + end. + +record_cons_send(#token{q_state = QState} = Token, ChPid, CTag, Len) -> + Token#token{q_state = record_send_q(CTag, Len, ChPid, QState)}. + %% Let the limiter know that the channel has received some acks from a %% consumer ack(Limiter, Count) -> maybe_cast(Limiter, {ack, Count}). @@ -119,6 +143,77 @@ unblock(Limiter) -> is_blocked(Limiter) -> maybe_call(Limiter, is_blocked, false). +inform(Limiter = #token{q_state = Credits}, + ChPid, Len, {basic_credit, CTag, Credit, Count, Drain, Reply}) -> + {Unblock, Credits2} = update_credit( + CTag, Len, ChPid, Credit, Count, Drain, Credits), + case Reply of + true -> rabbit_channel:send_credit_reply(ChPid, Len); + false -> ok + end, + {Unblock, Limiter#token{q_state = Credits2}}. + +forget_consumer(Limiter = #token{q_state = Credits}, CTag) -> + Limiter#token{q_state = dict:erase(CTag, Credits)}. + +copy_queue_state(#token{q_state = Credits}, Token) -> + Token#token{q_state = Credits}. + +%%---------------------------------------------------------------------------- +%% Queue-local code +%%---------------------------------------------------------------------------- + +%% We want to do all the AMQP 1.0-ish link level credit calculations in the +%% queue (to do them elsewhere introduces a ton of races). However, it's a big +%% chunk of code that is conceptually very linked to the limiter concept. So +%% we get the queue to hold a bit of state for us (#token.q_state), and +%% maintain a fiction that the limiter is making the decisions... + +record_send_q(CTag, Len, ChPid, Credits) -> + case dict:find(CTag, Credits) of + {ok, Cred} -> + decr_credit(CTag, Len, ChPid, Cred, Credits); + error -> + Credits + end. + +decr_credit(CTag, Len, ChPid, Cred, Credits) -> + #credit{credit = Credit, count = Count, drain = Drain} = Cred, + {NewCredit, NewCount} = maybe_drain(Len - 1, Drain, CTag, ChPid, + Credit - 1, serial_add(Count, 1)), + write_credit(CTag, NewCredit, NewCount, Drain, Credits). + +maybe_drain(0, true, CTag, ChPid, Credit, Count) -> + %% Drain, so advance til credit = 0 + NewCount = serial_add(Count, Credit - 2), + send_drained(ChPid, CTag, NewCount), + {0, NewCount}; %% Magic reduction to 0 + +maybe_drain(_, _, _, _, Credit, Count) -> + {Credit, Count}. + +send_drained(ChPid, CTag, Count) -> + rabbit_channel:send_drained(ChPid, CTag, Count). + +update_credit(CTag, Len, ChPid, Credit, Count0, Drain, Credits) -> + Count = case dict:find(CTag, Credits) of + %% Use our count if we can, more accurate + {ok, #credit{ count = LocalCount }} -> LocalCount; + %% But if this is new, take it from the adapter + _ -> Count0 + end, + {NewCredit, NewCount} = maybe_drain(Len, Drain, CTag, ChPid, Credit, Count), + NewCredits = write_credit(CTag, NewCredit, NewCount, Drain, Credits), + case NewCredit > 0 of + true -> {[CTag], NewCredits}; + false -> {[], NewCredits} + end. + +write_credit(CTag, Credit, Count, Drain, Credits) -> + dict:store(CTag, #credit{credit = Credit, + count = Count, + drain = Drain}, Credits). + %%---------------------------------------------------------------------------- %% gen_server callbacks %%---------------------------------------------------------------------------- diff --git a/src/rabbit_misc.erl b/src/rabbit_misc.erl index c36fb147..135f6443 100644 --- a/src/rabbit_misc.erl +++ b/src/rabbit_misc.erl @@ -69,6 +69,7 @@ -export([interval_operation/4]). -export([ensure_timer/4, stop_timer/2]). -export([get_parent/0]). +-export([serial_add/2, serial_compare/2, serial_diff/2]). %% Horrible macro to use in guards -define(IS_BENIGN_EXIT(R), @@ -83,6 +84,7 @@ -ifdef(use_specs). -export_type([resource_name/0, thunk/1]). +-export_type([serial_number/0]). -type(ok_or_error() :: rabbit_types:ok_or_error(any())). -type(thunk(T) :: fun(() -> T)). @@ -95,6 +97,8 @@ fun ((atom(), [term()]) -> [{digraph:vertex(), digraph_label()}])). -type(graph_edge_fun() :: fun ((atom(), [term()]) -> [{digraph:vertex(), digraph:vertex()}])). +-type(serial_number() :: non_neg_integer()). +-type(serial_compare_result() :: 'equal' | 'less' | 'greater'). -spec(method_record_type/1 :: (rabbit_framing:amqp_method_record()) -> rabbit_framing:amqp_method_name()). @@ -246,6 +250,12 @@ -spec(ensure_timer/4 :: (A, non_neg_integer(), non_neg_integer(), any()) -> A). -spec(stop_timer/2 :: (A, non_neg_integer()) -> A). -spec(get_parent/0 :: () -> pid()). +-spec(serial_add/2 :: (serial_number(), non_neg_integer()) -> + serial_number()). +-spec(serial_compare/2 :: (serial_number(), serial_number()) -> + serial_compare_result()). +-spec(serial_diff/2 :: (serial_number(), serial_number()) -> + integer()). -endif. %%---------------------------------------------------------------------------- @@ -1099,3 +1109,44 @@ whereis_name(Name) -> %% End copypasta from gen_server2.erl %% ------------------------------------------------------------------------- + +%% Serial arithmetic for unsigned ints. +%% http://www.faqs.org/rfcs/rfc1982.html +%% SERIAL_BITS = 32 + +%% 2 ^ SERIAL_BITS +-define(SERIAL_MAX, 16#100000000). +%% 2 ^ (SERIAL_BITS - 1) - 1 +-define(SERIAL_MAX_ADDEND, 16#7fffffff). + +serial_add(S, N) when N =< ?SERIAL_MAX_ADDEND -> + (S + N) rem ?SERIAL_MAX; +serial_add(S, N) -> + exit({out_of_bound_serial_addition, S, N}). + +serial_compare(A, B) -> + if A =:= B -> + equal; + (A < B andalso B - A < ?SERIAL_MAX_ADDEND) orelse + (A > B andalso A - B > ?SERIAL_MAX_ADDEND) -> + less; + (A < B andalso B - A > ?SERIAL_MAX_ADDEND) orelse + (A > B andalso B - A < ?SERIAL_MAX_ADDEND) -> + greater; + true -> exit({indeterminate_serial_comparison, A, B}) + end. + +-define(SERIAL_DIFF_BOUND, 16#80000000). + +serial_diff(A, B) -> + Diff = A - B, + if Diff > (?SERIAL_DIFF_BOUND) -> + %% B is actually greater than A + - (?SERIAL_MAX - Diff); + Diff < - (?SERIAL_DIFF_BOUND) -> + ?SERIAL_MAX + Diff; + Diff < ?SERIAL_DIFF_BOUND andalso Diff > -?SERIAL_DIFF_BOUND -> + Diff; + true -> + exit({indeterminate_serial_diff, A, B}) + end. diff --git a/src/rabbit_tests.erl b/src/rabbit_tests.erl index f5ea4fba..e2af7efd 100644 --- a/src/rabbit_tests.erl +++ b/src/rabbit_tests.erl @@ -50,6 +50,7 @@ all_tests() -> passed = test_table_codec(), passed = test_content_framing(), passed = test_content_transcoding(), + passed = test_serial_arithmetic(), passed = test_topic_matching(), passed = test_log_management(), passed = test_app_management(), @@ -559,6 +560,29 @@ sequence_with_content(Sequence) -> rabbit_framing_amqp_0_9_1), Sequence). +test_serial_arithmetic() -> + 1 = rabbit_misc:serial_add(0, 1), + 16#7fffffff = rabbit_misc:serial_add(0, 16#7fffffff), + 0 = rabbit_misc:serial_add(16#ffffffff, 1), + %% Cannot add more than 2 ^ 31 - 1 + case catch rabbit_misc:serial_add(200, 16#80000000) of + {'EXIT', {out_of_bound_serial_addition, _, _}} -> ok; + _ -> exit(fail_out_of_bound_serial_addition) + end, + + 1 = rabbit_misc:serial_diff(1, 0), + 2 = rabbit_misc:serial_diff(1, 16#ffffffff), + -2 = rabbit_misc:serial_diff(16#ffffffff, 1), + case catch rabbit_misc:serial_diff(0, 16#80000000) of + {'EXIT', {indeterminate_serial_diff, _, _}} -> ok; + _ -> exit(fail_indeterminate_serial_difference) + end, + case catch rabbit_misc:serial_diff(16#ffffffff, 16#7fffffff) of + {'EXIT', {indeterminate_serial_diff, _, _}} -> ok; + _ -> exit(fail_indeterminate_serial_difference) + end, + passed. + test_topic_matching() -> XName = #resource{virtual_host = <<"/">>, kind = exchange, |