this repo has no description
Remove get_cluster/refresh_cluster
+1
-35
-33
lib/hobbes.ex
-33
lib/hobbes.ex
···
1
1
defmodule Hobbes do
2
-
alias Hobbes.Servers.{Coordinator, Manager}
3
-
alias Hobbes.Structs.Cluster
4
-
5
-
@spec get_cluster([atom] | [{atom, atom}]) :: {:ok, Cluster.t} | {:error, :coordinator_timeout | :manager_timeout | :manager_recovering}
6
-
def get_cluster(coordinators) when is_list(coordinators) do
7
-
coordinators
8
-
|> Enum.map(&Coordinator.get_manager_send/1)
9
-
|> Enum.map(&Coordinator.get_manager_receive/1)
10
-
|> Enum.reduce([], fn
11
-
{:ok, result}, acc -> [result | acc]
12
-
{:error, _err}, acc -> acc
13
-
end)
14
-
|> case do
15
-
[_ | _] = results ->
16
-
{manager_pid, _generation} = Enum.max_by(results, fn {_pid, generation} -> generation end)
17
-
case Manager.get_cluster(manager_pid) do
18
-
{:ok, %Cluster{} = cluster} -> {:ok, cluster}
19
-
{:error, :recovering} -> {:error, :manager_recovering}
20
-
{:error, :timeout} -> {:error, :manager_timeout}
21
-
end
22
-
23
-
[] ->
24
-
{:error, :coordinator_timeout}
25
-
end
26
-
end
27
-
28
-
@spec refresh_cluster(Cluster.t) :: Cluster.t
29
-
def refresh_cluster(%Cluster{} = cluster) do
30
-
case get_cluster(cluster.coordinators) do
31
-
{:ok, %Cluster{} = new_cluster} -> new_cluster
32
-
{:error, _err} -> cluster
33
-
end
34
-
end
35
2
end
+1
-2
lib/workloads/shard_move.ex
+1
-2
lib/workloads/shard_move.ex
···
74
74
{:error, _error} -> state
75
75
end
76
76
else
77
-
{:error, :timeout} -> %{state | cluster: Hobbes.refresh_cluster(cluster)}
78
77
{:error, _err} -> state
79
78
end
80
79
end
···
85
84
]
86
85
87
86
def run(%{cluster: %Cluster{} = cluster}, opts) do
87
+
# TODO: use cluster_name instead of cluster
88
88
delay_ms = Keyword.get(opts, :delay_ms, 1_000)
89
89
num_clients = Keyword.get(opts, :clients, 1)
90
90
duration_ms = Keyword.get(opts, :duration_ms, 10_000)
···
103
103
|> Enum.map(&SimServer.receive_response(&1, :infinity))
104
104
|> Enum.map(fn {:reply, reply} -> reply end)
105
105
106
-
cluster = Hobbes.refresh_cluster(cluster)
107
106
shard_moves = list_shard_moves(cluster)
108
107
109
108
shard_moves_table =