this repo has no description
Separate timeouts from the queue for performance (much faster)
+61
-37
+1
-1
ROADMAP.md
+1
-1
ROADMAP.md
···
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
76
- [X] Add timeouts to SimServer calls
77
-
- [ ] Optimize timeouts to avoid filling up queue (very slow)
77
+
- [X] Optimize timeouts to avoid filling up queue (very slow)
78
78
79
79
80
80
### Testing
+60
-36
lib/construct/scheduler.ex
+60
-36
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
-
31
26
defmodule State do
32
27
@type t :: %__MODULE__{
33
28
clock: non_neg_integer,
34
29
current: {pid, reference} | nil,
35
30
queue: map,
36
31
awaiting_message: %{pid => %Resume{}},
32
+
timeouts: [{non_neg_integer, pid}],
37
33
log_server_pid: pid,
38
34
resumes_without_send: non_neg_integer,
39
35
}
···
43
39
current: nil,
44
40
queue: %{},
45
41
awaiting_message: %{},
42
+
timeouts: [],
43
+
46
44
log_server_pid: nil,
47
45
resumes_without_send: 0,
48
46
]
···
175
173
end
176
174
177
175
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 =
176
+
# Add new timeout if needed
177
+
timeouts =
182
178
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})
179
+
:infinity -> state.timeouts
180
+
timeout_ms -> state.timeouts ++ [{state.clock + timeout_ms, pid}]
189
181
end
182
+
183
+
# Add awaiting_message entry and update timeouts
184
+
state = %State{state |
185
+
awaiting_message: Map.put(state.awaiting_message, pid, {check_fun, resume}),
186
+
timeouts: timeouts,
187
+
}
190
188
191
189
{:noreply, state}
192
190
end
···
196
194
nil -> :noop
197
195
{^pid, mref} -> Process.demonitor(mref, [:flush])
198
196
end
199
-
{:noreply, %State{state | current: nil} |> perform_next_task()}
197
+
{:noreply, %State{state | current: nil} |> do_next()}
200
198
end
201
199
202
200
def handle_info({:DOWN, mref, :process, pid, _reason} = message, %State{} = state) do
203
201
case state.current do
204
202
{^pid, ^mref} ->
205
-
{:noreply, %State{state | current: nil} |> perform_next_task()}
203
+
state = purge_timeouts(state, pid)
204
+
{:noreply, %State{state | current: nil} |> do_next()}
206
205
207
206
_ ->
208
207
raise """
···
220
219
%State{state | queue: queue}
221
220
end
222
221
222
+
defp do_next(%State{} = state) do
223
+
# Try to perform a timeout first, and then perform
224
+
# a task if there are no expired timeouts
225
+
case maybe_perform_timeout(state) do
226
+
:noop ->
227
+
perform_next_task(state)
228
+
229
+
{:performed, %State{} = state} ->
230
+
state
231
+
end
232
+
end
233
+
234
+
defp maybe_perform_timeout(%State{timeouts: timeouts} = state) do
235
+
case Enum.find(timeouts, fn {expires, _pid} -> expires < state.clock end) do
236
+
nil ->
237
+
:noop
238
+
239
+
{_expires, timed_out_pid} ->
240
+
{_check_fun, %Resume{} = resume} = Map.fetch!(state.awaiting_message, timed_out_pid)
241
+
242
+
# Resume the timed out process with a timeout message
243
+
# The process will then call exit(:timeout) (see await_resume/1)
244
+
state = monitor_current(state, resume.pid)
245
+
send resume.pid, {:timeout, resume.ref}
246
+
247
+
# Clean up the timeout
248
+
state = purge_timeouts(state, timed_out_pid)
249
+
{:performed, state}
250
+
end
251
+
end
252
+
223
253
defp perform_next_task(%State{queue: queue} = state) when map_size(queue) == 0 do
224
254
raise """
225
255
Attempted to call `resume_next/1` but the queue is empty! Possible deadlock?
···
281
311
log(state, {:send, time, send.to_pid, send.message})
282
312
state = send_message(state, send)
283
313
284
-
perform_next_task(state)
285
-
end
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
314
+
do_next(state)
301
315
end
302
316
303
317
defp perform(%State{} = state, _time, %Resume{} = resume) do
···
322
336
nil ->
323
337
state
324
338
325
-
{_ref, check_fun, %Resume{} = resume} when is_function(check_fun) ->
339
+
{check_fun, %Resume{} = resume} when is_function(check_fun) ->
340
+
# If check_fun returns true for the message then it matches what
341
+
# the process is waiting for and we can resume it
326
342
case check_fun.(send.message) do
327
343
false ->
328
344
state
329
345
330
346
true ->
331
-
state = %State{state | awaiting_message: Map.delete(state.awaiting_message, send.to_pid)}
347
+
# Remove the awaiting_message and timeouts entries
348
+
state =
349
+
%State{state | awaiting_message: Map.delete(state.awaiting_message, send.to_pid)}
350
+
|> purge_timeouts(send.to_pid)
351
+
# Queue a resume for the process
332
352
queue_task(state, state.clock, resume)
333
353
end
334
354
end
355
+
end
356
+
357
+
defp purge_timeouts(%State{} = state, pid) when is_pid(pid) do
358
+
%State{state | timeouts: Enum.reject(state.timeouts, &(elem(&1, 1) == pid))}
335
359
end
336
360
337
361
defp log(%State{} = state, event) do