this repo has no description
Refactor Distributor scan and shard splits to use ShardInfoMap
+63
-81
+5
lib/dense_shard_map.ex
+5
lib/dense_shard_map.ex
+43
-80
lib/servers/distributor.ex
+43
-80
lib/servers/distributor.ex
···
4
4
import ExUnit.Assertions, only: [assert: 1]
5
5
6
6
alias Hobbes.{Transaction, MetaStore, ShardInfoMap}
7
+
alias Hobbes.ShardInfoMap.Shard
7
8
alias Hobbes.Structs.{Cluster, Server, ShardStats}
8
9
alias Hobbes.Servers.Storage
9
10
alias Hobbes.Transaction.TxnState
···
175
176
defp scan_shards(%State{} = state) do
176
177
assert state.cluster.status == :normal
177
178
178
-
with {:ok, txn} <- Transaction.new(state.cluster),
179
-
{:ok, {pairs, _txn}} <- Transaction.read_range(txn, key_servers_prefix(), key_servers_end())
180
-
do
181
-
do_scan_shards(state, pairs)
182
-
else
183
-
{:error, _error} -> state
184
-
end
185
-
end
179
+
shards = ShardInfoMap.list_shards(state.shard_map)
186
180
187
-
defp do_scan_shards(%State{} = state, pairs) when is_list(pairs) do
188
-
shard_sizes =
189
-
pairs
190
-
|> Enum.map(fn {"\xFF/key_servers/" <> shard_key, servers_encoded} ->
191
-
{from, to} = MetaStore.decode_key_servers(servers_encoded)
192
-
{shard_key, from, to}
193
-
end)
194
-
|> then(fn shards ->
195
-
shards ++ [{all_keys_end(), nil, nil}]
196
-
end)
197
-
|> Enum.chunk_every(2, 1, :discard)
198
-
|> Enum.map(fn [{start_key, from, _to}, {end_key, _, _}] ->
199
-
from
200
-
|> Enum.map(fn id ->
201
-
case Map.fetch(state.storage_servers, id) do
202
-
{:ok, %StorageInfo{pid: pid}} -> pid
203
-
:error -> nil
204
-
end
205
-
end)
206
-
|> Enum.reject(&is_nil/1)
207
-
|> case do
208
-
[_ | _] = from_pids ->
209
-
storage_pid = SimServer.deterministic_random(from_pids)
181
+
{shards_to_split, _shards} = Enum.split_with(shards, &should_split_shard?/1)
210
182
211
-
{:ok, %ShardStats{} = stats} = too_new_backoff(fn ->
212
-
Storage.get_shard_stats(storage_pid, start_key, end_key)
213
-
end)
214
-
{{start_key, end_key}, stats}
215
-
216
-
[] -> nil
217
-
end
218
-
end)
219
-
|> Enum.reject(&is_nil/1)
220
-
221
-
oversize_shards = Enum.filter(shard_sizes, fn {_shard, %ShardStats{} = stats} ->
222
-
stats.size_bytes > @shard_max_size_bytes
223
-
end)
224
-
225
-
Enum.each(oversize_shards, fn {{start_key, end_key}, %ShardStats{} = stats} ->
226
-
split_shard(start_key, end_key, stats.midpoint_key, state)
183
+
Enum.each(shards_to_split, fn shard ->
184
+
split_shard(state, shard)
227
185
end)
228
186
229
187
state
230
188
end
231
189
232
-
defp split_shard(shard_key, end_key, at_key, %State{} = state) when is_binary(shard_key) and is_binary(at_key) do
233
-
# TODO: we will have to take a lock for this
234
-
{:ok, txn} = Transaction.new(state.cluster)
235
-
{:ok, {shard_value, txn}} = Transaction.read(txn, "\xFF/key_servers/" <> shard_key)
190
+
defp should_split_shard?(%Shard{stats: nil}), do: false
191
+
defp should_split_shard?(%Shard{to_server_ids: to}) when to != [], do: false
192
+
defp should_split_shard?(%Shard{} = shard), do: shard.stats.size_bytes > @shard_max_size_bytes
236
193
237
-
case MetaStore.decode_key_servers(shard_value) do
238
-
{[_ | _] = from, nil} ->
239
-
pairs =
240
-
Enum.map(from, fn id ->
241
-
[
242
-
{server_keys_prefix() <> Integer.to_string(id) <> "/" <> shard_key, "complete/" <> at_key},
243
-
{server_keys_prefix() <> Integer.to_string(id) <> "/" <> at_key, "complete/" <> end_key},
244
-
]
245
-
end)
246
-
|> Enum.concat()
194
+
defp split_shard(%State{} = state, %Shard{} = shard) do
195
+
assert shard.to_server_ids == []
196
+
%ShardStats{midpoint_key: midpoint} = shard.stats
247
197
248
-
split_key = "\xFF/key_servers/" <> at_key
249
-
split_value = MetaStore.encode_key_servers(from, nil)
250
-
pairs = [{split_key, split_value}] ++ pairs
198
+
new_shard = %Shard{
199
+
start_key: midpoint,
200
+
end_key: shard.end_key,
201
+
from_server_ids: shard.from_server_ids,
202
+
to_server_ids: [],
203
+
stats: nil,
204
+
}
205
+
shard = %{shard | end_key: midpoint, stats: nil}
251
206
252
-
case Transaction.read(txn, split_key) do
253
-
{:ok, {nil, txn}} ->
254
-
txn
255
-
|> Transaction.write(pairs)
256
-
|> Transaction.commit()
257
-
|> case do
258
-
{:ok, _txn} -> :ok
259
-
{:error, _} -> {:error, :commit_error}
260
-
end
261
-
262
-
{:ok, {existing, _txn}} when is_binary(existing) ->
263
-
{:error, :split_shard_already_exists}
264
-
end
265
-
266
-
{[_ | _], [_ | _]} ->
267
-
{:error, :shard_moving}
207
+
# TODO: open a transaction with no read version and only handle commit error
208
+
with {:ok, txn} <- Transaction.new(state.cluster),
209
+
txn = Transaction.write(txn, [to_ks_pair(new_shard)]),
210
+
{:ok, _txn} <- Transaction.commit(txn)
211
+
do
212
+
ShardInfoMap.put(state.shard_map, shard.start_key, shard)
213
+
ShardInfoMap.put(state.shard_map, new_shard.start_key, new_shard)
214
+
:ok
215
+
else
216
+
# TODO: we only need to exit for errors which are not retryable (e.g. commit timed out)
217
+
# This is necessary to maintain consistency of the in-memory shard map
218
+
_ -> exit(:shutdown)
268
219
end
269
220
end
270
221
···
399
350
400
351
defp server_keys_key(server_id, shard_key) when is_integer(server_id) and is_binary(shard_key) do
401
352
server_keys_prefix() <> Integer.to_string(server_id) <> "/" <> shard_key
353
+
end
354
+
355
+
defp to_ks_pair(%Shard{} = shard) do
356
+
to_server_ids = case shard.to_server_ids do
357
+
[] -> nil
358
+
list -> list
359
+
end
360
+
361
+
key = key_servers_prefix() <> shard.start_key
362
+
value = MetaStore.encode_key_servers(shard.from_server_ids, to_server_ids)
363
+
364
+
{key, value}
402
365
end
403
366
end
+15
-1
lib/shard_info_map.ex
+15
-1
lib/shard_info_map.ex
···
11
11
end_key: binary,
12
12
from_server_ids: [integer],
13
13
to_server_ids: [integer],
14
-
stats: ShardStats.t,
14
+
stats: ShardStats.t | nil,
15
15
}
16
16
@enforce_keys [
17
17
:start_key,
···
29
29
@enforce_keys [:shard_map]
30
30
defstruct @enforce_keys
31
31
32
+
@spec new :: t
32
33
def new do
33
34
%ShardInfoMap{
34
35
shard_map: DenseShardMap.new(),
35
36
}
36
37
end
37
38
39
+
def put(%ShardInfoMap{shard_map: dsm}, start_key, %Shard{} = shard) do
40
+
DenseShardMap.put(dsm, start_key, shard)
41
+
end
42
+
43
+
@spec list_shards(t) :: [Shard.t]
44
+
def list_shards(%ShardInfoMap{shard_map: dsm} = _sim) do
45
+
DenseShardMap.list_shards(dsm)
46
+
|> Enum.map(fn {_sk, %Shard{} = shard} -> shard end)
47
+
end
48
+
49
+
@spec load(t, [{binary, binary}]) :: :ok
38
50
def load(%ShardInfoMap{shard_map: dsm}, key_servers_pairs) when is_list(key_servers_pairs) do
39
51
key_servers_pairs ++ [{key_servers_prefix() <> "\xFF\xFF", ""}]
40
52
|> Enum.chunk_every(2, 1, :discard)
···
46
58
:ok
47
59
end
48
60
61
+
@spec update_shard_stats(t, ShardStats.t) :: :ok | :error
49
62
def update_shard_stats(%ShardInfoMap{shard_map: dsm}, %ShardStats{} = stats) do
50
63
sk = stats.start_key
51
64
ek = stats.end_key
···
59
72
end
60
73
end
61
74
75
+
@spec shard_from_ks_pairs({binary, binary}, {binary, binary}) :: Shard.t
62
76
defp shard_from_ks_pairs({key_servers_prefix() <> start_key, value}, {key_servers_prefix() <> end_key, _value}) do
63
77
{from, to} = MetaStore.decode_key_servers(value)
64
78
to = to || []