this repo has no description
Add links
+69
-10
+65
-6
lib/trinity/scheduler.ex
+65
-6
lib/trinity/scheduler.ex
···
1
1
defmodule Trinity.Scheduler do
2
2
defmodule Simulation do
3
-
@enforce_keys [:queue, :now, :supervisor_pid]
3
+
@enforce_keys [
4
+
:queue,
5
+
:proc_links,
6
+
:now,
7
+
:supervisor_pid,
8
+
]
4
9
defstruct @enforce_keys
5
10
end
6
11
···
39
44
def start do
40
45
sim = %Simulation{
41
46
queue: :ets.new(__MODULE__, [:ordered_set, :public]),
47
+
proc_links: :ets.new(__MODULE__, [:set, :public]),
42
48
now: :atomics.new(1, signed: false),
43
49
supervisor_pid: nil,
44
50
}
···
65
71
end
66
72
67
73
@spec spawn_and_yield(function) :: pid
68
-
def spawn_and_yield(fun) do
69
-
%Simulation{queue: queue, now: now} = sim = get_sim()
70
-
yield_ref = enqueue_self(queue, now, 0)
74
+
def spawn_and_yield(fun, link? \\ false) do
75
+
sim = get_sim()
76
+
parent_pid = self()
77
+
spawn_ref = make_ref()
71
78
72
79
spawned_pid = spawn(fn ->
80
+
# In the new process, we first register with the supervisor
81
+
# and inject the sim
73
82
SimulationSupervisor.register_self(sim.supervisor_pid)
74
83
put_sim(sim)
75
84
85
+
# We then enqueue the new process and unsuspend the parent
86
+
# so that it can register any links before yielding back
87
+
# to us (eventually, if there are others in queue at `time=now`)
88
+
yield_ref = enqueue_self(sim.queue, sim.now, 0)
89
+
send parent_pid, spawn_ref
90
+
receive do
91
+
^yield_ref -> :noop
92
+
end
93
+
94
+
# Once we are properly resumed we can execute `fun`
76
95
fun.()
77
96
end)
78
97
98
+
# We suspend the parent process while waiting for the child
99
+
# to register itself
79
100
receive do
80
-
^yield_ref -> spawned_pid
101
+
^spawn_ref -> :noop
102
+
end
103
+
# Register the link, if requested
104
+
# Note that at this point the child process has registered itself
105
+
# but has not begun executing its `fun` (and so cannot have crashed)
106
+
if link?, do: add_link(spawned_pid)
107
+
108
+
# Yield to the child (or anything else in queue before it)
109
+
yield(0)
110
+
spawned_pid
111
+
end
112
+
113
+
@spec add_link(pid) :: :ok
114
+
def add_link(to_pid) do
115
+
# TODO: noproc if `to_pid` has exited
116
+
%Simulation{proc_links: proc_links} = get_sim()
117
+
from_pid = self()
118
+
119
+
from_links =
120
+
case :ets.lookup(proc_links, from_pid) do
121
+
[{^from_pid, links}] -> links
122
+
_ -> []
123
+
end
124
+
125
+
to_links =
126
+
case :ets.lookup(proc_links, to_pid) do
127
+
[{^to_pid, links}] -> links
128
+
_ -> []
129
+
end
130
+
131
+
case to_pid in from_links do
132
+
false ->
133
+
:ets.insert(proc_links, {from_pid, [to_pid | from_links]})
134
+
:ets.insert(proc_links, {to_pid, [from_pid | to_links]})
135
+
_ -> :noop
81
136
end
137
+
138
+
:ok
82
139
end
83
140
84
141
defp perform_next(queue, now) do
···
124
181
end
125
182
126
183
@doc false
127
-
def dump(%Simulation{} = sim) do
184
+
def dump do
185
+
sim = get_sim()
128
186
%{
129
187
queue: :ets.tab2list(sim.queue),
188
+
proc_links: :ets.tab2list(sim.proc_links),
130
189
now: read_now(sim.now),
131
190
supervisor_pid: sim.supervisor_pid,
132
191
}
+4
-4
test/trinity_test.exs
+4
-4
test/trinity_test.exs
···
6
6
test "scheduler" do
7
7
Scheduler.start()
8
8
9
-
pid1 = Scheduler.spawn_and_yield(fn -> dbg "hello 1" end)
10
-
pid2 = Scheduler.spawn_and_yield(fn -> dbg "hello 2" end)
11
-
pid3 = Scheduler.spawn_and_yield(fn -> dbg "hello 3" end)
9
+
pid1 = Scheduler.spawn_and_yield(fn -> dbg "hello 1" end, true)
10
+
pid2 = Scheduler.spawn_and_yield(fn -> dbg "hello 2" end, false)
11
+
pid3 = Scheduler.spawn_and_yield(fn -> dbg "hello 3" end, true)
12
12
dbg {pid1, pid2, pid3}
13
13
14
14
Scheduler.yield(100)
15
-
dbg "foo"
15
+
dbg Scheduler.dump()
16
16
end
17
17
end