Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feat: Use ets for last-known server-time. #136

Merged
merged 13 commits into from
Oct 22, 2024
2 changes: 1 addition & 1 deletion Makefile
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ doc:

tests:
docker run --name ld-test-redis -p 6379:6379 -d redis
@$(REBAR3) ct --dir="test,test-redis" --logdir logs/ct --cover
@$(REBAR3) ct --dir="test,test-redis" --logdir logs/ct
docker rm --force ld-test-redis

#This is used in running releaser. In this environment we do not want to run the redis tests.
Expand Down
29 changes: 18 additions & 11 deletions src/ldclient_backoff.erl
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@
-module(ldclient_backoff).

%% API
-export([init/4, fail/1, succeed/1, fire/1]).
-export([init/4, init/5, fail/1, succeed/1, fire/1]).

%% Types

Expand All @@ -19,7 +19,8 @@
active_since => integer() | undefined,
destination => pid(),
value => term(),
max_exp => float()
max_exp => float(),
uniform => fun(() -> float())
}.

-define(JITTER_RATIO, 0.5).
Expand All @@ -36,6 +37,11 @@

-spec init(Initial :: non_neg_integer(), Max :: non_neg_integer(), Destination :: pid(), Value :: term()) -> backoff().
init(Initial, Max, Destination, Value) ->
init(Initial, Max, Destination, Value, fun() -> rand:uniform() end).

%% This version of the function exists for testing and allows injecting the random number source.
-spec init(Initial :: non_neg_integer(), Max :: non_neg_integer(), Destination :: pid(), Value :: term(), Uniform :: fun(() -> float())) -> backoff().
init(Initial, Max, Destination, Value, Uniform) ->
SafeInitial = lists:max([Initial, 1]),
#{
initial => SafeInitial, %% Do not allow initial delay to be 0 or negative.
Expand All @@ -47,10 +53,12 @@ init(Initial, Max, Destination, Value) ->
value => Value,
%% The exponent at which the backoff delay will exceed the maximum.
%% Beyond this limit the backoff can be set to the max.
max_exp => math:ceil(math:log2(Max/SafeInitial))
max_exp => math:ceil(math:log2(Max/SafeInitial)),
%% For reasonable values this should ensure we never overflow.
%% Note that while integers can be arbitrarily large the math library uses C functions
%% that are implemented with floats.
%% Allow for alternate random number source.
uniform => Uniform
}.

%% @doc Get an updated backoff with updated delay. Does not start a timer automatically.
Expand Down Expand Up @@ -98,17 +106,16 @@ update_backoff(#{attempt := Attempt} = Backoff, _ActiveDuration) ->
Backoff#{current => delay(NewAttempt, Backoff), attempt => NewAttempt, active_since => undefined}.

-spec delay(Attempt :: non_neg_integer(), Backoff :: backoff()) -> non_neg_integer().
delay(Attempt, #{initial := Initial, max := Max, max_exp := MaxExp} = _Backoff)
delay(Attempt, #{initial := Initial, max := Max, max_exp := MaxExp, uniform := Uniform} = _Backoff)
when Attempt - 1 < MaxExp ->
jitter(min(backoff(Initial, Attempt), Max));
delay(_Attempt, #{max := Max} = _Backoff) ->
jitter(Max).
jitter(min(backoff(Initial, Attempt), Max), Uniform);
delay(_Attempt, #{max := Max, uniform := Uniform} = _Backoff) ->
jitter(Max, Uniform).

-spec backoff(Initial :: non_neg_integer(), Attempt :: non_neg_integer()) -> non_neg_integer().
backoff(Initial, Attempt) ->
trunc(Initial * (math:pow(2, Attempt - 1))).

-spec jitter(Value :: non_neg_integer()) -> non_neg_integer().
jitter(Value) ->
trunc(Value - (rand:uniform() * ?JITTER_RATIO * Value)).

-spec jitter(Value :: non_neg_integer(), Uniform :: fun(() -> float())) -> non_neg_integer().
jitter(Value, Uniform) ->
trunc(Value - (Uniform() * ?JITTER_RATIO * Value)).
33 changes: 23 additions & 10 deletions src/ldclient_event_process_server.erl
Original file line number Diff line number Diff line change
Expand Up @@ -27,10 +27,11 @@
global_private_attributes := ldclient_config:private_attributes(),
events_uri := string(),
tag := atom(),
dispatcher_state := any(),
last_server_time := integer()
dispatcher_state := any()
}.

-define(TABLE_PREFIX, "event_process_state").

%%===================================================================
%% API
%%===================================================================
Expand All @@ -46,8 +47,17 @@ send_events(Tag, Events, SummaryEvent) ->

-spec get_last_server_time(Tag :: atom()) -> integer().
get_last_server_time(Tag) ->
Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Instead of calling into the process this now looks up a value from ets.

ServerName = get_local_reg_name(Tag),
gen_server:call(ServerName, {get_last_server_time}).
TableName = ets_table_name(Tag),
case ets:info(TableName) of
undefined ->
0;
_ ->
case ets:lookup(TableName, last_known_server_time) of
[] -> 0;
[{last_known_server_time, LastKnownServerTime}] -> LastKnownServerTime
end
end.


%%===================================================================
%% Supervision
Expand All @@ -67,6 +77,7 @@ start_link(Tag) ->
{ok, State :: state()} | {ok, State :: state(), timeout() | hibernate} |
{stop, Reason :: term()} | ignore.
init([Tag]) ->
_Tid = ets:new(ets_table_name(Tag), [set, named_table, {read_concurrency, true}]),
Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This theoretically should be a read-heavy operation.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If this data were large, then write_concurrency would probably be reasonable, but it is very small.

SdkKey = ldclient_config:get_value(Tag, sdk_key),
Dispatcher = ldclient_config:get_value(Tag, events_dispatcher),
GlobalPrivateAttributes = ldclient_config:get_value(Tag, private_attributes),
Expand All @@ -77,8 +88,7 @@ init([Tag]) ->
global_private_attributes => GlobalPrivateAttributes,
events_uri => EventsUri,
tag => Tag,
dispatcher_state => Dispatcher:init(Tag, SdkKey),
last_server_time => 0
dispatcher_state => Dispatcher:init(Tag, SdkKey)
},
{ok, State}.

Expand All @@ -90,8 +100,6 @@ init([Tag]) ->
-spec handle_call(Request :: term(), From :: from(), State :: state()) ->
{reply, Reply :: term(), NewState :: state()} |
{stop, normal, {error, atom(), term()}, state()}.
handle_call({get_last_server_time}, _From, #{last_server_time := LastServerTime} = State) ->
{reply, LastServerTime, State};
handle_call(_Request, _From, State) ->
{reply, ok, State}.

Expand All @@ -101,7 +109,8 @@ handle_cast({send_events, Events, SummaryEvent},
dispatcher := Dispatcher,
global_private_attributes := GlobalPrivateAttributes,
events_uri := Uri,
dispatcher_state := DispatcherState
dispatcher_state := DispatcherState,
tag := Tag
} = State) ->
FormattedSummaryEvent = format_summary_event(SummaryEvent),
FormattedEvents = format_events(Events, GlobalPrivateAttributes),
Expand All @@ -111,7 +120,8 @@ handle_cast({send_events, Events, SummaryEvent},
ok ->
State;
{ok, Date} ->
State#{last_server_time => Date};
ets:insert(ets_table_name(Tag), {last_known_server_time, Date}),
State;
{error, temporary, _Reason} ->
erlang:send_after(1000, self(), {send, OutputEvents, PayloadId}),
State;
Expand Down Expand Up @@ -321,3 +331,6 @@ send(Dispatcher, DispatcherState, OutputEvents, PayloadId, Uri) ->
-spec get_local_reg_name(Tag :: atom()) -> atom().
get_local_reg_name(Tag) ->
list_to_atom("ldclient_event_process_server_" ++ atom_to_list(Tag)).

-spec ets_table_name(Tag :: atom()) -> atom().
ets_table_name(Tag) -> list_to_atom(?TABLE_PREFIX ++ atom_to_list(Tag)).
10 changes: 4 additions & 6 deletions test/ldclient_backoff_SUITE.erl
Original file line number Diff line number Diff line change
Expand Up @@ -60,9 +60,7 @@ end_per_testcase(_, _Config) ->
%%====================================================================

backoff_client(InitialDelay, Ratio) ->
meck:new(rand, [unstick]),
meck:expect(rand, uniform, fun() -> Ratio end),
ldclient_backoff:init(InitialDelay, 30000, 0, listen).
ldclient_backoff:init(InitialDelay, 30000, 0, listen, fun() -> Ratio end).

backoff_client(InitialDelay) ->
backoff_client(InitialDelay, 0).
Expand Down Expand Up @@ -100,7 +98,7 @@ delay_doubles_consecutive_failures(_) ->
#{current := 16000} = FifthUpdate,
SixthUpdate = ldclient_backoff:fail(FifthUpdate),
#{current := 30000} = SixthUpdate.


backoff_respects_max(_) ->
Backoff =
Expand Down Expand Up @@ -166,15 +164,15 @@ handles_initial_greater_than_max(_) ->
30000 = ldclient_backoff:delay(1, Backoff),
30000 = ldclient_backoff:delay(2, Backoff).

handles_bad_initial_retry(_) ->
handles_bad_initial_retry(_) ->
Backoff = backoff_client(0),
1 = ldclient_backoff:delay(1, Backoff),
2 = ldclient_backoff:delay(2, Backoff),
4 = ldclient_backoff:delay(3, Backoff),
16384 = ldclient_backoff:delay(15, Backoff),
30000 = ldclient_backoff:delay(16, Backoff),
%% Second backoff we do not use backoff_client as meck is already setup.
Backoff2 = ldclient_backoff:init(-100, 30000, 0, listen),
Backoff2 = backoff_client(-100),
1 = ldclient_backoff:delay(1, Backoff2),
2 = ldclient_backoff:delay(2, Backoff2),
4 = ldclient_backoff:delay(3, Backoff2),
Expand Down