%% This Source Code Form is subject to the terms of the Mozilla Public %% License, v. 2.0. If a copy of the MPL was not distributed with this %% file, You can obtain one at https://mozilla.org/MPL/2.0/. %% %% Copyright (c) 2017-2025 Broadcom. All Rights Reserved. The term Broadcom refers to Broadcom Inc. and/or its subsidiaries. %% %% @hidden -module(ra_li). %% Leaky integrator for rate estimation. %% Accumulates values over time while exponentially decaying toward zero. %% Useful for capturing smoothed rates (e.g., messages/second, bytes/second). -export([ new/1, update/2, update/3, read/1, read/2, rate/1, rate/2, reset/1 ]). -record(?MODULE, {decay_time_ms :: pos_integer(), value = 0.0 :: float(), last_update :: integer() | undefined}). -opaque state() :: #?MODULE{}. -export_type([ state/0 ]). %% @doc Create a new leaky integrator with the specified decay time in milliseconds. -spec new(DecayTimeMs :: pos_integer()) -> state(). new(DecayTimeMs) when is_integer(DecayTimeMs), DecayTimeMs > 0 -> #?MODULE{decay_time_ms = DecayTimeMs}. %% @doc Update the integrator with a value, using current monotonic time. -spec update(Amount :: number(), state()) -> state(). update(Amount, State) -> update(Amount, ts(), State). %% @doc Update the integrator with a value at the specified timestamp. -spec update(Amount :: number(), Ts :: integer(), state()) -> state(). update(Amount, Ts, #?MODULE{last_update = undefined} = State) -> State#?MODULE{value = float(Amount), last_update = Ts}; update(Amount, Ts, #?MODULE{decay_time_ms = DecayTimeMs, value = Value, last_update = LastUpdate} = State) when Ts >= LastUpdate -> Elapsed = Ts - LastUpdate, Decayed = decay(Value, Elapsed, DecayTimeMs), State#?MODULE{value = Decayed + Amount, last_update = Ts}; update(Amount, _Ts, #?MODULE{value = Value} = State) -> %% Timestamp went backwards - add amount but don't update last_update State#?MODULE{value = Value + Amount}. %% @doc Read the current decayed value using current monotonic time. -spec read(state()) -> float(). read(State) -> read(ts(), State). %% @doc Read the current decayed value at the specified timestamp. -spec read(Ts :: integer(), state()) -> float(). read(_Ts, #?MODULE{last_update = undefined}) -> 0.0; read(Ts, #?MODULE{decay_time_ms = DecayTimeMs, value = Value, last_update = LastUpdate}) -> Elapsed = max(0, Ts - LastUpdate), decay(Value, Elapsed, DecayTimeMs). %% @doc Return the current rate (value / decay_time_seconds) using current monotonic time. -spec rate(state()) -> float(). rate(State) -> rate(ts(), State). %% @doc Return the current rate at the specified timestamp. %% Rate is calculated as value / decay_time, giving units per decay period. %% For a 1000ms decay time, this gives units per second. -spec rate(Ts :: integer(), state()) -> float(). rate(Ts, #?MODULE{decay_time_ms = DecayTimeMs} = State) -> Value = read(Ts, State), %% Convert to rate per second Value / DecayTimeMs * 1000. %% @doc Reset the integrator to zero. -spec reset(state()) -> state(). reset(#?MODULE{} = State) -> State#?MODULE{value = 0.0, last_update = undefined}. %% Internal functions ts() -> erlang:monotonic_time(millisecond). decay(Value, Elapsed, _DecayTimeMs) when Elapsed =< 0 -> Value; decay(Value, Elapsed, DecayTimeMs) -> Value * math:exp(-Elapsed / DecayTimeMs). -ifdef(TEST). -include_lib("eunit/include/eunit.hrl"). new_test() -> S = new(1000), ?assertEqual(1000, S#?MODULE.decay_time_ms), ?assertEqual(0.0, S#?MODULE.value), ?assertEqual(undefined, S#?MODULE.last_update), ok. update_first_test() -> S0 = new(1000), S1 = update(100, 0, S0), ?assertEqual(100.0, S1#?MODULE.value), ?assertEqual(0, S1#?MODULE.last_update), ok. update_with_decay_test() -> S0 = new(1000), S1 = update(100, 0, S0), %% After one decay time, value should be ~36.8% (1/e) S2 = update(0, 1000, S1), ?assert(abs(S2#?MODULE.value - 100 * math:exp(-1)) < 0.001), ok. update_accumulates_test() -> S0 = new(1000), S1 = update(100, 0, S0), S2 = update(100, 0, S1), %% same timestamp, no decay ?assertEqual(200.0, S2#?MODULE.value), ok. read_empty_test() -> S0 = new(1000), V = read(0, S0), ?assertEqual(0.0, V), ok. read_with_decay_test() -> S0 = new(1000), S1 = update(100, 0, S0), %% After 500ms (half decay time), value should be ~60.6% V = read(500, S1), Expected = 100 * math:exp(-0.5), ?assert(abs(V - Expected) < 0.001), %% State should be unchanged (read is non-mutating) ?assertEqual(0, S1#?MODULE.last_update), ok. rate_test() -> S0 = new(1000), S1 = update(1000, 0, S0), %% With 1000 value and 1000ms decay time, rate should be 1000/s R = rate(0, S1), ?assertEqual(1000.0, R), ok. rate_with_decay_test() -> S0 = new(1000), S1 = update(1000, 0, S0), %% After one decay time R = rate(1000, S1), Expected = 1000 * math:exp(-1), ?assert(abs(R - Expected) < 0.001), ok. reset_test() -> S0 = new(1000), S1 = update(100, 0, S0), S2 = reset(S1), ?assertEqual(0.0, S2#?MODULE.value), ?assertEqual(undefined, S2#?MODULE.last_update), ?assertEqual(1000, S2#?MODULE.decay_time_ms), ok. negative_elapsed_test() -> %% If timestamp goes backwards - add amount but keep last_update unchanged S0 = new(1000), S1 = update(100, 1000, S0), S2 = update(50, 500, S1), %% earlier timestamp %% Amount should be added, but last_update should stay at 1000 ?assertEqual(150.0, S2#?MODULE.value), ?assertEqual(1000, S2#?MODULE.last_update), %% Subsequent update should decay from the original last_update S3 = update(0, 2000, S2), Expected = 150.0 * math:exp(-1), %% 1000ms elapsed from last_update=1000 ?assert(abs(S3#?MODULE.value - Expected) < 0.001), ok. -endif.