this repo has no description
Add shard info map to Distributor
+214
-57
+8
lib/dense_shard_map.ex
+8
lib/dense_shard_map.ex
···
37
37
:ok
38
38
end
39
39
40
+
@spec get(t, binary) :: term | nil
41
+
def get(table, start_key) when is_binary(start_key) do
42
+
case :ets.lookup(table, start_key) do
43
+
[{^start_key, value}] -> value
44
+
[] -> nil
45
+
end
46
+
end
47
+
40
48
@spec shard_for_key(t, binary) :: {binary, binary, term}
41
49
def shard_for_key(table, key) when is_database_key(key) do
42
50
# Raises if table is empty, which is fine
+57
-17
lib/servers/distributor.ex
+57
-17
lib/servers/distributor.ex
···
3
3
4
4
import ExUnit.Assertions, only: [assert: 1]
5
5
6
-
alias Hobbes.{Transaction, MetaStore}
6
+
alias Hobbes.{Transaction, MetaStore, ShardInfoMap}
7
7
alias Hobbes.Structs.{Cluster, Server, ShardStats}
8
8
alias Hobbes.Servers.Storage
9
9
alias Hobbes.Transaction.TxnState
···
38
38
id: non_neg_integer,
39
39
cluster: Cluster.t,
40
40
storage_servers: %{non_neg_integer => StorageInfo.t},
41
+
shard_map: ShardInfoMap.t,
41
42
shard_moves: [ShardMove.t],
42
43
}
43
44
@enforce_keys [
44
45
:id,
45
46
:cluster,
46
47
:storage_servers,
48
+
:shard_map,
47
49
:shard_moves,
48
50
]
49
51
defstruct @enforce_keys
···
60
62
SimServer.call(server, {:move_shard, shard_key, to_servers}, 300_000)
61
63
end
62
64
63
-
@spec storage_ping(pid, non_neg_integer) :: :ok
64
-
def storage_ping(server, storage_id) when is_integer(storage_id) do
65
-
SimServer.cast(server, {:storage_ping, self(), storage_id})
65
+
@spec storage_ping(pid, non_neg_integer, list) :: :ok
66
+
def storage_ping(server, storage_id, shard_stats) when is_integer(storage_id) when is_list(shard_stats) do
67
+
SimServer.cast(server, {:storage_ping, self(), storage_id, shard_stats})
66
68
end
67
69
68
70
def init(%{id: id, cluster: %Cluster{} = cluster}) do
···
70
72
id: id,
71
73
cluster: cluster,
72
74
storage_servers: %{},
75
+
shard_map: ShardInfoMap.new(),
73
76
shard_moves: [],
74
77
}
75
78
76
-
SimServer.send_after(self(), :tick_scan, @tick_scan_interval_ms)
77
-
SimServer.send_after(self(), :tick_shard_moves, @tick_shard_moves_interval_ms)
78
-
79
+
SimServer.send_after self(), :tick_setup, 0
79
80
{:ok, state}
80
81
end
81
82
···
87
88
end
88
89
end
89
90
90
-
def handle_cast({:storage_ping, storage_pid, storage_id}, %State{} = state) do
91
+
def handle_cast({:storage_ping, storage_pid, storage_id, shard_stats}, %State{} = state) do
91
92
prev_info = Map.get(state.storage_servers, storage_id)
92
93
93
94
info = %StorageInfo{id: storage_id, pid: storage_pid, last_ping_timestamp: SimServer.current_time()}
94
95
state = %{state | storage_servers: Map.put(state.storage_servers, storage_id, info)}
96
+
97
+
:ok = update_shard_stats(state, shard_stats)
95
98
96
99
if prev_info == nil or (prev_info.id != storage_id) do
97
-
# Send updated storage server pid map to CommitBuffers
98
-
# Note: in theory this message could be lost, and we do not currently re-send
99
-
# This is not a correctness issue, but reads would be blocked
100
-
server_map = state.storage_servers |> Map.values() |> Map.new(&{&1.id, &1.pid})
101
-
102
-
get_servers(state.cluster, Hobbes.Servers.CommitBuffer)
103
-
|> Enum.each(fn %Server{pid: buf_pid} ->
104
-
SimServer.send buf_pid, {:update_storage_servers, server_map}
105
-
end)
100
+
broadcast_storage_map(state)
106
101
end
107
102
108
103
{:noreply, state}
109
104
end
110
105
106
+
def handle_info(:tick_setup, %State{} = state) do
107
+
case setup(state) do
108
+
{:ok, state} ->
109
+
SimServer.send_after(self(), :tick_scan, @tick_scan_interval_ms)
110
+
SimServer.send_after(self(), :tick_shard_moves, @tick_shard_moves_interval_ms)
111
+
112
+
{:noreply, state}
113
+
114
+
:error ->
115
+
SimServer.send_after self(), :tick_setup, 100
116
+
{:noreply, state}
117
+
end
118
+
end
119
+
111
120
def handle_info(:tick_scan, %State{} = state) do
112
121
state = scan_shards(state)
113
122
SimServer.send_after(self(), :tick_scan, @tick_scan_interval_ms)
···
126
135
cluster.generation == state.cluster.generation -> {:noreply, %{state | cluster: cluster}}
127
136
cluster.generation > state.cluster.generation -> exit(:shutdown)
128
137
end
138
+
end
139
+
140
+
defp setup(%State{} = state) do
141
+
with {:ok, txn} <- Transaction.new(state.cluster),
142
+
{:ok, {key_servers_pairs, txn}} <- Transaction.read_range(txn, key_servers_prefix(), key_servers_end()),
143
+
{:ok, _txn} = Transaction.commit(txn)
144
+
do
145
+
ShardInfoMap.load(state.shard_map, key_servers_pairs)
146
+
{:ok, state}
147
+
else
148
+
_ -> :error
149
+
end
150
+
end
151
+
152
+
defp update_shard_stats(%State{shard_map: shard_map}, shard_stats) when is_list(shard_stats) do
153
+
Enum.each(shard_stats, fn stats ->
154
+
ShardInfoMap.update_shard_stats(shard_map, stats)
155
+
end)
156
+
:ok
157
+
end
158
+
159
+
defp broadcast_storage_map(%State{} = state) do
160
+
# Send updated storage server pid map to CommitBuffers
161
+
# Note: in theory this message could be lost, and we do not currently re-send
162
+
# This is not a correctness issue, but reads would be blocked
163
+
server_map = state.storage_servers |> Map.values() |> Map.new(&{&1.id, &1.pid})
164
+
165
+
get_servers(state.cluster, Hobbes.Servers.CommitBuffer)
166
+
|> Enum.each(fn %Server{pid: buf_pid} ->
167
+
SimServer.send buf_pid, {:update_storage_servers, server_map}
168
+
end)
129
169
end
130
170
131
171
defp scan_shards(%State{} = state) when state.cluster.status != :normal do
+9
-2
lib/servers/sequencer.ex
+9
-2
lib/servers/sequencer.ex
···
28
28
@spec get_commit_version(pid) :: {integer, integer}
29
29
def get_commit_version(server), do: SimServer.call(server, :get_commit_version)
30
30
31
-
@spec get_read_version(pid) :: integer
32
-
def get_read_version(server), do: SimServer.call(server, :get_read_version)
31
+
@spec get_read_version(pid) :: {:ok, integer} | {:error, :timeout}
32
+
def get_read_version(server) do
33
+
try do
34
+
version = SimServer.call(server, :get_read_version)
35
+
{:ok, version}
36
+
catch
37
+
:exit, {:timeout, _mfa} -> {:error, :timeout}
38
+
end
39
+
end
33
40
34
41
@spec notify_committed(pid, integer) :: :ok
35
42
def notify_committed(server, commit_version) when is_integer(commit_version) do
+41
-29
lib/servers/storage.ex
+41
-29
lib/servers/storage.ex
···
214
214
end
215
215
216
216
def handle_call({:get_shard_stats, start_key, end_key}, _from, %State{} = state) do
217
-
pairs = ByteSample.scan(state.byte_sample, start_key, end_key)
218
-
219
-
size =
220
-
pairs
221
-
|> Enum.reduce(0, fn {_k, bytes}, acc -> acc + bytes end)
222
-
|> round()
223
-
224
-
half_size = div(size, 2)
225
-
# TODO: we use the full midpoint key, but we could instead use the shortest key which
226
-
# separates the midpoint and the next key (which would make the shard map smaller)
227
-
midpoint =
228
-
Enum.reduce_while(pairs, 0, fn {k, bytes}, acc ->
229
-
acc = acc + bytes
230
-
case acc > half_size do
231
-
true -> {:halt, k}
232
-
false -> {:cont, acc}
233
-
end
234
-
end)
235
-
|> case do
236
-
midpoint when is_binary(midpoint) -> midpoint
237
-
# If the shard is too small to have any byte sample keys
238
-
0 -> start_key
239
-
end
240
-
241
-
stats = %ShardStats{size_bytes: size, midpoint_key: midpoint}
217
+
stats = compute_shard_stats(state, start_key, end_key)
242
218
{:reply, {:ok, stats}, state}
243
219
end
244
220
···
360
336
# If calls we make time out, we may receive their responses later
361
337
def handle_info(_message, state), do: {:noreply, state}
362
338
363
-
def ping_distributor(%State{} = state) do
339
+
defp ping_distributor(%State{} = state) do
364
340
case get_servers(state.cluster, Hobbes.Servers.Distributor) do
365
-
[%Server{pid: distributor_pid}] ->
366
-
Distributor.storage_ping(distributor_pid, state.id)
367
-
341
+
[%Server{pid: distributor_pid}] -> do_ping_distributor(state, distributor_pid)
368
342
[] -> :noop
369
343
end
370
344
371
345
state
346
+
end
347
+
348
+
defp do_ping_distributor(%State{} = state, distributor_pid) do
349
+
# TODO: this is very inefficient, we need to continuously update shard sizes in-memory
350
+
# TODO: only send oversize/undersize shards?
351
+
shard_stats =
352
+
SparseShardMap.list_shards(state.shard_map)
353
+
|> Enum.map(fn {sk, ek} -> compute_shard_stats(state, sk, ek) end)
354
+
355
+
Distributor.storage_ping(distributor_pid, state.id, shard_stats)
372
356
end
373
357
374
358
defp flush(%State{kv: kv} = state) do
···
691
675
shard_import = %{shard_import | nonce: nil, status: :complete, completed_version: shard_import.current_read_version}
692
676
put_in(state.imports[shard_import.id], shard_import)
693
677
end
678
+
end
679
+
680
+
defp compute_shard_stats(%State{} = state, start_key, end_key) when is_binary(start_key) and is_binary(end_key) do
681
+
pairs = ByteSample.scan(state.byte_sample, start_key, end_key)
682
+
683
+
size =
684
+
pairs
685
+
|> Enum.reduce(0, fn {_k, bytes}, acc -> acc + bytes end)
686
+
|> round()
687
+
688
+
half_size = div(size, 2)
689
+
# TODO: we use the full midpoint key, but we could instead use the shortest key which
690
+
# separates the midpoint and the next key (which would make the shard map smaller)
691
+
midpoint =
692
+
Enum.reduce_while(pairs, 0, fn {k, bytes}, acc ->
693
+
acc = acc + bytes
694
+
case acc > half_size do
695
+
true -> {:halt, k}
696
+
false -> {:cont, acc}
697
+
end
698
+
end)
699
+
|> case do
700
+
midpoint when is_binary(midpoint) -> midpoint
701
+
# If the shard is too small to have any byte sample keys
702
+
0 -> start_key
703
+
end
704
+
705
+
%ShardStats{start_key: start_key, end_key: end_key, size_bytes: size, midpoint_key: midpoint}
694
706
end
695
707
696
708
defp check_up_to_date(%State{} = state, version) do
+79
lib/shard_info_map.ex
+79
lib/shard_info_map.ex
···
1
+
defmodule Hobbes.ShardInfoMap do
2
+
alias Hobbes.{DenseShardMap, ShardInfoMap}
3
+
alias Hobbes.Structs.ShardStats
4
+
alias Hobbes.MetaStore
5
+
6
+
import Hobbes.Utils
7
+
8
+
defmodule Shard do
9
+
@type t :: %__MODULE__{
10
+
start_key: binary,
11
+
end_key: binary,
12
+
from_server_ids: [integer],
13
+
to_server_ids: [integer],
14
+
stats: ShardStats.t,
15
+
}
16
+
@enforce_keys [
17
+
:start_key,
18
+
:end_key,
19
+
:from_server_ids,
20
+
:to_server_ids,
21
+
:stats,
22
+
]
23
+
defstruct @enforce_keys
24
+
end
25
+
26
+
@type t :: %__MODULE__{
27
+
shard_map: DenseShardMap.t,
28
+
}
29
+
@enforce_keys [:shard_map]
30
+
defstruct @enforce_keys
31
+
32
+
def new do
33
+
%ShardInfoMap{
34
+
shard_map: DenseShardMap.new(),
35
+
}
36
+
end
37
+
38
+
def load(%ShardInfoMap{shard_map: dsm}, key_servers_pairs) when is_list(key_servers_pairs) do
39
+
key_servers_pairs ++ [{key_servers_prefix() <> "\xFF\xFF", ""}]
40
+
|> Enum.chunk_every(2, 1, :discard)
41
+
|> Enum.each(fn [pair, next] ->
42
+
%Shard{} = info = shard_from_ks_pairs(pair, next)
43
+
DenseShardMap.put(dsm, info.start_key, info)
44
+
end)
45
+
46
+
:ok
47
+
end
48
+
49
+
def update_shard_stats(%ShardInfoMap{shard_map: dsm}, %ShardStats{} = stats) do
50
+
sk = stats.start_key
51
+
ek = stats.end_key
52
+
53
+
case DenseShardMap.get(dsm, sk) do
54
+
%Shard{start_key: ^sk, end_key: ^ek} = existing ->
55
+
DenseShardMap.put(dsm, sk, %{existing | stats: stats})
56
+
:ok
57
+
58
+
_ -> :error
59
+
end
60
+
end
61
+
62
+
defp shard_from_ks_pairs({key_servers_prefix() <> start_key, value}, {key_servers_prefix() <> end_key, _value}) do
63
+
{from, to} = MetaStore.decode_key_servers(value)
64
+
to = to || []
65
+
66
+
%Shard{
67
+
start_key: start_key,
68
+
end_key: end_key,
69
+
from_server_ids: from,
70
+
to_server_ids: to,
71
+
stats: nil,
72
+
}
73
+
end
74
+
75
+
@doc false
76
+
def dump(%ShardInfoMap{} = sim) do
77
+
DenseShardMap.dump(sim.shard_map)
78
+
end
79
+
end
+5
lib/sparse_shard_map.ex
+5
lib/sparse_shard_map.ex
+3
-1
lib/structs.ex
+3
-1
lib/structs.ex
···
133
133
134
134
defmodule ShardStats do
135
135
@type t :: %__MODULE__{
136
+
start_key: binary,
137
+
end_key: binary,
136
138
size_bytes: non_neg_integer,
137
139
midpoint_key: binary,
138
140
}
139
-
@enforce_keys [:size_bytes, :midpoint_key]
141
+
@enforce_keys [:start_key, :end_key, :size_bytes, :midpoint_key]
140
142
defstruct @enforce_keys
141
143
end
142
144
end
+12
-8
lib/transaction.ex
+12
-8
lib/transaction.ex
···
29
29
else
30
30
case get_read_version(cluster) do
31
31
{:ok, read_version} -> {:ok, %TxnState{cluster: cluster, read_version: read_version}}
32
-
{:error, :timeout} -> {:error, :timeout}
32
+
{:error, _err} = error -> error
33
33
end
34
34
end
35
35
end
···
41
41
end
42
42
43
43
defp get_read_version(%Cluster{} = cluster) do
44
-
[%Server{pid: pid}] = get_servers(cluster, Hobbes.Servers.Sequencer)
45
-
try do
46
-
rv = Sequencer.get_read_version(pid)
47
-
{:ok, rv}
48
-
catch
49
-
:exit, {:timeout, _} ->
50
-
{:error, :timeout}
44
+
with {:ok, pid} <- get_sequencer_pid(cluster),
45
+
{:ok, _read_version} = result <- Sequencer.get_read_version(pid)
46
+
do
47
+
result
48
+
end
49
+
end
50
+
51
+
defp get_sequencer_pid(%Cluster{} = cluster) do
52
+
case get_servers(cluster, Hobbes.Servers.Sequencer) do
53
+
[%Server{pid: pid}] -> {:ok, pid}
54
+
[] -> {:error, :no_sequencer}
51
55
end
52
56
end
53
57