1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
|
defmodule Limiter do
require Logger
@moduledoc """
# Limiter
A concurrency limiter. Limits the number of concurrent invocations possible, without using a worker pool or different processes.
It supports two storage methods:
* **`[atomics](https://erlang.org/doc/man/atomics.html)`** recommended and default if your OTP is > 21.2.
* **`[ets](https://erlang.org/doc/man/ets.html)`** either with a single table per Limiter (faster) or a shared table (better for a large number of limiters).
You would however always want to use atomics, ets is mostly there for backwards compatibility.
"""
@doc """
Initializes a `Limiter`.
"""
@spec new(name, max_running, max_waiting, options) :: :ok | {:error, :existing}
when name: atom(),
max_running: non_neg_integer(),
max_waiting: non_neg_integer() | :infinity,
options: [option],
option: {:wait, non_neg_integer()} | backend,
backend: :atomics | ets_backend,
ets_backend: :ets | {:ets, atom()} | {:ets, ets_name :: atom(), ets_options :: []}
def new(name, max_running, max_waiting, options \\ []) do
name = atom_name(name)
if defined?(name) do
{:error, :existing}
else
wait = Keyword.get(options, :wait, 150)
backend = Keyword.get(options, :backend, default_backend())
{:ok, backend} = setup_backend(backend)
:persistent_term.put(name, {__MODULE__, max_running, max_waiting, backend, wait})
:ok
end
end
@spec set(name, new_max_running, new_max_waiting, options) :: :ok | :error
when name: atom(),
new_max_running: non_neg_integer(),
new_max_waiting: non_neg_integer() | :infinity,
options: [option],
option: {:wait, non_neg_integer()}
@doc "Adjust the limiter limits at runtime"
def set(name, new_max_running, new_max_waiting, options \\ []) do
name = atom_name(name)
if defined?(name) do
new_wait = Keyword.get(options, :wait)
{__MODULE__, max_running, max_waiting, backend, wait} = :persistent_term.get(name)
new =
{__MODULE__, new_max_running || max_running, new_max_waiting || max_waiting, backend,
new_wait || wait}
:persistent_term.put(name, new)
:ok
else
:error
end
end
@spec limit(atom(), function()) :: {:error, :overload} | any()
@doc "Limits invocation of `fun`."
def limit(name, fun) do
do_limit(atom_name(name), fun)
end
defp do_limit(name, fun) do
{__MODULE__, max_running, max_waiting, backend, wait} = :persistent_term.get(name)
max = max_running + max_waiting
counter = inc(backend, name)
cond do
counter <= max_running ->
try do
fun.()
after
dec(backend, name)
end
counter > max ->
dec(backend, name)
{:error, :overload}
counter > max_running ->
wait(backend, name, wait, fun)
end
end
defp wait(backend, name, wait, fun) do
Process.sleep(wait)
dec(backend, name)
do_limit(name, fun)
end
defp inc({:ets, ets}, name) do
:ets.update_counter(ets, name, {2, 1}, {name, 0})
end
defp inc({:atomics, ref}, _) do
:atomics.add_get(ref, 1, 1)
end
def dec({:ets, ets}, name) do
:ets.update_counter(ets, name, {2, -1}, {name, 0})
end
def dec({:atomics, ref}, _) do
:atomics.sub_get(ref, 1, 1)
end
defp atom_name(suffix), do: Module.concat(__MODULE__, suffix)
defp defined?(name) do
{__MODULE__, _, _, _, _, _} = :persistent_term.get(name)
true
rescue
_ -> false
end
defp default_backend() do
if Code.ensure_loaded?(:atomics) do
:atomics
else
Logger.debug("Limiter: atomics not available, using ETS backend")
:ets
end
end
defp setup_backend(:ets) do
setup_backend({:ets, ETS})
end
defp setup_backend({:ets, name}) do
setup_backend({:ets, name, [{:write_concurrency, true}, {:read_concurrency, true}]})
end
defp setup_backend({:ets, name, options}) do
ets_name = atom_name(name)
case :ets.whereis(ets_name) do
:undefined -> :ets.new(ets_name, [:public, :named_table] ++ options)
_ -> nil
end
{:ok, {:ets, ets_name}}
end
defp setup_backend(:atomics) do
{:ok, {:atomics, :atomics.new(1, signed: true)}}
end
end
|
