this repo has no description
Trigger view change during coordinator workload
+52
-6
+15
-1
lib/construct/scheduler.ex
+15
-1
lib/construct/scheduler.ex
···
149
149
GenServer.call(scheduler, {:register_process, self(), pid, name})
150
150
end
151
151
152
+
def whereis(scheduler, name) when is_atom(name) do
153
+
GenServer.call(scheduler, {:whereis, self(), name})
154
+
end
155
+
152
156
@spec get_file_store(pid) :: FileStore.t
153
157
def get_file_store(scheduler) do
154
158
GenServer.call(scheduler, {:get_file_store, self()})
···
216
220
{:reply, :ok, state}
217
221
{:error, _error} ->
218
222
{:reply, :error, state}
223
+
end
224
+
end
225
+
226
+
def handle_call({:whereis, _from_pid, name}, _from, %State{} = state) do
227
+
# TODO: use node of from_pid
228
+
node = :nonode
229
+
230
+
case ProcRegistry.whereis(state.proc_registry, node, name) do
231
+
{:ok, pid} -> {:reply, pid, state}
232
+
{:error, :not_found} -> {:reply, nil, state}
219
233
end
220
234
end
221
235
···
406
420
407
421
case ProcStore.fetch_state(state.proc_store, exit.to_pid) do
408
422
{:ok, %ProcState{} = proc} ->
409
-
case proc.trap_exit do
423
+
case proc.trap_exit and exit.reason != :kill do
410
424
true ->
411
425
state
412
426
|> send_message(%Send{dest: proc.pid, message: {:EXIT, exit.from_pid, exit.reason}})
+10
lib/construct/sim_server.ex
+10
lib/construct/sim_server.ex
···
370
370
end
371
371
end
372
372
373
+
@spec whereis(atom) :: pid | nil
374
+
def whereis(name) do
375
+
case get_scheduler_pid() do
376
+
scheduler_pid when is_pid(scheduler_pid) ->
377
+
Scheduler.whereis(scheduler_pid, name)
378
+
nil ->
379
+
Process.whereis(name)
380
+
end
381
+
end
382
+
373
383
@doc """
374
384
Returns a monotonic microsecond timestamp.
375
385
+5
-2
lib/servers/coordinator.ex
+5
-2
lib/servers/coordinator.ex
···
187
187
@spec request(server, {module, atom, list}) :: term | request_error
188
188
def request(server, {m, f, a} = command) when is_atom(m) and is_atom(f) and is_list(a) do
189
189
try do
190
-
SimServer.call(server, {:request, command})
190
+
SimServer.call(server, {:request, command}, 1000)
191
191
catch
192
192
:exit, {:timeout, _mfa} -> {:error, :timeout}
193
193
end
···
445
445
446
446
{:noreply, state}
447
447
448
-
false -> {:noreply, state}
448
+
# TODO: get rid of this once we have a proper supervision tree
449
+
# (for now we need to be able to kill coordinators for testing)
450
+
false -> exit(:normal)
451
+
#false -> {:noreply, state}
449
452
end
450
453
end
451
454
+7
-2
lib/workloads/coordinator_read_write.ex
+7
-2
lib/workloads/coordinator_read_write.ex
···
68
68
do
69
69
%State{} = state
70
70
else
71
-
{:error, :timeout} -> inc(state, :timeout)
72
-
{:error, {:not_primary, new_primary}} -> %{state | current_coordinator: new_primary} |> inc(:not_primary)
71
+
{:error, :timeout} ->
72
+
# Current coordinator is not responding, try another one
73
+
new_coordinator = List.delete(state.cluster.coordinators, state.current_coordinator) |> Enum.random()
74
+
%{state | current_coordinator: new_coordinator} |> inc(:timeout)
75
+
76
+
{:error, {:not_primary, new_primary}} ->
77
+
%{state | current_coordinator: new_primary} |> inc(:not_primary)
73
78
end
74
79
end
75
80
+14
-1
lib/workloads/kill_servers.ex
+14
-1
lib/workloads/kill_servers.ex
···
10
10
def run(%{cluster: %Cluster{} = cluster}, opts) do
11
11
delay_ms = Keyword.get(opts, :delay_ms, 0)
12
12
duration_ms = Keyword.get(opts, :duration_ms, 0)
13
+
server_type = Keyword.get(opts, :server_type, Hobbes.Servers.TLog)
13
14
14
15
SimServer.sleep(delay_ms)
15
16
16
-
%Server{pid: kill_pid} = get_servers(cluster, Hobbes.Servers.TLog) |> hd()
17
+
kill_pid =
18
+
case server_type do
19
+
Hobbes.Servers.Coordinator ->
20
+
hd(cluster.coordinators)
21
+
|> case do
22
+
pid when is_pid(pid) -> pid
23
+
name when is_atom(name) -> SimServer.whereis(name)
24
+
end
25
+
other ->
26
+
%Server{pid: pid} = get_servers(cluster, other) |> hd()
27
+
pid
28
+
end
29
+
17
30
SimServer.exit(kill_pid, :shutdown)
18
31
19
32
SimServer.sleep(duration_ms)