this repo has no description
Convert Coordinator to XKS
+62
-51
+62
-51
lib/servers/coordinator.ex
+62
-51
lib/servers/coordinator.ex
···
1
1
defmodule Hobbes.Servers.Coordinator do
2
2
use GenServer
3
-
alias Trinity.{Sim, SimProcess, SimServer, SimLogger}
3
+
alias Trinity.{SimProcess, SimServer, SimLogger}
4
4
require SimLogger
5
5
6
6
import ExUnit.Assertions, only: [assert: 1]
7
7
8
8
alias Hobbes.Servers.{ServerSupervisor, Manager}
9
9
alias Hobbes.Servers.Coordinator.Commands
10
-
alias Hobbes.KV.{FlatKV, FlatStorageKV}
11
-
alias Hobbes.Structs.{TLogGeneration, RangeResult}
10
+
alias Hobbes.Structs.TLogGeneration
11
+
12
+
alias Hobbes.XKS
12
13
13
14
import Hobbes.Utils
14
15
···
62
63
# Replication state
63
64
op_number: non_neg_integer,
64
65
commit_number: non_neg_integer,
65
-
kv_mod: module,
66
-
kv: term,
66
+
version: non_neg_integer,
67
+
xks: XKS.t,
67
68
68
69
# Primary replication state
69
70
current_op: {non_neg_integer, [{binary, binary}]} | nil,
···
102
103
103
104
op_number: -1,
104
105
commit_number: -1,
105
-
kv_mod: nil,
106
-
kv: nil,
106
+
version: nil,
107
+
xks: nil,
107
108
108
109
current_op: nil,
109
110
pending_ops: %{},
···
286
287
287
288
# Server
288
289
290
+
@state_partition 0
291
+
289
292
def init(%{id: id, replicas: replicas}) do
290
293
SimProcess.flag(:trap_exit, true)
291
294
···
297
300
replica_ids = Enum.to_list(0..(length(replicas) - 1))
298
301
quorum_size = div(length(replica_ids), 2) + 1
299
302
300
-
{kv_mod, kv} = case false && Sim.simulated?() do
301
-
true -> {FlatStorageKV, FlatStorageKV.new("/coordinator_#{Integer.to_string(id)}.kv")}
302
-
false -> {FlatKV, FlatKV.new()}
303
-
end
303
+
# TODO: larger blocks outside tests
304
+
xks = XKS.new(block_size: 1024, lsm_subtable_size: 1024)
305
+
# TODO: init @state_partition once required
304
306
307
+
# TODO: load version from XKS
308
+
load_read_version = 1
305
309
persisted_state =
306
-
case load_state(kv_mod, kv) do
310
+
case load_state(xks, load_read_version) do
307
311
{:ok, map} ->
308
312
map
309
313
:error ->
···
329
333
op_number: persisted_state.op_number,
330
334
commit_number: persisted_state.commit_number,
331
335
332
-
kv_mod: kv_mod,
333
-
kv: kv,
336
+
version: load_read_version,
337
+
xks: xks,
334
338
335
339
replica_map: replica_map,
336
340
replica_ids: replica_ids,
···
484
488
485
489
# KV
486
490
487
-
defp put(%State{kv_mod: kv_mod, kv: kv}, key, value) when is_binary(key) and is_binary(value) do
488
-
kv_mod.put(kv, key, value)
489
-
:ok
491
+
defp get(%State{xks: xks, version: version}, key) when is_binary(key) do
492
+
XKS.Reader.get(xks, @state_partition, version, key)
490
493
end
491
494
492
-
defp get(%State{kv_mod: kv_mod, kv: kv}, key) when is_binary(key) do
493
-
kv_mod.get(kv, key)
495
+
defp scan(%State{xks: xks, version: version}, start_key, end_key) when is_binary(start_key) and is_binary(end_key) do
496
+
XKS.Reader.scan(xks, @state_partition, version, start_key, end_key)
494
497
end
495
498
496
-
defp scan(%State{kv_mod: kv_mod, kv: kv}, start_key, end_key) when is_binary(start_key) and is_binary(end_key) do
497
-
%RangeResult{pairs: pairs} = kv_mod.scan(kv, start_key, end_key)
498
-
pairs
499
-
end
500
-
501
-
defp create_checkpoint(%State{kv_mod: kv_mod, kv: kv}) do
502
-
%RangeResult{pairs: pairs} = kv_mod.scan(kv, "", special_prefix())
503
-
pairs
499
+
defp create_checkpoint(%State{xks: xks, version: version}) do
500
+
XKS.Reader.scan(xks, @state_partition, version, "", special_prefix())
504
501
end
505
502
506
-
defp load_checkpoint(%State{kv_mod: kv_mod, kv: kv} = state, checkpoint) when is_list(checkpoint) do
507
-
kv_mod.delete_range(kv, "", special_prefix())
503
+
defp load_checkpoint(%State{} = state, checkpoint) when is_list(checkpoint) do
504
+
%{xks: xks, version: prev_version} = state
508
505
509
-
Enum.each(checkpoint, fn {k, v} ->
510
-
assert is_binary(k)
511
-
assert is_binary(v)
512
-
assert k < special_prefix()
513
-
514
-
kv_mod.put(kv, k, v)
506
+
mutations = Enum.map(checkpoint, fn {k, v} ->
507
+
{:write, k, v}
515
508
end)
509
+
mutations = [{:clear_range, "", special_prefix()} | mutations]
516
510
517
-
state
511
+
version = prev_version + 1
512
+
XKS.apply_batch(xks, @state_partition, version, mutations)
513
+
%{state | version: version}
518
514
end
519
515
520
-
defp commit_storage(%State{kv_mod: kv_mod, kv: kv} = state) do
521
-
kv_mod.commit(kv)
516
+
defp commit_storage(%State{xks: xks} = state) do
517
+
# TODO: compact all memtables
518
+
XKS.commit(xks)
522
519
state
523
520
end
524
521
525
522
@persist_keys [:id, :status, :view_number, :last_normal_view_number, :op_number, :commit_number]
526
523
527
-
defp save_state_and_commit(%State{kv_mod: kv_mod, kv: kv} = state) do
528
-
Enum.each(@persist_keys, fn key ->
524
+
defp save_state_and_commit(%State{} = state) do
525
+
%{xks: xks, version: prev_version} = state
526
+
527
+
mutations = Enum.map(@persist_keys, fn key ->
529
528
value = Map.fetch!(state, key)
530
-
kv_mod.put(kv, special_prefix() <> "/" <> Atom.to_string(key), serialize(value))
529
+
{
530
+
:write,
531
+
special_prefix() <> "/" <> Atom.to_string(key),
532
+
serialize(value),
533
+
}
531
534
end)
532
535
536
+
version = prev_version + 1
537
+
XKS.apply_batch(xks, @state_partition, version, mutations)
538
+
539
+
state = %{state | version: version}
533
540
commit_storage(state)
534
541
end
535
542
536
-
@spec load_state(module, term) :: {:ok, map} | :error
537
-
defp load_state(kv_mod, kv) when is_atom(kv_mod) do
538
-
case kv_mod.get(kv, special_prefix() <> "/id") do
543
+
@spec load_state(XKS.t, non_neg_integer) :: {:ok, map} | :error
544
+
defp load_state(%XKS{} = xks, version) do
545
+
case XKS.Reader.get(xks, @state_partition, version, special_prefix() <> "/id") do
539
546
nil ->
540
547
:error
541
548
542
549
_value ->
543
550
map = Map.new(@persist_keys, fn key ->
544
-
value = kv_mod.get(kv, special_prefix() <> "/" <> Atom.to_string(key))
551
+
value = XKS.Reader.get(xks, @state_partition, version, special_prefix() <> "/" <> Atom.to_string(key))
545
552
assert is_binary(value)
546
553
{key, deserialize(key, value)}
547
554
end)
···
576
583
end
577
584
578
585
defp apply_writes(%State{} = state, writes) when is_list(writes) do
579
-
Enum.each(writes, fn {k, v} ->
580
-
put(state, k, v)
581
-
end)
582
-
state
586
+
%{xks: xks} = state
587
+
version = state.version + 1
588
+
mutations = Enum.map(writes, fn {k, v} -> {:write, k, v} end)
589
+
XKS.apply_batch(xks, @state_partition, version, mutations)
590
+
591
+
%{state | version: version}
583
592
end
584
593
585
594
# Request
···
674
683
assert commit.view_number == state.view_number
675
684
assert commit.commit_number >= state.commit_number
676
685
677
-
%{state | commit_number: commit.commit_number}
678
-
|> save_state_and_commit()
686
+
case commit.commit_number == state.commit_number do
687
+
true -> state
688
+
false -> %{state | commit_number: commit.commit_number} |> save_state_and_commit()
689
+
end
679
690
|> track_primary_message()
680
691
end
681
692