this repo has no description
Re-enqueue input level if it's still oversize
+25
-8
+10
-1
lib/xks/fuzz/write_info.ex
+10
-1
lib/xks/fuzz/write_info.ex
···
8
8
@enforce_keys [
9
9
:xks,
10
10
:version,
11
+
:bytes_written,
11
12
:opts,
12
13
]
13
14
defstruct @enforce_keys
···
29
30
state = %State{
30
31
xks: XKS.new(opts.xks_opts),
31
32
version: 0,
33
+
bytes_written: 0,
32
34
opts: opts,
33
35
}
34
36
···
38
40
39
41
max_block = XKS.FreeList.get_max_block(state.xks.free_list)
40
42
xks_size_bytes = state.xks.opts.block_size * max_block
43
+
44
+
dbg state.bytes_written / 1_000_000
41
45
dbg xks_size_bytes / 1_000_000
42
46
43
47
dbg XKS.info_manifest(state.xks, 0)
···
90
94
:ok = XKS.set_max_persist_version(xks, version)
91
95
92
96
batch = make_batch(opts)
97
+
size = batch_size(batch)
93
98
partitioned_batches = [{0, batch}]
94
99
95
100
:ok = XKS.apply_partitioned_batches(xks, version, partitioned_batches)
96
101
97
-
state
102
+
%{state | bytes_written: state.bytes_written + size}
98
103
end
99
104
100
105
defp make_batch(_opts) do
···
105
110
{:write, key, value}
106
111
end)
107
112
end
113
+
114
+
defp batch_size(mutations, acc \\ 0)
115
+
defp batch_size([], acc), do: acc
116
+
defp batch_size([{:write, k, v} | rest], acc), do: batch_size(rest, acc + byte_size(k) + byte_size(v))
108
117
end
+13
-6
lib/xks/xks.ex
+13
-6
lib/xks/xks.ex
···
804
804
ManifestLog.update_table(manifest_log, table_info)
805
805
end)
806
806
807
-
level_table_count = Manifest.get_table_count_for_level(manifest, partition, output_level)
808
-
target_table_count = level_target(output_level)
809
-
case level_table_count > target_table_count do
810
-
true -> WorkQueue.enqueue_compact_level(work_queue, partition, output_level)
811
-
false -> :noop
812
-
end
807
+
# subtracting here is kind-of ugly but should always be correct
808
+
input_level = output_level - 1
809
+
maybe_enqueue_compact_level(manifest, work_queue, partition, input_level)
810
+
maybe_enqueue_compact_level(manifest, work_queue, partition, output_level)
813
811
814
812
# Update epoch *after* all writes are complete so that readers cannot observe torn writes
815
813
:atomics.put(xks.epoch_atomic, 1, new_epoch)
816
814
817
815
:ok
816
+
end
817
+
818
+
defp maybe_enqueue_compact_level(manifest, work_queue, partition, level) do
819
+
level_table_count = Manifest.get_table_count_for_level(manifest, partition, level)
820
+
target_table_count = level_target(level)
821
+
case level_table_count > target_table_count do
822
+
true -> WorkQueue.enqueue_compact_level(work_queue, partition, level)
823
+
false -> :noop
824
+
end
818
825
end
819
826
820
827
@level_tables_target {0, 2, 4, 8, 16, 32, 64}
+2
-1
test/xks_test.exs
+2
-1
test/xks_test.exs
···
255
255
path = :memory
256
256
257
257
Hobbes.XKS.Fuzz.WriteInfo.run(100, [
258
-
iterations: 200,
258
+
iterations: 300,
259
259
xks_opts: [
260
260
path: path,
261
261
block_size: 128 * 1024,
262
262
extent_block_count: 10,
263
+
lsm_table_data_blocks_max: 16,
263
264
lsm_subtable_size: 32 * 1024,
264
265
lsm_memtable_size_limit: 4 * 1024 * 1024,
265
266
],