this repo has no description
Turn Cache into a GenServer
+84
-37
+80
-36
lib/cache.ex
+80
-36
lib/cache.ex
···
1
1
defmodule Hobbes.Cache do
2
+
use GenServer
3
+
alias Trinity.{SimServer, SimPersistentTerm}
4
+
2
5
alias Hobbes.Servers.{Coordinator, Manager}
3
6
alias Hobbes.Structs.Cluster
4
7
5
-
alias Trinity.SimPersistentTerm
6
-
7
8
@type cluster_name :: binary
8
9
9
10
@clusters_table_key :_hobbes_clusters_table
10
11
12
+
@spec get_clusters_table :: :ets.table
11
13
defp get_clusters_table do
12
14
case SimPersistentTerm.get(@clusters_table_key, nil) do
13
-
nil ->
14
-
# TODO: race condition obviously
15
-
table = :ets.new(__MODULE__, [:set, :public])
16
-
:ok = SimPersistentTerm.put(@clusters_table_key, table)
17
-
table
18
-
table ->
19
-
table
15
+
nil -> raise "Hobbes.Cache has not been started"
16
+
table -> table
20
17
end
21
18
end
22
19
20
+
defmodule CachedCluster do
21
+
@enforce_keys [
22
+
:coordinators,
23
+
:cluster,
24
+
:counter,
25
+
]
26
+
defstruct @enforce_keys
27
+
end
28
+
29
+
defmodule State do
30
+
@enforce_keys [
31
+
:clusters_table,
32
+
]
33
+
defstruct @enforce_keys
34
+
end
35
+
36
+
@spec start_link(nil) :: GenServer.on_start
37
+
def start_link(_arg \\ nil) do
38
+
SimServer.start_link(__MODULE__, nil, name: __MODULE__)
39
+
end
40
+
23
41
@spec lookup_cluster(cluster_name) :: {:ok, Cluster.t} | {:error, atom}
24
42
def lookup_cluster(name) do
25
43
clusters_table = get_clusters_table()
26
44
27
45
case :ets.lookup(clusters_table, name) do
28
-
[{^name, %{cluster: %Cluster{} = cluster}}] ->
46
+
[{^name, %CachedCluster{cluster: %Cluster{} = cluster}}] ->
29
47
{:ok, cluster}
30
-
31
-
[{^name, %{coordinators: coordinators}}] ->
32
-
case get_cluster(coordinators) do
33
-
{:ok, %Cluster{} = cluster} ->
34
-
write_cached_cluster(name, cluster)
35
-
{:ok, cluster}
36
-
{:error, _err} = error -> error
37
-
end
38
-
48
+
[{^name, %CachedCluster{counter: counter}}] ->
49
+
refresh_cluster(name, counter)
50
+
{:error, :cluster_not_loaded}
39
51
[] ->
40
52
{:error, :unknown_cluster}
41
53
end
42
54
end
43
55
44
-
@spec refresh_cluster(cluster_name) :: :ok | {:error, :cluster_not_found | :coordinator_timeout | :manager_timeout | :manager_recovering}
45
-
def refresh_cluster(name) do
46
-
with {:ok, %Cluster{} = cluster} <- lookup_cluster(name),
47
-
{:ok, %Cluster{} = cluster} <- get_cluster(cluster.coordinators)
48
-
do
49
-
:ok = write_cached_cluster(name, cluster)
50
-
{:ok, cluster}
51
-
end
56
+
@spec refresh_cluster(cluster_name, non_neg_integer) :: :ok
57
+
def refresh_cluster(name, counter) do
58
+
SimServer.cast(__MODULE__, {:refresh_cluster, name, counter})
52
59
end
53
60
54
61
@spec write_coordinators(cluster_name, list) :: :ok
55
62
def write_coordinators(name, coordinators) when is_list(coordinators) do
56
-
clusters_table = get_clusters_table()
57
-
:ets.insert(clusters_table, {name, %{coordinators: coordinators}})
58
-
:ok
63
+
try do
64
+
SimServer.call(__MODULE__, {:write_coordinators, name, coordinators}, 1000)
65
+
catch
66
+
:exit, {:timeout, _mfa} -> {:error, :timeout}
67
+
end
68
+
end
69
+
70
+
def init(nil) do
71
+
clusters_table = :ets.new(__MODULE__, [:set, :protected])
72
+
SimPersistentTerm.put(@clusters_table_key, clusters_table)
73
+
74
+
state = %State{
75
+
clusters_table: clusters_table,
76
+
}
77
+
{:ok, state}
59
78
end
60
79
61
-
@spec write_cached_cluster(cluster_name, Cluster.t) :: :ok
62
-
defp write_cached_cluster(name, %Cluster{} = cluster) when is_binary(name) do
63
-
clusters_table = get_clusters_table()
80
+
def handle_call({:write_coordinators, name, coordinators}, _from, %State{} = state) do
81
+
%{clusters_table: clusters_table} = state
64
82
65
-
value = %{cluster: cluster, coordinators: cluster.coordinators}
66
-
:ets.insert(clusters_table, {name, value})
67
-
:ok
83
+
case :ets.member(clusters_table, name) do
84
+
false ->
85
+
cached_cluster = %CachedCluster{coordinators: coordinators, cluster: nil, counter: 0}
86
+
:ets.insert(clusters_table, {name, cached_cluster})
87
+
{:reply, :ok, state}
88
+
89
+
true ->
90
+
{:reply, {:error, :name_already_taken}, state}
91
+
end
92
+
end
93
+
94
+
def handle_cast({:refresh_cluster, name, counter}, %State{} = state) do
95
+
%{clusters_table: clusters_table} = state
96
+
97
+
case :ets.lookup(clusters_table, name) do
98
+
[{^name, %CachedCluster{} = cached_cluster}] when cached_cluster.counter <= counter ->
99
+
cached_cluster =
100
+
case get_cluster(cached_cluster.coordinators) do
101
+
{:ok, %Cluster{} = cluster} -> %{cached_cluster | cluster: cluster}
102
+
{:error, _err} -> cached_cluster
103
+
end
104
+
105
+
cached_cluster = %{cached_cluster | counter: cached_cluster.counter + 1}
106
+
:ets.insert(clusters_table, {name, cached_cluster})
107
+
108
+
_ -> :noop
109
+
end
110
+
111
+
{:noreply, state}
68
112
end
69
113
70
114
@spec get_cluster([atom] | [{atom, atom}]) :: {:ok, Cluster.t} | {:error, :coordinator_timeout | :manager_timeout | :manager_recovering}
+2
-1
lib/sandbox.ex
+2
-1
lib/sandbox.ex
+2
lib/workloads.ex
+2
lib/workloads.ex
···
146
146
{cluster_opts, _opts} = Keyword.pop(opts, :cluster_opts, [])
147
147
cluster_name = "cluster"
148
148
149
+
{:ok, _pid} = Hobbes.Cache.start_link()
150
+
149
151
# Use a distributed cluster in simulation
150
152
cluster_opts = Keyword.put(cluster_opts, :distributed, Sim.simulated?())
151
153
cluster_opts = Keyword.put(cluster_opts, :cluster_name, cluster_name)