this repo has no description
Convert Storage server to XKS
+82
-104
+74
-95
lib/servers/storage.ex
+74
-95
lib/servers/storage.ex
···
5
5
6
6
import ExUnit.Assertions, only: [assert: 1]
7
7
8
-
alias Hobbes.{HybridKV, SparseShardMap}
8
+
alias Hobbes.SparseShardMap
9
9
alias Hobbes.KV.{MutationLog, ByteSample}
10
10
alias Hobbes.Structs.{Cluster, TLogGeneration, Server, PeekResult, RangeResult, StorageStats, ShardStats}
11
11
alias Hobbes.Servers.{CommitBuffer, TLog, Storage, Distributor}
12
12
alias Hobbes.Encoding.Keyset
13
+
14
+
alias Hobbes.XKS
13
15
14
16
alias Hobbes.Utils
15
17
import Hobbes.Utils
···
50
52
51
53
peek_nonce: reference,
52
54
55
+
xks: XKS.t,
53
56
kv: term,
54
57
shard_map: SparseShardMap.t,
55
58
byte_sample: ByteSample.t,
···
68
71
:peek_nonce,
69
72
:peek_tlog_i,
70
73
74
+
:xks,
71
75
:kv,
72
76
:shard_map,
73
77
:byte_sample,
···
141
145
end
142
146
end
143
147
144
-
def init(%{cluster: %Cluster{} = cluster, path: path, id: id, storage_team_id: storage_team_id}) do
148
+
@state_partition 0
149
+
@special_partiton -1
150
+
# TODO: partition per shard
151
+
@data_partition 1
152
+
defmacro durable_version_key, do: "durable_version"
153
+
154
+
def init(%{cluster: %Cluster{} = cluster, path: _path, id: id, storage_team_id: storage_team_id}) do
145
155
# Init new storage server
146
156
assert is_integer(id)
147
-
path = case false && Sim.simulated?() do
148
-
true -> path
149
-
false -> nil
150
-
end
151
-
kv = HybridKV.new(path: path)
157
+
xks = XKS.new()
158
+
159
+
# TODO: init state_partition (currently initialized by new())
160
+
XKS.init_partition(xks, @special_partiton)
161
+
XKS.init_partition(xks, @data_partition)
152
162
153
-
# Write id to storage and commit
154
-
HybridKV.put_storage(kv, special_id_key(), Keyset.pack([id]))
155
-
HybridKV.put_storage(kv, special_storage_team_id_key(), Keyset.pack([storage_team_id]))
156
-
HybridKV.commit(kv)
163
+
XKS.apply_batch(xks, @state_partition, 1, [
164
+
{:write, special_id_key(), Keyset.pack([id])},
165
+
{:write, special_storage_team_id_key(), Keyset.pack([storage_team_id])},
166
+
{:write, durable_version_key(), Keyset.pack([0])},
167
+
])
168
+
# TODO: force compaction here
169
+
XKS.commit(xks)
157
170
158
-
do_init(cluster, kv)
171
+
do_init(cluster, xks)
159
172
end
160
173
161
-
def init(%{cluster: %Cluster{} = cluster, path: path}) do
174
+
def init(%{cluster: %Cluster{} = _cluster, path: _path}) do
162
175
# Init existing storage server from storage KV
176
+
raise "disabled"
163
177
assert Sim.simulated?()
164
-
kv = HybridKV.new(path: path)
165
178
166
-
do_init(cluster, kv)
179
+
#do_init(cluster, kv)
167
180
end
168
181
169
-
def do_init(%Cluster{} = cluster, %HybridKV{} = kv) do
170
-
[id] = HybridKV.get(kv, 1, special_id_key()) |> Keyset.unpack()
171
-
[storage_team_id] = HybridKV.get(kv, 1, special_storage_team_id_key()) |> Keyset.unpack()
182
+
def do_init(%Cluster{} = cluster, %XKS{} = xks) do
183
+
# TODO: use XKS version
184
+
load_read_version = 1
185
+
186
+
[id] = XKS.Reader.get(xks, @state_partition, load_read_version, special_id_key()) |> Keyset.unpack()
187
+
[storage_team_id] = XKS.Reader.get(xks, @state_partition, load_read_version, special_storage_team_id_key()) |> Keyset.unpack()
188
+
[durable_version] = XKS.Reader.get(xks, @state_partition, load_read_version, durable_version_key()) |> Keyset.unpack()
172
189
assert is_integer(id)
173
190
assert is_integer(storage_team_id)
191
+
assert is_integer(durable_version)
174
192
175
193
# We are reading from storage here no matter what so the read at version=1 is fine
176
-
startup_version = String.to_integer(HybridKV.get(kv, 1, special_prefix() <> "durable_version") || "0")
194
+
startup_version = durable_version
177
195
178
196
state = %State{
179
197
id: id,
···
188
206
# TODO: would be good to randomize this but we don't know the replication factor here
189
207
peek_tlog_i: 0,
190
208
191
-
kv: kv,
209
+
xks: xks,
210
+
kv: nil,
192
211
shard_map: SparseShardMap.new(),
193
212
byte_sample: ByteSample.new(),
194
213
···
197
216
198
217
# Load the shard map and imports from the special keyspace
199
218
shard_map = state.shard_map
200
-
%RangeResult{pairs: sk_pairs} = HybridKV.scan(kv, 1, special_storage_keys_prefix(), special_storage_keys_end())
219
+
sk_pairs = XKS.Reader.scan(xks, @special_partiton, load_read_version, special_storage_keys_prefix(), special_storage_keys_end())
201
220
202
221
state =
203
222
Enum.reduce(sk_pairs, state, fn {k, v}, state ->
···
211
230
SparseShardMap.put(shard_map, start_key, end_key)
212
231
state
213
232
214
-
[shard_move_id, end_key, "fetching"] ->
215
-
# Load the import's status from the special_shard_import keyspace
216
-
case HybridKV.get(kv, 1, special_shard_import_prefix() <> Keyset.pack([shard_move_id])) do
217
-
nil ->
218
-
# This shard was not fully imported, restart the import
219
-
begin_import(state, shard_move_id, start_key, end_key)
220
-
221
-
si_value ->
222
-
# This shard was fully imported at completed_version,
223
-
# add the :complete ShardImport to the state so that we can ack to the Distributor
224
-
[completed_version] = Keyset.unpack(si_value)
225
-
226
-
si = %ShardImport{
227
-
id: shard_move_id,
228
-
nonce: nil,
229
-
completed_version: completed_version,
230
-
start_key: start_key,
231
-
end_key: end_key,
232
-
current_end_key: end_key,
233
-
current_read_version: completed_version,
234
-
status: :complete,
235
-
}
236
-
put_in(state.imports[si.id], si)
237
-
end
233
+
[_shard_move_id, _end_key, "fetching"] ->
234
+
# TODO: initialize shard import
235
+
state
238
236
end
239
237
end)
240
-
241
-
byte_sample_pairs = HybridKV.scan(kv, 1, special_byte_sample_prefix(), special_byte_sample_end()).pairs
242
-
ByteSample.load(state.byte_sample, byte_sample_pairs)
243
238
244
239
SimProcess.send_after(self(), :tick_ping, 0)
245
240
SimProcess.send_after(self(), :flush, @flush_interval_ms)
···
251
246
with :ok <- check_up_to_date(state, read_version),
252
247
:ok <- check_contains_keys(state, keys)
253
248
do
254
-
kv = state.kv
255
-
values = Enum.map(keys, &HybridKV.get(kv, read_version, &1))
249
+
xks = state.xks
250
+
values = Enum.map(keys, &XKS.Reader.get(xks, @data_partition, read_version, &1))
256
251
Sim.simulate_work([1, 2, 3, 10, 30])
257
252
{:reply, {:ok, values}, state}
258
253
else
···
265
260
with :ok <- check_up_to_date(state, read_version),
266
261
:ok <- check_contains_range(state, start_key, end_key)
267
262
do
268
-
%RangeResult{} = result = HybridKV.scan(state.kv, read_version, start_key, end_key, limit: limit, reverse: reverse)
263
+
pairs = XKS.Reader.scan(state.xks, @data_partition, read_version, start_key, end_key, reverse, limit)
264
+
result = %RangeResult{
265
+
pairs: pairs,
266
+
# TODO
267
+
more: false,
268
+
}
269
269
Sim.simulate_work([1, 2, 3, 10, 30])
270
270
{:reply, {:ok, result}, state}
271
271
else
···
294
294
end
295
295
end
296
296
297
+
# TODO: remove probably
297
298
def handle_cast({:read_range_async, from_pid, nonce, read_version, start_key, end_key, limit}, %State{} = state) do
298
299
with :ok <- check_up_to_date(state, read_version),
299
300
:ok <- check_contains_range(state, start_key, end_key)
300
301
do
301
-
%RangeResult{} = result = HybridKV.scan(state.kv, read_version, start_key, end_key, limit: limit, reverse: false)
302
+
reverse = false
303
+
pairs = XKS.Reader.scan(state.xks, @data_partition, read_version, start_key, end_key, reverse, limit)
304
+
result = %RangeResult{
305
+
pairs: pairs,
306
+
# TODO
307
+
more: false,
308
+
}
302
309
Sim.simulate_work([1, 2, 3, 10, 30])
303
310
SimProcess.send from_pid, {:read_range_result, nonce, result}
304
311
else
···
429
436
Distributor.storage_ping(distributor_pid, state.id, storage_stats, shard_stats)
430
437
end
431
438
432
-
defp flush(%State{kv: kv} = state) do
433
-
# We must maintain the invariant that
434
-
# state.durable_version < shard_import.current_read_version
435
-
# for all in-flight shard import reads
436
-
#
437
-
# If this invariant is violated we could clobber newer versions with older reads in storage
438
-
#
439
-
# This could only happen if a read took longer than the mvcc window to complete, which
440
-
# is extremely unlikely in practice because the import read timeout window is considerably
441
-
# shorter than the mvcc window
442
-
min_import_version =
443
-
state.imports
444
-
|> Map.values()
445
-
|> Enum.reject(fn %ShardImport{} = si -> si.status == :complete end)
446
-
|> Enum.map(fn %ShardImport{} = si -> si.current_read_version end)
447
-
|> Enum.min(&<=/2, fn -> state.data_version end)
448
-
449
-
flush_version =
450
-
state.data_version - mvcc_window()
451
-
|> min(min_import_version - 1)
452
-
|> max(0)
453
-
454
-
kv = HybridKV.flush(kv, flush_version)
455
-
:ok = HybridKV.put_storage(kv, special_prefix() <> "durable_version", Integer.to_string(flush_version))
456
-
:ok = HybridKV.commit(kv)
457
-
state = %{state | kv: kv}
458
-
459
-
Sim.simulate_work([1, 10, 20, 30, 100])
439
+
defp flush(%State{} = state) do
440
+
# TODO: compute durable_version from XKS
441
+
durable_version = max(state.data_version - mvcc_window(), 0)
460
442
461
443
# Send pops to tlogs
462
444
# Note: pop is a cast (async) because the result doesn't matter for correctness
463
445
# (if a pop gets lost the next one will cover it)
464
446
tlogs = get_servers(state.cluster, Hobbes.Servers.TLog)
465
447
Enum.each(tlogs, fn %Server{pid: pid} ->
466
-
TLog.pop(pid, state.id, flush_version)
448
+
TLog.pop(pid, state.id, durable_version)
467
449
end)
468
450
469
-
%State{state | durable_version: flush_version}
451
+
%{state | durable_version: durable_version}
470
452
end
471
453
472
454
defp peek_logs(%State{} = state, _rotate_tlog?) when state.cluster.status != :normal do
···
549
531
|> apply_data_mutations(commit_version, mutations)
550
532
end
551
533
552
-
defp apply_data_mutations(%State{kv: kv} = state, version, data_mutations) when is_list(data_mutations) do
553
-
kv = HybridKV.apply_batch(kv, version, data_mutations)
554
-
555
-
byte_sample_mutations = ByteSample.apply_batch(state.byte_sample, data_mutations)
556
-
MutationLog.append(state.kv.mutation_log, version, byte_sample_mutations)
557
-
558
-
%{state | kv: kv}
534
+
defp apply_data_mutations(%State{} = state, version, data_mutations) when is_list(data_mutations) do
535
+
XKS.apply_batch(state.xks, @data_partition, version, data_mutations)
536
+
state
559
537
end
560
538
561
539
defp apply_special_mutations(%State{} = state, version, mutations) do
562
-
kv = state.kv
540
+
xks = state.xks
563
541
564
-
Enum.reduce(mutations, state, fn
542
+
state = Enum.reduce(mutations, state, fn
565
543
{:write, k, v}, state ->
566
-
existing = HybridKV.get(kv, version, k)
567
-
HybridKV.put(kv, version, k, v)
544
+
existing = XKS.Reader.get(xks, @special_partiton, version, k)
568
545
handle_special_diff(k, existing, v, version, state)
569
546
{:clear, k}, state ->
570
-
existing = HybridKV.get(kv, version, k)
571
-
HybridKV.delete(kv, version, k)
547
+
existing = XKS.Reader.get(xks, @special_partiton, version, k)
572
548
handle_special_diff(k, existing, nil, version, state)
573
549
# Note: range clears not supported by special mutations
574
550
end)
551
+
552
+
XKS.apply_batch(xks, @special_partiton, version, mutations)
553
+
state
575
554
end
576
555
577
556
defp handle_special_diff(special_storage_keys_prefix() <> rest, old_value, new_value, version, %State{} = state) do
+7
-8
lib/xks/manifest.ex
+7
-8
lib/xks/manifest.ex
···
9
9
10
10
defmodule MemtableInfo do
11
11
@type t :: %__MODULE__{
12
-
partition: non_neg_integer,
12
+
partition: integer,
13
13
epoch_min: non_neg_integer,
14
14
epoch_max: non_neg_integer | :infinity,
15
15
memtable: Memtable.t,
···
28
28
29
29
defmodule TableInfo do
30
30
@type t :: %__MODULE__{
31
-
partition: non_neg_integer,
31
+
partition: integer,
32
32
level: non_neg_integer,
33
33
# -1 for sentinel
34
34
epoch: non_neg_integer | -1,
···
79
79
@spec insert_memtable(t, MemtableInfo.t) :: :ok
80
80
def insert_memtable(manifest, %MemtableInfo{} = info) do
81
81
assert is_integer(info.partition)
82
-
assert info.partition >= 0
83
82
:ets.insert(manifest, {key_for(info), info})
84
83
:ok
85
84
end
···
115
114
:ok
116
115
end
117
116
118
-
@spec get_memtable_from_min_epoch(t, non_neg_integer, non_neg_integer) :: %MemtableInfo{}
117
+
@spec get_memtable_from_min_epoch(t, integer, non_neg_integer) :: %MemtableInfo{}
119
118
def get_memtable_from_min_epoch(manifest, partition, min_epoch) do
120
119
case :ets.lookup(manifest, {partition, @memtable_level, min_epoch}) do
121
120
[{_key, %MemtableInfo{} = info}] ->
···
127
126
128
127
defp memtable_start_sentinel(partition), do: {partition, @memtable_level, -1}
129
128
130
-
@spec list_memtables(t, non_neg_integer, non_neg_integer) :: [MemtableInfo.t]
129
+
@spec list_memtables(t, integer, non_neg_integer) :: [MemtableInfo.t]
131
130
def list_memtables(manifest, partition, epoch) do
132
131
# Returns memtables in descending order of `epoch_min`,
133
132
# i.e. the most recent memtables are ordered first
···
248
247
%TableInfo{partition: partition, level: level, epoch: -1, first_key: {"", 0}, last_key: {"", 0}, last_block_address: c_null_address()}
249
248
end
250
249
251
-
@spec seek_prev_table(t, non_neg_integer, non_neg_integer, non_neg_integer, binary, non_neg_integer) :: {:ok, TableInfo.t} | :error
250
+
@spec seek_prev_table(t, non_neg_integer, integer, non_neg_integer, binary, non_neg_integer) :: {:ok, TableInfo.t} | :error
252
251
def seek_prev_table(manifest, epoch, partition, level, key, version) do
253
252
# version + 1 because :ets.prev() is exclusive and we need a table with first_key <= seek_key
254
253
# epoch = 0 because we are trying to seek the previous key in the manifest keyspace
···
274
273
end
275
274
end
276
275
277
-
@spec seek_next_table(t, non_neg_integer, non_neg_integer, non_neg_integer, binary, non_neg_integer) :: {:ok, TableInfo.t} | :error
276
+
@spec seek_next_table(t, non_neg_integer, integer, non_neg_integer, binary, non_neg_integer) :: {:ok, TableInfo.t} | :error
278
277
def seek_next_table(manifest, epoch, partition, level, key, version) do
279
278
# We first seek backwards to see if there is a table containing `{key, version}`
280
279
#
···
346
345
end
347
346
end
348
347
349
-
@spec list_overlapping_tables(t, non_neg_integer, non_neg_integer, non_neg_integer, {binary, non_neg_integer}, {binary, non_neg_integer}) :: [TableInfo.t]
348
+
@spec list_overlapping_tables(t, non_neg_integer, integer, non_neg_integer, {binary, non_neg_integer}, {binary, non_neg_integer}) :: [TableInfo.t]
350
349
def list_overlapping_tables(manifest, epoch, partition, level, first_key, last_key) do
351
350
{fk_key, fk_ver} = first_key
352
351
+1
-1
lib/xks/xks.ex
+1
-1
lib/xks/xks.ex