this repo has no description
Remove shard import and byte sample from Storage
+24
-228
+24
-228
lib/servers/storage.ex
+24
-228
lib/servers/storage.ex
···
6
6
import ExUnit.Assertions, only: [assert: 1]
7
7
8
8
alias Hobbes.SparseShardMap
9
-
alias Hobbes.KV.{MutationLog, ByteSample}
10
9
alias Hobbes.Structs.{Cluster, TLogGeneration, Server, PeekResult, RangeResult, StorageStats, ShardStats}
11
-
alias Hobbes.Servers.{CommitBuffer, TLog, Storage, Distributor}
10
+
alias Hobbes.Servers.{TLog, Distributor}
12
11
alias Hobbes.Encoding.Keyset
13
12
14
13
alias Hobbes.XKS
···
53
52
peek_nonce: reference,
54
53
55
54
xks: XKS.t,
56
-
kv: term,
57
55
shard_map: SparseShardMap.t,
58
-
byte_sample: ByteSample.t,
59
-
60
-
imports: [ShardImport.t],
61
56
}
62
57
@enforce_keys [
63
58
:id,
···
72
67
:peek_tlog_i,
73
68
74
69
:xks,
75
-
:kv,
76
70
:shard_map,
77
-
:byte_sample,
78
-
79
-
:imports,
80
71
]
81
72
defstruct @enforce_keys
82
73
end
···
174
165
def init(%{cluster: %Cluster{} = _cluster, path: _path}) do
175
166
# Init existing storage server from storage KV
176
167
raise "disabled"
177
-
assert Sim.simulated?()
178
-
179
-
#do_init(cluster, kv)
180
168
end
181
169
182
170
def do_init(%Cluster{} = cluster, %XKS{} = xks) do
···
207
195
peek_tlog_i: 0,
208
196
209
197
xks: xks,
210
-
kv: nil,
211
198
shard_map: SparseShardMap.new(),
212
-
byte_sample: ByteSample.new(),
213
-
214
-
imports: %{},
215
199
}
216
200
217
201
# Load the shard map and imports from the special keyspace
···
275
259
end
276
260
277
261
def handle_call({:get_shard_stats, start_key, end_key}, _from, %State{} = state) do
278
-
stats = compute_shard_stats(state, start_key, end_key)
262
+
# TODO
263
+
stats = %ShardStats{
264
+
start_key: start_key,
265
+
end_key: end_key,
266
+
midpoint_key: start_key,
267
+
size_bytes: 0,
268
+
}
279
269
{:reply, {:ok, stats}, state}
280
270
end
281
271
282
-
def handle_call({:check_import_complete, shard_move_id}, _from, %State{} = state) do
283
-
case Map.fetch(state.imports, shard_move_id) do
284
-
{:ok, %ShardImport{status: :complete, completed_version: completed_version}} when is_integer(completed_version) ->
285
-
# An import is only complete once the imported keys have been written out to storage
286
-
completed? = completed_version <= state.durable_version
287
-
{:reply, {:ok, completed?}, state}
288
-
289
-
{:ok, %ShardImport{status: :importing}} ->
290
-
{:reply, {:ok, false}, state}
291
-
292
-
:error ->
293
-
{:reply, {:error, :unknown_shard_move}, state}
294
-
end
272
+
def handle_call({:check_import_complete, _shard_move_id}, _from, %State{} = state) do
273
+
# TODO
274
+
{:reply, {:error, :unknown_shard_move}, state}
295
275
end
296
276
297
277
# TODO: remove probably
···
351
331
{:noreply, state}
352
332
end
353
333
354
-
def handle_info({:read_range_result, nonce, %RangeResult{} = result}, %State{} = state) do
355
-
case fetch_import_by_nonce(state, nonce) do
356
-
{:ok, %ShardImport{} = shard_import} ->
357
-
{:noreply, on_import_result(shard_import, result, state)}
358
-
359
-
# Ignore old reply
360
-
:error -> {:noreply, state}
361
-
end
362
-
end
363
-
364
-
def handle_info({:read_range_result, nonce, {:error, _error}}, %State{} = state) do
365
-
case fetch_import_by_nonce(state, nonce) do
366
-
{:ok, %ShardImport{} = import} ->
367
-
# TODO: maybe some sort of delay or backoff here?
368
-
{:noreply, tick_import(import, state)}
369
-
370
-
# Ignore old reply
371
-
:error -> {:noreply, state}
372
-
end
373
-
end
374
-
375
-
def handle_info({:import_read_range_timeout, nonce}, %State{} = state) do
376
-
case fetch_import_by_nonce(state, nonce) do
377
-
{:ok, %ShardImport{} = import} ->
378
-
{:noreply, tick_import(import, state)}
379
-
380
-
# Ignore timeout for old nonce
381
-
:error -> {:noreply, state}
382
-
end
383
-
end
384
-
385
334
def handle_info(:flush, state) do
386
335
state = flush(state)
387
336
SimProcess.send_after(self(), :flush, @flush_interval_ms)
···
416
365
end
417
366
418
367
defp do_ping_distributor(%State{} = state, distributor_pid) do
419
-
# TODO: this is very inefficient, we need to continuously update shard sizes in-memory
420
-
# TODO: only send oversize/undersize shards?
421
-
shard_stats =
422
-
SparseShardMap.list_shards(state.shard_map)
423
-
|> Enum.map(fn {sk, ek} -> compute_shard_stats(state, sk, ek) end)
368
+
# TODO
369
+
shard_stats = []
424
370
425
371
# TODO: real storage capacity
426
372
total_storage_bytes = 1_000_000_000
···
553
499
state
554
500
end
555
501
556
-
defp handle_special_diff(special_storage_keys_prefix() <> rest, old_value, new_value, version, %State{} = state) do
502
+
defp handle_special_diff(special_storage_keys_prefix() <> rest, old_value, new_value, _version, %State{} = state) do
557
503
[team_id, start_key] = unpack_server_keys_key(rest)
558
504
assert team_id == state.storage_team_id
559
505
···
573
519
574
520
# Handle shard imports
575
521
case {old_unpacked, new_unpacked} do
576
-
{nil, [shard_move_id, end_key, "fetching"]} ->
577
-
begin_import(state, shard_move_id, start_key, end_key)
522
+
{nil, [_shard_move_id, _end_key, "fetching"]} ->
523
+
# TODO: begin shard import
524
+
state
578
525
579
526
{[_sm, _ek, "fetching"], [_, _, "complete"]} ->
580
-
remove_import(start_key, state)
527
+
# TODO: in-progress import was completed
528
+
state
581
529
582
-
{[_sm, end_key, "fetching"], nil} ->
583
-
clear_shard(version, start_key, end_key, state)
584
-
remove_import(start_key, state)
530
+
{[_sm, _end_key, "fetching"], nil} ->
531
+
# TODO: import was aborted, delete shard partition
532
+
state
585
533
586
-
{[_sm, end_key, "complete"], nil} ->
587
-
clear_shard(version, start_key, end_key, state)
534
+
{[_sm, _end_key, "complete"], nil} ->
535
+
# TODO: shard was removed, delete shard partition
588
536
state
589
537
590
538
# Shard split, we do nothing
···
592
540
# Shard split or initial database seed transaction
593
541
{nil, [_sm, _ek, "complete"]} -> state
594
542
end
595
-
end
596
-
597
-
defp clear_shard(version, start_key, end_key, %State{} = state) when is_binary(start_key) and is_binary(end_key) do
598
-
mutations = [
599
-
{:clear_range, start_key, end_key},
600
-
{:clear_range, special_byte_sample_prefix() <> start_key, special_byte_sample_prefix() <> end_key},
601
-
]
602
-
MutationLog.append(state.kv.mutation_log, version, mutations)
603
-
ByteSample.delete_range(state.byte_sample, start_key, end_key)
604
-
605
-
:ok
606
-
end
607
-
608
-
defp fetch_import_by_nonce(%State{} = state, nonce) when is_reference(nonce) do
609
-
find_import_by(state, &(&1.nonce == nonce))
610
-
end
611
-
612
-
defp fetch_import_by_start_key(%State{} = state, start_key) when is_binary(start_key) do
613
-
find_import_by(state, &(&1.start_key == start_key))
614
-
end
615
-
616
-
defp find_import_by(%State{} = state, fun) when is_function(fun, 1) do
617
-
case Map.values(state.imports) |> Enum.filter(fun) do
618
-
[%ShardImport{} = si] -> {:ok, si}
619
-
[] -> :error
620
-
end
621
-
end
622
-
623
-
defp begin_import(%State{} = state, shard_move_id, start_key, end_key)
624
-
when is_integer(shard_move_id) and is_binary(start_key) and is_binary(end_key) do
625
-
shard_import = %ShardImport{
626
-
id: shard_move_id,
627
-
nonce: make_ref(),
628
-
completed_version: nil,
629
-
start_key: start_key,
630
-
end_key: end_key,
631
-
current_end_key: start_key,
632
-
current_read_version: 0,
633
-
status: :importing,
634
-
}
635
-
636
-
assert not Map.has_key?(state.imports, shard_import.id)
637
-
638
-
SimProcess.send_after(self(), {:import_read_range_timeout, shard_import.nonce}, 0)
639
-
put_in(state.imports[shard_import.id], shard_import)
640
-
end
641
-
642
-
defp remove_import(start_key, %State{} = state) when is_binary(start_key) do
643
-
{:ok, %ShardImport{} = shard_import} = fetch_import_by_start_key(state, start_key)
644
-
%{state | imports: Map.delete(state.imports, shard_import.id)}
645
-
end
646
-
647
-
defp tick_import(%ShardImport{} = shard_import, %State{} = state) when state.cluster.status != :normal do
648
-
# Cluster is recovering, try again later
649
-
shard_import = %{shard_import | nonce: make_ref()}
650
-
SimProcess.send_after self(), {:import_read_range_timeout, shard_import.nonce}, 300
651
-
state
652
-
end
653
-
654
-
# TODO: buggify down in sim
655
-
@import_key_limit 4
656
-
#@import_key_limit 5_000
657
-
658
-
defp tick_import(%ShardImport{} = shard_import, %State{} = state) do
659
-
assert state.cluster.status == :normal
660
-
661
-
# TODO: maybe better to read at a slightly older version to prevent too_new errors?
662
-
shard_import = %{shard_import | nonce: make_ref(), current_read_version: state.data_version}
663
-
state = put_in(state.imports[shard_import.id], shard_import)
664
-
665
-
%Server{pid: buf_pid} = state.cluster |> get_servers(Hobbes.Servers.CommitBuffer) |> Enum.random()
666
-
667
-
with {:ok, [[{_sk, _ek, {_ids, from_pids}}]]} <- CommitBuffer.get_shards_multi(buf_pid, [{shard_import.start_key, shard_import.end_key}]),
668
-
{:ok, from_storage_pid} <- random_not_nil(from_pids)
669
-
do
670
-
Storage.read_range_async(from_storage_pid, shard_import.nonce, shard_import.current_read_version, shard_import.current_end_key, shard_import.end_key, limit: @import_key_limit)
671
-
672
-
SimProcess.send_after(self(), {:import_read_range_timeout, shard_import.nonce}, 300)
673
-
state
674
-
else
675
-
# Failure cases:
676
-
# get_shards_multi times out
677
-
# get_shards_multi returns multiple shards (only possible if CommitBuffer is lagging behind, very unlikely)
678
-
# from_pids is all nil because CommitBuffer does not know current storage pids
679
-
_ ->
680
-
# Retry
681
-
SimProcess.send_after(self(), {:import_read_range_timeout, shard_import.nonce}, 100)
682
-
state
683
-
end
684
-
end
685
-
686
-
defp on_import_result(%ShardImport{} = shard_import, %RangeResult{} = result, %State{} = state) do
687
-
%{start_key: start_key, end_key: end_key} = shard_import
688
-
mutations = Enum.map(result.pairs, fn {k, v} ->
689
-
assert k >= start_key
690
-
assert k < end_key
691
-
{:write, k, v}
692
-
end)
693
-
694
-
MutationLog.append(state.kv.mutation_log, shard_import.current_read_version, mutations)
695
-
696
-
case result.more do
697
-
true ->
698
-
{last_key, _value} = List.last(result.pairs)
699
-
shard_import = %{shard_import | current_end_key: last_key <> "\x00"}
700
-
state = put_in(state.imports[shard_import.id], shard_import)
701
-
702
-
# TODO: maybe add a delay here to throttle?
703
-
tick_import(shard_import, state)
704
-
705
-
false ->
706
-
completed_version = shard_import.current_read_version
707
-
708
-
MutationLog.append(state.kv.mutation_log, completed_version, [
709
-
{
710
-
:write,
711
-
special_shard_import_prefix() <> Keyset.pack([shard_import.id]),
712
-
Keyset.pack([completed_version]),
713
-
}
714
-
])
715
-
716
-
shard_import = %{shard_import | nonce: nil, status: :complete, completed_version: completed_version}
717
-
put_in(state.imports[shard_import.id], shard_import)
718
-
end
719
-
end
720
-
721
-
defp compute_shard_stats(%State{} = state, start_key, end_key) when is_binary(start_key) and is_binary(end_key) do
722
-
pairs = ByteSample.scan(state.byte_sample, start_key, end_key)
723
-
724
-
size =
725
-
pairs
726
-
|> Enum.reduce(0, fn {_k, bytes}, acc -> acc + bytes end)
727
-
|> round()
728
-
729
-
half_size = div(size, 2)
730
-
# TODO: we use the full midpoint key, but we could instead use the shortest key which
731
-
# separates the midpoint and the next key (which would make the shard map smaller)
732
-
midpoint =
733
-
Enum.reduce_while(pairs, 0, fn {k, bytes}, acc ->
734
-
acc = acc + bytes
735
-
case acc > half_size do
736
-
true -> {:halt, k}
737
-
false -> {:cont, acc}
738
-
end
739
-
end)
740
-
|> case do
741
-
midpoint when is_binary(midpoint) -> midpoint
742
-
# If the shard is too small to have any byte sample keys
743
-
0 -> start_key
744
-
end
745
-
746
-
%ShardStats{start_key: start_key, end_key: end_key, size_bytes: size, midpoint_key: midpoint}
747
543
end
748
544
749
545
defp check_up_to_date(%State{} = state, version) do