this repo has no description
Add name registration
+101
-27
+7
-16
lib/trinity/scheduler.ex
+7
-16
lib/trinity/scheduler.ex
···
30
30
end
31
31
32
32
defmacro simulation_key, do: :_trinity_simulation
33
+
defmacro sim_node_key, do: :_trinity_simulation_node
33
34
34
35
@spec get_sim :: Simulation.t
35
36
defp get_sim, do: Process.get(simulation_key())
···
75
76
76
77
@spec spawn_node_and_yield(function, atom, boolean) :: pid
77
78
def spawn_node_and_yield(fun, node, link?) do
78
-
%{proc_nodes: proc_nodes} = sim = get_sim()
79
-
node = node || get_proc_node(proc_nodes)
79
+
sim = get_sim()
80
+
node = node || get_proc_node()
80
81
81
82
parent_pid = self()
82
83
spawn_ref = make_ref()
···
230
231
linked
231
232
end
232
233
233
-
def handle_sent(dest_ref) when is_reference(dest_ref) do
234
-
%{proc_aliases: proc_aliases} = get_sim()
235
-
# Look up the alias ref in the aliases table to get its pid
236
-
case :ets.lookup(proc_aliases, dest_ref) do
237
-
[{_key, dest_pid}] -> handle_sent(dest_pid)
238
-
# This alias has no known pid (alias was unaliased), do nothing
239
-
[] -> :noop
240
-
end
241
-
end
242
-
243
234
def handle_sent(dest_pid) when is_pid(dest_pid) do
244
235
%{queue: queue, proc_queue_keys: proc_queue_keys, now: now} = get_sim()
245
236
···
386
377
end
387
378
end
388
379
389
-
defp get_proc_node(proc_nodes) do
390
-
[{_pid, node}] = :ets.lookup(proc_nodes, self())
391
-
node
392
-
end
380
+
defp get_proc_node, do: Process.get(sim_node_key())
393
381
394
382
defp put_proc_node(proc_nodes, node_procs, node) do
395
383
pid = self()
384
+
385
+
Process.put(sim_node_key(), node)
396
386
:ets.insert(proc_nodes, {pid, node})
387
+
397
388
existing_procs = case :ets.lookup(node_procs, node) do
398
389
[{_node, procs}] -> procs
399
390
[] -> []
+85
-11
lib/trinity/sim_process.ex
+85
-11
lib/trinity/sim_process.ex
···
2
2
alias Trinity.Scheduler
3
3
alias Trinity.Scheduler.Simulation
4
4
5
-
import Trinity.Scheduler, only: [simulation_key: 0]
5
+
import Trinity.Scheduler, only: [simulation_key: 0, sim_node_key: 0]
6
6
7
7
@spec get_sim :: Simulation.t | nil
8
8
defp get_sim, do: Process.get(simulation_key())
9
9
10
+
@spec get_proc_node :: atom
11
+
defp get_proc_node, do: Process.get(sim_node_key())
12
+
13
+
@spec send(pid | reference | atom | {atom, node}, term) :: term
10
14
def send(dest, message) do
11
-
Scheduler.handle_sent(dest)
12
-
Kernel.send(dest, message)
15
+
case get_sim() do
16
+
nil -> Kernel.send(dest, message)
17
+
_sim -> sim_send(dest, message)
18
+
end
19
+
end
20
+
21
+
@spec node :: node
22
+
def node do
23
+
case get_sim() do
24
+
nil -> Kernel.node()
25
+
_sim -> get_proc_node()
26
+
end
13
27
end
14
28
15
29
@spec spawn_node(atom, function) :: pid
···
28
42
end
29
43
end
30
44
45
+
@spec unalias(reference) :: boolean
46
+
def unalias(alias) do
47
+
case get_sim() do
48
+
nil -> Process.unalias(alias)
49
+
_sim -> sim_unalias(alias)
50
+
end
51
+
end
52
+
53
+
@spec register(pid, atom) :: true
54
+
def register(pid, name) do
55
+
case get_sim() do
56
+
nil -> Process.register(pid, name)
57
+
_sim -> sim_register(pid, name)
58
+
end
59
+
end
60
+
61
+
defp sim_send(pid, message) when is_pid(pid) do
62
+
Scheduler.handle_sent(pid)
63
+
Kernel.send(pid, message)
64
+
end
65
+
66
+
defp sim_send(alias, message) when is_reference(alias) do
67
+
%{proc_aliases: proc_aliases} = get_sim()
68
+
case :ets.lookup(proc_aliases, alias) do
69
+
[{_key, pid}] -> Scheduler.handle_sent(pid)
70
+
# If the alias doesn't exist, do nothing
71
+
[] -> :noop
72
+
end
73
+
74
+
Kernel.send(alias, message)
75
+
end
76
+
77
+
defp sim_send(name, message) when is_atom(name) do
78
+
%{proc_aliases: proc_aliases} = get_sim()
79
+
node = get_proc_node()
80
+
case :ets.lookup(proc_aliases, {name, node}) do
81
+
[{_key, pid}] ->
82
+
Scheduler.handle_sent(pid)
83
+
Kernel.send(pid, message)
84
+
85
+
[] -> raise ArgumentError, "Invalid destination #{inspect(name)}"
86
+
end
87
+
end
88
+
89
+
defp sim_send({_name, _node} = name_node, message) do
90
+
%{proc_aliases: proc_aliases} = get_sim()
91
+
case :ets.lookup(proc_aliases, name_node) do
92
+
[{_key, pid}] ->
93
+
Scheduler.handle_sent(pid)
94
+
Kernel.send(pid, message)
95
+
96
+
[] -> raise ArgumentError, "Invalid destination #{inspect(name_node)}"
97
+
end
98
+
end
99
+
31
100
defp sim_alias do
32
101
%Simulation{proc_aliases: proc_aliases} = get_sim()
33
102
···
37
106
alias
38
107
end
39
108
40
-
@spec unalias(reference) :: boolean
41
-
def unalias(alias) do
42
-
case get_sim() do
43
-
nil -> Process.unalias(alias)
44
-
_sim -> sim_unalias(alias)
45
-
end
46
-
end
47
-
48
109
defp sim_unalias(alias) do
49
110
%Simulation{proc_aliases: proc_aliases} = get_sim()
50
111
:ets.delete(proc_aliases, alias)
51
112
52
113
Process.unalias(alias)
114
+
end
115
+
116
+
defp sim_register(pid, name) do
117
+
%{proc_aliases: proc_aliases} = get_sim()
118
+
node = get_proc_node()
119
+
120
+
key = {name, node}
121
+
case :ets.lookup(proc_aliases, key) do
122
+
[] -> :ets.insert(proc_aliases, {key, pid})
123
+
[{_key, _pid}] -> raise ArgumentError, "Could not register #{inspect(pid)} with name #{inspect(name)}"
124
+
end
125
+
126
+
true
53
127
end
54
128
end
+9
test/trinity_test.exs
+9
test/trinity_test.exs
···
26
26
nodes = [:n1, :n2, :n3]
27
27
pids =
28
28
Enum.map(nodes, fn node ->
29
+
name = String.to_atom(Atom.to_string(node) <> "_proc")
29
30
parent = self()
30
31
ref = make_ref()
31
32
32
33
SimProcess.spawn_node(node, fn ->
34
+
SimProcess.register(self(), name)
35
+
36
+
receive_yield do
37
+
:begin -> dbg {"began", name}
38
+
end
39
+
33
40
pids = Enum.map(1..10, fn i ->
34
41
{:ok, pid} = Counter.start_link(i)
35
42
pid
···
37
44
38
45
SimProcess.send parent, {ref, pids}
39
46
end)
47
+
48
+
SimProcess.send {name, node}, :begin
40
49
41
50
receive_yield do
42
51
{^ref, pids} -> pids