this repo has no description
Recruit servers evenly across supervisors
+59
-12
+59
-12
lib/servers/manager.ex
+59
-12
lib/servers/manager.ex
···
339
339
dbg {"Recovery", state.cluster.generation, max_kcv, min_dv}
340
340
#dbg {meta_pairs, state.cluster.tlog_generations}
341
341
342
-
ids = allocate_server_ids(state, state.config.num_tlogs + state.config.num_begin_buffers + state.config.num_commit_buffers + 3)
343
-
{tlog_ids, ids} = Enum.split(ids, state.config.num_tlogs)
344
-
{begin_buffer_ids, ids} = Enum.split(ids, state.config.num_begin_buffers)
345
-
{commit_buffer_ids, ids} = Enum.split(ids, state.config.num_commit_buffers)
342
+
supervisors_stateless = Enum.map(state.supervisors, fn {pid, %{stateless: stateless_slots}} -> {pid, stateless_slots} end)
343
+
supervisors_tlogs = Enum.map(state.supervisors, fn {pid, %{tlog: tlog_slots}} -> {pid, tlog_slots} end)
344
+
345
+
open_stateless_slots = Enum.sum_by(supervisors_stateless, fn {_pid, stateless_slots} -> stateless_slots end)
346
+
open_tlog_slots = Enum.sum_by(supervisors_tlogs, fn {_pid, tlog_slots} -> tlog_slots end)
347
+
348
+
# Sequencer, BeginBuffer, CommitBuffer, Resolver, Distributor
349
+
assert open_stateless_slots >= 5
350
+
assert open_tlog_slots >= state.config.num_replicas
351
+
352
+
ids = allocate_server_ids(state, open_tlog_slots + 5)
353
+
{tlog_ids, ids} = Enum.split(ids, open_tlog_slots)
354
+
{begin_buffer_ids, ids} = Enum.split(ids, 1)
355
+
{commit_buffer_ids, ids} = Enum.split(ids, 1)
346
356
[sequencer_id, resolver_id, distributor_id] = ids
347
357
348
358
# Create new TLog generation
···
356
366
# Recruit new TLogs
357
367
cluster = state.cluster
358
368
gen = state.cluster.generation
359
-
state = recruit(state, gen, Hobbes.Servers.TLog, tlog_ids, %{cluster: state.cluster, prev_version: last_generation_end_version, meta_pairs: meta_pairs})
369
+
370
+
state =
371
+
Enum.map(tlog_ids, fn id -> {Hobbes.Servers.TLog, id, %{cluster: state.cluster, prev_version: last_generation_end_version, meta_pairs: meta_pairs}} end)
372
+
|> recruit_servers(supervisors_tlogs, gen)
373
+
|> Enum.reduce(state, fn %Server{} = server, state ->
374
+
put_in(state.cluster.servers[server.id], server)
375
+
end)
360
376
361
377
# Copy mutations in range [max_kcv + 1, min_dv] to the new TLogs
362
378
shard_map = ShardTagMap.new(new_tlog_generation)
···
392
408
shard_moves_pairs = FlatKV.scan(meta_kv, shard_moves_prefix(), shard_moves_end()).pairs
393
409
[next_shard_move_id] = FlatKV.get(meta_kv, next_shard_move_id_key()) |> Keyset.unpack()
394
410
395
-
# Recruit the rest of the new transaction system
411
+
# Recruit stateless servers
396
412
state =
397
-
state
398
-
|> recruit(gen, Hobbes.Servers.BeginBuffer, begin_buffer_ids, %{cluster: cluster})
399
-
|> recruit(gen, Hobbes.Servers.CommitBuffer, commit_buffer_ids, %{cluster: cluster, key_servers_pairs: key_servers_pairs})
400
-
|> recruit(gen, Hobbes.Servers.Sequencer, [sequencer_id], %{cluster: cluster, prev_version: recovery_commit_version})
401
-
|> recruit(gen, Hobbes.Servers.Resolver, [resolver_id], %{cluster: cluster, prev_version: recovery_commit_version})
402
-
|> recruit(gen, Hobbes.Servers.Distributor, [distributor_id], %{cluster: cluster, key_servers_pairs: key_servers_pairs, shard_moves_pairs: shard_moves_pairs, next_shard_move_id: next_shard_move_id})
413
+
[
414
+
[{Hobbes.Servers.Sequencer, sequencer_id, %{cluster: cluster, prev_version: recovery_commit_version}}],
415
+
[{Hobbes.Servers.Resolver, resolver_id, %{cluster: cluster, prev_version: recovery_commit_version}}],
416
+
[{Hobbes.Servers.Distributor, distributor_id, %{cluster: cluster, key_servers_pairs: key_servers_pairs, shard_moves_pairs: shard_moves_pairs, next_shard_move_id: next_shard_move_id}}],
417
+
Enum.map(begin_buffer_ids, fn id -> {Hobbes.Servers.BeginBuffer, id, %{cluster: cluster}} end),
418
+
Enum.map(commit_buffer_ids, fn id -> {Hobbes.Servers.CommitBuffer, id, %{cluster: cluster, key_servers_pairs: key_servers_pairs}} end),
419
+
]
420
+
|> Enum.concat()
421
+
|> recruit_servers(supervisors_stateless, gen)
422
+
|> Enum.reduce(state, fn %Server{} = server, state ->
423
+
put_in(state.cluster.servers[server.id], server)
424
+
end)
403
425
404
426
# Write the new TLog generation to the Coordinators
405
427
# Once complete, the effect of this recovery is permanent, and any future recoveries will have to start from these TLogs
···
522
544
end)
523
545
524
546
state
547
+
end
548
+
549
+
@spec recruit_servers([tuple], [{pid, non_neg_integer}], non_neg_integer, non_neg_integer, list) :: list
550
+
defp recruit_servers(servers, supervisors, generation, i \\ 0, acc \\ [])
551
+
552
+
defp recruit_servers([], _supervisors, _generation, _i, acc) do
553
+
Enum.reverse(acc)
554
+
end
555
+
556
+
defp recruit_servers(servers, supervisors, generation, i, acc) do
557
+
sup_i = rem(i, length(supervisors))
558
+
559
+
case Enum.at(supervisors, sup_i) do
560
+
{sup_pid, open_slots} when open_slots > 0 ->
561
+
[{module, id, arg} | servers] = servers
562
+
arg = Map.put(arg, :id, id)
563
+
{:ok, server_pid} = ServerSupervisor.start_child_server(sup_pid, generation, module, arg)
564
+
565
+
supervisors = List.replace_at(supervisors, sup_i, {sup_pid, open_slots - 1})
566
+
acc = [%Server{type: module, id: id, pid: server_pid} | acc]
567
+
recruit_servers(servers, supervisors, generation, i + 1, acc)
568
+
569
+
{_pid, 0} ->
570
+
recruit_servers(servers, supervisors, generation, i + 1, acc)
571
+
end
525
572
end
526
573
527
574
defp recruit(%State{} = state, generation, module, ids, args) when is_integer(generation) and is_atom(module) and is_list(ids) and is_map(args) do