+6

lib/dense_shard_map.ex

reviewed

··· 19 19 :ets.new(__MODULE__, [:ordered_set, :private]) 20 20 end 21 21 22 22 + @spec destroy(t) :: :ok 23 23 + def destroy(table) do 24 24 + :ets.delete(table) 25 25 + :ok 26 26 + end 27 27 + 22 28 @spec put(t, binary, term) :: :ok 23 29 def put(table, start_key, value) when is_database_key(start_key) do 24 30 :ets.insert(table, {start_key, value})

+39 -48

lib/servers/manager.ex

reviewed

··· 3 3 4 4 import ExUnit.Assertions, only: [assert: 1] 5 5 6 6 - alias Hobbes.MetaStore 6 6 + alias Hobbes.{MetaStore, ShardTagMap} 7 7 alias Hobbes.Structs.{Cluster, TLogGeneration, Server, TLogStatus, LogBatch} 8 8 alias Hobbes.Servers.{Coordinator, ServerSupervisor, TLog} 9 9 ··· 282 282 |> recruit(gen, Hobbes.Servers.Resolver, [resolver_id], %{cluster: cluster, prev_version: 1}) 283 283 |> recruit(gen, Hobbes.Servers.Distributor, [distributor_id], %{cluster: cluster}) 284 284 285 285 - seed_meta_store(state, meta_pairs) 285 285 + seed_meta_store(state, first_tlog_generation, meta_pairs) 286 286 287 287 # Write the first generation into the Coordinators 288 288 # Once complete, any future recoveries will have to start from these TLogs ··· 334 334 state = recruit(state, gen, Hobbes.Servers.TLog, tlog_ids, %{cluster: state.cluster, meta_pairs: meta_pairs}) 335 335 336 336 # Copy mutations in range [max_kcv + 1, min_dv] to the new TLogs 337 337 - {state, last_batch_version} = copy_mutations_to_new_generation(state, prev_tlog_generation, new_tlog_generation, meta_pairs, new_tlog_generation.start_version, min_dv) 337 337 + shard_map = ShardTagMap.new(new_tlog_generation) 338 338 + ShardTagMap.load_meta_pairs(shard_map, meta_pairs) 339 339 + 340 340 + {state, last_batch_version} = copy_mutations_to_new_generation(state, prev_tlog_generation, new_tlog_generation, meta_pairs, shard_map, new_tlog_generation.start_version, min_dv) 338 341 if last_batch_version != nil, do: assert last_batch_version > max_kcv and last_batch_version <= min_dv 339 342 340 343 # If there are no batches on the TLogs in the range (max_kcv, min_dv], ··· 352 355 recovery_commit_version = max(new_tlog_generation.start_version, last_batch_version + 1) 353 356 state = write_recovery_commit(state, new_tlog_generation, last_batch_version, recovery_commit_version) 354 357 #dbg {last_batch_version, recovery_commit_version} 358 358 + 359 359 + # TODO: CommitBuffers do not currently receive the meta mutations which were copied, which will break correctness 355 360 356 361 # Recruit the rest of the new transaction system 357 362 state = ··· 379 384 |> Enum.sort() 380 385 end 381 386 382 382 - defp copy_mutations_to_new_generation(%State{} = state, %TLogGeneration{} = old_gen, %TLogGeneration{} = new_gen, meta_pairs, start_version, end_version) do 387 387 + defp copy_mutations_to_new_generation(%State{} = state, %TLogGeneration{} = old_gen, %TLogGeneration{} = new_gen, meta_pairs, %ShardTagMap{} = shard_map, start_version, end_version) do 383 388 tags = extract_tags(meta_pairs) 384 389 385 390 recovered_old_tlog_pids = ··· 409 414 end) 410 415 |> merge_batches() 411 416 |> then(fn batches -> 417 417 + # We lie about the first prev_version being 0 since that's what new TLogs expect 412 418 [{0, []} | batches] 413 419 end) 414 420 |> Enum.chunk_every(2, 1, :discard) ··· 420 426 } 421 427 end) 422 428 423 423 - meta_store = MetaStore.new() 424 424 - MetaStore.load(meta_store, meta_pairs) 425 425 - 426 429 Enum.each(batches, fn batch -> 427 427 - write_batch_to_tlogs(state.cluster, new_gen, batch, meta_store) 430 430 + write_batch_to_tlogs(state.cluster, new_gen, batch, shard_map) 428 431 end) 429 432 430 433 last_batch_version = ··· 436 439 {state, last_batch_version} 437 440 end 438 441 439 439 - defp write_batch_to_tlogs(%Cluster{} = cluster, %TLogGeneration{} = tlog_gen, batch, meta_store) when is_map(batch) do 440 440 - %{prev_version: prev_version, commit_version: commit_version, mutations: mutations} = batch 442 442 + defp write_batch_to_tlogs(%Cluster{} = cluster, %TLogGeneration{} = tlog_gen, batch, %ShardTagMap{} = shard_map) when is_map(batch) do 443 443 + %{prev_version: prev_version, commit_version: commit_version} = batch 441 444 442 442 - # Apply meta mutations from the batch first 443 443 - meta_mutations = 444 444 - mutations 445 445 - |> Enum.filter(fn {_i, mut} -> meta_mutation?(mut) end) 446 446 - |> Enum.map(fn {_i, mut} -> mut end) 447 447 - MetaStore.apply_meta_mutations(meta_store, commit_version, meta_mutations) 445 445 + # Unwrap mutations 446 446 + mutations = Enum.map(batch.mutations, fn {_i, mut} -> mut end) 448 447 449 449 - tagged_mutations = 450 450 - mutations 451 451 - |> Enum.map(fn {_i, mut} = num_mut -> 452 452 - tags = MetaStore.get_key_server_mutation_tags(meta_store, commit_version, mutation_key(mut)) 453 453 - {tags, num_mut} 454 454 - end) 455 455 - 456 456 - sliced_mutations = slice_mutations_for_tlogs(tagged_mutations, tlog_gen.tlog_ids, 3) 448 448 + ShardTagMap.apply_metadata_mutations(shard_map, Enum.filter(mutations, &meta_mutation?/1)) 449 449 + sliced_mutations = ShardTagMap.tag_and_slice_mutations(shard_map, mutations) 457 450 458 451 tlog_gen.tlog_ids 459 452 |> Enum.map(fn tlog_id -> ··· 513 506 end) 514 507 end 515 508 516 516 - defp seed_meta_store(%State{} = state, meta_pairs) when is_list(meta_pairs) do 517 517 - # Create temporary MetaStore 518 518 - # TODO: destroy after 519 519 - meta_store = MetaStore.new() 520 520 - MetaStore.apply_meta_mutations(meta_store, 0, Enum.map(meta_pairs, fn {k, v} -> {:write, k, v} end)) 509 509 + defp seed_meta_store(%State{} = state, %TLogGeneration{} = tlog_gen, meta_pairs) when is_list(meta_pairs) do 510 510 + shard_map = ShardTagMap.new(tlog_gen) 511 511 + ShardTagMap.load_meta_pairs(shard_map, meta_pairs) 521 512 522 522 - tagged_mutations = 513 513 + sliced_mutations = 523 514 meta_pairs 524 515 |> Enum.map(fn {k, v} -> {:write, k, v} end) 525 516 |> then(fn mutations -> 526 517 mutations ++ compute_special_mutations(mutations) 527 518 end) 528 528 - |> Enum.with_index() 529 529 - |> Enum.map(fn {{:write, k, v}, i} -> 530 530 - # Note: we remove the meta tag because TLogs are already directly seeded with these pairs 531 531 - tags = MetaStore.get_key_server_mutation_tags(meta_store, 0, k) |> List.delete(meta_tag()) 532 532 - {tags, {i, {:write, k, v}}} 533 533 - end) 519 519 + |> then(&ShardTagMap.tag_and_slice_mutations(shard_map, &1)) 534 520 535 535 - batch = %LogBatch{ 536 536 - commit_buffer_id: nil, 537 537 - commit_version: 1, 538 538 - prev_commit_version: 0, 539 539 - tagged_mutations: tagged_mutations, 540 540 - last_committed_version: 0, 541 541 - } 521 521 + ShardTagMap.destroy(shard_map) 542 522 543 523 # Seed the meta mutations directly into the TLogs with a manual "transaction" at version=1 544 524 # 545 545 - # Send to all tlogs for simplicity (instead of choosing the correct ones) 546 546 - # Storage servers pop from all tlogs anyway 547 547 - get_servers(state.cluster, Hobbes.Servers.TLog) 548 548 - |> Enum.map(fn %Server{pid: pid} -> 549 549 - TLog.write_batch_send(pid, batch) 525 525 + # Note: Meta pairs will be sent to TLogs again even though they were already seeded at init 526 526 + # (this should be fine due to idempotence) 527 527 + tlog_gen.tlog_ids 528 528 + |> Enum.map(fn tlog_id -> 529 529 + batch = %LogBatch{ 530 530 + commit_buffer_id: nil, 531 531 + commit_version: 1, 532 532 + prev_commit_version: 0, 533 533 + tagged_mutations: Map.fetch!(sliced_mutations, tlog_id), 534 534 + last_committed_version: 0, 535 535 + } 536 536 + 537 537 + %Server{pid: tlog_pid} = Map.fetch!(state.cluster.servers, tlog_id) 538 538 + TLog.write_batch_send(tlog_pid, batch) 550 539 end) 551 540 |> Enum.each(fn req_id -> 552 541 :ok = TLog.write_batch_receive(req_id) 553 542 end) 543 543 + 544 544 + :ok 554 545 end 555 546 556 547 @spec allocate_server_ids(State.t, non_neg_integer) :: [non_neg_integer]

+18

lib/shard_tag_map.ex

reviewed

··· 23 23 } 24 24 end 25 25 26 26 + @spec destroy(t) :: :ok 27 27 + def destroy(%ShardTagMap{} = stm) do 28 28 + DenseShardMap.destroy(stm.shard_map) 29 29 + :ok 30 30 + end 31 31 + 32 32 + @spec load_meta_pairs(t, [{binary, binary}]) :: :ok 33 33 + def load_meta_pairs(%ShardTagMap{} = stm, pairs) when is_list(pairs) do 34 34 + # We could skip wrapping the mutations and load them directly but this is simpler 35 35 + # (this function is only called on the recovery path) 36 36 + apply_metadata_mutations(stm, Enum.map(pairs, fn {k, v} -> {:write, k, v} end)) 37 37 + end 38 38 + 26 39 @spec apply_metadata_mutations(t, [Utils.mutation]) :: :ok 27 40 def apply_metadata_mutations(%ShardTagMap{} = stm, mutations) when is_list(mutations) do 28 41 %{shard_map: shard_map, tlog_ids: tlog_ids, replication_factor: rf} = stm ··· 116 129 117 130 false -> ids 118 131 end 132 132 + end 133 133 + 134 134 + @doc false 135 135 + def dump(%ShardTagMap{} = stm) do 136 136 + DenseShardMap.dump(stm.shard_map) 119 137 end 120 138 end