this repo has no description
Scan manifest memtables at epoch
+39
-29
+19
-18
lib/xks/manifest.ex
+19
-18
lib/xks/manifest.ex
···
30
30
:ok
31
31
end
32
32
33
-
@spec list_memtables(t) :: [{non_neg_integer, Memtable.t}]
34
-
def list_memtables(manifest) do
35
-
# TODO: scan at epoch
36
-
scan_memtables(manifest, {@memtable_level, -1, -1}, [])
33
+
@spec list_memtables(t, non_neg_integer) :: [{non_neg_integer, Memtable.t}]
34
+
def list_memtables(manifest, epoch) do
35
+
scan_memtables(manifest, epoch, {@memtable_level, -1, -1}, [])
37
36
|> Enum.reverse()
38
-
|> Enum.reduce([], fn {id, value}, acc ->
39
-
case value do
40
-
:tombstone ->
41
-
# Tombstone cancels out memtable
42
-
[{^id, _} | rest] = acc
43
-
rest
44
-
value ->
45
-
[{id, value} | acc]
46
-
end
47
-
end)
48
37
end
49
38
50
-
defp scan_memtables(manifest, prev_key, acc) do
39
+
defp scan_memtables(manifest, epoch, prev_key, acc) do
51
40
case :ets.next_lookup(manifest, prev_key) do
52
-
{{@memtable_level, id, _epoch} = key, [{_key, value}]} ->
53
-
scan_memtables(manifest, key, [{id, value} | acc])
41
+
{{@memtable_level, _id, ep} = key, _object} when ep > epoch ->
42
+
# This entry is not visible at epoch
43
+
scan_memtables(manifest, epoch, key, acc)
54
44
55
-
_ -> acc
45
+
{{@memtable_level, id, _ep} = key, [{_key, :tombstone}]} ->
46
+
# This entry is a tombstone and cancels out the previous memtable
47
+
[{^id, _value} | acc] = acc
48
+
scan_memtables(manifest, epoch, key, acc)
49
+
50
+
{{@memtable_level, id, _ep} = key, [{_key, value}]} ->
51
+
# This entry is a memtable, accumulate
52
+
scan_memtables(manifest, epoch, key, [{id, value} | acc])
53
+
54
+
_ ->
55
+
# Reached the end of the memtable space
56
+
acc
56
57
end
57
58
end
58
59
+18
-11
lib/xks/xks.ex
+18
-11
lib/xks/xks.ex
···
50
50
51
51
@spec apply_batch(t, non_neg_integer, list) :: :ok
52
52
def apply_batch(%XKS{} = xks, version, mutations) do
53
-
[{_id, active_memtable} | _] = Manifest.list_memtables(xks.manifest)
53
+
epoch = :atomics.get(xks.epoch_atomic, 1)
54
+
55
+
[{_id, active_memtable} | _] = Manifest.list_memtables(xks.manifest, epoch)
54
56
Memtable.apply_mutations(active_memtable, version, mutations)
55
57
56
58
:ok
···
58
60
59
61
@spec get(t, non_neg_integer, binary) :: binary | nil
60
62
def get(%XKS{} = xks, version, key) do
61
-
Manifest.list_memtables(xks.manifest)
63
+
epoch = :atomics.get(xks.epoch_atomic, 1)
64
+
65
+
Manifest.list_memtables(xks.manifest, epoch)
62
66
|> Enum.find_value(:error, fn {_id, memtable} ->
63
67
case Memtable.get(memtable, version, key) do
64
68
nil -> false
···
74
78
75
79
@spec scan(t, non_neg_integer, binary, binary) :: [{binary, binary}]
76
80
def scan(%XKS{} = xks, version, start_key, end_key) do
81
+
epoch = :atomics.get(xks.epoch_atomic, 1)
82
+
77
83
iterators =
78
-
Manifest.list_memtables(xks.manifest)
84
+
Manifest.list_memtables(xks.manifest, epoch)
79
85
|> Enum.map(fn {_id, memtable} ->
80
86
Merge.iterator_for_memtable(memtable, start_key)
81
87
end)
···
117
123
@spec compact_memtables(t) :: :ok
118
124
def compact_memtables(%XKS{} = xks) do
119
125
manifest = xks.manifest
126
+
prev_epoch = :atomics.get(xks.epoch_atomic, 1)
120
127
121
-
input_tables = Manifest.list_memtables(manifest)
128
+
input_tables = Manifest.list_memtables(manifest, prev_epoch)
122
129
output_tables = Compaction.compact(xks, input_tables)
123
130
124
-
epoch = :atomics.get(xks.epoch_atomic, 1)
125
131
# We can't increment the atomic until after the memtable has been updated
126
-
epoch = epoch + 1
132
+
new_epoch = prev_epoch + 1
127
133
128
134
# Delete input tables
129
135
Enum.each(input_tables, fn {id, _memtable} ->
130
-
Manifest.delete_memtable(manifest, epoch, id)
136
+
Manifest.delete_memtable(manifest, new_epoch, id)
131
137
end)
132
138
133
139
# Insert output tables
134
140
level = 1
135
141
Enum.each(output_tables, fn {:table, block_index, block_checksum, first_key, last_key} ->
136
-
Manifest.insert_table(manifest, epoch, level, {block_index, block_checksum}, first_key, last_key)
142
+
Manifest.insert_table(manifest, new_epoch, level, {block_index, block_checksum}, first_key, last_key)
137
143
end)
138
144
139
145
# Update epoch *after* all writes are complete so that readers cannot observe torn writes
140
-
:atomics.put(xks.epoch_atomic, 1, epoch)
146
+
:atomics.put(xks.epoch_atomic, 1, new_epoch)
141
147
142
148
dbg Manifest.dump(manifest)
143
149
···
151
157
152
158
@doc false
153
159
def dump(%XKS{} = xks) do
154
-
xks.manifest
155
-
|> Manifest.list_memtables()
160
+
epoch = :atomics.get(xks.epoch_atomic, 1)
161
+
162
+
Manifest.list_memtables(xks.manifest, epoch)
156
163
|> Enum.map(fn {id, mt} ->
157
164
{id, mt, Memtable.dump(mt)}
158
165
end)