this repo has no description
Apply partitioned batches in tlog append
+62
-38
+62
-38
lib/servers/tlog.ex
+62
-38
lib/servers/tlog.ex
···
573
573
end)
574
574
end
575
575
576
-
defp append_batch(%State{} = state, %LogBatch{} = batch, known_committed_version) do
576
+
defp append_batch(%State{} = state, %LogBatch{} = log_batch, known_committed_version) do
577
577
groups =
578
-
case batch.tagged_mutations do
578
+
case log_batch.tagged_mutations do
579
579
[] -> [{-2, []}]
580
580
groups -> groups
581
581
end
582
582
583
-
state = do_append_groups(groups, state, batch.commit_version, known_committed_version)
583
+
{state, partitioned_batches} = do_append_groups(groups, state, log_batch.commit_version, known_committed_version, [], [], [])
584
+
XKS.apply_partitioned_batches(state.xks, log_batch.commit_version, partitioned_batches)
584
585
585
586
XKS.maybe_rotate_memtable(state.xks, @meta_partition)
586
587
XKS.maybe_rotate_memtable(state.xks, @queue_partition)
···
591
592
# group_size + commit_version + known_committed_version + prev_checksum + entry_checksum
592
593
defmacrop c_entry_overhead_bytes, do: 8 + 8 + 8 + 16 + 16
593
594
594
-
defp do_append_groups([], %State{} = state, _commit_version, _known_committed_version) do
595
-
state
595
+
defp do_append_groups([], %State{} = state, _commit_version, _known_committed_version, state_acc, queue_acc, meta_acc) do
596
+
partitioned_batches = [
597
+
{@state_partition, Enum.reverse(state_acc)},
598
+
{@queue_partition, Enum.reverse(queue_acc)},
599
+
{@meta_partition, Enum.reverse(meta_acc)},
600
+
]
601
+
{state, partitioned_batches}
596
602
end
597
603
598
-
defp do_append_groups([group | groups_rest], %State{} = state, commit_version, known_committed_version) do
604
+
defp do_append_groups([group | groups_rest], %State{} = state, commit_version, known_committed_version, state_acc, queue_acc, meta_acc) do
599
605
%{xks: xks} = state
600
606
opt_extent_size = xks.opts.extent_block_count * xks.opts.block_size
601
607
602
-
{tag, _mutations} = group
608
+
{tag, group_mutations} = group
603
609
group_data = :erlang.term_to_binary(group)
604
610
group_size = byte_size(group_data)
605
611
entry_size = c_entry_overhead_bytes() + group_size
606
612
607
613
# Rotate extent if needed
608
-
state =
614
+
{state, state_acc} =
609
615
case (state.current_extent.size + entry_size) > opt_extent_size do
610
-
true -> rotate_extent(state, commit_version)
611
-
false -> state
616
+
true ->
617
+
{state, mut} = rotate_extent(state)
618
+
{state, [mut | state_acc]}
619
+
false -> {state, state_acc}
612
620
end
613
621
614
622
# Encode log entry
···
626
634
:ok = XKS.set_max_persist_version(xks, commit_version)
627
635
628
636
# Write queue entry and meta mutations, unless this is a blank log entry
629
-
case tag do
630
-
-2 ->
631
-
:noop
632
-
_ ->
633
-
:ok = write_queue(xks, tag, commit_version, extent_index, position, entry_size, entry_checksum)
634
-
:ok = maybe_write_meta(group, commit_version, xks)
635
-
end
637
+
{queue_acc, meta_acc} =
638
+
case tag do
639
+
-2 ->
640
+
# Blank group, don't write to queue
641
+
{queue_acc, meta_acc}
642
+
643
+
meta_tag() ->
644
+
{
645
+
[pack_queue_mutation(tag, commit_version, extent_index, position, entry_size, entry_checksum) | queue_acc],
646
+
Enum.reverse(group_mutations, meta_acc),
647
+
}
648
+
649
+
_ ->
650
+
{
651
+
[pack_queue_mutation(tag, commit_version, extent_index, position, entry_size, entry_checksum) | queue_acc],
652
+
meta_acc,
653
+
}
654
+
end
636
655
637
656
state = %{state |
638
657
current_extent: %{state.current_extent | size: position + entry_size},
639
658
prev_checksum: entry_checksum,
640
659
}
641
660
642
-
do_append_groups(groups_rest, state, commit_version, known_committed_version)
661
+
do_append_groups(groups_rest, state, commit_version, known_committed_version, state_acc, queue_acc, meta_acc)
643
662
end
644
663
645
664
defp encode_entry(group_data, prev_checksum, commit_version, known_committed_version) do
···
660
679
{entry_data, entry_checksum}
661
680
end
662
681
682
+
defp pack_queue_mutation(tag, commit_version, extent_index, position, entry_size, checksum) do
683
+
{
684
+
:write,
685
+
Keyset.pack([tag, commit_version]),
686
+
# TODO: we could use a fixed encoding for better performance
687
+
Keyset.pack([extent_index, position, entry_size, checksum]),
688
+
}
689
+
end
690
+
663
691
defp write_queue(%XKS{} = xks, tag, commit_version, extent, position, entry_size, checksum) do
664
-
XKS.apply_batch(xks, @queue_partition, commit_version, [
665
-
{
666
-
:write,
667
-
Keyset.pack([tag, commit_version]),
668
-
# TODO: this is a performance-critical path we can use a fixed encoding here instead of Keyset
669
-
Keyset.pack([extent, position, entry_size, checksum]),
670
-
},
671
-
])
692
+
XKS.apply_batch(xks, @queue_partition, commit_version, [pack_queue_mutation(tag, commit_version, extent, position, entry_size, checksum)])
672
693
:ok
673
694
end
674
695
···
845
866
state
846
867
end
847
868
848
-
defp rotate_extent(%State{} = state, version) do
869
+
defp rotate_extent(%State{} = state) do
849
870
%{
850
871
xks: xks,
851
872
current_extent: current_extent,
···
858
879
nonce: make_nonce(),
859
880
}
860
881
861
-
:ok = write_new_extent(xks, version, new_extent)
862
882
863
-
%{state |
883
+
state = %{state |
864
884
xks: xks,
865
885
current_extent: new_extent,
866
886
prev_checksum: new_extent.nonce,
867
887
# Mark dirty so that the new extent will be committed with this batch
868
888
extent_dirty?: true,
869
889
}
890
+
891
+
mutation = pack_extent_mutation(new_extent)
892
+
{state, mutation}
870
893
end
871
894
872
895
defp make_nonce do
···
875
898
<<nonce_i::integer-64, nonce_i::integer-64>>
876
899
end
877
900
878
-
defp write_new_extent(%XKS{} = xks, version, %{i: i, extent: extent, size: size, nonce: nonce}) do
879
-
XKS.apply_batch(xks, @state_partition, version, [
880
-
{
881
-
:write,
882
-
extents_prefix() <> Keyset.pack([i]),
883
-
Keyset.pack([extent, size, nonce]),
884
-
},
885
-
])
886
-
:ok
901
+
defp write_new_extent(%XKS{} = xks, version, extent) do
902
+
XKS.apply_batch(xks, @state_partition, version, [pack_extent_mutation(extent)])
903
+
end
904
+
905
+
defp pack_extent_mutation(%{i: i, extent: extent, size: size, nonce: nonce} = _extent) do
906
+
{
907
+
:write,
908
+
extents_prefix() <> Keyset.pack([i]),
909
+
Keyset.pack([extent, size, nonce]),
910
+
}
887
911
end
888
912
889
913
defp unpack_extent_pair({extents_prefix() <> key, value} = _pair) do