this repo has no description
Refactor to split generation recruitment back up
+39
-31
+39
-31
lib/servers/manager.ex
+39
-31
lib/servers/manager.ex
···
251
251
assert state.cluster.tlog_generations == []
252
252
253
253
%Config{} = config = state.config
254
-
storage_ids = allocate_server_ids(state, config.num_storage)
254
+
255
+
ids = allocate_server_ids(state, config.num_storage + config.num_tlogs + config.num_commit_buffers + 3)
256
+
{storage_ids, ids} = Enum.split(ids, config.num_storage)
257
+
{tlog_ids, ids} = Enum.split(ids, config.num_tlogs)
258
+
{commit_buffer_ids, ids} = Enum.split(ids, config.num_commit_buffers)
259
+
[sequencer_id, resolver_id, distributor_id] = ids
260
+
255
261
meta_pairs = build_seed_meta(config, storage_ids)
256
262
257
-
counts = Map.take(config, [:num_tlogs, :num_commit_buffers])
263
+
cluster = state.cluster
264
+
gen = cluster.generation
265
+
258
266
state =
259
267
state
260
-
|> recruit(nil, Hobbes.Servers.Storage, storage_ids, %{cluster: state.cluster})
261
-
|> recruit_transaction_generation(counts, meta_pairs)
268
+
|> recruit(nil, Hobbes.Servers.Storage, storage_ids, %{cluster: cluster})
269
+
|> recruit(gen, Hobbes.Servers.TLog, tlog_ids, %{cluster: cluster, meta_pairs: meta_pairs})
270
+
|> recruit(gen, Hobbes.Servers.CommitBuffer, commit_buffer_ids, %{cluster: cluster, meta_pairs: meta_pairs})
271
+
|> recruit(gen, Hobbes.Servers.Sequencer, [sequencer_id], %{})
272
+
|> recruit(gen, Hobbes.Servers.Resolver, [resolver_id], %{})
273
+
|> recruit(gen, Hobbes.Servers.Distributor, [distributor_id], %{cluster: cluster})
262
274
263
275
seed_meta_store(state, meta_pairs)
276
+
277
+
# Write the first generation into the Coordinators
278
+
# Once complete, any future recoveries will have to start from these TLogs
279
+
first_tlog_generation = %TLogGeneration{generation: state.cluster.generation, tlog_ids: tlog_ids}
280
+
{:ok, :ok} = Coordinator.write_generation(state.primary_coordinator, first_tlog_generation.generation, first_tlog_generation.tlog_ids)
281
+
282
+
state = put_in(state.cluster.tlog_generations, [first_tlog_generation])
264
283
265
284
put_in(state.cluster.status, :normal)
266
285
end
···
286
305
287
306
dbg {max_kcv, recovery_version, meta_pairs}
288
307
289
-
counts = %{
290
-
num_tlogs: state.config.num_tlogs,
291
-
num_commit_buffers: state.config.num_commit_buffers,
292
-
}
293
-
294
-
state = recruit_transaction_generation(state, counts, meta_pairs)
295
-
296
-
put_in(state.cluster.status, :normal)
297
-
end
298
-
299
-
defp recruit_transaction_generation(%State{} = state, counts, meta_pairs) when is_map(counts) and is_list(meta_pairs) do
300
-
%{num_tlogs: num_tlogs, num_commit_buffers: num_commit_buffers} = counts
301
-
ids = allocate_server_ids(state, num_tlogs + num_commit_buffers + 3)
302
-
303
-
{tlog_ids, ids} = Enum.split(ids, num_tlogs)
304
-
{commit_buffer_ids, ids} = Enum.split(ids, num_commit_buffers)
308
+
ids = allocate_server_ids(state, state.config.num_tlogs + state.config.num_commit_buffers + 3)
309
+
{tlog_ids, ids} = Enum.split(ids, state.config.num_tlogs)
310
+
{commit_buffer_ids, ids} = Enum.split(ids, state.config.num_commit_buffers)
305
311
[sequencer_id, resolver_id, distributor_id] = ids
306
312
307
313
cluster = state.cluster
308
-
gen = cluster.generation
314
+
gen = state.cluster.generation
315
+
316
+
state = recruit(state, gen, Hobbes.Servers.TLog, tlog_ids, %{cluster: state.cluster, meta_pairs: meta_pairs})
317
+
318
+
# TODO: copy mutations to new tlogs here
309
319
310
320
state =
311
321
state
312
-
|> recruit(gen, Hobbes.Servers.TLog, tlog_ids, %{cluster: cluster, meta_pairs: meta_pairs})
313
322
|> recruit(gen, Hobbes.Servers.CommitBuffer, commit_buffer_ids, %{cluster: cluster, meta_pairs: meta_pairs})
314
323
|> recruit(gen, Hobbes.Servers.Sequencer, [sequencer_id], %{})
315
324
|> recruit(gen, Hobbes.Servers.Resolver, [resolver_id], %{})
316
325
|> recruit(gen, Hobbes.Servers.Distributor, [distributor_id], %{cluster: cluster})
317
326
318
-
# Write the new TLog generation into the coordinators
319
-
# Once this is complete, the effect of this recovery is permanent, and any future recoveries will
320
-
# have to start from these TLogs
321
-
#
322
-
# Note that tlog_ids is always sorted here (sliced from a monotonic list)
323
-
{:ok, :ok} = Coordinator.write_generation(state.primary_coordinator, state.cluster.generation, tlog_ids)
327
+
# Write the new TLog generation to the Coordinators
328
+
# 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
+
{:ok, :ok} = Coordinator.write_generation(state.primary_coordinator, new_tlog_generation.generation, new_tlog_generation.tlog_ids)
324
331
325
-
# Add the generation to the cluster
326
-
update_in(state.cluster.tlog_generations, fn generations when is_list(generations) ->
327
-
[%TLogGeneration{generation: state.cluster.generation, tlog_ids: tlog_ids} | generations]
332
+
state = update_in(state.cluster.tlog_generations, fn generations when is_list(generations) ->
333
+
[new_tlog_generation | generations]
328
334
end)
335
+
336
+
put_in(state.cluster.status, :normal)
329
337
end
330
338
331
339
defp recruit(%State{} = state, generation, module, ids, args) when is_atom(module) and is_list(ids) and is_map(args) do