1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
|
defprotocol Shard.Manifest do
@moduledoc"""
A shard manifest is a data structure that uniquely defines the identity of the shard.
The hash of the manifest is the unique identifier of that shard on the network.
The Manifest protocol is a protocol implemented by the manifest structs for the
different shard types. It contains an operation start() that is able to launch the
correct process for this shard and connect to other peers that use it.
"""
@doc"""
Start the corresponding Shard process
"""
def start(manifest)
end
defmodule Shard.Manager do
@moduledoc"""
Maintains several important tables :
- :shard_db (persistent with DETS)
List of
{ id, manifest, pid | nil }
- :shard_state (persistent with DETS)
List of
{ id, state }
- :peer_db (persistent with DETS)
Mult-list of
{ shard_id, peer_info } # TODO: add health info (last seen, ping, etc)
peer_info := {:inet, ip, port} | {:inet6, ip, port} | {:onion, name}
- :shard_procs (not persistent)
List of
{ {id, path}, pid }
"""
use GenServer
require Logger
@shard_db [Application.get_env(:shard, :data_path), "shard_db"] |> Path.join |> String.to_atom
@shard_state [Application.get_env(:shard, :data_path), "shard_state"] |> Path.join |> String.to_atom
@peer_db [Application.get_env(:shard, :data_path), "peer_db"] |> Path.join |> String.to_atom
def start_link(_) do
GenServer.start_link(__MODULE__, nil, name: __MODULE__)
end
def init(_) do
:dets.open_file(@shard_db, [type: :set])
for [{id, manifest, _pid}] <- :dets.match @shard_db, :"$1" do
:dets.insert @shard_db, {id, manifest, nil}
spawn fn -> Shard.Manifest.start manifest end
end
:dets.open_file(@shard_state, [type: :set])
:dets.open_file(@peer_db, [type: :bag])
:ets.new(:shard_procs, [:set, :protected, :named_table])
{:ok, nil}
end
def handle_call({:register, shard_id, manifest, pid}, _from, state) do
will_live = case :dets.lookup(@shard_db, shard_id) do
[{ ^shard_id, _, pid }] when pid != nil -> not Process.alive?(pid)
_ -> true
end
reply = if will_live do
Process.monitor(pid)
:dets.insert(@shard_db, {shard_id, manifest, pid})
:ok
else
:redundant
end
{:reply, reply, state}
end
def handle_cast({:dispatch_to, shard_id, path, pid}, state) do
:ets.insert(:shard_procs, { {shard_id, path}, pid })
Process.monitor(pid)
{:noreply, state}
end
def handle_cast({:interested, conn_pid, peer_info, auth, shards}, state) do
for shard_id <- shards do
case :dets.lookup(@shard_db, shard_id) do
[{ ^shard_id, _, pid }] ->
:dets.insert(@peer_db, {shard_id, peer_info})
GenServer.cast(pid, {:interested, conn_pid, auth})
[] -> nil
end
end
{:noreply, state}
end
def handle_cast({:not_interested, peer_info, shard_id}, state) do
:dets.match_delete(@peer_db, {shard_id, peer_info})
{:noreply, state}
end
def handle_cast({:shard_peer_db_insert, shard_id, peer_info}, state) do
:dets.insert(@peer_db, {shard_id, peer_info})
{:noreply, state}
end
def handle_info({:DOWN, _, :process, pid, _}, state) do
:ets.match_delete(:shard_procs, {:_, pid})
{:noreply, state}
end
# ======================
# CALLED BY SNet.TcpConn
# ======================
@doc"""
Dispatch incoming message to correct shard process
"""
def incoming(conn_pid, peer_info, auth, {:interested, shards}) do
GenServer.cast(__MODULE__, {:interested, conn_pid, peer_info, auth, shards})
end
def incoming(_conn_pid, peer_info, _auth, {:not_interested, shard}) do
GenServer.cast(__MODULE__, {:not_interested, peer_info, shard})
end
def incoming(conn_pid, peer_info, auth, {shard_id, path, msg}) do
case :dets.lookup(@shard_db, shard_id) do
[] ->
GenServer.cast(conn_pid, {:send_msg, {:not_interested, shard_id}})
[_] ->
case :dets.match(@peer_db, {shard_id, peer_info}) do
[] ->
GenServer.cast(__MODULE__, {:shard_peer_db_insert, shard_id, peer_info})
_ -> nil
end
case :ets.lookup(:shard_procs, {shard_id, path}) do
[{ {^shard_id, ^path}, pid }] ->
GenServer.cast(pid, {:msg, conn_pid, auth, shard_id, path, msg})
[] ->
Logger.info("Warning: dropping message for #{inspect shard_id}/#{inspect path}, no handler running.\n\t#{inspect msg}")
end
end
end
# ================
# CALLED BY Sapp.*
# ================
@doc"""
Register a process as the main process for a shard.
Returns either :ok or :redundant, in which case the process must exit.
"""
def register(shard_id, manifest, pid) do
GenServer.call(__MODULE__, {:register, shard_id, manifest, pid})
end
@doc"""
Register a process as the handler for shard packets for a given path.
"""
def dispatch_to(shard_id, path, pid) do
GenServer.cast(__MODULE__, {:dispatch_to, shard_id, path, pid})
end
@doc"""
Return the list of all peer info for peers that are interested in a certain shard
"""
def get_shard_peers(shard_id) do
for {_, peer_info} <- :dets.lookup(@peer_db, shard_id), do: peer_info
end
@doc"""
Return the saved state value for a shard
"""
def load_state(shard_id) do
case :dets.lookup(@shard_state, shard_id) do
[{^shard_id, state}] -> state
_ -> nil
end
end
@doc"""
Save a state value for a shard
"""
def save_state(shard_id, state) do
:dets.insert(@shard_state, {shard_id, state})
end
# ================
# CALLED BY ANYONE
# ================
@doc"""
Returns the pid for a shard if it exists
"""
def find_proc(shard_id) do
case :dets.lookup(@shard_db, shard_id) do
[{^shard_id, _, pid}] -> pid
_ -> nil
end
end
@doc"""
Return the list of all shards.
"""
def list_shards() do
for [x] <- :dets.match(@shard_db, :"$1"), do: x
end
end
|
