this repo has no description
Add extent functions to Blocks
+51
-28
+1
-18
lib/servers/tlog.ex
+1
-18
lib/servers/tlog.ex
···
467
467
468
468
false ->
469
469
position = size
470
-
write_group(xks, extent, position, group_data)
470
+
XKS.Blocks.write_extent(xks.block_store, extent, position, group_data)
471
471
472
472
state = put_in(state.current_extent.size, size + group_size)
473
473
do_append_groups(groups_rest, state)
474
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
475
end
493
476
494
477
defp group_mutations(tagged_mutations) do
+43
-4
lib/xks/blocks.ex
+43
-4
lib/xks/blocks.ex
···
42
42
checksum
43
43
end
44
44
45
-
def checksum(block_data) when is_binary(block_data) do
46
-
compute_checksum(block_data)
47
-
end
48
-
49
45
defp do_write({:memory, memory_store}, index, block_data) do
46
+
assert byte_size(block_data) == MemoryStore.get_block_size(memory_store)
50
47
MemoryStore.put(memory_store, index, block_data)
51
48
end
52
49
···
87
84
end
88
85
end
89
86
87
+
@spec checksum(binary) :: checksum
88
+
def checksum(block_data) when is_binary(block_data) do
89
+
compute_checksum(block_data)
90
+
end
91
+
90
92
defp compute_checksum(binary) do
91
93
<<checksum::binary-16, _rest::binary>> = :crypto.hash(@hash_function, binary)
92
94
checksum
95
+
end
96
+
97
+
@spec write_extent(XKS.block_store, non_neg_integer, non_neg_integer, binary) :: :ok
98
+
def write_extent(block_store, extent_index, position, data) do
99
+
do_write_extent(block_store, extent_index, position, data)
100
+
:ok
101
+
end
102
+
103
+
defp do_write_extent({:memory, memory_store}, extent_index, position, data) do
104
+
existing = case MemoryStore.fetch(memory_store, extent_index) do
105
+
{:ok, existing} -> existing
106
+
:error -> ""
107
+
end
108
+
# Only appends are supported right now
109
+
assert byte_size(existing) == position
110
+
111
+
MemoryStore.put(memory_store, extent_index, existing <> data)
112
+
end
113
+
114
+
@spec read_extent(XKS.block_store, non_neg_integer, non_neg_integer, non_neg_integer) :: binary
115
+
def read_extent(block_store, extent_index, position, size) do
116
+
do_read_extent(block_store, extent_index, position, size)
117
+
end
118
+
119
+
defp do_read_extent({:memory, memory_store}, extent_index, position, size) do
120
+
data = case MemoryStore.fetch(memory_store, extent_index) do
121
+
{:ok, data} -> data
122
+
:error -> ""
123
+
end
124
+
125
+
<<
126
+
_prefix::binary-size(position),
127
+
read_data::binary-size(size),
128
+
_rest::binary,
129
+
>> = data
130
+
131
+
read_data
93
132
end
94
133
end
+7
-6
lib/xks/memory_store.ex
+7
-6
lib/xks/memory_store.ex
···
1
1
defmodule Hobbes.XKS.MemoryStore do
2
-
import ExUnit.Assertions, only: [assert: 1]
3
2
@type t :: :ets.table
4
3
5
4
@type new :: t
···
14
13
def destroy(store) do
15
14
:ets.delete(store)
16
15
:ok
16
+
end
17
+
18
+
@spec get_block_size(t) :: pos_integer
19
+
def get_block_size(store) do
20
+
[{:block_size, block_size}] = :ets.lookup(store, :block_size)
21
+
block_size
17
22
end
18
23
19
24
@spec put_superblock(t, non_neg_integer, binary) :: :ok
···
32
37
end
33
38
end
34
39
35
-
@spec put(t, non_neg_integer, binary) :: :ok
40
+
@spec put(t, non_neg_integer, binary) :: true
36
41
def put(store, index, block_data) when is_binary(block_data) do
37
-
[{:block_size, block_size}] = :ets.lookup(store, :block_size)
38
-
assert byte_size(block_data) == block_size
39
-
40
42
:ets.insert(store, {index, block_data})
41
-
:ok
42
43
end
43
44
44
45
@spec fetch(t, non_neg_integer) :: {:ok, binary} | :error