this repo has no description
Peek and merge batches for each tag during recovery
+93
-13
+45
-2
lib/servers/manager.ex
+45
-2
lib/servers/manager.ex
···
313
313
cluster = state.cluster
314
314
gen = state.cluster.generation
315
315
316
+
# Recruit new TLogs
316
317
state = recruit(state, gen, Hobbes.Servers.TLog, tlog_ids, %{cluster: state.cluster, meta_pairs: meta_pairs})
317
318
318
-
# TODO: copy mutations to new tlogs here
319
+
# Copy mutations to new TLogs
320
+
new_tlog_generation = %TLogGeneration{generation: state.cluster.generation, tlog_ids: tlog_ids}
321
+
tags = extract_tags(meta_pairs)
322
+
state = copy_mutations_to_new_generation(state, hd(state.cluster.tlog_generations), new_tlog_generation, tags, max_kcv, recovery_version)
319
323
324
+
# Recruit the rest of the new transaction system
320
325
state =
321
326
state
322
327
|> recruit(gen, Hobbes.Servers.CommitBuffer, commit_buffer_ids, %{cluster: cluster, meta_pairs: meta_pairs})
···
326
331
327
332
# Write the new TLog generation to the Coordinators
328
333
# Once complete, the effect of this recovery is permanent, and any future recoveries will have to start from these TLogs
329
-
new_tlog_generation = %TLogGeneration{generation: state.cluster.generation, tlog_ids: tlog_ids}
330
334
{:ok, :ok} = Coordinator.write_generation(state.primary_coordinator, new_tlog_generation.generation, new_tlog_generation.tlog_ids)
331
335
332
336
state = update_in(state.cluster.tlog_generations, fn generations when is_list(generations) ->
···
334
338
end)
335
339
336
340
put_in(state.cluster.status, :normal)
341
+
end
342
+
343
+
defp extract_tags(meta_pairs) do
344
+
meta_pairs
345
+
|> Enum.reduce([], fn {k, _v}, acc ->
346
+
case k do
347
+
servers_prefix() <> id -> [String.to_integer(id) | acc]
348
+
_ -> acc
349
+
end
350
+
end)
351
+
|> Enum.sort()
352
+
end
353
+
354
+
defp copy_mutations_to_new_generation(%State{} = state, %TLogGeneration{} = old_gen, %TLogGeneration{} = _new_gen, tags, start_version, end_version) do
355
+
recovered_old_tlog_pids =
356
+
old_gen.tlog_ids
357
+
|> Enum.filter(fn id ->
358
+
Enum.any?(state.recovered_tlogs, &(&1.id == id))
359
+
end)
360
+
|> Enum.map(fn id ->
361
+
%Server{pid: pid} = state.cluster.servers[id]
362
+
pid
363
+
end)
364
+
365
+
Enum.each(tags, fn tag ->
366
+
batches = peek_merge(recovered_old_tlog_pids, tag, start_version, end_version)
367
+
dbg {tag, batches}
368
+
# TODO: write to new TLogs
369
+
end)
370
+
371
+
state
372
+
end
373
+
374
+
defp peek_merge(tlog_pids, tag, start_version, end_version) when is_list(tlog_pids) and is_integer(tag) do
375
+
Enum.map(tlog_pids, fn pid ->
376
+
{{_sv, _ev}, batches} = TLog.peek(pid, tag, start_version, end_version)
377
+
batches
378
+
end)
379
+
|> merge_batches()
337
380
end
338
381
339
382
defp recruit(%State{} = state, generation, module, ids, args) when is_atom(module) and is_list(ids) and is_map(args) do
+7
-6
lib/servers/tlog.ex
+7
-6
lib/servers/tlog.ex
···
43
43
def start_link(arg), do: SimServer.start_link(__MODULE__, arg)
44
44
45
45
@doc """
46
-
Peeks logs for `tag` starting at `start_version`.
46
+
Peeks logs for `tag` starting at `start_version` and optionally ending at `end_version` (both inclusive).
47
47
48
48
Returns `{{start_version, end_version}, {meta_mutations, data_mutations}}`.
49
49
"""
50
-
@spec peek(pid, term, non_neg_integer | nil) :: {{non_neg_integer, non_neg_integer}, {non_neg_integer, [Utils.mutation]}}
51
-
def peek(server, tag, start_version) do
52
-
SimServer.call(server, {:peek, tag, start_version})
50
+
@spec peek(pid, term, non_neg_integer, non_neg_integer | nil) :: {{non_neg_integer, non_neg_integer}, {non_neg_integer, [Utils.numbered_mutation]}}
51
+
def peek(server, tag, start_version, end_version \\ nil) do
52
+
SimServer.call(server, {:peek, tag, start_version, end_version})
53
53
end
54
54
55
55
@doc """
···
93
93
{:ok, state}
94
94
end
95
95
96
-
def handle_call({:peek, tag, start_version}, _from, state) when is_integer(start_version) do
97
-
end_version = state.version
96
+
def handle_call({:peek, tag, start_version, end_version}, _from, state)
97
+
when is_integer(start_version) and (is_nil(end_version) or is_integer(end_version)) do
98
+
end_version = end_version || state.version
98
99
99
100
batches = TaggedQueue.peek(state.tagged_queue, tag, start_version, end_version)
100
101
{:reply, {{start_version, end_version}, batches}, state}
+36
lib/utils.ex
+36
lib/utils.ex
···
172
172
defp interleave([], [], acc), do: acc
173
173
174
174
@doc """
175
+
Merges a list of lists of versioned batches of numbered mutations (e.g. TLog peek replies).
176
+
177
+
## Examples
178
+
179
+
iex> tlog_pids |> Enum.map(&TLog.peek(&1, tag, sv, ev)) |> merge_batches()
180
+
[
181
+
{
182
+
100,
183
+
[
184
+
{0, {:write, "foo", "bar"}},
185
+
{1, {:clear, "foo"}},
186
+
]
187
+
},
188
+
{101, ...},
189
+
{102, ...},
190
+
]
191
+
"""
192
+
@spec merge_batches([[{non_neg_integer, [Utils.numbered_mutation]}]]) :: [{non_neg_integer, [Utils.numbered_mutation]}]
193
+
def merge_batches(batches) do
194
+
batches
195
+
|> Enum.concat()
196
+
|> Enum.group_by(fn {version, _mutations} -> version end, fn {_version, mutations} -> mutations end)
197
+
|> Enum.map(fn {version, mut_batches} ->
198
+
# TODO: this could be done more efficiently with a K-way merge since all batches are sorted
199
+
# But, for now, this function is not really on the hot path (only needed after TLog failure)
200
+
mutations =
201
+
mut_batches
202
+
|> Enum.concat()
203
+
|> Enum.uniq_by(fn {i, _mut} -> i end)
204
+
|> Enum.sort()
205
+
{version, mutations}
206
+
end)
207
+
|> Enum.sort()
208
+
end
209
+
210
+
@doc """
175
211
Attempts to run `fun`, but backs off and retries
176
212
if `{:error, :read_version_too_new}` is received.
177
213
+5
-5
test/hobbes_test.exs
+5
-5
test/hobbes_test.exs
···
111
111
@tag :disable
112
112
test "Cycle Recovery", %{test: test} do
113
113
Workloads.run([
114
-
#{Workloads.Cycle, [
115
-
#keys: 20,
116
-
#clients: 10,
117
-
#duration_ms: 20_000,
118
-
#]},
114
+
{Workloads.Cycle, [
115
+
keys: 20,
116
+
clients: 10,
117
+
duration_ms: 20_000,
118
+
]},
119
119
{Workloads.KillServers, [
120
120
delay_ms: 3_000,
121
121
duration_ms: 20_000,