Skip to content

Commit 90f8d6a

Browse files
ckampfeckampfe
committed
ad sync_strategy and fsync all merge write files
1 parent 1b1b167 commit 90f8d6a

3 files changed

Lines changed: 144 additions & 113 deletions

File tree

‎lib/b4.ex‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -8,8 +8,14 @@ defmodule B4 do
88
- `target_file_size`: when the size of the current write file reaches this size,
99
the system will attempt to close it and start a fresh one.
1010
Defaults to `2 ** 31` bytes (2 GiB).
11+
- `sync_strategy`: when to flush the operating system write buffers to disk.
12+
If `:every_write`, an fsync is performed after every single insert and delete.
13+
This is slower, but safer in the event of e.g., power loss, whole system failure, etc.
14+
If any other value, or unset, allow the operating system to flush when it wants to,
15+
which is faster but not resilient to power loss, system failure, etc.
16+
Defaults to `:every_write`
1117
"""
12-
def new(directory, options \\ [target_file_size: 2 ** 31]) do
18+
def new(directory, options \\ [target_file_size: 2 ** 31, sync_strategy: :every_write]) do
1319
case DatabasesSupervisor.start_database(directory, options) do
1420
{:ok, _pid} -> :ok
1521
other -> other

‎lib/b4/keydir_owner.ex‎

Lines changed: 117 additions & 105 deletions
Original file line numberDiff line numberDiff line change
@@ -23,7 +23,7 @@ defmodule B4.KeydirOwner do
2323
end
2424

2525
@impl GenServer
26-
def init(%{directory: directory, options: [target_file_size: target_file_size]} = _init_arg) do
26+
def init(%{directory: directory, options: options} = _init_arg) do
2727
tid = Keydir.new()
2828

2929
:ok = :persistent_term.put({:b4_keydir_tid, directory}, tid)
@@ -34,7 +34,7 @@ defmodule B4.KeydirOwner do
3434
Files.apply_file_to_keydir(path, tid)
3535
end)
3636

37-
{:ok, %State{directory: directory, tid: tid, target_file_size: target_file_size}}
37+
{:ok, %State{directory: directory, tid: tid, target_file_size: options[:target_file_size]}}
3838
end
3939

4040
@impl GenServer
@@ -66,114 +66,126 @@ defmodule B4.KeydirOwner do
6666
{:ok, %{write_file: merge_write_file, file_id: merge_write_file_id}} =
6767
Writer.new_write_file(directory)
6868

69-
Enum.reduce(read_only_database_files, %{merge_file_ids: MapSet.new()}, fn path, outer_acc ->
70-
acc_for_file =
71-
path
72-
|> Files.stream_entries()
73-
|> Enum.reduce(
74-
%{
75-
merge_write_file: merge_write_file,
76-
merge_write_file_id: merge_write_file_id,
77-
merge_file_ids: MapSet.new(),
78-
merge_write_file_position: 0
79-
},
80-
fn %{
81-
entry: %{
82-
crc32: crc32,
83-
entry_id: on_disk_entry_id,
84-
key_size: key_size,
85-
value_size: value_size,
86-
key_bytes: key_bytes,
87-
value_bytes: value_bytes
69+
merge_state =
70+
Enum.reduce(read_only_database_files, %{merge_file_ids: MapSet.new()}, fn path, outer_acc ->
71+
acc_for_file =
72+
path
73+
|> Files.stream_entries()
74+
|> Enum.reduce(
75+
%{
76+
merge_write_file: merge_write_file,
77+
merge_write_file_id: merge_write_file_id,
78+
merge_file_ids: MapSet.new(),
79+
merge_write_file_position: 0
80+
},
81+
fn %{
82+
entry: %{
83+
crc32: crc32,
84+
entry_id: on_disk_entry_id,
85+
key_size: key_size,
86+
value_size: value_size,
87+
key_bytes: key_bytes,
88+
value_bytes: value_bytes
89+
},
90+
meta: %{}
8891
},
89-
meta: %{}
90-
},
91-
%{
92-
merge_write_file: merge_write_file,
93-
merge_write_file_id: merge_write_file_id,
94-
merge_write_file_position: merge_write_file_position
95-
} =
96-
acc ->
97-
key = :erlang.binary_to_term(key_bytes)
98-
99-
case Keydir.fetch(tid, key) do
100-
{:ok, {_key, _file_id, _entry_size, _file_position, keydir_entry_id}}
101-
when keydir_entry_id == on_disk_entry_id ->
102-
{:ok,
103-
%{
104-
merge_write_file: merge_write_file,
105-
merge_write_file_id: merge_write_file_id,
106-
merge_write_file_position: merge_write_file_position
107-
}} =
108-
if merge_write_file_position >= target_file_size do
109-
{:ok, %{write_file: merge_write_file, file_id: merge_write_file_id}} =
110-
Writer.new_write_file(directory)
111-
112-
{:ok,
113-
%{
114-
merge_write_file: merge_write_file,
115-
merge_write_file_id: merge_write_file_id,
116-
merge_write_file_position: 0
117-
}}
118-
else
119-
{:ok,
120-
%{
121-
merge_write_file: merge_write_file,
122-
merge_write_file_id: merge_write_file_id,
123-
merge_write_file_position: merge_write_file_position
124-
}}
125-
end
126-
127-
entry =
128-
[
129-
Writer.int_to_u32_bytes(crc32),
130-
Writer.int_to_u128_bytes(on_disk_entry_id),
131-
Writer.int_to_u32_bytes(key_size),
132-
Writer.int_to_u32_bytes(value_size),
133-
key_bytes,
134-
value_bytes
135-
]
136-
137-
:ok = :file.write(merge_write_file, entry)
138-
139-
entry_size = :erlang.iolist_size(entry)
140-
141-
true =
142-
Keydir.insert(
143-
tid,
144-
key,
145-
merge_write_file_id,
146-
entry_size,
147-
merge_write_file_position,
148-
on_disk_entry_id
149-
)
150-
151-
%{
92+
%{
93+
merge_write_file: merge_write_file,
94+
merge_write_file_id: merge_write_file_id,
95+
merge_write_file_position: merge_write_file_position
96+
} =
97+
acc ->
98+
key = :erlang.binary_to_term(key_bytes)
99+
100+
case Keydir.fetch(tid, key) do
101+
{:ok, {_key, _file_id, _entry_size, _file_position, keydir_entry_id}}
102+
when keydir_entry_id == on_disk_entry_id ->
103+
{:ok,
104+
%{
105+
merge_write_file: merge_write_file,
106+
merge_write_file_id: merge_write_file_id,
107+
merge_write_file_position: merge_write_file_position
108+
}} =
109+
if merge_write_file_position >= target_file_size do
110+
:file.close(merge_write_file)
111+
112+
{:ok, %{write_file: merge_write_file, file_id: merge_write_file_id}} =
113+
Writer.new_write_file(directory)
114+
115+
{:ok,
116+
%{
117+
merge_write_file: merge_write_file,
118+
merge_write_file_id: merge_write_file_id,
119+
merge_write_file_position: 0
120+
}}
121+
else
122+
{:ok,
123+
%{
124+
merge_write_file: merge_write_file,
125+
merge_write_file_id: merge_write_file_id,
126+
merge_write_file_position: merge_write_file_position
127+
}}
128+
end
129+
130+
entry =
131+
[
132+
Writer.int_to_u32_bytes(crc32),
133+
Writer.int_to_u128_bytes(on_disk_entry_id),
134+
Writer.int_to_u32_bytes(key_size),
135+
Writer.int_to_u32_bytes(value_size),
136+
key_bytes,
137+
value_bytes
138+
]
139+
140+
# TODO should we be fsyncing here?
141+
:ok = :file.write(merge_write_file, entry)
142+
143+
entry_size = :erlang.iolist_size(entry)
144+
145+
true =
146+
Keydir.insert(
147+
tid,
148+
key,
149+
merge_write_file_id,
150+
entry_size,
151+
merge_write_file_position,
152+
on_disk_entry_id
153+
)
154+
155+
%{
156+
acc
157+
| merge_write_file: merge_write_file,
158+
merge_write_file_id: merge_write_file_id,
159+
merge_write_file_position: acc.merge_write_file_position + entry_size,
160+
merge_file_ids: MapSet.put(acc.merge_file_ids, merge_write_file_id)
161+
}
162+
163+
# the entry isn't in the keydir,
164+
# so it isn't live anymore,
165+
# so skip it
166+
:error ->
152167
acc
153-
| merge_write_file: merge_write_file,
154-
merge_write_file_id: merge_write_file_id,
155-
merge_write_file_position: acc.merge_write_file_position + entry_size,
156-
merge_file_ids: MapSet.put(acc.merge_file_ids, merge_write_file_id)
157-
}
158-
159-
# the entry isn't in the keydir,
160-
# so it isn't live anymore,
161-
# so skip it
162-
:error ->
163-
acc
164-
165-
# the ids for the given key didn't match,
166-
# so they are old version of that key,
167-
# so we ignore them
168-
_ ->
169-
acc
168+
169+
# the ids for the given key didn't match,
170+
# so they are old version of that key,
171+
# so we ignore them
172+
_ ->
173+
acc
174+
end
170175
end
171-
end
172-
)
176+
)
173177

174-
Map.update!(outer_acc, :merge_file_ids, fn merge_file_ids ->
175-
MapSet.union(merge_file_ids, acc_for_file[:merge_file_ids])
178+
Map.update!(outer_acc, :merge_file_ids, fn merge_file_ids ->
179+
MapSet.union(merge_file_ids, acc_for_file[:merge_file_ids])
180+
end)
176181
end)
182+
183+
# fsync all merge files
184+
Enum.each(merge_state[:merge_file_ids], fn merge_file_id ->
185+
merge_file_path = Path.join([directory, "#{merge_file_id}.b4"])
186+
{:ok, f} = :file.open(merge_file_path, [:raw, :binary, :read])
187+
:ok = :file.sync(f)
188+
:ok = :file.close(f)
177189
end)
178190

179191
# |> IO.inspect(label: "merge file acc state")

‎lib/b4/writer.ex‎

Lines changed: 20 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,8 @@ defmodule B4.Writer do
1212
:file_id,
1313
:file_position,
1414
:target_file_size,
15-
:merge_in_progress?
15+
:merge_in_progress?,
16+
:sync_strategy
1617
]
1718
defstruct [
1819
:directory,
@@ -21,7 +22,8 @@ defmodule B4.Writer do
2122
:file_id,
2223
:file_position,
2324
:target_file_size,
24-
:merge_in_progress?
25+
:merge_in_progress?,
26+
:sync_strategy
2527
]
2628
end
2729

@@ -46,7 +48,7 @@ defmodule B4.Writer do
4648
end
4749

4850
@impl GenServer
49-
def init(%{directory: directory, options: [target_file_size: target_file_size]}) do
51+
def init(%{directory: directory, options: options}) do
5052
{:ok, %{write_file: write_file, file_id: file_id}} = new_write_file(directory)
5153

5254
tid = KeydirOwner.get_keydir_tid(directory)
@@ -58,8 +60,9 @@ defmodule B4.Writer do
5860
write_file: write_file,
5961
file_id: file_id,
6062
file_position: 0,
61-
target_file_size: target_file_size,
62-
merge_in_progress?: false
63+
target_file_size: options[:target_file_size],
64+
merge_in_progress?: false,
65+
sync_strategy: options[:sync_strategy]
6366
}}
6467
end
6568

@@ -74,7 +77,8 @@ defmodule B4.Writer do
7477
file_id: file_id,
7578
file_position: file_position,
7679
target_file_size: target_file_size,
77-
merge_in_progress?: merge_in_progress?
80+
merge_in_progress?: merge_in_progress?,
81+
sync_strategy: sync_strategy
7882
} =
7983
state
8084
) do
@@ -105,6 +109,10 @@ defmodule B4.Writer do
105109

106110
:ok = :file.write(write_file, entry)
107111

112+
if sync_strategy == :every_write do
113+
:ok = :file.sync(write_file)
114+
end
115+
108116
entry_size = :erlang.iolist_size(entry)
109117

110118
true = Keydir.insert(tid, key, file_id, entry_size, file_position, entry_id)
@@ -123,7 +131,8 @@ defmodule B4.Writer do
123131
file_position: file_position,
124132
file_id: file_id,
125133
target_file_size: target_file_size,
126-
merge_in_progress?: merge_in_progress?
134+
merge_in_progress?: merge_in_progress?,
135+
sync_strategy: sync_strategy
127136
} =
128137
state
129138
) do
@@ -154,6 +163,10 @@ defmodule B4.Writer do
154163

155164
:ok = :file.write(write_file, entry)
156165

166+
if sync_strategy == :every_write do
167+
:ok = :file.sync(write_file)
168+
end
169+
157170
entry_size = :erlang.iolist_size(entry)
158171

159172
Keydir.delete(tid, key)

0 commit comments

Comments
 (0)