this repo has no description
Add timeouts to SimServer calls and receives
+97
-39
+2
-1
ROADMAP.md
+2
-1
ROADMAP.md
···
73
73
- [X] Optimization: avoid waking a process unless it has received messages
74
74
- [X] Use a single clock source in Scheduler (above optimization slowed per-process clock progression)
75
75
- [X] Optimization: only wake a process when it receives the message it's waiting for
76
-
- [ ] Add timeouts to SimServer calls
76
+
- [X] Add timeouts to SimServer calls
77
+
- [ ] Optimize timeouts to avoid filling up queue (very slow)
77
78
78
79
79
80
### Testing
+53
-20
lib/construct/scheduler.ex
+53
-20
lib/construct/scheduler.ex
···
23
23
defstruct @enforce_keys
24
24
end
25
25
26
+
defmodule Timeout do
27
+
@enforce_keys [:pid, :ref]
28
+
defstruct @enforce_keys
29
+
end
30
+
26
31
defmodule State do
27
32
@type t :: %__MODULE__{
28
33
clock: non_neg_integer,
···
54
59
GenServer.cast(scheduler, {:queue_task, self(), delay, resume})
55
60
GenServer.cast(scheduler, {:yield, self()})
56
61
57
-
ref = resume.ref
58
-
receive do
59
-
{:resume, ^ref, nil} -> :ok
60
-
end
62
+
nil = await_resume(resume)
63
+
:ok
61
64
end
62
65
63
-
@spec yield_until_message(pid, function) :: :ok
64
-
def yield_until_message(scheduler, check_fun) when is_function(check_fun) do
66
+
@spec yield_until_message(pid, function, non_neg_integer | :infinity) :: :ok
67
+
def yield_until_message(scheduler, check_fun, timeout_ms)
68
+
when is_function(check_fun) and ((is_integer(timeout_ms) and timeout_ms >= 0) or timeout_ms == :infinity) do
65
69
resume = %Resume{ref: make_ref(), pid: self(), value: nil}
66
70
67
-
GenServer.cast(scheduler, {:await_message, self(), check_fun, resume})
71
+
GenServer.cast(scheduler, {:await_message, self(), check_fun, timeout_ms, resume})
68
72
GenServer.cast(scheduler, {:yield, self()})
69
73
70
-
ref = resume.ref
71
-
receive do
72
-
{:resume, ^ref, nil} -> :ok
73
-
end
74
+
nil = await_resume(resume)
75
+
:ok
74
76
end
75
77
76
78
@spec spawn_and_yield(term, :link | :nolink, module, atom, [term]) :: pid
···
81
83
GenServer.cast(scheduler, {:queue_task, self(), 0, spawn})
82
84
GenServer.cast(scheduler, {:yield, self()})
83
85
84
-
ref = resume.ref
85
-
spawned_pid = receive do
86
-
{:resume, ^ref, spawned_pid} -> spawned_pid
87
-
end
88
-
86
+
spawned_pid = await_resume(resume)
89
87
case link do
90
88
:link -> Process.link(spawned_pid)
91
89
:nolink -> :noop
92
90
end
93
91
94
92
spawned_pid
93
+
end
94
+
95
+
defp await_resume(%Resume{ref: ref} = _resume) do
96
+
receive do
97
+
{:resume, ^ref, value} -> value
98
+
{:timeout, ^ref} -> exit(:timeout)
99
+
end
95
100
end
96
101
97
102
@spec send(term, pid, term) :: :ok
···
169
174
{:noreply, state}
170
175
end
171
176
172
-
def handle_cast({:await_message, pid, check_fun, %Resume{} = resume}, state) when is_pid(pid) and is_function(check_fun) do
173
-
awaiting_message = Map.put(state.awaiting_message, pid, {check_fun, resume})
174
-
{:noreply, %State{state | awaiting_message: awaiting_message}}
177
+
def handle_cast({:await_message, pid, check_fun, timeout_ms, %Resume{} = resume}, state) when is_pid(pid) and is_function(check_fun) do
178
+
ref = make_ref()
179
+
state = %State{state | awaiting_message: Map.put(state.awaiting_message, pid, {ref, check_fun, resume})}
180
+
181
+
state =
182
+
case timeout_ms do
183
+
:infinity ->
184
+
state
185
+
186
+
timeout_ms when is_integer(timeout_ms) and timeout_ms >= 0 ->
187
+
expires_after = state.clock + timeout_ms
188
+
queue_task(state, expires_after, %Timeout{pid: pid, ref: ref})
189
+
end
190
+
191
+
{:noreply, state}
175
192
end
176
193
177
194
def handle_cast({:yield, pid}, %State{} = state) when is_pid(pid) do
···
267
284
perform_next_task(state)
268
285
end
269
286
287
+
defp perform(%State{} = state, _time, %Timeout{pid: pid, ref: ref}) do
288
+
case Map.get(state.awaiting_message, pid) do
289
+
# The process is still waiting, so we trigger the timeout
290
+
{^ref, _check_fun, %Resume{} = resume} ->
291
+
state = monitor_current(state, pid)
292
+
send resume.pid, {:timeout, resume.ref}
293
+
294
+
state
295
+
296
+
# The process is either not waiting anymore, or waiting for
297
+
# a different Timeout
298
+
_ ->
299
+
perform_next_task(state)
300
+
end
301
+
end
302
+
270
303
defp perform(%State{} = state, _time, %Resume{} = resume) do
271
304
#log(state, {:resume, time, resume.pid, resume.value})
272
305
···
289
322
nil ->
290
323
state
291
324
292
-
{check_fun, %Resume{} = resume} when is_function(check_fun) ->
325
+
{_ref, check_fun, %Resume{} = resume} when is_function(check_fun) ->
293
326
case check_fun.(send.message) do
294
327
false ->
295
328
state
+23
-9
lib/construct/sim_server.ex
+23
-9
lib/construct/sim_server.ex
···
22
22
end
23
23
24
24
"""
25
-
defmacro yield_receive(scheduler_pid, [do: do_block]) do
25
+
defmacro yield_receive(scheduler_pid, timeout \\ :infinity, [do: do_block]) do
26
26
# Here, we use `do_block` to build a function which
27
27
# checks if a value matches the intended receive block
28
28
#
···
82
82
case receive_func.() do
83
83
^ref ->
84
84
# If there is no matching message, we yield until one is sent to us
85
-
Hobbes.Construct.Scheduler.yield_until_message(unquote(scheduler_pid), check_func)
85
+
Hobbes.Construct.Scheduler.yield_until_message(unquote(scheduler_pid), check_func, unquote(timeout))
86
86
87
87
# Now that a message has arrived, we extract it with the receive
88
88
# The case is a sanity check
···
159
159
nil ->
160
160
GenServer.call(server, request, timeout)
161
161
scheduler_pid when is_pid(scheduler_pid) ->
162
-
sim_call(scheduler_pid, server, request)
162
+
sim_call(scheduler_pid, server, request, timeout)
163
163
end
164
164
end
165
165
···
202
202
end
203
203
end
204
204
205
-
defp sim_call(scheduler_pid, server, request) do
205
+
defp sim_call(scheduler_pid, server, request, timeout) do
206
206
request_id = sim_send_request(scheduler_pid, server, request)
207
-
sim_receive_response(scheduler_pid, request_id)
207
+
try do
208
+
yield_receive(scheduler_pid, timeout) do
209
+
{^request_id, reply} -> reply
210
+
end
211
+
catch
212
+
:exit, reason ->
213
+
# GenServer: {:timeout, {m, f, a}}
214
+
# SimServer: {{:timeout, {m, f, a}}, stacktrace}
215
+
exit({{reason, {__MODULE__, :call, [server, request]}}, __STACKTRACE__})
216
+
end
208
217
end
209
218
210
219
defp sim_send_request(scheduler_pid, server, request) do
···
217
226
ref
218
227
end
219
228
220
-
defp sim_receive_response(scheduler_pid, request_id, _timeout \\ 5000) do
221
-
# TODO: timeout?
222
-
yield_receive(scheduler_pid) do
223
-
{^request_id, reply} -> reply
229
+
defp sim_receive_response(scheduler_pid, request_id, timeout) do
230
+
# TODO: catch timeout exit
231
+
try do
232
+
yield_receive(scheduler_pid, timeout) do
233
+
{^request_id, reply} -> reply
234
+
end
235
+
catch
236
+
:exit, reason ->
237
+
exit({{reason, {__MODULE__, :receive_response, [request_id]}}, __STACKTRACE__})
224
238
end
225
239
end
226
240
+1
-1
lib/servers/distributor.ex
+1
-1
lib/servers/distributor.ex
···
31
31
def start_link, do: SimServer.start_link(__MODULE__, nil)
32
32
33
33
def move_shard(server, shard_key, to_servers) when is_binary(shard_key) and is_list(to_servers) do
34
-
SimServer.call(server, {:move_shard, shard_key, to_servers})
34
+
SimServer.call(server, {:move_shard, shard_key, to_servers}, 20_000)
35
35
end
36
36
37
37
def init(nil) do
+1
lib/servers/storage.ex
+1
lib/servers/storage.ex
+16
-7
lib/workloads.ex
+16
-7
lib/workloads.ex
···
92
92
93
93
failed_message =
94
94
failed
95
-
|> Enum.map(fn {{:exit, {exception, stacktrace}}, seed} ->
96
-
"""
97
-
- Seed #{inspect(seed)} failed with error:
98
-
#{Exception.format_banner(:error, exception, stacktrace) |> color(:red) |> indent()}
99
-
#{"stacktrace:" |> color(:cyan) |> indent()}
100
-
#{Exception.format_stacktrace(stacktrace)}
101
-
"""
95
+
|> Enum.map(fn
96
+
# TODO: stacktrace only comes from SimServer, not GenServer
97
+
{{:exit, {{:timeout, _mfa} = reason, stacktrace}}, seed} ->
98
+
"""
99
+
- Seed #{inspect(seed)} timed out
100
+
#{Exception.format_exit(reason) |> color(:red) |> indent()}
101
+
#{Exception.format_stacktrace(stacktrace)}
102
+
"""
103
+
104
+
{{:exit, {exception, stacktrace}}, seed} ->
105
+
"""
106
+
- Seed #{inspect(seed)} failed with error:
107
+
#{Exception.format_banner(:error, exception, stacktrace) |> color(:red) |> indent()}
108
+
#{"stacktrace:" |> color(:cyan) |> indent()}
109
+
#{Exception.format_stacktrace(stacktrace)}
110
+
"""
102
111
end)
103
112
|> Enum.join("\n")
104
113
+1
-1
lib/workloads/shard_move.ex
+1
-1
lib/workloads/shard_move.ex