this repo has no description
Redesign TaggedQueue to use ets (bench tps 25% higher)
+72
-78
+20
-19
lib/servers/tlog.ex
+20
-19
lib/servers/tlog.ex
···
85
85
86
86
def handle_cast({:pop, tag, up_to_version}, state)
87
87
when is_integer(tag) and is_integer(up_to_version) and up_to_version >= 0 do
88
-
tq = TaggedQueue.pop(state.tagged_queue, tag, up_to_version)
89
-
{:noreply, %State{state | tagged_queue: tq}}
88
+
:ok = TaggedQueue.pop(state.tagged_queue, tag, up_to_version)
89
+
{:noreply, state}
90
90
end
91
91
92
92
@spec maybe_write_next(%State{}) :: %State{}
···
106
106
107
107
@spec do_write_batch(%State{}, term, %LogBatch{}) :: %State{}
108
108
defp do_write_batch(%State{} = state, from, %LogBatch{} = batch) do
109
-
tagged_queue =
110
-
batch.tagged_mutations
111
-
|> Enum.reduce(%{}, fn {tags, mutation}, acc ->
112
-
Enum.reduce(tags, acc, fn tag, acc ->
113
-
case acc[tag] do
114
-
nil -> Map.put(acc, tag, [mutation])
115
-
# Note: this reverses mutation order, which would violate correctness
116
-
# if we did not restore the order below
117
-
existing -> Map.put(acc, tag, [mutation | existing])
118
-
end
119
-
end)
109
+
batch.tagged_mutations
110
+
|> Enum.reduce(%{}, fn {tags, mutation}, acc ->
111
+
Enum.reduce(tags, acc, fn tag, acc ->
112
+
case acc[tag] do
113
+
nil -> Map.put(acc, tag, [mutation])
114
+
# Note: this reverses mutation order, which would violate correctness
115
+
# if we did not restore the order below
116
+
existing -> Map.put(acc, tag, [mutation | existing])
117
+
end
120
118
end)
121
-
|> Enum.sort_by(&elem(&1, 0))
122
-
|> Enum.reduce(state.tagged_queue, fn {tag, mutations}, tq ->
123
-
# Note: we restore the mutation order with Enum.reverse here
124
-
TaggedQueue.append_batch(tq, tag, batch.commit_version, Enum.reverse(mutations))
125
-
end)
119
+
end)
120
+
# TODO: this sort is not strictly needed for determinism because
121
+
# TaggedQueue ets insert order should not affect anything
122
+
|> Enum.sort_by(&elem(&1, 0))
123
+
|> Enum.each(fn {tag, mutations} ->
124
+
# Note: we restore the mutation order with Enum.reverse here
125
+
:ok = TaggedQueue.append_batch(state.tagged_queue, tag, batch.commit_version, Enum.reverse(mutations))
126
+
end)
126
127
127
128
# Ack to CommitBuffer
128
129
SimServer.reply(from, :ok)
129
130
130
-
%State{state | version: batch.commit_version, tagged_queue: tagged_queue}
131
+
%State{state | version: batch.commit_version}
131
132
end
132
133
end
+27
-49
lib/tagged_queue.ex
+27
-49
lib/tagged_queue.ex
···
7
7
alias Hobbes.TaggedQueue
8
8
9
9
@type t :: %__MODULE__{
10
-
tagged_mutations: %{integer => :queue.queue({integer, [{binary, binary}]})},
10
+
table: :ets.table,
11
11
}
12
-
@enforce_keys [:tagged_mutations]
12
+
@enforce_keys [:table]
13
13
defstruct @enforce_keys
14
14
15
15
@doc """
···
17
17
"""
18
18
@spec new :: t
19
19
def new do
20
-
%TaggedQueue{tagged_mutations: %{}}
20
+
table = :ets.new(:tagged_queue, [:ordered_set, :private])
21
+
%TaggedQueue{table: table}
21
22
end
22
23
23
-
@spec append_batch(%TaggedQueue{}, integer, non_neg_integer, [{binary, binary}]) :: %TaggedQueue{}
24
+
@spec append_batch(%TaggedQueue{}, integer, non_neg_integer, [{binary, binary}]) :: :ok
24
25
def append_batch(%TaggedQueue{} = tq, tag, version, mutations) when is_integer(tag) and is_integer(version) and is_list(mutations) do
25
-
queue = get_in(tq.tagged_mutations[tag]) || :queue.new()
26
-
queue = :queue.in({version, mutations}, queue)
27
-
28
-
put_in(tq.tagged_mutations[tag], queue)
26
+
true = :ets.insert(tq.table, {{tag, version}, mutations})
27
+
:ok
29
28
end
30
29
31
30
@doc """
···
33
32
"""
34
33
@spec peek(%TaggedQueue{}, integer, non_neg_integer, non_neg_integer) :: []
35
34
def peek(%TaggedQueue{} = tq, tag, start_version, end_version) when is_integer(tag) and is_integer(start_version) and is_integer(end_version) do
36
-
case tq.tagged_mutations[tag] do
37
-
nil ->
38
-
[]
39
-
40
-
queue ->
41
-
peek_queue(queue, start_version, end_version, [])
42
-
|> Enum.reverse()
43
-
end
35
+
do_scan(tq.table, tag, end_version, start_version - 1, [])
36
+
|> Enum.reverse()
44
37
end
45
38
46
-
defp peek_queue(queue, start_version, end_version, acc) do
47
-
case :queue.out(queue) do
48
-
{:empty, _} ->
39
+
defp do_scan(table, tag, end_version, prev_version, acc) do
40
+
case :ets.next_lookup(table, {tag, prev_version}) do
41
+
{_key, [{{^tag, ver}, mutations}]} when ver <= end_version ->
42
+
do_scan(table, tag, end_version, ver, [{ver, mutations} | acc])
43
+
_ ->
49
44
acc
50
-
51
-
{{:value, {ver, _mut} = batch}, queue} ->
52
-
nil
53
-
cond do
54
-
ver > end_version ->
55
-
acc
56
-
ver < start_version ->
57
-
peek_queue(queue, start_version, end_version, acc)
58
-
true ->
59
-
peek_queue(queue, start_version, end_version, [batch | acc])
60
-
end
61
45
end
62
46
end
63
47
64
48
@doc """
65
49
Pops (removes) all mutations for `tag` with a version up to `end_version` (inclusive).
66
50
"""
67
-
@spec pop(%TaggedQueue{}, integer, non_neg_integer) :: %TaggedQueue{}
51
+
@spec pop(%TaggedQueue{}, integer, non_neg_integer) :: :ok
68
52
def pop(%TaggedQueue{} = tq, tag, end_version) when is_integer(tag) and is_integer(end_version) do
69
-
case tq.tagged_mutations[tag] do
70
-
nil ->
71
-
tq
72
-
queue ->
73
-
queue = pop_queue(queue, end_version)
74
-
put_in(tq.tagged_mutations[tag], queue)
75
-
end
53
+
:ets.select_delete(tq.table, [{
54
+
{{:"$1", :"$2"}, :"$3"},
55
+
[
56
+
{:"==", :"$1", tag},
57
+
{:"=<", :"$2", end_version},
58
+
],
59
+
[true],
60
+
}])
61
+
:ok
76
62
end
77
63
78
-
defp pop_queue(queue, end_version) do
79
-
case :queue.out(queue) do
80
-
{:empty, _} ->
81
-
queue
82
-
83
-
{{:value, {ver, _mut}}, popped_queue} ->
84
-
case ver > end_version do
85
-
true -> queue
86
-
false -> pop_queue(popped_queue, end_version)
87
-
end
88
-
end
64
+
@doc false
65
+
def dump(%TaggedQueue{table: table}) do
66
+
:ets.tab2list(table)
89
67
end
90
68
end
+17
lib/utils.ex
+17
lib/utils.ex
···
183
183
end)
184
184
end)
185
185
end
186
+
187
+
@doc """
188
+
Times a function and logs how long it takes.
189
+
190
+
## Examples
191
+
192
+
iex> time_fn("hello world", fn -> :foo end)
193
+
:foo
194
+
195
+
$ "hello world took 0 ms"
196
+
197
+
"""
198
+
def time_fn(label, func) when is_binary(label) and is_function(func) do
199
+
{time, result} = :timer.tc(func)
200
+
IO.inspect "#{label} took #{time / 1000} ms"
201
+
result
202
+
end
186
203
end
+8
-10
test/tagged_queue_test.exs
+8
-10
test/tagged_queue_test.exs
···
11
11
end
12
12
13
13
test "appends, peeks, and pops", %{tq: tq} do
14
-
tq =
15
-
tq
16
-
|> TaggedQueue.append_batch(1, 101, [{"foo1", "bar1"}])
17
-
|> TaggedQueue.append_batch(1, 102, [{"foo2", "bar2"}])
18
-
|> TaggedQueue.append_batch(1, 103, [{"foo3", "bar3"}])
19
-
|> TaggedQueue.append_batch(1, 104, [{"foo4", "bar4"}])
20
-
|> TaggedQueue.append_batch(2, 101, [{"2foo1", "2bar1"}])
21
-
|> TaggedQueue.append_batch(2, 102, [{"2foo2", "2bar2"}])
14
+
assert :ok = TaggedQueue.append_batch(tq, 1, 101, [{"foo1", "bar1"}])
15
+
assert :ok = TaggedQueue.append_batch(tq, 1, 102, [{"foo2", "bar2"}])
16
+
assert :ok = TaggedQueue.append_batch(tq, 1, 103, [{"foo3", "bar3"}])
17
+
assert :ok = TaggedQueue.append_batch(tq, 1, 104, [{"foo4", "bar4"}])
18
+
assert :ok = TaggedQueue.append_batch(tq, 2, 101, [{"2foo1", "2bar1"}])
19
+
assert :ok = TaggedQueue.append_batch(tq, 2, 102, [{"2foo2", "2bar2"}])
22
20
23
21
assert TaggedQueue.peek(tq, 1, 101, 104) == [
24
22
{101, [{"foo1", "bar1"}]},
···
35
33
{102, [{"2foo2", "2bar2"}]},
36
34
]
37
35
38
-
tq = TaggedQueue.pop(tq, 1, 102)
36
+
assert :ok = TaggedQueue.pop(tq, 1, 102)
39
37
assert TaggedQueue.peek(tq, 1, 0, 1000) == [
40
38
{103, [{"foo3", "bar3"}]},
41
39
{104, [{"foo4", "bar4"}]},
···
45
43
{102, [{"2foo2", "2bar2"}]},
46
44
]
47
45
48
-
tq = TaggedQueue.pop(tq, 1, 104)
46
+
assert :ok = TaggedQueue.pop(tq, 1, 104)
49
47
assert [] = TaggedQueue.peek(tq, 1, 0, 1000)
50
48
assert [{101, _}, {102, _}] = TaggedQueue.peek(tq, 2, 0, 1000)
51
49
end