this repo has no description
Refactor log batch write
+30
-26
+30
-26
lib/servers/tlog.ex
+30
-26
lib/servers/tlog.ex
···
574
574
end
575
575
576
576
defp append_batch(%State{} = state, %LogBatch{} = batch, known_committed_version) do
577
-
state =
577
+
groups =
578
578
case batch.tagged_mutations do
579
-
[] ->
580
-
blank_group = {-2, []}
581
-
write_log_entry(state, batch.commit_version, known_committed_version, blank_group)
582
-
groups ->
583
-
do_append_groups(groups, batch.commit_version, known_committed_version, state)
579
+
[] -> [{-2, []}]
580
+
groups -> groups
584
581
end
582
+
583
+
state = do_append_groups(groups, state, batch.commit_version, known_committed_version)
585
584
586
585
XKS.maybe_rotate_memtable(state.xks, @meta_partition)
587
586
XKS.maybe_rotate_memtable(state.xks, @queue_partition)
···
592
591
# group_size + commit_version + known_committed_version + prev_checksum + entry_checksum
593
592
defmacrop c_entry_overhead_bytes, do: 8 + 8 + 8 + 16 + 16
594
593
595
-
defp do_append_groups([], _commit_version, _known_committed_version, %State{} = state), do: state
596
-
597
-
defp do_append_groups([group | groups_rest], commit_version, known_committed_version, %State{} = state) do
598
-
state = write_log_entry(state, commit_version, known_committed_version, group)
599
-
do_append_groups(groups_rest, commit_version, known_committed_version, state)
594
+
defp do_append_groups([], %State{} = state, _commit_version, _known_committed_version) do
595
+
state
600
596
end
601
597
602
-
defp write_log_entry(%State{} = state, commit_version, known_committed_version, group) do
598
+
defp do_append_groups([group | groups_rest], %State{} = state, commit_version, known_committed_version) do
603
599
%{xks: xks} = state
604
600
opt_extent_size = xks.opts.extent_block_count * xks.opts.block_size
605
601
···
616
612
end
617
613
618
614
# Encode log entry
619
-
prev_checksum = state.prev_checksum
620
-
entry_data = <<
621
-
group_size::integer-64,
622
-
commit_version::integer-64,
623
-
known_committed_version::integer-64,
624
-
prev_checksum::binary-16,
625
-
group_data::binary,
626
-
>>
627
-
entry_checksum = XKS.Blocks.checksum(entry_data)
628
-
entry_data = <<
629
-
entry_data::binary,
630
-
entry_checksum::binary-16,
631
-
>>
615
+
{entry_data, entry_checksum} = encode_entry(group_data, state.prev_checksum, commit_version, known_committed_version)
632
616
assert byte_size(entry_data) == entry_size
633
617
assert entry_size <= opt_extent_size
634
618
···
650
634
:ok = maybe_write_meta(group, commit_version, xks)
651
635
end
652
636
653
-
%{state |
637
+
state = %{state |
654
638
current_extent: %{state.current_extent | size: position + entry_size},
655
639
prev_checksum: entry_checksum,
656
640
}
641
+
642
+
do_append_groups(groups_rest, state, commit_version, known_committed_version)
643
+
end
644
+
645
+
defp encode_entry(group_data, prev_checksum, commit_version, known_committed_version) do
646
+
group_size = byte_size(group_data)
647
+
entry_data = <<
648
+
group_size::integer-64,
649
+
commit_version::integer-64,
650
+
known_committed_version::integer-64,
651
+
prev_checksum::binary-16,
652
+
group_data::binary,
653
+
>>
654
+
entry_checksum = XKS.Blocks.checksum(entry_data)
655
+
656
+
entry_data = <<
657
+
entry_data::binary,
658
+
entry_checksum::binary-16,
659
+
>>
660
+
{entry_data, entry_checksum}
657
661
end
658
662
659
663
defp write_queue(%XKS{} = xks, tag, commit_version, extent, position, entry_size, checksum) do