this repo has no description
Seal memtables at the current version
+36
-3
+26
lib/xks/manifest.ex
+26
lib/xks/manifest.ex
···
14
14
epoch_max: non_neg_integer | :infinity,
15
15
memtable: Memtable.t,
16
16
size_bytes: non_neg_integer,
17
+
mutable: boolean,
18
+
last_version: non_neg_integer | nil,
17
19
}
18
20
@enforce_keys [
19
21
:partition,
···
23
25
defstruct [
24
26
epoch_max: :infinity,
25
27
size_bytes: 0,
28
+
mutable: true,
29
+
last_version: nil,
26
30
] ++ @enforce_keys
27
31
end
28
32
···
90
94
case :ets.lookup(manifest, key) do
91
95
[{_key, %MemtableInfo{} = info}] ->
92
96
info = %{info | size_bytes: info.size_bytes + inc_bytes}
97
+
:ets.insert(manifest, {key, info})
98
+
info
99
+
100
+
[] ->
101
+
raise "Memtable #{inspect(update_info)} not found"
102
+
end
103
+
end
104
+
105
+
@spec seal_memtable(t, MemtableInfo.t, non_neg_integer) :: MemtableInfo.t
106
+
def seal_memtable(manifest, %MemtableInfo{} = update_info, at_version) do
107
+
assert at_version >= 0
108
+
109
+
key = key_for(update_info)
110
+
case :ets.lookup(manifest, key) do
111
+
[{_key, %MemtableInfo{} = info}] ->
112
+
assert info.mutable == true
113
+
assert info.last_version == nil
114
+
115
+
info = %{info |
116
+
mutable: false,
117
+
last_version: at_version,
118
+
}
93
119
:ets.insert(manifest, {key, info})
94
120
info
95
121
+10
-3
lib/xks/xks.ex
+10
-3
lib/xks/xks.ex
···
242
242
243
243
[%MemtableInfo{} = active_info | _] = Manifest.list_memtables(manifest, partition, epoch)
244
244
assert active_info.epoch_max == :infinity
245
+
assert active_info.mutable == true
246
+
assert active_info.last_version == nil
245
247
246
248
batch_size_bytes = Memtable.apply_mutations(xks, active_info.memtable, partition, version, mutations)
247
249
active_info = Manifest.inc_memtable_size(manifest, active_info, batch_size_bytes)
···
280
282
end)
281
283
end
282
284
283
-
@spec rotate_memtable(t, non_neg_integer) :: :ok
285
+
@spec rotate_memtable(t, integer) :: :ok
284
286
def rotate_memtable(%XKS{} = xks, partition) when is_integer(partition) do
285
-
manifest = xks.manifest
286
-
prev_epoch = :atomics.get(xks.epoch_atomic, 1)
287
+
%{manifest: manifest, epoch_atomic: epoch_atomic, version_atomic: version_atomic} = xks
288
+
prev_epoch = :atomics.get(epoch_atomic, 1)
289
+
version = :atomics.get(version_atomic, 1)
287
290
288
291
[%MemtableInfo{} = active_info | _] = Manifest.list_memtables(manifest, partition, prev_epoch)
289
292
assert active_info.epoch_max == :infinity
293
+
assert active_info.mutable == true
294
+
assert active_info.last_version == nil
295
+
296
+
active_info = Manifest.seal_memtable(manifest, active_info, version)
290
297
WorkQueue.enqueue_compact_memtable(xks.work_queue, active_info)
291
298
292
299
new_epoch = prev_epoch + 1