this repo has no description
Make coordinator path configurable
+32
-12
+10
-2
lib/cluster_node.ex
+10
-2
lib/cluster_node.ex
···
16
16
17
17
:coordinator_names,
18
18
:coordinator_id,
19
+
:coordinator_path,
19
20
:initial_cluster_opts,
20
21
21
22
:in_memory?,
···
28
29
%NodeConfig{
29
30
cluster_name: Keyword.fetch!(opts, :cluster),
30
31
coordinator_names: Keyword.fetch!(opts, :coordinators),
32
+
# TODO: combine these two
31
33
coordinator_id: Keyword.get(opts, :coordinator_id),
34
+
coordinator_path: Keyword.fetch!(opts, :coordinator_path),
32
35
initial_cluster_opts: Keyword.fetch!(opts, :initial_cluster_opts),
33
36
in_memory?: Keyword.fetch!(opts, :in_memory?),
34
37
slots: Keyword.fetch!(opts, :slots),
···
68
71
69
72
defp start_coordinator(%{coordinator_id: nil}), do: nil
70
73
71
-
defp start_coordinator(%{coordinator_id: id, coordinator_names: coordinator_names})
74
+
defp start_coordinator(%{coordinator_id: id, coordinator_path: coordinator_path, coordinator_names: coordinator_names})
72
75
when is_integer(id) and is_list(coordinator_names) do
73
76
assert id < length(coordinator_names)
74
77
75
78
# TODO: use cluster name
76
79
name = String.to_atom("coordinator-#{id}")
77
-
{:ok, pid} = Coordinator.start_link(id, coordinator_names, name: name)
80
+
arg = %{
81
+
id: id,
82
+
replicas: coordinator_names,
83
+
path: coordinator_path,
84
+
}
85
+
{:ok, pid} = Coordinator.start_link(arg, name: name)
78
86
pid
79
87
end
80
88
+10
lib/sandbox.ex
+10
lib/sandbox.ex
···
50
50
coordinators
51
51
|> Enum.with_index()
52
52
|> Enum.each(fn {{_name, node}, i} ->
53
+
coordinator_path = case cluster_path do
54
+
:memory -> :memory
55
+
path -> "#{path}/coordinator_#{i}"
56
+
end
53
57
node_config = [
54
58
cluster: Keyword.fetch!(opts, :cluster_name),
55
59
coordinators: coordinators,
56
60
coordinator_id: i,
61
+
coordinator_path: coordinator_path,
57
62
initial_cluster_opts: opts,
58
63
59
64
in_memory?: in_memory?,
···
82
87
coordinators
83
88
|> Enum.with_index()
84
89
|> Enum.each(fn {_name, i} ->
90
+
coordinator_path = case cluster_path do
91
+
:memory -> :memory
92
+
path -> "#{path}/coordinator_#{i}"
93
+
end
85
94
node_config = [
86
95
cluster: Keyword.fetch!(opts, :cluster_name),
87
96
coordinators: coordinators,
88
97
coordinator_id: i,
98
+
coordinator_path: coordinator_path,
89
99
initial_cluster_opts: opts,
90
100
91
101
in_memory?: in_memory?,
+12
-10
lib/servers/coordinator.ex
+12
-10
lib/servers/coordinator.ex
···
1
1
defmodule Hobbes.Servers.Coordinator do
2
2
use GenServer
3
-
alias Trinity.{Sim, SimProcess, SimServer, SimFile, SimLogger}
3
+
alias Trinity.{SimProcess, SimServer, SimFile, SimLogger}
4
4
require SimLogger
5
5
6
6
import ExUnit.Assertions, only: [assert: 1]
···
130
130
@request_start_view_min_elapsed_ms 1_000
131
131
@clear_svc_replicas_timeout_ms 3_000
132
132
133
-
def start_link(id, replicas, opts \\ []) when is_integer(id) and is_list(replicas) do
134
-
SimServer.start_link(__MODULE__, %{id: id, replicas: replicas}, opts)
133
+
def start_link(arg, opts \\ []) do
134
+
SimServer.start_link(__MODULE__, arg, opts)
135
135
end
136
136
137
137
def configure_replicas(server, replicas) do
···
301
301
@state_partition 0
302
302
@work_interval_ms 1000
303
303
304
-
def init(%{id: id, replicas: replicas}) do
304
+
def init(%{id: id, path: dir_path, replicas: replicas}) do
305
305
SimProcess.flag(:trap_exit, true)
306
306
307
+
path = case dir_path do
308
+
dir_path when is_binary(dir_path) ->
309
+
SimFile.mkdir_p(dir_path)
310
+
"#{dir_path}/coordinator_#{id}.exs"
311
+
312
+
:memory -> :memory
313
+
end
314
+
307
315
replica_map =
308
316
replicas
309
317
|> Enum.with_index()
···
311
319
312
320
replica_ids = Enum.to_list(0..(length(replicas) - 1))
313
321
quorum_size = div(length(replica_ids), 2) + 1
314
-
315
-
path =
316
-
case Sim.simulated?() do
317
-
true -> "/coordinator_#{id}.xks"
318
-
false -> :memory
319
-
end
320
322
321
323
xks =
322
324
case path != :memory and SimFile.exists?(path) do