this repo has no description
Optimize CommitBuffer shard map
+371
-85
+3
ROADMAP.md
+3
ROADMAP.md
···
91
91
- [X] Update TaggedQueue to use ets instead of erlang queues (tlog batch appends would randomly spike to 50ms, maybe gc? this fixed it)
92
92
- [ ] Optimize resolver
93
93
- [ ] Store byte sample in a separate FlatKV and add mutations to HybridKV mutation log
94
+
- [X] Optimize CommitBuffer to use unversioned shard map with pre-computed tlogs and tags
95
+
- CommitBuffer reductions on bench_rw went from 85 million down to 25 million!
96
+
- Txn/s went from about 24.8k to about 25.8k! (obviously CommitBuffer is not the bottleneck)
94
97
95
98
96
99
### Construct
+79
lib/dense_shard_map.ex
+79
lib/dense_shard_map.ex
···
1
+
defmodule Hobbes.DenseShardMap do
2
+
@moduledoc ~S"""
3
+
This module implements a *Dense* Shard Map, where the entire key space of the database is covered.
4
+
5
+
Shards are stored in a tree with keys of `start_key`, with the `end_key` implied by
6
+
the `start_key` of the following shard.
7
+
The final `end_key` is implied to be `Utils.all_keys_end()` (`\xFF\xFF`).
8
+
9
+
This mirrors how shards are stored in the actual Hobbes meta keyspace (the `key_servers` space),
10
+
and `DenseShardMap` is designed to function as a local coherent cache of that space.
11
+
"""
12
+
13
+
import Hobbes.Utils
14
+
15
+
@type t :: :ets.table
16
+
17
+
@spec new :: t
18
+
def new do
19
+
:ets.new(__MODULE__, [:ordered_set, :private])
20
+
end
21
+
22
+
@spec put(t, binary, term) :: :ok
23
+
def put(table, start_key, value) when is_database_key(start_key) do
24
+
:ets.insert(table, {start_key, value})
25
+
:ok
26
+
end
27
+
28
+
@spec delete(t, binary) :: :ok
29
+
def delete(table, start_key) when is_database_key(start_key) do
30
+
:ets.delete(table, start_key)
31
+
:ok
32
+
end
33
+
34
+
@spec shard_for_key(t, binary) :: {binary, binary, term}
35
+
def shard_for_key(table, key) when is_database_key(key) do
36
+
# Raises if table is empty, which is fine
37
+
{_sk, [{sk, value}]} = :ets.prev_lookup(table, next_key(key))
38
+
ek = shard_end_key(table, sk)
39
+
40
+
{sk, ek, value}
41
+
end
42
+
43
+
@spec shards_for_range(t, binary, binary) :: [{{binary, binary}, term}]
44
+
def shards_for_range(table, start_key, end_key) when is_database_range(start_key, end_key) do
45
+
{_sk, [{first_sk, _value} = first]} = :ets.prev_lookup(table, next_key(start_key))
46
+
47
+
scan_shards(table, end_key, first_sk, [first])
48
+
|> Enum.reverse()
49
+
|> Enum.chunk_every(2, 1, :discard)
50
+
|> Enum.map(fn [{sk, value}, {ek, _value}] ->
51
+
{sk, ek, value}
52
+
end)
53
+
end
54
+
55
+
defp scan_shards(table, end_key, prev_key, acc) do
56
+
case :ets.next_lookup(table, prev_key) do
57
+
{_sk, [{sk, _value} = tup]} ->
58
+
case sk < end_key do
59
+
true -> scan_shards(table, end_key, sk, [tup | acc])
60
+
false -> [tup | acc]
61
+
end
62
+
63
+
:"$end_of_table" -> [{all_keys_end(), nil} | acc]
64
+
end
65
+
end
66
+
67
+
@spec shard_end_key(t, binary) :: binary
68
+
defp shard_end_key(table, start_key) when is_binary(start_key) do
69
+
case :ets.next(table, start_key) do
70
+
:"$end_of_table" -> all_keys_end()
71
+
ek when is_binary(ek) -> ek
72
+
end
73
+
end
74
+
75
+
@doc false
76
+
def dump(table) do
77
+
:ets.tab2list(table)
78
+
end
79
+
end
+45
-68
lib/servers/commit_buffer.ex
+45
-68
lib/servers/commit_buffer.ex
···
3
3
4
4
import ExUnit.Assertions, only: [assert: 1]
5
5
6
-
alias Hobbes.MetaStore
7
-
alias Hobbes.Structs.{Cluster, Server, ResolveBatch, CommitTxn, LogBatch}
6
+
alias Hobbes.ShardTagMap
7
+
alias Hobbes.Structs.{Cluster, TLogGeneration, Server, ResolveBatch, CommitTxn, LogBatch}
8
8
9
9
alias Hobbes.Servers.{Sequencer, Resolver, TLog}
10
10
···
16
16
@flush_interval_ms 1
17
17
@max_buffer_size 300
18
18
19
-
@tlog_replication_factor 3
20
-
21
19
defmodule State do
20
+
@type t :: %__MODULE__{
21
+
id: non_neg_integer,
22
+
cluster: Cluster.t,
23
+
shard_map: ShardTagMap.t,
24
+
last_committed_version: non_neg_integer,
25
+
26
+
buffer: list,
27
+
buffer_size: non_neg_integer,
28
+
29
+
storage_servers: map,
30
+
}
31
+
22
32
@enforce_keys [
23
33
:id,
24
34
:cluster,
25
-
:meta_store,
35
+
:shard_map,
26
36
:last_committed_version,
27
37
28
38
:buffer,
···
87
97
end
88
98
89
99
def init(%{id: id, cluster: %Cluster{} = cluster, meta_pairs: meta_pairs}) do
100
+
%TLogGeneration{} = current_generation = hd(cluster.tlog_generations)
101
+
assert current_generation.generation == cluster.generation
102
+
90
103
state = %State{
91
104
id: id,
92
105
cluster: cluster,
93
-
meta_store: MetaStore.new(),
106
+
shard_map: ShardTagMap.new(current_generation),
94
107
last_committed_version: 0,
95
108
96
109
buffer: [],
···
100
113
}
101
114
102
115
seed_meta = Enum.map(meta_pairs, fn {k, v} -> {:write, k, v} end)
103
-
MetaStore.apply_meta_mutations(state.meta_store, 0, seed_meta)
116
+
ShardTagMap.apply_metadata_mutations(state.shard_map, seed_meta)
104
117
105
118
SimServer.send_after(self(), :flush, @flush_interval_ms)
106
119
{:ok, state}
107
120
end
108
121
109
122
def handle_call({:get_shards, keys_or_ranges}, _from, state) when is_list(keys_or_ranges) do
110
-
ms = state.meta_store
111
-
version = state.last_committed_version
123
+
stm = state.shard_map
112
124
storage_servers = state.storage_servers
113
125
114
126
shard_lists =
115
-
Enum.map(keys_or_ranges, &MetaStore.get_shards(ms, version, &1))
116
-
|> Enum.map(fn shards ->
117
-
Enum.map(shards, fn {sk, ek, ids} ->
118
-
pids = Enum.map(ids, &Map.get(storage_servers, &1))
119
-
{sk, ek, {ids, pids}}
127
+
Enum.map(keys_or_ranges, fn key_or_range ->
128
+
ShardTagMap.shards_for_key_or_range(stm, key_or_range)
129
+
|> Enum.map(fn {sk, ek, {_tlogs, _all_tags, from_tags}} ->
130
+
pids = Enum.map(from_tags, &Map.get(storage_servers, &1))
131
+
{sk, ek, {from_tags, pids}}
120
132
end)
121
133
end)
134
+
122
135
{:reply, {:ok, shard_lists}, state}
123
136
end
124
137
···
187
200
txn.read_version,
188
201
txn.read_conflicts,
189
202
txn.write_conflicts,
190
-
extract_meta(txn.mutations),
203
+
Enum.filter(txn.mutations, &meta_mutation?/1),
191
204
} | acc]
192
205
end)
193
206
···
202
215
{txn_results_reversed, meta_log} = Resolver.resolve_batch(resolver_pid, batch)
203
216
204
217
# Apply meta mutations received, including our own from this batch
205
-
Enum.each(meta_log, fn {commit_version, mutations} ->
206
-
MetaStore.apply_meta_mutations(state.meta_store, commit_version, mutations)
218
+
Enum.each(meta_log, fn {_commit_version, mutations} ->
219
+
ShardTagMap.apply_metadata_mutations(state.shard_map, mutations)
207
220
end)
208
221
209
222
{allowed_transactions, rejected_transactions} =
···
216
229
217
230
# If the database is locked, filter out all non-meta transactions
218
231
# TODO: use a lock-aware flag instead like FDB
219
-
{allowed_transactions, locked_transactions} =
220
-
case MetaStore.locked?(state.meta_store, commit_version) do
221
-
true -> Enum.split_with(allowed_transactions, fn txn -> has_meta?(txn.mutations) end)
222
-
false -> {allowed_transactions, []}
223
-
end
232
+
# TODO: removed for now, bring back later without MetaStore
233
+
#{allowed_transactions, locked_transactions} =
234
+
# case MetaStore.locked?(state.meta_store, commit_version) do
235
+
# true -> Enum.split_with(allowed_transactions, fn txn -> has_meta?(txn.mutations) end)
236
+
# false -> {allowed_transactions, []}
237
+
# end
238
+
locked_transactions = []
224
239
225
240
# Add storage tags to each mutation (including special meta tag for meta mutations)
226
-
meta_store = state.meta_store
227
-
tagged_mutations =
241
+
tlog_mutations =
228
242
allowed_transactions
229
243
|> Enum.map(fn %CommitTxn{} = txn -> txn.mutations end)
230
244
|> Enum.concat()
231
245
|> then(fn mutations ->
232
246
mutations ++ compute_special_mutations(mutations)
233
247
end)
234
-
|> Enum.with_index()
235
-
|> Enum.map(fn {mut, i} ->
236
-
# TODO: will have to split range clears
237
-
tags = MetaStore.get_key_server_mutation_tags(meta_store, commit_version, mutation_key(mut))
238
-
{tags, {i, mut}}
248
+
|> then(fn mutations ->
249
+
ShardTagMap.tag_and_slice_mutations(state.shard_map, mutations)
239
250
end)
240
251
241
-
# TODO: get latest generation only
242
-
assert state.cluster.status == :normal
243
-
tlogs =
244
-
hd(state.cluster.tlog_generations).tlog_ids
245
-
|> Enum.map(fn id -> Map.fetch!(state.cluster.servers, id) end)
246
-
247
-
num_tlogs = length(tlogs)
248
-
all_tlogs = Enum.to_list(0..(num_tlogs - 1))
249
-
# TODO: maybe a tuple is better? there are not that many tlogs
250
-
log_mutations_reversed = Map.new(all_tlogs, fn i -> {i, []} end)
251
-
252
-
# Slice up mutations for tlogs
253
-
# Note: prepending reverses the mutations, hence the name
254
-
# (they are reversed again when the batch is created below)
255
-
log_mutations_reversed =
256
-
Enum.reduce(tagged_mutations, log_mutations_reversed, fn {tags, _mut} = tm, acc ->
257
-
tlogs =
258
-
case -1 in tags do
259
-
# Send meta mutations (tagged with -1) to all tlogs
260
-
true -> all_tlogs
261
-
false -> tlogs_for_tags(num_tlogs, @tlog_replication_factor, tags)
262
-
end
263
-
Enum.reduce(tlogs, acc, fn tlog_i, acc ->
264
-
# Prepend mutation onto list for each tlog batch
265
-
Map.put(acc, tlog_i, [tm | Map.fetch!(acc, tlog_i)])
266
-
end)
267
-
end)
252
+
tlog_ids = hd(state.cluster.tlog_generations).tlog_ids
268
253
269
254
# Send sliced mutations to each tlog
270
-
all_tlogs
271
-
|> Enum.map(fn tlog_i ->
272
-
tagged_mutations = Enum.reverse(Map.fetch!(log_mutations_reversed, tlog_i))
255
+
tlog_ids
256
+
|> Enum.map(fn tlog_id ->
257
+
tagged_mutations = Map.fetch!(tlog_mutations, tlog_id)
273
258
274
259
log_batch = %LogBatch{
275
260
commit_buffer_id: state.id,
···
279
264
last_committed_version: state.last_committed_version,
280
265
}
281
266
282
-
%Server{pid: tlog_pid} = Enum.at(tlogs, tlog_i)
267
+
%Server{pid: tlog_pid} = Map.fetch!(state.cluster.servers, tlog_id)
283
268
TLog.write_batch_send(tlog_pid, log_batch)
284
269
end)
285
270
|> Enum.each(fn req_id ->
···
313
298
end)
314
299
315
300
%State{state | last_committed_version: commit_version, buffer: [], buffer_size: 0}
316
-
end
317
-
318
-
defp has_meta?(mutations) do
319
-
Enum.any?(mutations, &meta_mutation?/1)
320
-
end
321
-
322
-
defp extract_meta(mutations) do
323
-
Enum.filter(mutations, &meta_mutation?/1)
324
301
end
325
302
end
+11
-11
lib/servers/manager.ex
+11
-11
lib/servers/manager.ex
···
267
267
268
268
meta_pairs = build_seed_meta(config, storage_ids)
269
269
270
+
first_tlog_generation = %TLogGeneration{generation: state.cluster.generation, start_version: 0, tlog_ids: tlog_ids}
271
+
state = put_in(state.cluster.tlog_generations, [first_tlog_generation])
272
+
270
273
cluster = state.cluster
271
274
gen = cluster.generation
272
275
···
283
286
284
287
# Write the first generation into the Coordinators
285
288
# Once complete, any future recoveries will have to start from these TLogs
286
-
first_tlog_generation = %TLogGeneration{generation: state.cluster.generation, start_version: 0, tlog_ids: tlog_ids}
287
289
{:ok, :ok} = Coordinator.write_generation(state.primary_coordinator, first_tlog_generation.generation, first_tlog_generation.start_version, first_tlog_generation.tlog_ids)
288
290
289
291
state = put_in(state.cluster.tlog_generations, [first_tlog_generation])
···
311
313
%Server{pid: old_tlog_pid} = Map.fetch!(state.cluster.servers, hd(state.recovered_tlogs).id)
312
314
{:ok, meta_pairs} = TLog.read_meta_store(old_tlog_pid, min_dv)
313
315
314
-
#dbg {state.cluster.generation, max_kcv, min_dv}
316
+
dbg {"Recovery", state.cluster.generation, max_kcv, min_dv}
315
317
#dbg {meta_pairs, state.cluster.tlog_generations}
316
318
317
319
ids = allocate_server_ids(state, state.config.num_tlogs + state.config.num_commit_buffers + 3)
···
319
321
{commit_buffer_ids, ids} = Enum.split(ids, state.config.num_commit_buffers)
320
322
[sequencer_id, resolver_id, distributor_id] = ids
321
323
322
-
cluster = state.cluster
323
-
gen = state.cluster.generation
324
+
# Create new TLog generation
325
+
new_tlog_generation = %TLogGeneration{generation: state.cluster.generation, start_version: max_kcv + 1, tlog_ids: tlog_ids}
326
+
assert new_tlog_generation.start_version > hd(state.cluster.tlog_generations).start_version
327
+
328
+
state = update_in(state.cluster.tlog_generations, &[new_tlog_generation | &1])
324
329
325
330
# Recruit new TLogs
331
+
cluster = state.cluster
332
+
gen = state.cluster.generation
326
333
state = recruit(state, gen, Hobbes.Servers.TLog, tlog_ids, %{cluster: state.cluster, meta_pairs: meta_pairs})
327
-
328
-
new_tlog_generation = %TLogGeneration{generation: state.cluster.generation, start_version: max_kcv + 1, tlog_ids: tlog_ids}
329
-
assert new_tlog_generation.start_version > hd(state.cluster.tlog_generations).start_version
330
334
331
335
# Copy mutations in range [max_kcv + 1, min_dv] to the new TLogs
332
336
{state, last_batch_version} = copy_mutations_to_new_generation(state, hd(state.cluster.tlog_generations), new_tlog_generation, meta_pairs, new_tlog_generation.start_version, min_dv)
···
359
363
# Write the new TLog generation to the Coordinators
360
364
# Once complete, the effect of this recovery is permanent, and any future recoveries will have to start from these TLogs
361
365
{:ok, :ok} = Coordinator.write_generation(state.primary_coordinator, new_tlog_generation.generation, new_tlog_generation.start_version, new_tlog_generation.tlog_ids)
362
-
363
-
state = update_in(state.cluster.tlog_generations, fn generations when is_list(generations) ->
364
-
[new_tlog_generation | generations]
365
-
end)
366
366
367
367
put_in(state.cluster.status, :normal)
368
368
end
+120
lib/shard_tag_map.ex
+120
lib/shard_tag_map.ex
···
1
+
defmodule Hobbes.ShardTagMap do
2
+
alias Hobbes.{DenseShardMap, ShardTagMap, Utils}
3
+
alias Hobbes.Structs.TLogGeneration
4
+
5
+
import Hobbes.Utils
6
+
import Hobbes.MetaStore, only: [decode_server_ids: 1]
7
+
8
+
@type t :: %__MODULE__{
9
+
tlog_ids: [non_neg_integer],
10
+
replication_factor: non_neg_integer,
11
+
shard_map: DenseShardMap.t,
12
+
}
13
+
@enforce_keys [:tlog_ids, :replication_factor, :shard_map]
14
+
defstruct @enforce_keys
15
+
16
+
@spec new(TLogGeneration.t) :: t
17
+
def new(%TLogGeneration{} = generation) do
18
+
%ShardTagMap{
19
+
# TODO: parameterize
20
+
tlog_ids: generation.tlog_ids,
21
+
replication_factor: 3,
22
+
shard_map: DenseShardMap.new(),
23
+
}
24
+
end
25
+
26
+
@spec apply_metadata_mutations(t, [Utils.mutation]) :: :ok
27
+
def apply_metadata_mutations(%ShardTagMap{} = stm, mutations) when is_list(mutations) do
28
+
%{shard_map: shard_map, tlog_ids: tlog_ids, replication_factor: rf} = stm
29
+
Enum.each(mutations, fn
30
+
{:write, key_servers_prefix() <> start_key, value} ->
31
+
{shard_all_tags, shard_from_tags} = server_tags_from_key_servers_value(value)
32
+
shard_tlog_ids = tlog_ids_for_servers(tlog_ids, shard_all_tags, rf)
33
+
34
+
DenseShardMap.put(shard_map, start_key, {shard_tlog_ids, shard_all_tags, shard_from_tags})
35
+
36
+
{:clear, key_servers_prefix() <> start_key} ->
37
+
DenseShardMap.delete(shard_map, start_key)
38
+
39
+
_mut -> :noop
40
+
end)
41
+
42
+
:ok
43
+
end
44
+
45
+
@spec tag_and_slice_mutations(t, [Utils.mutation]) :: %{non_neg_integer => [Utils.tagged_mutation]}
46
+
def tag_and_slice_mutations(%ShardTagMap{} = stm, mutations) when is_list(mutations) do
47
+
%{shard_map: shard_map, tlog_ids: all_tlog_ids} = stm
48
+
tlog_mutations = Map.new(stm.tlog_ids, fn id -> {id, []} end)
49
+
50
+
Enum.reduce(mutations, {tlog_mutations, 0}, fn mut, {acc, i} ->
51
+
{tlogs, tags} =
52
+
case mutation_key(mut) do
53
+
special_server_keys_prefix() <> rest ->
54
+
# Special server_keys mutations are sent only to that particular Storage server
55
+
[id_str, _] = String.split(rest, "/")
56
+
tags = [String.to_integer(id_str)]
57
+
# Send to all TLogs for simplicity (shard moves should never be a bottleneck)
58
+
{all_tlog_ids, tags}
59
+
60
+
meta_prefix() <> _ = key ->
61
+
{_sk, _ek, {_tlogs, tags, _from_tags}} = DenseShardMap.shard_for_key(shard_map, key)
62
+
# Tag meta mutations with meta tag (-1) and send to all tlogs
63
+
tags = [meta_tag() | tags]
64
+
{all_tlog_ids, tags}
65
+
66
+
key ->
67
+
{_sk, _ek, {tlogs, tags, _from_tags}} = DenseShardMap.shard_for_key(shard_map, key)
68
+
# Send to pre-computed tlogs/tags for this shard
69
+
{tlogs, tags}
70
+
end
71
+
72
+
tagged_mut = {tags, {i, mut}}
73
+
acc = Enum.reduce(tlogs, acc, fn tlog_id, acc ->
74
+
Map.update!(acc, tlog_id, &[tagged_mut | &1])
75
+
end)
76
+
{acc, i + 1}
77
+
end)
78
+
|> then(fn {tlog_mutations, _i} ->
79
+
Map.new(tlog_mutations, fn {tlog_id, mutations_reversed} -> {tlog_id, Enum.reverse(mutations_reversed)} end)
80
+
end)
81
+
end
82
+
83
+
@spec shards_for_key_or_range(t, binary | {binary, binary}) :: [{binary, binary, {list, Utils.tag_list, Utils.tag_list}}]
84
+
def shards_for_key_or_range(%ShardTagMap{} = stm, key) when is_binary(key) do
85
+
[DenseShardMap.shard_for_key(stm.shard_map, key)]
86
+
end
87
+
88
+
def shards_for_key_or_range(%ShardTagMap{} = stm, {start_key, end_key}) do
89
+
DenseShardMap.shards_for_range(stm.shard_map, start_key, end_key)
90
+
end
91
+
92
+
# Returns {all_tags, from_tags} where all_tags is (from_tags ++ to_tags)
93
+
defp server_tags_from_key_servers_value(value) do
94
+
case String.split(value, "/") do
95
+
[from_str, to_str] ->
96
+
from_ids = decode_server_ids(from_str)
97
+
{from_ids ++ decode_server_ids(to_str), from_ids}
98
+
99
+
[from_str] ->
100
+
from_ids = decode_server_ids(from_str)
101
+
{from_ids, from_ids}
102
+
end
103
+
end
104
+
105
+
@doc false
106
+
@spec tlog_ids_for_servers([non_neg_integer], [non_neg_integer], non_neg_integer) :: [non_neg_integer]
107
+
def tlog_ids_for_servers(tlog_ids, server_ids, min_replicas) do
108
+
ids = server_ids |> Enum.map(&buddy_tlog_id(tlog_ids, &1)) |> Enum.uniq()
109
+
length = length(ids)
110
+
111
+
case length < min_replicas do
112
+
true ->
113
+
needed = min_replicas - length
114
+
remaining = tlog_ids -- ids
115
+
ids ++ Enum.take_random(remaining, needed)
116
+
117
+
false -> ids
118
+
end
119
+
end
120
+
end
+3
-2
lib/sparse_shard_map.ex
+3
-2
lib/sparse_shard_map.ex
···
1
1
defmodule Hobbes.SparseShardMap do
2
2
@moduledoc """
3
-
This module implements a sparse shard map on top of an ETS table.
4
-
The shard map is used by storage servers to track which shards it can safely serve reads for.
3
+
This module implements a *Sparse* Shard Map where only some shards are present.
5
4
6
5
Shards are stored in a tree as key/value pairs of `{start_key, end_key}`.
7
6
The shards are required to be disjoint, but no safety checks are performed as it is assumed
8
7
that all shard updates coming from the database are correct.
8
+
9
+
`SparseShardMap` is used by `Storage` servers to keep track of which shards they can serve.
9
10
"""
10
11
11
12
@type t :: :ets.table
+24
-4
lib/utils.ex
+24
-4
lib/utils.ex
···
9
9
@type mutation :: {:write, binary, binary} | {:clear, binary} | {:clear_range, binary, binary}
10
10
@type numbered_mutation :: {non_neg_integer, mutation}
11
11
@type tag_list :: [integer]
12
+
@type tagged_mutation :: {tag_list, numbered_mutation}
12
13
14
+
defmacro normal_prefix, do: ""
15
+
defmacro normal_end, do: "\xFF"
13
16
defmacro meta_prefix, do: "\xFF"
14
17
defmacro meta_end, do: "\xFF\xFF"
18
+
defmacro database_prefix, do: normal_prefix()
19
+
defmacro database_end, do: meta_end()
20
+
21
+
# TODO: remove all_keys in favor of database
15
22
defmacro all_keys_prefix, do: ""
16
23
defmacro all_keys_end, do: meta_end()
17
24
···
31
38
def meta_tag, do: -1
32
39
33
40
def mvcc_window, do: 5_000_000
41
+
42
+
defguard is_database_key(key) when is_binary(key) and key >= database_prefix() and key < database_end()
43
+
44
+
defguard is_database_range(start_key, end_key)
45
+
when is_binary(start_key) and is_binary(end_key)
46
+
and start_key < end_key
47
+
and start_key >= database_prefix() and start_key < database_end()
48
+
and end_key <= database_end()
49
+
50
+
@spec next_key(binary) :: binary
51
+
def next_key(key) when is_binary(key), do: key <> "\x00"
34
52
35
53
@spec mutation_key(mutation) :: binary
36
54
def mutation_key({:write, key, _value}), do: key
···
109
127
rem(tag, num_tlogs)
110
128
end
111
129
112
-
@spec buddy_tlog_id(TLogGeneration.t, non_neg_integer) :: non_neg_integer
113
-
def buddy_tlog_id(%TLogGeneration{} = generation, server_id) when is_integer(server_id) do
114
-
index = rem(server_id, length(generation.tlog_ids))
115
-
Enum.at(generation.tlog_ids, index)
130
+
@spec buddy_tlog_id(TLogGeneration.t | [non_neg_integer], non_neg_integer) :: non_neg_integer
131
+
def buddy_tlog_id(%TLogGeneration{} = generation, server_id), do: buddy_tlog_id(generation.tlog_ids, server_id)
132
+
133
+
def buddy_tlog_id(tlog_ids, server_id) when is_list(tlog_ids) and is_integer(server_id) do
134
+
index = rem(server_id, length(tlog_ids))
135
+
Enum.at(tlog_ids, index)
116
136
end
117
137
118
138
@spec tlogs_for_tags(pos_integer, pos_integer, [integer]) :: [non_neg_integer]
+60
test/dense_shard_map_test.exs
+60
test/dense_shard_map_test.exs
···
1
+
defmodule Hobbes.DenseShardMapTest do
2
+
use ExUnit.Case, async: true
3
+
4
+
alias Hobbes.DenseShardMap
5
+
6
+
@moduletag :dense_shard_map
7
+
8
+
setup do
9
+
%{dsm: DenseShardMap.new()}
10
+
end
11
+
12
+
describe "shard_for_key/2" do
13
+
test "returns shard", %{dsm: dsm} do
14
+
DenseShardMap.put(dsm, "", 1)
15
+
DenseShardMap.put(dsm, "a", 2)
16
+
DenseShardMap.put(dsm, "d", 3)
17
+
DenseShardMap.put(dsm, "z", 4)
18
+
19
+
assert DenseShardMap.shard_for_key(dsm, "") == {"", "a", 1}
20
+
assert DenseShardMap.shard_for_key(dsm, "0") == {"", "a", 1}
21
+
assert DenseShardMap.shard_for_key(dsm, "a") == {"a", "d", 2}
22
+
assert DenseShardMap.shard_for_key(dsm, "aardvark") == {"a", "d", 2}
23
+
assert DenseShardMap.shard_for_key(dsm, "balloon") == {"a", "d", 2}
24
+
assert DenseShardMap.shard_for_key(dsm, "echidna") == {"d", "z", 3}
25
+
assert DenseShardMap.shard_for_key(dsm, "zebra") == {"z", "\xFF\xFF", 4}
26
+
assert DenseShardMap.shard_for_key(dsm, "\xFF/foo") == {"z", "\xFF\xFF", 4}
27
+
end
28
+
end
29
+
30
+
describe "shards_for_range" do
31
+
test "returns shards", %{dsm: dsm} do
32
+
DenseShardMap.put(dsm, "", 1)
33
+
DenseShardMap.put(dsm, "a", 2)
34
+
DenseShardMap.put(dsm, "d", 3)
35
+
DenseShardMap.put(dsm, "z", 4)
36
+
37
+
assert DenseShardMap.shards_for_range(dsm, "a", "z") == [
38
+
{"a", "d", 2},
39
+
{"d", "z", 3},
40
+
]
41
+
42
+
assert DenseShardMap.shards_for_range(dsm, "balloon", "banana") == [
43
+
{"a", "d", 2},
44
+
]
45
+
46
+
assert DenseShardMap.shards_for_range(dsm, "balloon", "zebra") == [
47
+
{"a", "d", 2},
48
+
{"d", "z", 3},
49
+
{"z", "\xFF\xFF", 4},
50
+
]
51
+
52
+
assert DenseShardMap.shards_for_range(dsm, "", "\xFF\xFF") == [
53
+
{"", "a", 1},
54
+
{"a", "d", 2},
55
+
{"d", "z", 3},
56
+
{"z", "\xFF\xFF", 4},
57
+
]
58
+
end
59
+
end
60
+
end
+3
test/hobbes_test.exs
+3
test/hobbes_test.exs
···
59
59
defmodule CycleLockWorkloadTest do
60
60
use ExUnit.Case, async: true
61
61
@tag :cycle_lock
62
+
@tag :disable
62
63
test "Cycle and Lock", %{test: test} do
63
64
Workloads.run([
64
65
{Workloads.Cycle, [
···
91
92
]},
92
93
], HobbesTest.SimOpts.sim_opts(name: test, cluster_opts: [
93
94
num_storage: 12,
95
+
num_tlogs: 6,
94
96
95
97
initial_shards: [
96
98
"",
···
304
306
setup [:setup_sim, :setup_cluster]
305
307
306
308
@tag :lock
309
+
@tag :disable
307
310
test "locks database", %{cluster: cluster} do
308
311
assert {:ok, _txn} =
309
312
Transaction.new!(cluster)
+23
test/shard_tag_map_test.exs
+23
test/shard_tag_map_test.exs
···
1
+
defmodule Hobbes.ShardTagMapTest do
2
+
use ExUnit.Case, async: true
3
+
4
+
alias Hobbes.ShardTagMap
5
+
6
+
@moduletag :shard_tag_map
7
+
8
+
describe "tlog_ids_for_servers/2" do
9
+
test "returns tlog ids" do
10
+
# TODO: this test will be flaky if it ever fails, it should really be a fuzz test
11
+
# but for now the function is so simple it's not worth testing further
12
+
tlog_ids = [0, 1, 2, 3, 4, 5]
13
+
14
+
assert [0, 1, 2] = ShardTagMap.tlog_ids_for_servers(tlog_ids, [0, 1, 2], 3)
15
+
16
+
assert [1, 2, _] = ids1 = ShardTagMap.tlog_ids_for_servers(tlog_ids, [1, 7, 8], 3)
17
+
assert length(Enum.uniq(ids1)) == 3
18
+
19
+
assert [1, _, _] = ids2 = ShardTagMap.tlog_ids_for_servers(tlog_ids, [1, 7, 13], 3)
20
+
assert length(Enum.uniq(ids2)) == 3
21
+
end
22
+
end
23
+
end