this repo has no description
Make Coordinator commands generic and move them to their own module
+113
-88
+42
-85
lib/servers/coordinator.ex
+42
-85
lib/servers/coordinator.ex
···
4
4
import ExUnit.Assertions, only: [assert: 1]
5
5
6
6
alias Hobbes.Servers.{ServerSupervisor, Manager}
7
+
alias Hobbes.Servers.Coordinator.Commands
7
8
alias Hobbes.KV.{FlatKV, FlatStorageKV}
8
9
alias Hobbes.Structs.RangeResult
9
10
10
11
import Hobbes.Utils
12
+
13
+
@type server :: pid | atom
14
+
@type not_primary_error :: {:error, {:not_primary, pid}}
11
15
12
16
@type id :: non_neg_integer
13
17
@type role :: :primary | :backup
···
136
140
137
141
# Public API
138
142
143
+
@spec write(server, binary, binary) :: :ok | not_primary_error
139
144
def write(server, key, value) when is_binary(key) and is_binary(value) do
140
145
write(server, [{key, value}])
141
146
end
142
147
148
+
@spec write(server, [{binary, binary}]) :: :ok | not_primary_error
143
149
def write(server, pairs) when is_list(pairs) do
144
-
SimServer.call(server, {:request, {:write, pairs}})
150
+
request(server, {Commands, :write, [pairs]})
145
151
end
146
152
153
+
@spec read(server, binary) :: {:ok, binary} | not_primary_error
147
154
def read(server, key) when is_binary(key) do
148
155
case read(server, [key]) do
149
156
{:ok, [{^key, value}]} -> {:ok, value}
150
-
{:error, _, _} = error -> error
157
+
{:error, _err} = error -> error
151
158
end
152
159
end
153
160
161
+
@spec read(server, [binary]) :: {:ok, [{binary, binary}]} | not_primary_error
154
162
def read(server, keys) when is_list(keys) do
155
-
SimServer.call(server, {:request, {:read, keys}})
163
+
request(server, {Commands, :read, [keys]})
164
+
end
165
+
166
+
@spec inc_generation(server) :: {:ok, map} | not_primary_error
167
+
def inc_generation(server) do
168
+
request(server, {Commands, :inc_generation, []})
169
+
end
170
+
171
+
@spec write_generation(server, non_neg_integer, non_neg_integer, [non_neg_integer]) :: :ok | not_primary_error
172
+
def write_generation(server, generation, start_version, tlog_ids)
173
+
when is_integer(generation) and is_integer(start_version) and is_list(tlog_ids) do
174
+
request(server, {Commands, :write_generation, [generation, start_version, tlog_ids]})
175
+
end
176
+
177
+
@spec allocate_ids(server, non_neg_integer, pos_integer) :: {:ok, {non_neg_integer, non_neg_integer}} | not_primary_error
178
+
def allocate_ids(server, generation, count) when is_integer(generation) and is_integer(count) do
179
+
request(server, {Commands, :allocate_ids, [generation, count]})
180
+
end
181
+
182
+
@spec request(server, {module, atom, list}) :: term | not_primary_error
183
+
def request(server, {m, f, a} = command) when is_atom(m) and is_atom(f) and is_list(a) do
184
+
SimServer.call(server, {:request, command})
156
185
end
157
186
158
187
def get_manager(server) do
···
172
201
end
173
202
end
174
203
175
-
def inc_generation(server) do
176
-
SimServer.call(server, {:request, :inc_generation})
177
-
end
178
-
179
-
def write_generation(server, generation, start_version, tlog_ids) when is_integer(generation) and is_integer(start_version) and is_list(tlog_ids) do
180
-
for id <- tlog_ids, do: assert(is_integer(id))
181
-
182
-
SimServer.call(server, {:request, {:write_generation, generation, start_version, tlog_ids}})
183
-
end
184
-
185
-
def allocate_ids(server, generation, count) when is_integer(generation) and is_integer(count) and count > 0 do
186
-
SimServer.call(server, {:request, {:allocate_ids, generation, count}})
187
-
end
188
-
189
204
def track_manager_generation(server, generation) when is_integer(generation) do
190
205
SimServer.call(server, {:track_manager_generation, self(), generation})
191
206
end
···
340
355
341
356
:backup ->
342
357
primary_pid = Map.fetch!(state.replica_map, state.primary_id)
343
-
{:reply, {:error, :not_primary, primary_pid}, state}
358
+
{:reply, {:error, {:not_primary, primary_pid}}, state}
344
359
end
345
360
end
346
361
···
532
547
533
548
# Commands
534
549
535
-
@spec command(term, State.t) :: {[{binary, binary}], term}
536
-
defp command({:write, pairs}, %State{} = _state) do
537
-
Enum.each(pairs, fn {k, v} ->
550
+
defp run_command(%State{} = state, {m, f, a} = _command) when is_atom(m) and is_atom(f) and is_list(a) do
551
+
get = fn k -> get(state, k) end
552
+
scan = fn sk, ek -> scan(state, sk, ek) end
553
+
554
+
{writes, result} = apply(m, f, [get, scan, a])
555
+
Enum.each(writes, fn w ->
556
+
assert {k, v} = w
538
557
assert is_binary(k)
539
558
assert is_binary(v)
540
559
end)
541
-
{pairs, :ok}
542
-
end
543
560
544
-
defp command({:read, keys}, %State{} = state) do
545
-
values = Enum.map(keys, fn k -> {k, get(state, k)} end)
546
-
{[], values}
547
-
end
548
-
549
-
defp command(:inc_generation, %State{} = state) do
550
-
case scan(state, "config/", "config0") do
551
-
[] ->
552
-
{[], {:error, :no_config}}
553
-
554
-
config_pairs ->
555
-
old_generation = String.to_integer(get(state, "generation") || "0")
556
-
generation = old_generation + 1
557
-
558
-
old_generations = scan(state, "gen/", "gen0")
559
-
560
-
{
561
-
[{"generation", Integer.to_string(generation)}],
562
-
{:ok, %{generation: generation, config_pairs: config_pairs, old_generations: old_generations}},
563
-
}
564
-
end
565
-
end
566
-
567
-
defp command({:write_generation, generation, start_version, tlog_ids}, %State{} = state) do
568
-
gen_str = Integer.to_string(generation)
569
-
570
-
case get(state, "generation") do
571
-
^gen_str ->
572
-
key = "gen/" <> gen_str
573
-
case get(state, key) do
574
-
nil ->
575
-
tlogs_encoded = tlog_ids |> Enum.map(&Integer.to_string/1) |> Enum.join(",")
576
-
start_version_encoded = Integer.to_string(start_version)
577
-
578
-
{
579
-
[{key, "#{start_version_encoded};#{tlogs_encoded}"}],
580
-
:ok
581
-
}
582
-
583
-
_value -> {[], {:error, :already_written}}
584
-
end
585
-
586
-
_other -> {[], {:error, :wrong_generation}}
587
-
end
588
-
end
589
-
590
-
defp command({:allocate_ids, generation, count}, %State{} = state) do
591
-
gen_str = Integer.to_string(generation)
592
-
593
-
case get(state, "generation") do
594
-
^gen_str ->
595
-
start_id = String.to_integer(get(state, "next_id") || "0")
596
-
end_id = start_id + count
597
-
598
-
{
599
-
[{"next_id", Integer.to_string(end_id)}],
600
-
{:ok, {start_id, end_id}},
601
-
}
602
-
603
-
_other -> {:error, :wrong_generation}
604
-
end
561
+
{writes, result}
605
562
end
606
563
607
564
defp apply_writes(%State{} = state, writes) when is_list(writes) do
···
617
574
assert state.role == :primary
618
575
619
576
op_number = state.op_number + 1
620
-
{writes, result} = command(command, state)
577
+
{writes, result} = run_command(state, command)
621
578
622
579
state =
623
580
%{state |
+68
lib/servers/coordinator/commands.ex
+68
lib/servers/coordinator/commands.ex
···
1
+
defmodule Hobbes.Servers.Coordinator.Commands do
2
+
def write(_get, _scan, [pairs]) when is_list(pairs) do
3
+
{pairs, :ok}
4
+
end
5
+
6
+
def read(get, _scan, [keys]) when is_list(keys) do
7
+
pairs = Enum.map(keys, fn k -> {k, get.(k)} end)
8
+
{[], pairs}
9
+
end
10
+
11
+
def inc_generation(get, scan, []) do
12
+
case scan.("config/", "config0") do
13
+
[] ->
14
+
{[], {:error, :no_config}}
15
+
16
+
config_pairs ->
17
+
old_generation = String.to_integer(get.("generation") || "0")
18
+
generation = old_generation + 1
19
+
20
+
old_generations = scan.("gen/", "gen0")
21
+
22
+
{
23
+
[{"generation", Integer.to_string(generation)}],
24
+
{:ok, %{generation: generation, config_pairs: config_pairs, old_generations: old_generations}},
25
+
}
26
+
end
27
+
end
28
+
29
+
def write_generation(get, _scan, [generation, start_version, tlog_ids]) do
30
+
gen_str = Integer.to_string(generation)
31
+
32
+
case get.("generation") do
33
+
^gen_str ->
34
+
key = "gen/" <> gen_str
35
+
case get.(key) do
36
+
nil ->
37
+
tlogs_encoded = tlog_ids |> Enum.map(&Integer.to_string/1) |> Enum.join(",")
38
+
start_version_encoded = Integer.to_string(start_version)
39
+
40
+
{
41
+
[{key, "#{start_version_encoded};#{tlogs_encoded}"}],
42
+
:ok
43
+
}
44
+
45
+
_value -> {[], {:error, :already_written}}
46
+
end
47
+
48
+
_other -> {[], {:error, :wrong_generation}}
49
+
end
50
+
end
51
+
52
+
def allocate_ids(get, _scan, [generation, count]) do
53
+
gen_str = Integer.to_string(generation)
54
+
55
+
case get.("generation") do
56
+
^gen_str ->
57
+
start_id = String.to_integer(get.("next_id") || "0")
58
+
end_id = start_id + count
59
+
60
+
{
61
+
[{"next_id", Integer.to_string(end_id)}],
62
+
{:ok, {start_id, end_id}},
63
+
}
64
+
65
+
_other -> {:error, :wrong_generation}
66
+
end
67
+
end
68
+
end
+3
-3
test/hobbes_test.exs
+3
-3
test/hobbes_test.exs
···
244
244
245
245
SimServer.sleep(1_000)
246
246
247
-
assert {:error, :not_primary, :"coordinator-0"} = Coordinator.write(c2, "foo", "bar")
248
-
assert {:error, :not_primary, :"coordinator-0"} = Coordinator.write(c3, "foo", "bar")
247
+
assert {:error, {:not_primary, :"coordinator-0"}} = Coordinator.write(c2, "foo", "bar")
248
+
assert {:error, {:not_primary, :"coordinator-0"}} = Coordinator.write(c3, "foo", "bar")
249
249
250
250
assert {:ok, _} = Coordinator.write(c1, "foo", "bar")
251
251
assert {:ok, _} = Coordinator.write(c1, "hello", "world")
···
257
257
258
258
SimServer.sleep(3_000)
259
259
260
-
assert {:error, :not_primary, :"coordinator-1"} = Coordinator.read(c3, "foo")
260
+
assert {:error, {:not_primary, :"coordinator-1"}} = Coordinator.read(c3, "foo")
261
261
262
262
assert {:ok, "bar"} = Coordinator.read(c2, "foo")
263
263