this repo has no description
Buffered write
+107
-14
+11
-8
lib/btree/btree.ex
+11
-8
lib/btree/btree.ex
···
19
19
page_store: :ets.table,
20
20
free_list: FreeList.t,
21
21
root_store: :ets.table,
22
+
write_buffer: :ets.table,
22
23
}
23
24
@enforce_keys [
24
25
:opts,
25
26
:page_store,
26
27
:free_list,
27
28
:root_store,
29
+
:write_buffer,
28
30
]
29
31
defstruct @enforce_keys
30
32
···
45
47
page_store: :ets.new(__MODULE__, [:set, :protected]),
46
48
free_list: FreeList.new(),
47
49
root_store: :ets.new(__MODULE__, [:ordered_set, :protected]),
50
+
write_buffer: :ets.new(__MODULE__, [:ordered_set, :private]),
48
51
}
49
52
:ok = create_root_pages(btree)
50
53
···
69
72
leaf_checksum = checksum(leaf_data)
70
73
:ok = write_page(page_store, leaf_index, leaf_data)
71
74
72
-
leaf_start_key = ""
75
+
leaf_end_key = "\xFF\xFF\xFF\xFF"
73
76
leaf_address = <<leaf_index::integer-64, leaf_checksum::binary-16>>
74
77
75
-
root_pairs = [{leaf_start_key, leaf_address}]
78
+
root_pairs = [{leaf_end_key, leaf_address}]
76
79
root_data = Page.encode_page_from_pairs(:inner, opt_page_size, root_pairs)
77
80
root_checksum = checksum(root_data)
78
81
:ok = write_page(page_store, root_index, root_data)
···
81
84
:ok
82
85
end
83
86
84
-
defp checksum(data) do
87
+
def checksum(data) do
85
88
<<hash::binary-16, _rest::binary-16>> = :crypto.hash(:sha256, data)
86
89
hash
87
90
end
88
91
89
-
defp write_page(page_store, page_index, data) do
92
+
def write_page(page_store, page_index, data) do
90
93
:ets.insert(page_store, {page_index, data})
91
94
:ok
92
95
end
93
96
94
-
defp read_page(page_store, page_index, page_checksum) do
97
+
def read_page(page_store, page_index, page_checksum) do
95
98
case :ets.lookup(page_store, page_index) do
96
99
[{_index, data}] ->
97
100
assert checksum(data) == page_checksum
···
150
153
{new_page_index, new_page_checksum}
151
154
end
152
155
153
-
defp write_pair(pairs, key, value, acc \\ [])
154
-
defp write_pair([], key, value, acc), do: Enum.reverse(acc, [{key, value}])
155
-
defp write_pair([{k, _v} = p | pairs_rest], key, value, acc) do
156
+
def write_pair(pairs, key, value, acc \\ [])
157
+
def write_pair([], key, value, acc), do: Enum.reverse(acc, [{key, value}])
158
+
def write_pair([{k, _v} = p | pairs_rest], key, value, acc) do
156
159
cond do
157
160
k < key -> write_pair(pairs_rest, key, value, [p | acc])
158
161
k == key -> Enum.reverse([{key, value} | acc], pairs_rest)
+84
lib/btree/writer.ex
+84
lib/btree/writer.ex
···
1
+
defmodule Hobbes.BTree.Writer do
2
+
alias Hobbes.BTree
3
+
alias Hobbes.BTree.{FreeList, Page}
4
+
5
+
@spec apply_batch(BTree.t, list) :: :ok
6
+
def apply_batch(%BTree{} = btree, mutations) do
7
+
%{
8
+
write_buffer: write_buffer,
9
+
root_store: root_store,
10
+
} = btree
11
+
12
+
:ok = do_write_buffer(mutations, write_buffer)
13
+
14
+
[{:root_address, {root_index, root_checksum}}] = :ets.lookup(root_store, :root_address)
15
+
16
+
[{_, <<new_ri::integer-64, new_rch::binary-16>>}] = do_flush(btree, root_index, root_checksum, "", "\xFF\xFF\xFF\xFF")
17
+
18
+
:ets.insert(root_store, {:root_address, {new_ri, new_rch}})
19
+
:ok
20
+
end
21
+
22
+
defp do_write_buffer([], _write_buffer), do: :ok
23
+
defp do_write_buffer([mut | mutations_rest], write_buffer) do
24
+
case mut do
25
+
{:write, k, v} -> :ets.insert(write_buffer, {k, v})
26
+
{:clear, k} -> :ets.insert(write_buffer, {k, :tombstone})
27
+
end
28
+
do_write_buffer(mutations_rest, write_buffer)
29
+
end
30
+
31
+
defp do_flush(%BTree{} = btree, page_index, page_checksum, page_sk, page_ek) do
32
+
{page_type, pairs} =
33
+
BTree.read_page(btree.page_store, page_index, page_checksum)
34
+
|> Page.decode_page()
35
+
36
+
pairs =
37
+
case page_type do
38
+
:inner -> fill_inner(btree, pairs, page_sk, page_ek, [])
39
+
:leaf -> fill_leaf(btree.write_buffer, page_ek, pairs)
40
+
end
41
+
42
+
new_page_data = Page.encode_page_from_pairs(page_type, btree.opts.page_size, pairs)
43
+
[new_page_index] = FreeList.reserve_pages(btree.free_list, 1)
44
+
new_page_checksum = BTree.checksum(new_page_data)
45
+
:ok = BTree.write_page(btree.page_store, new_page_index, new_page_data)
46
+
47
+
[{page_ek, <<new_page_index::integer-64, new_page_checksum::binary-16>>}]
48
+
end
49
+
50
+
defp fill_inner(%BTree{} = btree, pairs, page_sk, page_ek, writes_acc) do
51
+
case :ets.first(btree.write_buffer) do
52
+
key when is_binary(key) and key < page_ek ->
53
+
{child_sk, child_ek, child_address} = find_child(pairs, key, page_sk)
54
+
<<child_index::integer-64, child_checksum::binary-16>> = child_address
55
+
56
+
writes = do_flush(btree, child_index, child_checksum, child_sk, child_ek)
57
+
writes_acc = Enum.reverse(writes, writes_acc)
58
+
# TODO: free(child_index)
59
+
fill_inner(btree, pairs, page_sk, page_ek, writes_acc)
60
+
61
+
_ ->
62
+
Enum.reduce(writes_acc, pairs, fn {k, v}, acc -> BTree.write_pair(acc, k, v) end)
63
+
end
64
+
end
65
+
66
+
defp find_child([{k, v} | pairs_rest], search_key, last_ek) do
67
+
case search_key < k do
68
+
true -> {last_ek, k, v}
69
+
false -> find_child(pairs_rest, search_key, k)
70
+
end
71
+
end
72
+
73
+
defp fill_leaf(write_buffer, page_ek, pairs_acc) do
74
+
case :ets.first_lookup(write_buffer) do
75
+
{_key, [{key, value}]} when key < page_ek ->
76
+
:ets.delete(write_buffer, key)
77
+
78
+
pairs_acc = BTree.write_pair(pairs_acc, key, value)
79
+
fill_leaf(write_buffer, page_ek, pairs_acc)
80
+
81
+
_ -> pairs_acc
82
+
end
83
+
end
84
+
end
+12
-6
test/btree_test.exs
+12
-6
test/btree_test.exs
···
7
7
8
8
describe "BTree" do
9
9
test "insert" do
10
-
mutations = [
11
-
{:write, "foo", "bar"},
12
-
{:write, "hello", "world"},
13
-
]
10
+
btree = BTree.new()
14
11
15
-
btree = BTree.new()
16
-
:ok = BTree.apply_batch(btree, mutations)
12
+
:ok = BTree.Writer.apply_batch(btree, [
13
+
{:write, "a", "1"},
14
+
{:write, "b", "2"},
15
+
{:write, "c", "3"},
16
+
])
17
+
18
+
:ok = BTree.Writer.apply_batch(btree, [
19
+
{:write, "d", "4"},
20
+
{:write, "e", "5"},
21
+
{:write, "f", "6"},
22
+
])
17
23
end
18
24
end
19
25
end