diff --git a/fluxer_gateway/src/gateway/gateway_dispatch_relay_batch.erl b/fluxer_gateway/src/gateway/gateway_dispatch_relay_batch.erl index df1f6c6aa..595ff5615 100644 --- a/fluxer_gateway/src/gateway/gateway_dispatch_relay_batch.erl +++ b/fluxer_gateway/src/gateway/gateway_dispatch_relay_batch.erl @@ -12,6 +12,7 @@ current_workers_tuple_normalized/0, normalize_workers_tuple/1, message_queue_len/1, + max_queue/0, start_workers/1, start_worker/1, worker_index/3 @@ -19,6 +20,7 @@ -define(STATE_KEY, {gateway_dispatch_relay, state}). -define(SYNC_TIMEOUT_MS, 5000). +-define(SLOT_KEY(Worker), {?MODULE, Worker}). -spec relay_or_direct_many([pid()], atom(), term()) -> ok. relay_or_direct_many(SessionPids, Event, Payload) -> @@ -39,7 +41,8 @@ relay_or_direct_many(SessionPids, Event, Payload) -> -spec relay_many_to_shards([pid()], atom(), term(), tuple(), pos_integer()) -> ok. relay_many_to_shards(SessionPids, Event, Payload, Workers, Count) -> ShardBuckets = build_shard_buckets(SessionPids, Count), - deliver_shard_buckets(1, Count, ShardBuckets, Event, Payload, Workers). + Deferred = deliver_shard_buckets(1, Count, ShardBuckets, Event, Payload, Workers, []), + deliver_deferred_shards(Deferred, Event, Payload). -spec build_shard_buckets([pid()], pos_integer()) -> tuple(). build_shard_buckets(SessionPids, Count) -> @@ -56,20 +59,35 @@ build_shard_buckets(SessionPids, Count) -> ). -spec deliver_shard_buckets( - pos_integer(), pos_integer(), tuple(), atom(), term(), tuple() -) -> ok. -deliver_shard_buckets(Index, Count, _Buckets, _Event, _Payload, _Workers) when Index > Count -> - ok; -deliver_shard_buckets(Index, Count, Buckets, Event, Payload, Workers) -> - case element(Index, Buckets) of - [] -> ok; - Pids -> deliver_shard(Index, Pids, Event, Payload, Workers) - end, - deliver_shard_buckets(Index + 1, Count, Buckets, Event, Payload, Workers). + pos_integer(), pos_integer(), tuple(), atom(), term(), tuple(), [{pid(), [pid()]}] +) -> [{pid(), [pid()]}]. +deliver_shard_buckets(Index, Count, _Buckets, _Event, _Payload, _Workers, Deferred) when + Index > Count +-> + Deferred; +deliver_shard_buckets(Index, Count, Buckets, Event, Payload, Workers, Deferred) -> + Next = + case element(Index, Buckets) of + [] -> Deferred; + Pids -> deliver_shard(Index, Pids, Event, Payload, Workers, Deferred) + end, + deliver_shard_buckets(Index + 1, Count, Buckets, Event, Payload, Workers, Next). --spec deliver_shard(pos_integer(), [pid()], atom(), term(), tuple()) -> ok. -deliver_shard(Index, Pids, Event, Payload, Workers) -> - enqueue(element(Index, Workers), {deliver_many, Pids, Event, Payload}). +-spec deliver_shard(pos_integer(), [pid()], atom(), term(), tuple(), [{pid(), [pid()]}]) -> + [{pid(), [pid()]}]. +deliver_shard(Index, Pids, Event, Payload, Workers, Deferred) -> + Worker = element(Index, Workers), + case enqueue_async(Worker, {deliver_many, Pids, Event, Payload}) of + ok -> Deferred; + full -> [{Worker, Pids} | Deferred] + end. + +-spec deliver_deferred_shards([{pid(), [pid()]}], atom(), term()) -> ok. +deliver_deferred_shards([], _Event, _Payload) -> + ok; +deliver_deferred_shards([{Worker, Pids} | Rest], Event, Payload) -> + enqueue_sync(Worker, {deliver_many, Pids, Event, Payload}), + deliver_deferred_shards(Rest, Event, Payload). -spec relay_or_direct(pid(), atom(), term()) -> ok. relay_or_direct(SessionPid, Event, Payload) -> @@ -82,9 +100,37 @@ relay_or_direct(SessionPid, Event, Payload) -> -spec enqueue(pid(), term()) -> ok. enqueue(Worker, Msg) -> - case is_over_max_queue(Worker) of - true -> enqueue_sync(Worker, Msg); - false -> gen_server:cast(Worker, Msg) + case enqueue_async(Worker, Msg) of + ok -> ok; + full -> enqueue_sync(Worker, Msg) + end. + +-spec enqueue_async(pid(), term()) -> ok | full. +enqueue_async(Worker, Msg) -> + case claim_queue_slot(Worker, max_queue()) of + ok -> gen_server:cast(Worker, Msg); + full -> full + end. + +-spec claim_queue_slot(pid(), pos_integer()) -> ok | full. +claim_queue_slot(Worker, MaxQueue) -> + case erlang:get(?SLOT_KEY(Worker)) of + Claimed when is_integer(Claimed), Claimed < MaxQueue -> + _ = erlang:put(?SLOT_KEY(Worker), Claimed + 1), + ok; + _ -> + sample_queue_slot(Worker, MaxQueue) + end. + +-spec sample_queue_slot(pid(), pos_integer()) -> ok | full. +sample_queue_slot(Worker, MaxQueue) -> + case message_queue_len(Worker) of + Sampled when Sampled < MaxQueue -> + _ = erlang:put(?SLOT_KEY(Worker), Sampled + 1), + ok; + _ -> + _ = erlang:erase(?SLOT_KEY(Worker)), + full end. -spec enqueue_sync(pid(), term()) -> ok. @@ -95,14 +141,13 @@ enqueue_sync(Worker, Msg) -> exit:_Reason -> ok end. --spec is_over_max_queue(pid()) -> boolean(). -is_over_max_queue(Worker) -> - MaxQueue = max_queue(), - MaxQueue > 0 andalso message_queue_len(Worker) >= MaxQueue. - --spec max_queue() -> non_neg_integer(). +-spec max_queue() -> pos_integer(). max_queue() -> - gateway_rollout_config:gateway_dispatch_relay_max_queue(). + Ceiling = process_health_watchdog:kill_threshold(), + case gateway_rollout_config:gateway_dispatch_relay_max_queue() of + Configured when is_integer(Configured), Configured > 0 -> min(Configured, Ceiling); + _ -> Ceiling + end. -spec current_workers() -> [pid()]. current_workers() -> diff --git a/fluxer_gateway/src/telemetry/process_health_watchdog.erl b/fluxer_gateway/src/telemetry/process_health_watchdog.erl index 07879a594..bc055600a 100644 --- a/fluxer_gateway/src/telemetry/process_health_watchdog.erl +++ b/fluxer_gateway/src/telemetry/process_health_watchdog.erl @@ -4,7 +4,7 @@ -typing([eqwalizer]). -behaviour(gen_server). --export([start_link/0]). +-export([start_link/0, kill_threshold/0]). -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). -define(CHECK_INTERVAL_MS, 10_000). @@ -17,6 +17,10 @@ -type queue_history() :: #{pid() => [non_neg_integer()]}. -type state() :: #{history := queue_history()}. +-spec kill_threshold() -> pos_integer(). +kill_threshold() -> + ?KILL_THRESHOLD. + -spec start_link() -> {ok, pid()} | {error, term()}. start_link() -> case gen_server:start_link({local, ?MODULE}, ?MODULE, [], []) of diff --git a/fluxer_gateway/test/gateway_dispatch_relay_bound_tests.erl b/fluxer_gateway/test/gateway_dispatch_relay_bound_tests.erl index 8e7ee29e6..fb1c8d321 100644 --- a/fluxer_gateway/test/gateway_dispatch_relay_bound_tests.erl +++ b/fluxer_gateway/test/gateway_dispatch_relay_bound_tests.erl @@ -10,6 +10,11 @@ -define(BOUND, 8). -define(FILL, 6). -define(EVENT_COUNT, 6). +-define(PROBE_BOUND, 1000). +-define(PROBE_EVENT_COUNT, 200). +-define(PROBE_BUDGET, 2). +-define(SHARD_COUNT, 2). +-define(FAST_SHARD_TIMEOUT_MS, 1500). dispatch_is_bounded_and_stays_ordered_test_() -> {timeout, 30, fun dispatch_is_bounded_and_stays_ordered/0}. @@ -20,8 +25,17 @@ dispatch_many_is_bounded_and_stays_ordered_test_() -> max_queue_config_governs_the_bound_test_() -> {timeout, 30, fun max_queue_config_governs_the_bound/0}. -max_queue_zero_disables_the_bound_test_() -> - {timeout, 30, fun max_queue_zero_disables_the_bound/0}. +max_queue_zero_keeps_the_bound_test_() -> + {timeout, 60, fun max_queue_zero_keeps_the_bound/0}. + +max_queue_is_capped_by_the_watchdog_kill_threshold_test_() -> + {timeout, 30, fun max_queue_is_capped_by_the_watchdog_kill_threshold/0}. + +dispatch_does_not_probe_the_worker_per_event_test_() -> + {timeout, 30, fun dispatch_does_not_probe_the_worker_per_event/0}. + +saturated_shard_does_not_delay_other_shards_test_() -> + {timeout, 60, fun saturated_shard_does_not_delay_other_shards/0}. dispatch_is_bounded_and_stays_ordered() -> assert_bounded_and_ordered(fun send_dispatch/2). @@ -39,8 +53,57 @@ max_queue_config_governs_the_bound() -> ?assertEqual(blocked, producer_status_with_bound(?BOUND, ?BOUND)), ?assertEqual(finished, producer_status_with_bound(?BOUND * 8, ?BOUND)). -max_queue_zero_disables_the_bound() -> - ?assertEqual(finished, producer_status_with_bound(0, ?BOUND)). +max_queue_zero_keeps_the_bound() -> + Ceiling = process_health_watchdog:kill_threshold(), + with_max_queue(0, fun() -> + ?assertEqual(Ceiling, gateway_dispatch_relay_batch:max_queue()) + end), + ?assertEqual(blocked, producer_status_with_bound(0, Ceiling)). + +max_queue_is_capped_by_the_watchdog_kill_threshold() -> + Ceiling = process_health_watchdog:kill_threshold(), + with_max_queue(Ceiling * 4, fun() -> + ?assertEqual(Ceiling, gateway_dispatch_relay_batch:max_queue()) + end), + with_max_queue(Ceiling - 1, fun() -> + ?assertEqual(Ceiling - 1, gateway_dispatch_relay_batch:max_queue()) + end). + +dispatch_does_not_probe_the_worker_per_event() -> + with_max_queue(?PROBE_BOUND, fun() -> + with_worker(fun(Worker) -> + Ref = make_ref(), + Session = spawn_session(self(), Ref), + Producer = spawn_gated_producer(Session, ?PROBE_EVENT_COUNT), + Probes = count_probes(Worker, Producer), + ?assertEqual( + lists:seq(1, ?PROBE_EVENT_COUNT), + collect_observed(Ref, ?PROBE_EVENT_COUNT, []) + ), + Session ! stop, + ?assert(Probes =< ?PROBE_BUDGET) + end) + end). + +saturated_shard_does_not_delay_other_shards() -> + with_max_queue(?BOUND, fun() -> + with_workers(?SHARD_COUNT, fun([Blocked, _Free]) -> + SlowRef = make_ref(), + FastRef = make_ref(), + Slow = session_for_shard(1, self(), SlowRef), + Fast = session_for_shard(2, self(), FastRef), + ok = sys:suspend(Blocked), + fill_queue(Blocked, ?BOUND), + Producer = spawn_fanout_producer([Slow, Fast]), + ?assertEqual([1], collect_within(FastRef, ?FAST_SHARD_TIMEOUT_MS)), + ?assertEqual(blocked, producer_status(Producer, 100)), + ok = sys:resume(Blocked), + ?assertEqual([1], collect_observed(SlowRef, 1, [])), + ?assertEqual(finished, producer_status(Producer, 20000)), + Slow ! stop, + Fast ! stop + end) + end). assert_bounded_and_ordered(SendFun) -> {Status, QueueLen, Observed} = run_over_bound(SendFun), @@ -90,6 +153,58 @@ spawn_producer(SendFun, Session) -> Parent ! {producer_finished, self()} end). +spawn_gated_producer(Session, Count) -> + Parent = self(), + spawn_link(fun() -> + receive + start -> ok + end, + lists:foreach(fun(N) -> ok = send_dispatch(Session, N) end, lists:seq(1, Count)), + Parent ! {producer_finished, self()} + end). + +spawn_fanout_producer(Sessions) -> + Parent = self(), + spawn_link(fun() -> + ok = gateway_dispatch_relay:dispatch_many(Sessions, relay_bound_event, #{<<"n">> => 1}), + Parent ! {producer_finished, self()} + end). + +session_for_shard(Index, Parent, Ref) -> + Session = spawn_session(Parent, Ref), + case erlang:phash2(Session, ?SHARD_COUNT) + 1 of + Index -> Session; + _ -> session_for_shard(Index, Parent, Ref) + end. + +count_probes(Worker, Producer) -> + 1 = erlang:trace(Producer, true, [call]), + _ = erlang:trace_pattern({erlang, process_info, 2}, true, [global]), + Producer ! start, + try + drain_probes(Worker, Producer, 0) + after + _ = erlang:trace_pattern({erlang, process_info, 2}, false, [global]), + untrace(Producer) + end. + +untrace(Producer) -> + try erlang:trace(Producer, false, [call]) of + _ -> ok + catch + error:badarg -> ok + end. + +drain_probes(Worker, Producer, Count) -> + receive + {trace, Producer, call, {erlang, process_info, [Worker, message_queue_len]}} -> + drain_probes(Worker, Producer, Count + 1); + {producer_finished, Producer} -> + Count + after 20000 -> + ?assert(false, {probe_count_timeout, Count}) + end. + await_producer(_Producer, finished) -> ok; await_producer(Producer, blocked) -> @@ -123,6 +238,13 @@ session_loop(Parent, Ref) -> ok end. +collect_within(Ref, Timeout) -> + receive + {relay_bound_received, Ref, N} -> [N] + after Timeout -> + [] + end. + collect_observed(_Ref, 0, Acc) -> lists:reverse(Acc); collect_observed(Ref, Remaining, Acc) -> @@ -133,11 +255,17 @@ collect_observed(Ref, Remaining, Acc) -> end. with_worker(Fun) -> - Worker = gateway_dispatch_relay_batch:start_worker(0), + with_workers(1, fun([Worker]) -> Fun(Worker) end). + +with_workers(Count, Fun) -> + Workers = [ + gateway_dispatch_relay_batch:start_worker(Index) + || Index <- lists:seq(0, Count - 1) + ], try - with_relay_workers([Worker], fun() -> Fun(Worker) end) + with_relay_workers(Workers, fun() -> Fun(Workers) end) after - stop_worker(Worker) + lists:foreach(fun stop_worker/1, Workers) end. with_relay_workers(Workers, Fun) -> diff --git a/fluxer_gateway/test/gateway_dispatch_relay_ordering_tests.erl b/fluxer_gateway/test/gateway_dispatch_relay_ordering_tests.erl index 38ec87e76..81df36882 100644 --- a/fluxer_gateway/test/gateway_dispatch_relay_ordering_tests.erl +++ b/fluxer_gateway/test/gateway_dispatch_relay_ordering_tests.erl @@ -36,7 +36,7 @@ ordered_delivery(Worker, SendFun) -> Ref = make_ref(), Parent = self(), SessionPid = spawn_link(fun() -> session_loop(Parent, Ref) end), - MaxQueue = gateway_rollout_config:gateway_dispatch_relay_max_queue(), + MaxQueue = gateway_dispatch_relay_batch:max_queue(), ok = sys:suspend(Worker), fill_queue(Worker, MaxQueue - ?EVENT_COUNT), lists:foreach(fun(N) -> ok = SendFun(SessionPid, N) end, lists:seq(1, ?EVENT_COUNT)),