this repo has no description
Number mutations in batches in preparation for tlog peek merge
+38
-22
+5
-8
lib/servers/commit_buffer.ex
+5
-8
lib/servers/commit_buffer.ex
···
196
196
allowed_transactions
197
197
|> Enum.map(fn %CommitTxn{} = txn -> txn.mutations end)
198
198
|> Enum.concat()
199
-
|> Enum.map(fn
200
-
{:write, k, _v} = mut ->
201
-
tags = MetaStore.get_key_server_mutation_tags(meta_store, commit_version, k)
202
-
{tags, mut}
203
-
{:clear, k} = mut ->
204
-
tags = MetaStore.get_key_server_mutation_tags(meta_store, commit_version, k)
205
-
{tags, mut}
206
-
# TODO: range clears (will have to be split)
199
+
|> Enum.with_index()
200
+
|> Enum.map(fn {mut, i} ->
201
+
# TODO: will have to split range clears
202
+
tags = MetaStore.get_key_server_mutation_tags(meta_store, commit_version, mutation_key(mut))
203
+
{tags, {i, mut}}
207
204
end)
208
205
209
206
# TODO: get latest generation only
+8
-9
lib/servers/manager.ex
+8
-9
lib/servers/manager.ex
···
348
348
meta_store = MetaStore.new()
349
349
MetaStore.apply_meta_mutations(meta_store, 0, Enum.map(meta_pairs, fn {k, v} -> {:write, k, v} end))
350
350
351
-
tagged_mutations = Enum.map(meta_pairs, fn {k, v} ->
352
-
tags = MetaStore.get_key_server_mutation_tags(meta_store, 0, k)
353
-
{
354
-
# TODO: remove -1 from tags once storage servers pull server_keys from their own tags
355
-
# (currently this replays the meta mutations against the TLog meta store, which is fine because of idempotence)
356
-
tags,
357
-
{:write, k, v},
358
-
}
359
-
end)
351
+
tagged_mutations =
352
+
meta_pairs
353
+
|> Enum.with_index()
354
+
|> Enum.map(fn {{k, v}, i} ->
355
+
# Note: we remove the meta tag because TLogs are already directly seeded with these pairs
356
+
tags = MetaStore.get_key_server_mutation_tags(meta_store, 0, k) |> List.delete(meta_tag())
357
+
{tags, {i, {:write, k, v}}}
358
+
end)
360
359
361
360
batch = %LogBatch{
362
361
commit_buffer_id: nil,
+2
lib/servers/storage.ex
+2
lib/servers/storage.ex
+2
-2
lib/servers/tlog.ex
+2
-2
lib/servers/tlog.ex
···
219
219
220
220
defp apply_meta(%State{meta_kv: kv}, version, meta_mutations) do
221
221
Enum.each(meta_mutations, fn
222
-
{:write, meta_prefix() <> _ = k, v} ->
222
+
{_i, {:write, meta_prefix() <> _ = k, v}} ->
223
223
HybridKV.put(kv, version, k, v)
224
-
{:clear, meta_prefix() <> _ = k} ->
224
+
{_i, {:clear, meta_prefix() <> _ = k}} ->
225
225
HybridKV.delete(kv, version, k)
226
226
end)
227
227
end
+1
-1
lib/structs.ex
+1
-1
lib/structs.ex
···
97
97
commit_buffer_id: integer,
98
98
commit_version: non_neg_integer,
99
99
prev_commit_version: non_neg_integer,
100
-
tagged_mutations: {list, [{binary, binary}]},
100
+
tagged_mutations: [{Utils.tag_list, Utils.numbered_mutation}],
101
101
last_committed_version: non_neg_integer,
102
102
}
103
103
@enforce_keys [:commit_buffer_id, :commit_version, :prev_commit_version, :tagged_mutations, :last_committed_version]
+14
-2
lib/tagged_queue.ex
+14
-2
lib/tagged_queue.ex
···
5
5
"""
6
6
7
7
alias Hobbes.TaggedQueue
8
+
alias Hobbes.Utils
8
9
9
10
@type t :: %__MODULE__{
10
11
table: :ets.table,
···
21
22
%TaggedQueue{table: table}
22
23
end
23
24
24
-
@spec append_batch(%TaggedQueue{}, integer, non_neg_integer, [{binary, binary}]) :: :ok
25
+
@spec append_batch(%TaggedQueue{}, integer, non_neg_integer, [Utils.numbered_mutation]) :: :ok
25
26
def append_batch(%TaggedQueue{} = tq, tag, version, mutations) when is_integer(tag) and is_integer(version) and is_list(mutations) do
26
27
true = :ets.insert(tq.table, {{tag, version}, mutations})
27
28
:ok
···
29
30
30
31
@doc """
31
32
Peeks mutations for `tag` from `start_version` to `end_version` (both inclusive).
33
+
34
+
iex> peek(tq, 1, 100, 200)
35
+
[
36
+
{
37
+
101,
38
+
[
39
+
{0, {:write, "foo", "bar"}},
40
+
{1, {:clear, "foo"}},
41
+
],
42
+
},
43
+
]
32
44
"""
33
-
@spec peek(%TaggedQueue{}, integer, non_neg_integer, non_neg_integer) :: []
45
+
@spec peek(%TaggedQueue{}, integer, non_neg_integer, non_neg_integer) :: [{non_neg_integer, [Utils.numbered_mutation]}]
34
46
def peek(%TaggedQueue{} = tq, tag, start_version, end_version) when is_integer(tag) and is_integer(start_version) and is_integer(end_version) do
35
47
do_scan(tq.table, tag, end_version, start_version - 1, [])
36
48
|> Enum.reverse()
+6
lib/utils.ex
+6
lib/utils.ex
···
7
7
alias Hobbes.Construct.SimServer
8
8
9
9
@type mutation :: {:write, binary, binary} | {:clear, binary} | {:clear_range, binary, binary}
10
+
@type numbered_mutation :: {non_neg_integer, mutation}
11
+
@type tag_list :: [integer]
10
12
11
13
defmacro meta_prefix, do: "\xFF"
12
14
defmacro meta_end, do: "\xFF\xFF"
···
19
21
defmacro server_keys_end, do: "\xFF/sk0"
20
22
21
23
def meta_tag, do: -1
24
+
25
+
@spec mutation_key(mutation) :: binary
26
+
def mutation_key({:write, key, _value}), do: key
27
+
def mutation_key({:clear, key}), do: key
22
28
23
29
@spec meta_mutation?(mutation) :: boolean
24
30
def meta_mutation?({:write, meta_prefix() <> _, _value}), do: true