this repo has no description
Refactor distributor to hold shard moves in memory
+58
-48
+58
-48
lib/servers/distributor.ex
+58
-48
lib/servers/distributor.ex
···
7
7
8
8
import Hobbes.MetaStore, only: [meta: 1]
9
9
10
+
defmodule ShardMove do
11
+
@enforce_keys [
12
+
:initiated_from,
13
+
:shard_key,
14
+
:to_servers,
15
+
]
16
+
defstruct @enforce_keys
17
+
end
18
+
10
19
defmodule State do
11
20
@enforce_keys [
12
21
:id,
13
22
:cluster,
14
23
]
15
24
defstruct [
16
-
shard_moves_from: %{},
25
+
shard_moves: [],
17
26
] ++@enforce_keys
18
27
end
19
28
20
-
@check_imports_interval_ms 1000
29
+
@tick_shard_moves_interval_ms 1000
21
30
22
31
def start_link, do: SimServer.start_link(__MODULE__, nil)
23
32
···
35
44
def handle_call({:setup, %{id: id, cluster: %Cluster{} = cluster}}, _from, state) do
36
45
state = %State{state | id: id, cluster: cluster}
37
46
38
-
SimServer.send_after(self(), :check_imports, @check_imports_interval_ms)
47
+
SimServer.send_after(self(), :tick_shard_moves, @tick_shard_moves_interval_ms)
39
48
40
49
{:reply, :ok, state}
41
50
end
42
51
43
52
def handle_call({:move_shard, shard_key, to_servers}, from, state) do
44
-
{result, %State{} = state} = begin_move_shard(shard_key, to_servers, state)
53
+
{result, %State{} = state} = begin_move_shard(from, shard_key, to_servers, state)
45
54
case result do
46
55
:ok ->
47
-
{:noreply, %State{state | shard_moves_from: Map.put(state.shard_moves_from, shard_key, from)}}
56
+
{:noreply, state}
48
57
{:error, _} = error ->
49
58
{:reply, error, state}
50
59
end
51
60
end
52
61
53
-
def handle_info(:check_imports, %State{} = state) do
54
-
state = check_imports(state)
55
-
SimServer.send_after(self(), :check_imports, @check_imports_interval_ms)
62
+
def handle_info(:tick_shard_moves, %State{} = state) do
63
+
state = tick_shard_moves(state)
64
+
SimServer.send_after(self(), :tick_shard_moves, @tick_shard_moves_interval_ms)
56
65
{:noreply, state}
57
66
end
58
67
59
-
defp begin_move_shard(shard_key, to_servers, %State{} = state) when is_list(to_servers) do
68
+
defp tick_shard_moves(%State{} = state) do
69
+
shard_moves =
70
+
state.shard_moves
71
+
|> Enum.map(fn %ShardMove{} = move ->
72
+
case shard_move_complete?(move, state) do
73
+
true ->
74
+
:ok = complete_shard_move(move, state)
75
+
nil
76
+
77
+
false -> move
78
+
end
79
+
end)
80
+
|> Enum.filter(&(&1 != nil))
81
+
82
+
%State{state | shard_moves: shard_moves}
83
+
end
84
+
85
+
defp shard_move_complete?(%ShardMove{shard_key: shard_key} = move, %State{cluster: cluster}) do
86
+
move.to_servers
87
+
|> Enum.map(fn id ->
88
+
%Server{pid: storage_pid} = Map.fetch!(cluster.servers, id)
89
+
Storage.check_import_complete_send(storage_pid, shard_key)
90
+
end)
91
+
|> Enum.map(&Storage.check_import_complete_receive/1)
92
+
|> Enum.all?(fn {:ok, completed?} -> completed? end)
93
+
end
94
+
95
+
defp begin_move_shard(initiated_from, shard_key, to_servers, %State{} = state) when is_binary(shard_key) and is_list(to_servers) do
60
96
txn = Transaction.new(state.cluster)
61
97
{value, txn} = Transaction.read!(txn, meta("key_servers/" <> shard_key))
62
98
···
69
105
{:ok, _txn} =
70
106
txn
71
107
|> Transaction.write(meta("key_servers/" <> shard_key), new_value)
72
-
|> Transaction.write(meta("shard_moves/" <> shard_key), MetaStore.encode_server_ids(to_servers))
73
108
|> Transaction.commit()
74
109
110
+
shard_move = %ShardMove{
111
+
initiated_from: initiated_from,
112
+
shard_key: shard_key,
113
+
to_servers: to_servers
114
+
}
115
+
state = %State{state | shard_moves: [shard_move | state.shard_moves]}
116
+
75
117
{:ok, state}
76
118
77
119
true ->
···
83
125
end
84
126
end
85
127
86
-
defp check_imports(%State{} = state) do
87
-
{:ok, {results, _txn}} =
88
-
Transaction.new(state.cluster)
89
-
|> Transaction.read_range("\xFF/shard_moves/", "\xFF/shard_moves/\xFF\xFF")
90
-
91
-
results
92
-
# TODO: remove filter once we can clear the completed move keys
93
-
|> Enum.filter(fn {_k, v} -> v != "" end)
94
-
|> Enum.reduce(state, fn {"\xFF/shard_moves/" <> start_key, servers_encoded}, state ->
95
-
servers_encoded
96
-
|> MetaStore.decode_server_ids()
97
-
|> Enum.map(fn id ->
98
-
%Server{pid: storage_pid} = Map.fetch!(state.cluster.servers, id)
99
-
Storage.check_import_complete_send(storage_pid, start_key)
100
-
end)
101
-
|> Enum.map(&Storage.check_import_complete_receive(&1))
102
-
|> then(fn results ->
103
-
case Enum.all?(results, fn {:ok, completed?} -> completed? end) do
104
-
true -> complete_shard_move(start_key, state)
105
-
false -> state
106
-
end
107
-
end)
108
-
end)
109
-
end
110
-
111
-
defp complete_shard_move(shard_key, %State{} = state) do
128
+
defp complete_shard_move(%ShardMove{shard_key: shard_key} = move, %State{} = state) do
112
129
txn = Transaction.new(state.cluster)
113
130
114
-
sm_key = meta("shard_moves/") <> shard_key
115
131
ks_key = meta("key_servers/") <> shard_key
116
-
117
-
{:ok, {%{^sm_key => _sm_value, ^ks_key => ks_value}, txn}} = Transaction.read(txn, [sm_key, ks_key])
132
+
{:ok, {ks_value, txn}} = Transaction.read(txn, ks_key)
118
133
119
134
{from_servers, to_servers} = MetaStore.decode_key_servers(ks_value)
120
135
if length(to_servers) == 0, do: raise "Attempted to move a shard to empty list of servers"
121
136
122
-
# This is NOT transactionally safe if there are multiple distributors
123
-
# because we are relying on the keys having not been modified since we scanned
124
-
# the shard_moves space and checked the storage servers' import statuses
125
-
#
126
137
# TODO: take a lock to prevent concurrent modification in the case of multiple
127
138
# distributors (after recovery is implemented)
128
139
{:ok, _txn} =
129
140
txn
130
-
# TODO: clear sm_key
131
-
|> Transaction.write(sm_key, "")
132
141
|> Transaction.write(ks_key, MetaStore.encode_key_servers(to_servers))
133
142
|> Transaction.commit()
134
143
135
-
{reply_to, shard_moves_from} = Map.pop!(state.shard_moves_from, shard_key)
136
-
SimServer.reply(reply_to, {:ok, {from_servers, to_servers}})
144
+
if move.initiated_from do
145
+
SimServer.reply(move.initiated_from, {:ok, {from_servers, to_servers}})
146
+
end
137
147
138
-
%State{state | shard_moves_from: shard_moves_from}
148
+
:ok
139
149
end
140
150
end