this repo has no description
Write shard clears directly into mutation log
+69
-63
+21
-2
lib/kv/byte_sample.ex
+21
-2
lib/kv/byte_sample.ex
···
27
27
:ok
28
28
end
29
29
30
+
@spec delete_range(t, binary, binary) :: :ok
31
+
def delete_range(table, start_key, end_key) when is_binary(start_key) and is_binary(end_key) do
32
+
:ets.delete(table, start_key)
33
+
do_scan_delete(table, end_key, start_key)
34
+
end
35
+
36
+
defp do_scan_delete(table, end_key, prev_key) do
37
+
case :ets.next(table, prev_key) do
38
+
key when is_binary(key) and key < end_key ->
39
+
:ets.delete(table, key)
40
+
do_scan_delete(table, end_key, key)
41
+
42
+
_ -> :ok
43
+
end
44
+
end
45
+
30
46
@spec scan(t, binary, binary) :: [{binary, float}]
31
47
def scan(table, start_key, end_key) do
32
48
acc =
···
82
98
false -> acc
83
99
end
84
100
85
-
{:clear_range, _k, _v}, _acc ->
86
-
raise "Not implemented"
101
+
{:clear_range, sk, ek}, acc ->
102
+
delete_range(table, sk, ek)
103
+
# TODO: if the range was empty in the byte sample we don't need this mutation
104
+
mut = {:clear_range, special_byte_sample_prefix() <> sk, special_byte_sample_prefix() <> ek}
105
+
[mut | acc]
87
106
end)
88
107
|> Enum.reverse()
89
108
end
+48
-61
lib/servers/storage.ex
+48
-61
lib/servers/storage.ex
···
28
28
defstruct @enforce_keys
29
29
end
30
30
31
-
defmodule ShardClear do
32
-
@type t :: %__MODULE__{}
33
-
@enforce_keys [:at_durable_version, :start_key, :end_key]
34
-
defstruct @enforce_keys
35
-
end
36
-
37
31
defmodule State do
38
32
@type t :: %__MODULE__{
39
33
id: non_neg_integer,
···
53
47
byte_sample: ByteSample.t,
54
48
55
49
imports: [ShardImport.t],
56
-
shard_clears: [ShardClear.t],
57
50
}
58
51
@enforce_keys [
59
52
:id,
···
73
66
:byte_sample,
74
67
75
68
:imports,
76
-
:shard_clears,
77
69
]
78
70
defstruct @enforce_keys
79
71
end
···
183
175
byte_sample: ByteSample.new(),
184
176
185
177
imports: %{},
186
-
shard_clears: [],
187
178
}
188
179
189
180
byte_sample_pairs = HybridKV.scan(kv, 1, special_byte_sample_prefix(), special_byte_sample_end()).pairs
···
381
372
end
382
373
383
374
defp flush(%State{kv: kv} = state) do
384
-
flush_version = max(state.data_version - mvcc_window(), 0)
385
-
386
-
{shards_to_clear, remaining_clears} =
387
-
Enum.split_with(state.shard_clears, fn %ShardClear{at_durable_version: adv} ->
388
-
adv <= flush_version
389
-
end)
390
-
391
-
# We process shard clears and flushes in total order
392
-
# Each shard clear is executed on storage at the exact moment when all of the
393
-
# relevant versions are in storage and out of memory
375
+
# We must maintain the invariant that
376
+
# state.durable_version < shard_import.current_read_version
377
+
# for all in-flight shard import reads
394
378
#
395
-
# Note: the vast majority of the time, there are no clears and this entire
396
-
# pipeline is a noop
397
-
shards_to_clear
398
-
|> Enum.sort_by(fn %ShardClear{at_durable_version: adv} -> adv end)
399
-
|> Enum.each(fn %ShardClear{at_durable_version: adv, start_key: start_key, end_key: end_key} ->
400
-
# If this invariant were to be violated, we would clear data written after the shard was removed
401
-
# from this server. Any mutations past that point would still be valid (if the shard was added back),
402
-
# so that is not acceptable
403
-
#
404
-
# We avoid violating this invariant by processing clears and flushes in total order
405
-
# The only way to violate this invariant is to queue a clear at a version < durable_version,
406
-
# which should not happen
407
-
if state.durable_version > adv, do: raise "Shard cleared too late"
379
+
# If this invariant is violated we could clobber newer versions with older reads in storage
380
+
#
381
+
# This could only happen if a read took longer than the mvcc window to complete, which
382
+
# is extremely unlikely in practice because the import read timeout window is considerably
383
+
# shorter than the mvcc window
384
+
min_import_version =
385
+
state.imports
386
+
|> Map.values()
387
+
|> Enum.reject(fn %ShardImport{} = si -> si.status == :complete end)
388
+
|> Enum.map(fn %ShardImport{} = si -> si.current_read_version end)
389
+
|> Enum.min(&<=/2, fn -> state.data_version end)
408
390
409
-
HybridKV.flush(kv, adv)
391
+
flush_version =
392
+
state.data_version - mvcc_window()
393
+
|> min(min_import_version - 1)
394
+
|> max(0)
410
395
411
-
HybridKV.delete_range_storage(kv, start_key, end_key)
412
-
# TODO: clear byte sample
413
-
end)
414
-
state = %State{state | shard_clears: remaining_clears}
415
-
416
-
# Now that all shard clears are complete, we can flush the remaining versions
417
396
HybridKV.flush(kv, flush_version)
418
397
HybridKV.put_storage(kv, special_prefix() <> "durable_version", Integer.to_string(flush_version))
419
398
HybridKV.commit(kv)
···
594
573
# Handle shard imports
595
574
case {old_value, new_value} do
596
575
{nil, "fetching/" <> end_key} ->
597
-
begin_import_shard(version, start_key, end_key, state)
576
+
begin_import(version, start_key, end_key, state)
598
577
599
578
{"fetching/" <> _, "complete/" <> _} ->
600
-
remove_import(start_key, false, state)
579
+
remove_import(start_key, state)
601
580
602
-
{"fetching/" <> _end_key, nil} ->
603
-
remove_import(start_key, true, state)
581
+
{"fetching/" <> end_key, nil} ->
582
+
clear_shard(version, start_key, end_key, state)
583
+
remove_import(start_key, state)
604
584
605
585
{"complete/" <> end_key, nil} ->
606
-
queue_clear_shard(version, start_key, end_key, state)
586
+
clear_shard(version, start_key, end_key, state)
587
+
state
607
588
608
589
# Shard split, we do nothing
609
590
{"complete/" <> _, "complete/" <> _} -> state
···
612
593
end
613
594
end
614
595
615
-
defp queue_clear_shard(version, start_key, end_key, %State{} = state) when is_binary(start_key) do
616
-
sc = %ShardClear{at_durable_version: version, start_key: start_key, end_key: end_key}
596
+
defp clear_shard(version, start_key, end_key, %State{} = state) when is_binary(start_key) and is_binary(end_key) do
597
+
mutations = [
598
+
{:clear_range, start_key, end_key},
599
+
{:clear_range, special_byte_sample_prefix() <> start_key, special_byte_sample_prefix() <> end_key},
600
+
]
601
+
MutationLog.append(state.kv.mutation_log, version, mutations)
602
+
ByteSample.delete_range(state.byte_sample, start_key, end_key)
617
603
618
-
%State{state | shard_clears: [sc | state.shard_clears]}
604
+
:ok
619
605
end
620
606
621
607
defp fetch_import_by_nonce(%State{} = state, nonce) when is_reference(nonce) do
622
-
case Map.values(state.imports) |> Enum.find(fn %ShardImport{} = import -> import.nonce == nonce end) do
623
-
%ShardImport{} = import -> {:ok, import}
624
-
nil -> :error
608
+
find_import_by(state, &(&1.nonce == nonce))
609
+
end
610
+
611
+
defp fetch_import_by_start_key(%State{} = state, start_key) when is_binary(start_key) do
612
+
find_import_by(state, &(&1.start_key == start_key))
613
+
end
614
+
615
+
defp find_import_by(%State{} = state, fun) when is_function(fun, 1) do
616
+
case Map.values(state.imports) |> Enum.filter(fun) do
617
+
[%ShardImport{} = si] -> {:ok, si}
618
+
[] -> :error
625
619
end
626
620
end
627
621
628
-
defp begin_import_shard(version, start_key, end_key, %State{} = state)
622
+
defp begin_import(version, start_key, end_key, %State{} = state)
629
623
when is_integer(version) and is_binary(start_key) do
630
624
shard_import = %ShardImport{
631
625
id: make_ref(),
···
643
637
put_in(state.imports[shard_import.id], shard_import)
644
638
end
645
639
646
-
defp remove_import(start_key, cancelled?, %State{} = state) when is_binary(start_key) and is_boolean(cancelled?) do
647
-
[%ShardImport{} = si] = Map.values(state.imports) |> Enum.filter(&(&1.start_key == start_key))
648
-
649
-
if cancelled? do
650
-
%ShardImport{start_key: start_key, end_key: end_key} = si
651
-
HybridKV.nuke_range(state.kv, start_key, end_key)
652
-
# TODO: clear byte sample
653
-
end
654
-
655
-
%State{state | imports: Map.delete(state.imports, si.id)}
640
+
defp remove_import(start_key, %State{} = state) when is_binary(start_key) do
641
+
{:ok, %ShardImport{} = shard_import} = fetch_import_by_start_key(state, start_key)
642
+
%{state | imports: Map.delete(state.imports, shard_import.id)}
656
643
end
657
644
658
645
defp tick_import(%ShardImport{} = shard_import, %State{} = state) do