this repo has no description
Add CoordinatorReadWrite workload test
+193
-10
+19
-10
lib/servers/coordinator.ex
+19
-10
lib/servers/coordinator.ex
···
11
11
import Hobbes.Utils
12
12
13
13
@type server :: pid | atom
14
-
@type not_primary_error :: {:error, {:not_primary, pid}}
14
+
@type request_error :: {:error, :timeout | {:not_primary, pid | atom}}
15
15
16
16
@type id :: non_neg_integer
17
17
@type role :: :primary | :backup
···
140
140
141
141
# Public API
142
142
143
-
@spec write(server, binary, binary) :: :ok | not_primary_error
143
+
@spec write(server, binary, binary) :: {:ok, :ok} | request_error
144
144
def write(server, key, value) when is_binary(key) and is_binary(value) do
145
145
write(server, [{key, value}])
146
146
end
147
147
148
-
@spec write(server, [{binary, binary}]) :: :ok | not_primary_error
148
+
@spec write(server, [{binary, binary}]) :: {:ok, :ok} | request_error
149
149
def write(server, pairs) when is_list(pairs) do
150
150
request(server, {Commands, :write, [pairs]})
151
151
end
152
152
153
-
@spec read(server, binary) :: {:ok, binary} | not_primary_error
153
+
@spec read(server, binary) :: {:ok, binary} | request_error
154
154
def read(server, key) when is_binary(key) do
155
155
case read(server, [key]) do
156
156
{:ok, [{^key, value}]} -> {:ok, value}
···
158
158
end
159
159
end
160
160
161
-
@spec read(server, [binary]) :: {:ok, [{binary, binary}]} | not_primary_error
161
+
@spec read(server, [binary]) :: {:ok, [{binary, binary}]} | request_error
162
162
def read(server, keys) when is_list(keys) do
163
163
request(server, {Commands, :read, [keys]})
164
164
end
165
165
166
-
@spec inc_generation(server) :: {:ok, map} | not_primary_error
166
+
@spec dump(server) :: {:ok, [{binary, binary}]} | request_error
167
+
def dump(server) do
168
+
request(server, {Commands, :dump, []})
169
+
end
170
+
171
+
@spec inc_generation(server) :: {:ok, map} | request_error
167
172
def inc_generation(server) do
168
173
request(server, {Commands, :inc_generation, []})
169
174
end
170
175
171
-
@spec write_generation(server, non_neg_integer, non_neg_integer, [non_neg_integer]) :: :ok | not_primary_error
176
+
@spec write_generation(server, non_neg_integer, non_neg_integer, [non_neg_integer]) :: :ok | request_error
172
177
def write_generation(server, generation, start_version, tlog_ids)
173
178
when is_integer(generation) and is_integer(start_version) and is_list(tlog_ids) do
174
179
request(server, {Commands, :write_generation, [generation, start_version, tlog_ids]})
175
180
end
176
181
177
-
@spec allocate_ids(server, non_neg_integer, pos_integer) :: {:ok, {non_neg_integer, non_neg_integer}} | not_primary_error
182
+
@spec allocate_ids(server, non_neg_integer, pos_integer) :: {:ok, {non_neg_integer, non_neg_integer}} | request_error
178
183
def allocate_ids(server, generation, count) when is_integer(generation) and is_integer(count) do
179
184
request(server, {Commands, :allocate_ids, [generation, count]})
180
185
end
181
186
182
-
@spec request(server, {module, atom, list}) :: term | not_primary_error
187
+
@spec request(server, {module, atom, list}) :: term | request_error
183
188
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})
189
+
try do
190
+
SimServer.call(server, {:request, command})
191
+
catch
192
+
:exit, {:timeout, _mfa} -> {:error, :timeout}
193
+
end
185
194
end
186
195
187
196
def get_manager(server) do
+7
lib/servers/coordinator/commands.ex
+7
lib/servers/coordinator/commands.ex
···
1
1
defmodule Hobbes.Servers.Coordinator.Commands do
2
+
import Hobbes.Utils
3
+
2
4
def write(_get, _scan, [pairs]) when is_list(pairs) do
3
5
{pairs, :ok}
4
6
end
5
7
6
8
def read(get, _scan, [keys]) when is_list(keys) do
7
9
pairs = Enum.map(keys, fn k -> {k, get.(k)} end)
10
+
{[], pairs}
11
+
end
12
+
13
+
def dump(_get, scan, []) do
14
+
pairs = scan.("", special_prefix())
8
15
{[], pairs}
9
16
end
10
17
+151
lib/workloads/coordinator_read_write.ex
+151
lib/workloads/coordinator_read_write.ex
···
1
+
defmodule Hobbes.Workloads.CoordinatorReadWrite do
2
+
alias Hobbes.Construct.SimServer
3
+
alias Hobbes.Structs.Cluster
4
+
5
+
import Hobbes.Utils
6
+
import Hobbes.Workloads
7
+
8
+
@behaviour Hobbes.Workloads.Workload
9
+
10
+
defmodule Client do
11
+
use Hobbes.Construct.SimServer
12
+
13
+
alias Hobbes.Servers.Coordinator
14
+
import Hobbes.Utils
15
+
16
+
defmodule State do
17
+
@enforce_keys [
18
+
:id,
19
+
:tick_ms,
20
+
:cluster,
21
+
:current_coordinator,
22
+
23
+
:write_count,
24
+
:stats,
25
+
:stopped,
26
+
]
27
+
defstruct @enforce_keys
28
+
end
29
+
30
+
def start_link(args), do: SimServer.start_link(__MODULE__, args)
31
+
32
+
def init(%{id: id, tick_ms: tick_ms, cluster: %Cluster{} = cluster}) do
33
+
state = %State{
34
+
id: id,
35
+
tick_ms: tick_ms,
36
+
cluster: cluster,
37
+
current_coordinator: hd(cluster.coordinators),
38
+
39
+
write_count: 0,
40
+
stats: %{},
41
+
stopped: false,
42
+
}
43
+
44
+
SimServer.send_after self(), :tick, state.tick_ms
45
+
{:ok, state}
46
+
end
47
+
48
+
def handle_call(:stop, _from, %State{} = state) do
49
+
stats = Map.put(state.stats, :write_count, state.write_count)
50
+
51
+
state = %{state | stopped: true}
52
+
{:reply, stats, state}
53
+
end
54
+
55
+
def handle_info(:tick, %State{stopped: true} = state), do: {:noreply, state}
56
+
57
+
def handle_info(:tick, %State{} = state) do
58
+
state = tick(state)
59
+
60
+
SimServer.send_after self(), :tick, state.tick_ms
61
+
{:noreply, state}
62
+
end
63
+
64
+
defp tick(%State{} = state) do
65
+
with {:ok, state} <- do_writes(state)
66
+
do
67
+
%State{} = state
68
+
else
69
+
{:error, :timeout} -> inc(state, :timeout)
70
+
{:error, {:not_primary, new_primary}} -> %{state | current_coordinator: new_primary} |> inc(:not_primary)
71
+
end
72
+
end
73
+
74
+
defp do_writes(%State{} = state) do
75
+
id = state.id
76
+
count = Enum.random(1..3)
77
+
si = state.write_count
78
+
ei = state.write_count + count
79
+
80
+
pairs = Enum.map(si..ei, fn i ->
81
+
{
82
+
"key #{id} #{i}",
83
+
"value #{id} #{i}",
84
+
}
85
+
end)
86
+
87
+
case Coordinator.write(state.current_coordinator, pairs) do
88
+
{:ok, :ok} -> {:ok, %{state | write_count: state.write_count + count}}
89
+
{:error, _err} = error -> error
90
+
end
91
+
end
92
+
93
+
defp inc(%State{} = state, stat) do
94
+
%{state | stats: inc_stat(state.stats, stat)}
95
+
end
96
+
end
97
+
98
+
def run(%{cluster: %Cluster{} = cluster}, opts) do
99
+
num_clients = Keyword.get(opts, :num_clients, 10)
100
+
client_tick_ms = Keyword.get(opts, :client_tick_ms, 300)
101
+
duration_ms = Keyword.get(opts, :duration_ms, 10_000)
102
+
103
+
clients = Enum.map(1..num_clients, fn i ->
104
+
{:ok, pid} = Client.start_link(%{id: i, tick_ms: client_tick_ms, cluster: cluster})
105
+
pid
106
+
end)
107
+
108
+
SimServer.sleep(duration_ms)
109
+
110
+
client_stats =
111
+
clients
112
+
|> Enum.map(&SimServer.send_request(&1, :stop))
113
+
|> Enum.map(&SimServer.receive_response/1)
114
+
|> Enum.map(fn {:reply, reply} -> reply end)
115
+
116
+
cols = [:write_count, :timeout, :not_primary]
117
+
118
+
client_stats_table =
119
+
client_stats
120
+
|> Enum.with_index()
121
+
|> Enum.map(fn {stats, i} ->
122
+
cols
123
+
|> Enum.map(&Map.get(stats, &1, 0))
124
+
|> Enum.map(&pretty_number/1)
125
+
|> then(fn row ->
126
+
[to_string(i) | row]
127
+
end)
128
+
end)
129
+
|> table(
130
+
cols: ["i" | Enum.map(cols, &Atom.to_string/1)]
131
+
)
132
+
133
+
total_stats = sum_stats(client_stats)
134
+
135
+
total_stats_table = table(
136
+
[Enum.map(cols, fn col -> Map.get(total_stats, col, 0) |> pretty_number() end)],
137
+
cols: Enum.map(cols, &Atom.to_string/1)
138
+
)
139
+
140
+
{
141
+
:ok,
142
+
"""
143
+
Stats (per client):
144
+
#{client_stats_table}
145
+
146
+
Stats (total):
147
+
#{total_stats_table}
148
+
""",
149
+
}
150
+
end
151
+
end
+16
test/hobbes_test.exs
+16
test/hobbes_test.exs
···
156
156
end
157
157
end
158
158
159
+
defmodule CoordinatorReadWriteWorkloadTest do
160
+
use ExUnit.Case, async: true
161
+
@tag :coord_read_write
162
+
@tag :coord
163
+
@tag :disable
164
+
test "CoordinatorReadWrite", %{test: test} do
165
+
Workloads.run([
166
+
{Workloads.CoordinatorReadWrite, [
167
+
clients: 10,
168
+
client_tick_ms: 300,
169
+
duration_ms: 10_000,
170
+
]},
171
+
], HobbesTest.SimOpts.sim_opts(name: test, preset: :read_write))
172
+
end
173
+
end
174
+
159
175
defmodule ReadWriteBenchmarkWorkloadTest do
160
176
use ExUnit.Case, async: true
161
177
@tag :bench_rw