Packages
finitomata
0.23.4
0.41.0
0.40.0
0.35.0
0.34.0
0.33.0
0.32.0
0.31.1
0.30.3
0.30.2
0.30.1
0.30.0
0.29.10
0.29.9
0.29.8
0.29.7
0.29.6
0.29.5
0.29.4
0.29.3
0.29.2
0.29.1
0.29.0
0.28.1
0.28.0
0.27.1
0.27.0
0.26.4
0.26.3
0.26.2
0.26.1
0.26.0
0.25.0
0.24.4
0.24.3
0.24.2
0.24.1
0.24.0
0.23.7
0.23.6
0.23.5
0.23.4
0.23.3
0.23.2
0.23.1
0.23.0
0.22.1
0.22.0
0.21.4
0.21.3
0.21.2
0.21.1
0.21.0
0.20.2
0.20.1
0.20.0
0.19.6
0.19.5
0.19.4
0.19.3
0.19.2
0.19.1
0.19.0
0.18.4
0.18.3
0.18.2
0.18.1
0.18.0
0.17.1
0.17.0
0.16.0
0.15.1
0.15.0
0.14.6
0.14.5
0.14.4
0.14.3
0.14.2
0.14.1
0.14.0
0.13.0
0.12.1
0.12.0
0.11.3
0.11.2
0.11.1
0.11.0
0.10.0
0.9.1
0.9.0
0.8.2
0.8.1
0.8.0
0.7.2
0.7.1
0.7.0
0.6.3
0.6.2
0.6.1
0.6.0
0.5.2
0.5.1
0.5.0
0.4.0
0.3.0
0.2.0
0.1.1
0.1.0
The FSM implementation generated from PlantUML textual representation.
Current section
Files
Jump to
Current section
Files
lib/finitomata/throttler.ex
defmodule Finitomata.Throttler do
@moduledoc """
The internal definition of the call to throttle.
`Finitomata.Throttler.call/3` is a blocking call similar to `GenServer.call/3`, but
served by the underlying `GenStage` producer-consumer pair.
Despite this implementation of throttling based on `GenStage` is provided
mostly for internal needs, it is generic enough to use wherever. Use the childspec
`{Finitomata.Throttler, name: name, initial: [], max_demand: 3, interval: 1_000}`
to start a throttling process and `Finitomata.Throttler.call/3` to perform throttled
synchronous calls from different processes.
### Usage
```elixir
{:ok, pid} = Finitomata.Throttler.start_link(name: Throttler)
Finitomata.Throttler.call(Throttler, {IO, :inspect, [42]})
42
#⇒ %Finitomata.Throttler{
# from: {#PID<0.335.0>, #Reference<0.3154300821.2643722246.59214>},
# fun: {IO, :inspect},
# args: ~c"*",
# result: 42,
# duration: 192402,
# payload: nil
# }
```
"""
@typedoc "The _in/out_ parameter for calls to `Finitomata.Throttler.call/3`"
@type t :: %{
__struct__: Finitomata.Throttler,
from: GenServer.from(),
fun: (keyword() -> any()),
args: keyword(),
result: any(),
duration: pos_integer(),
payload: any()
}
@typedoc "The simplified _in_ parameter for calls to `Finitomata.Throttler.call/3`"
@type throttlee :: t() | {(keyword() -> any()), [any()]}
defstruct ~w|from fun args result duration payload|a
use Supervisor
require Logger
alias Finitomata.Throttler.{Consumer, Producer}
@doc """
Starts the throttler with the underlying producer-consumer stages.
Accepted options are:
- `name` the base name for the throttler to be used in calls to `call/3`
- `initial` the initial load of requests (avoid using it unless really needed)
- `max_demand`, `initial` the options to be passed directly to `GenStage`’s consumer
"""
def start_link(opts) do
name =
opts
|> Keyword.get(:name)
|> Finitomata.Supervisor.throttler_name()
opts = Keyword.put_new(opts, :name, name)
Supervisor.start_link(__MODULE__, opts, name: name)
end
@doc false
@impl Supervisor
def init(opts) do
{initial, opts} = Keyword.pop(opts, :initial, [])
{name, opts} = Keyword.pop!(opts, :name)
children = [
{Producer, initial},
{Consumer, opts}
]
flags = Supervisor.init(children, strategy: :one_for_one)
Task.start_link(fn ->
opts =
opts
|> Keyword.take(~w|max_demand interval|a)
|> Keyword.put_new(:to, producer(name))
name
|> consumer()
|> GenStage.sync_subscribe(opts)
end)
flags
end
@doc """
Synchronously executes the function, using throttling based on `GenStage`.
This function has a default timeout `:infinity` because of its nature
(throttling is supposed to take a while,) but it might be passed as the third
argument in a call to `call/3`.
If a list of functions is given, executes all of them in parallel,
collects the results, and then returns them to the caller.
The function might be given as `t:Finitomata.Throttler.t/0` or
in a simplified form as `{function_of_arity_1, arg}` or `{mod, fun, args}`.
"""
@spec call(Finitomata.id(), t() | {(any() -> any()), arg} | {module(), atom(), [arg]}) :: any()
when arg: any()
def call(name \\ nil, request, timeout \\ :infinity)
def call(name, requests, timeout) when is_list(requests) do
requests
|> Enum.map(&Task.async(Finitomata.Throttler, :call, [name, &1, timeout]))
|> Task.await_many()
end
@utc_now_truncate_to if(Version.compare(System.version(), "1.15.0") == :lt,
do: Calendar.ISO,
else: :microsecond
)
def call(name, request, timeout) do
name
|> producer()
|> GenStage.call({:add, request}, timeout)
|> then(
&%Finitomata.Throttler{
&1
| duration:
DateTime.diff(DateTime.utc_now(@utc_now_truncate_to), &1.duration, :microsecond)
}
)
end
@doc false
def producer(name \\ nil), do: lookup(Producer, name)
@doc false
def consumer(name \\ nil), do: lookup(Consumer, name)
@doc false
def debug(any, opts \\ []) do
any
|> inspect(opts)
|> Logger.debug()
end
defp lookup(who, name) do
name
|> Finitomata.Supervisor.throttler_name()
|> Supervisor.which_children()
|> Enum.find(&match?({_name, _pid, :worker, [^who]}, &1))
|> case do
{_, pid, _, _} when is_pid(pid) -> pid
_ -> nil
end
end
end