this repo has no description
Write log entries to extent
+68
-4
+60
-4
lib/servers/tlog.ex
+60
-4
lib/servers/tlog.ex
···
335
335
336
336
# Write mutations to storage
337
337
# TODO: do we need to write empty batches?
338
-
append_batch(state, batch)
338
+
state = append_batch(state, batch)
339
339
340
340
# Update versions
341
341
state = %{state |
···
441
441
XKS.commit(xks)
442
442
end
443
443
444
-
defp append_batch(%State{} = _state, %LogBatch{} = _batch) do
445
-
# TODO
444
+
defp append_batch(%State{} = state, %LogBatch{} = batch) do
445
+
# TODO: mutations should come from CommitBuffer grouped this way
446
+
groups = group_mutations(batch.tagged_mutations)
447
+
state = do_append_groups(groups, state)
448
+
state
449
+
end
450
+
451
+
defp do_append_groups([], %State{} = state), do: state
452
+
453
+
defp do_append_groups([group | groups_rest], %State{} = state) do
454
+
%{xks: xks, current_extent: current_extent} = state
455
+
%{extent: extent, size: size} = current_extent
456
+
extent_size = xks.opts.extent_block_count * xks.opts.block_size
457
+
458
+
group_data = :erlang.term_to_binary(group)
459
+
group_size = byte_size(group_data)
460
+
461
+
case size + group_size > extent_size do
462
+
true ->
463
+
# Rotate extent and start this iteration over
464
+
# TODO: this is not tested
465
+
state = rotate_extent(state)
466
+
do_append_groups([group | groups_rest], state)
467
+
468
+
false ->
469
+
position = size
470
+
write_group(xks, extent, position, group_data)
471
+
472
+
state = put_in(state.current_extent.size, size + group_size)
473
+
do_append_groups(groups_rest, state)
474
+
end
475
+
end
476
+
477
+
defp write_group(%XKS{} = xks, extent, position, group_data) do
478
+
# TODO: this is a hack to write to blocks, need a raw read/write API
479
+
block_index = extent + div(position, xks.opts.block_size)
480
+
block_pos = rem(position, xks.opts.block_size)
481
+
482
+
existing_block_data = XKS.Blocks.read_raw(xks.block_store, block_index)
483
+
484
+
<<
485
+
prefix::binary-size(block_pos),
486
+
_::binary-size(byte_size(group_data)),
487
+
suffix::binary,
488
+
>> = existing_block_data
489
+
490
+
new_block_data = <<prefix::binary, group_data::binary, suffix::binary>>
491
+
XKS.Blocks.write(xks.block_store, block_index, new_block_data)
492
+
end
493
+
494
+
defp group_mutations(tagged_mutations) do
495
+
Enum.reduce(tagged_mutations, %{}, fn {tags, mut}, acc ->
496
+
Enum.reduce(tags, acc, fn t, acc ->
497
+
Map.update(acc, t, [mut], &[mut | &1])
498
+
end)
499
+
end)
500
+
|> Enum.sort_by(&elem(&1, 0))
501
+
|> Enum.map(fn {k, v} -> {k, Enum.reverse(v)} end)
446
502
end
447
503
448
504
defp rotate_extent(%State{} = state) do
···
454
510
455
511
new_extent = %{
456
512
i: current_extent.i + 1,
457
-
extent: FreeList.reserve_extent(xks),
513
+
extent: FreeList.reserve_extent(xks.free_list),
458
514
size: 0,
459
515
}
460
516
+8
lib/xks/blocks.ex
+8
lib/xks/blocks.ex
···
72
72
end
73
73
end
74
74
75
+
def read_raw({:memory, memory_store}, index) do
76
+
[{:block_size, block_size}] = :ets.lookup(memory_store, :block_size)
77
+
case MemoryStore.fetch(memory_store, index) do
78
+
{:ok, block_data} -> block_data
79
+
:error -> <<0::integer-unit(8)-size(block_size)>>
80
+
end
81
+
end
82
+
75
83
defp do_read({:memory, memory_store}, index) do
76
84
# TODO: return zeros for missing blocks in memory store?
77
85
case MemoryStore.fetch(memory_store, index) do