this repo has no description
Apply initial pairs as partitioned batches in tlog
+25
-13
+12
-12
lib/servers/tlog.ex
+12
-12
lib/servers/tlog.ex
···
154
154
end
155
155
xks = XKS.new(path: path, block_size: 1024, extent_block_count: extent_block_count, lsm_subtable_size: 1024, lsm_memtable_size_limit: 4096)
156
156
157
-
# Write initial meta pairs
158
-
meta_mutations = Enum.map(meta_pairs, fn {k, v} -> {:write, k, v} end)
159
-
XKS.apply_batch(xks, @meta_partition, prev_version, meta_mutations)
160
-
XKS.maybe_rotate_memtable(xks, @meta_partition)
161
-
162
-
# Write the first extent
163
-
write_new_extent(xks, prev_version, %{
157
+
# Write first extent and meta pairs
158
+
first_extent = %{
164
159
i: 0,
165
160
extent: FreeList.reserve_extent(xks.free_list),
166
161
size: 0,
167
162
nonce: make_nonce(),
168
-
})
163
+
}
164
+
165
+
state_mutations = [pack_extent_mutation(first_extent)]
166
+
meta_mutations = Enum.map(meta_pairs, fn {k, v} -> {:write, k, v} end)
167
+
168
+
partitioned_batches = [
169
+
{@state_partition, state_mutations},
170
+
{@meta_partition, meta_mutations},
171
+
]
172
+
XKS.apply_partitioned_batches(xks, prev_version, partitioned_batches)
169
173
170
174
# Put and commit the initial state of the TLog
171
175
state_fields = %{
···
896
900
# TODO: strong_rand_bytes(16)
897
901
nonce_i = Enum.random(1..1_000_000_000)
898
902
<<nonce_i::integer-64, nonce_i::integer-64>>
899
-
end
900
-
901
-
defp write_new_extent(%XKS{} = xks, version, extent) do
902
-
XKS.apply_batch(xks, @state_partition, version, [pack_extent_mutation(extent)])
903
903
end
904
904
905
905
defp pack_extent_mutation(%{i: i, extent: extent, size: size, nonce: nonce} = _extent) do
+13
-1
lib/xks/xks.ex
+13
-1
lib/xks/xks.ex
···
295
295
},
296
296
} = xks
297
297
prev_version = :atomics.get(version_atomic, 1)
298
-
assert version > prev_version
298
+
case version == 0 and xks.sequence == 0 do
299
+
true ->
300
+
# Allow a write at zero, one time, as long as this is a new XKS
301
+
#
302
+
# It would be nicer to initialize `version_atomic` to -1 and avoid this,
303
+
# but that would require a signed version
304
+
# TODO: find a less hacky way to ensure this is the first write
305
+
assert :ets.first(xks.manifest) == :"$end_of_table"
306
+
false ->
307
+
assert version != 0
308
+
# Ensure writes are strictly monotonic
309
+
assert version > prev_version
310
+
end
299
311
300
312
Enum.each(partitioned_batches, fn {partition, mutations} ->
301
313
assert is_integer(partition)