defmodule ExWal do
@moduledoc "README.md"
|> File.read!()
|> String.split("<!-- MDOC !-->")
|> Enum.fetch!(1)
use GenServer
alias ExWal.LRU
alias ExWal.Models.Block
alias ExWal.Models.Entry
alias ExWal.Models.Segment
alias ExWal.Store
alias ExWal.Uvarint
@external_resource "README.md"
@wal_option_schema [
path: [
type: :string,
doc: "The path of the WAL. Default is `./wal`.",
default: "./wal"
],
nosync: [
type: :boolean,
doc: "If true, WAL will not sync to disk. Default is false.",
default: false
],
segment_size: [
type: :integer,
doc: "The size of each segment file. Default is 16MB.",
default: 16 * 1024 * 1024
],
segment_cache_size: [
type: :integer,
doc: "The size of the segment cache. Default is 100.",
default: 2
],
store: [
type: :atom,
doc: "The store module which implement `ExWal.Store`. Default is `ExWal.Store.File`.",
default: Store.File
],
store_opts: [
type: :keyword_list,
doc: "The options for the store module.",
default: []
],
dir_permission: [
type: :integer,
doc: "The permission of the directory. Default is 0o750.",
default: 0o750
],
file_permission: [
type: :integer,
doc: "The permission of the file. Default is 0o640.",
default: 0o640
],
name: [
type: :atom,
doc: "The name of the WAL. Default is `wal`.",
default: __MODULE__
]
]
defstruct data_path: "",
hot: %Segment{},
cold: :array.new(),
first_index: 0,
last_index: 0,
lru_cache: nil,
store: nil,
tail_store_handler: nil,
opts: [
nosync: false,
segment_size: 64 * 1024 * 1024,
dir_permission: 0o750,
file_permission: 0o640
]
@type t :: %__MODULE__{
data_path: String.t(),
hot: Segment.t(),
cold: :array.array(),
first_index: non_neg_integer(),
last_index: non_neg_integer(),
lru_cache: atom(),
store: Store.t(),
tail_store_handler: Store.handler(),
opts: [
nosync: boolean(),
segment_size: non_neg_integer(),
dir_permission: non_neg_integer(),
file_permission: non_neg_integer()
]
}
@type index :: non_neg_integer()
@type wal_option_schema_t :: [unquote(NimbleOptions.option_typespec(@wal_option_schema))]
@doc """
Start a WAL process
## Options
#{NimbleOptions.docs(@wal_option_schema)}
"""
@spec start_link(wal_option_schema_t()) :: GenServer.on_start()
def start_link(opts) do
opts = NimbleOptions.validate!(opts, @wal_option_schema)
GenServer.start_link(__MODULE__, opts, name: opts[:name])
end
@doc """
Stop a WAL process
"""
@spec stop(atom() | pid()) :: :ok
def stop(name_or_pid) do
GenServer.stop(name_or_pid)
end
@doc """
Write entries to WAL, the entries must be strictly consecutive and incremental,
and the index of the first entry must be WAL's last_index + 1.
## Examples
iex> last_index = ExWal.last_index(:wal_name)
iex> enties = last_index..(last_index+10) |> Enum.map(fn i -> Entry.new(i, "some data") end)
iex> :ok = ExWal.write(:wal_name, entries)
"""
@spec write(atom() | pid(), [Entry.t()], non_neg_integer()) :: :ok
def write(name_or_pid, entries, timeout \\ 5000) do
GenServer.call(name_or_pid, {:write, entries}, timeout)
end
@doc """
Read entry content from WAL by index. if index is not found, return {:error, :index_not_found}
## Examples
iex> {:ok, data} = ExWal.read(:wal_name, 1)
"""
@spec read(atom() | pid(), index()) :: {:ok, binary()} | {:error, :index_not_found}
def read(name_or_pid, index, timeout \\ 5000) do
GenServer.call(name_or_pid, {:read, index}, timeout)
end
@doc """
Get the last index of WAL
"""
@spec last_index(atom() | pid()) :: index()
def last_index(name_or_pid) do
GenServer.call(name_or_pid, :last_index)
end
@doc """
Get the first index of WAL
"""
@spec first_index(atom() | pid()) :: index()
def first_index(name_or_pid) do
GenServer.call(name_or_pid, :first_index)
end
@doc """
Get the segment count of WAL
"""
@spec segment_count(atom() | pid()) :: non_neg_integer()
def segment_count(name_or_pid) do
GenServer.call(name_or_pid, :segment_count)
end
@doc """
Truncates the write-ahead log (WAL) after a specific index.
## Examples
iex> MyApp.WAL.truncate_after(:my_wal, 10)
:ok
## Parameters
* `name_or_pid` - The name or process identifier of the WAL GenServer.
* `index` - The index after which to truncate the WAL.
* `timeout` (optional) - The timeout value in milliseconds (default: 5000).
## Returns
- `:ok` - If the truncation is successful.
- `{:error, :index_out_of_range}` - If the provided index is out of range.
"""
@spec truncate_after(atom() | pid(), index(), non_neg_integer()) :: :ok | {:error, :index_out_of_range}
def truncate_after(name_or_pid, index, timeout \\ 5000) do
GenServer.call(name_or_pid, {:truncate_after, index}, timeout)
end
@doc """
Truncates the write-ahead log (WAL) before a specific index.
## Examples
iex> MyApp.WAL.truncate_before(:my_wal, 10)
:ok
## Parameters
* `name_or_pid` - The name or process identifier of the WAL GenServer.
* `index` - The index before which to truncate the WAL.
* `timeout` (optional) - The timeout value in milliseconds (default: 5000).
## Returns
- `:ok` - If the truncation is successful.
- `{:error, :index_out_of_range}` - If the provided index is out of range.
"""
@spec truncate_before(atom() | pid(), index(), non_neg_integer()) :: :ok | {:error, :index_out_of_range}
def truncate_before(name_or_pid, index, timeout \\ 5000) do
GenServer.call(name_or_pid, {:truncate_before, index}, timeout)
end
@doc """
Clears the write-ahead log (WAL) by removing all entries.
## Examples
iex> MyApp.WAL.clear(:my_wal)
:ok
## Parameters
* `name_or_pid` - The name or process identifier of the WAL GenServer.
## Returns
- `:ok` - If the clearing is successful.
"""
@spec clear(atom() | pid()) :: :ok
def clear(name_or_pid) do
GenServer.cast(name_or_pid, :clear)
end
# ----------------- Server -----------------
@impl GenServer
def init(opts) do
lru_name = :"#{opts[:name]}_lru"
{:ok, _} = LRU.start_link(lru_name, opts[:segment_cache_size])
store = apply(opts[:store], :new, [opts[:store_opts]])
path = Path.absname(opts[:path])
Store.mkdir(store, path, opts[:dir_permission])
segments =
path
|> Path.join("*")
|> Path.wildcard()
|> Enum.reject(fn x -> File.dir?(x) end)
|> Enum.map(&Path.basename(&1))
|> Enum.sort(:desc)
|> Enum.reject(&(String.length(&1) < 20))
|> Enum.map(fn x ->
%Segment{
path: Path.join(path, x),
index: parse_segment_filename(x)
}
end)
if Enum.empty?(segments) do
seg1_path = Path.join(path, segment_filename(1))
{:ok, h} =
Store.open(
store,
seg1_path,
[:read, :append],
opts[:file_permission]
)
{:ok,
%__MODULE__{
data_path: path,
hot: %Segment{path: seg1_path, index: 1},
cold: :array.new(),
first_index: 1,
last_index: 0,
lru_cache: lru_name,
store: store,
tail_store_handler: h,
opts: [
nosync: opts[:nosync],
segment_size: opts[:segment_size],
dir_permission: opts[:dir_permission],
file_permission: opts[:file_permission]
]
}}
else
# segment is in reverse order
first_index = List.last(segments).index
[%Segment{path: spath} = seg | t] = segments
{:ok, h} = Store.open(store, spath, [:read, :append], opts[:file_permission])
{:ok,
%__MODULE__{
data_path: path,
hot: seg,
cold: :array.from_list(t),
first_index: first_index,
lru_cache: lru_name,
store: store,
tail_store_handler: h,
opts: [
nosync: opts[:nosync],
segment_size: opts[:segment_size],
dir_permission: opts[:dir_permission],
file_permission: opts[:file_permission]
]
}, {:continue, :load_segment}}
end
end
@impl GenServer
def terminate(_reason, %__MODULE__{lru_cache: lru, store: store, tail_store_handler: h}) do
LRU.stop(lru)
Store.sync(store, h)
Store.close(store, h)
end
@impl GenServer
def handle_continue(:load_segment, %__MODULE__{hot: seg, store: store} = state) do
%Segment{block_count: bc, index: begin_index} =
seg =
load_segment(seg, store)
{:noreply, %__MODULE__{state | hot: seg, last_index: begin_index + bc - 1}}
end
@impl GenServer
def handle_call(:last_index, _, %__MODULE__{last_index: last_index} = state), do: {:reply, last_index, state}
def handle_call(:first_index, _, %__MODULE__{first_index: first_index} = state), do: {:reply, first_index, state}
def handle_call(:segment_count, _, %__MODULE__{cold: cold} = state) do
{:reply, :array.size(cold) + 1, state}
end
def handle_call(
{:write, entries},
_from,
%__MODULE__{last_index: last_index, hot: %Segment{buf: buf0}, opts: opts} = state
) do
# check
entries
|> Enum.with_index(last_index + 1)
|> Enum.find(fn {%Entry{index: index}, i} -> index != i end)
|> is_nil() || raise ArgumentError, "invalid index"
# cycle if needed
state =
if byte_size(buf0) > opts[:segment_size] do
cycle(state)
else
state
end
{:reply, :ok, write_entries(entries, state)}
end
def handle_call({:read, index}, _from, %__MODULE__{first_index: first_index, last_index: last_index} = state)
when index < first_index or index > last_index do
{:reply, {:error, :index_not_found}, state}
end
def handle_call({:read, index}, _from, %__MODULE__{} = state) do
%Segment{index: begin_index, blocks: blocks, buf: buf} = find_segment(index, state)
%Block{offset: offset, size: size} = :array.get(index - begin_index, blocks)
# ExWal.Debug.stacktrace(%{
# target_index: index,
# begin_index: begin_index,
# block_offset: offset,
# block_size: size,
# buf_size: byte_size(buf)
# })
data = binary_part(buf, offset, size)
{data_size, _, data} = Uvarint.decode(data)
# ExWal.Debug.debug(data)
byte_size(data) == data_size || raise ArgumentError, "invalid data"
{:reply, {:ok, data}, state}
end
def handle_call({:truncate_after, 0}, _from, state), do: {:reply, {:error, :index_out_of_range}, state}
def handle_call({:truncate_after, index}, _from, %__MODULE__{first_index: first_index, last_index: last_index} = state)
when index < first_index or index > last_index do
{:reply, {:error, :index_out_of_range}, state}
end
def handle_call({:truncate_after, index}, _from, %__MODULE__{last_index: last_index} = state) when index == last_index,
do: {:reply, :ok, state}
def handle_call({:truncate_after, index}, _from, state) do
{:reply, :ok, __truncate_after(index, state)}
end
def handle_call({:truncate_before, 0}, _from, state), do: {:reply, {:error, :index_out_of_range}, state}
def handle_call({:truncate_before, _}, _from, %__MODULE__{last_index: 0} = state),
do: {:reply, {:error, :index_out_of_range}, state}
def handle_call({:truncate_before, index}, _from, %__MODULE__{last_index: last_index, first_index: first_index} = state)
when index < first_index or index > last_index do
{:reply, {:error, :index_out_of_range}, state}
end
def handle_call({:truncate_before, index}, _from, %__MODULE__{first_index: first_index} = state)
when index == first_index,
do: {:reply, :ok, state}
def handle_call({:truncate_before, index}, _from, state) do
{:reply, :ok, __truncate_before(index, state)}
end
@impl GenServer
def handle_cast(
:clear,
%__MODULE__{
data_path: data_path,
lru_cache: lru,
hot: %Segment{path: path},
store: store,
tail_store_handler: h,
cold: cold,
opts: opts
} = state
) do
# rm cache
LRU.clear(lru)
# rm hot
:ok = Store.close(store, h)
:ok = Store.rm(store, path)
# rm cold
if :array.size(cold) > 0 do
Enum.each(0..(:array.size(cold) - 1), fn i ->
%Segment{path: path} = :array.get(i, cold)
:ok = Store.rm(store, path)
end)
end
# reinit
seg1_path = Path.join(data_path, segment_filename(1))
{:ok, h} =
Store.open(
store,
seg1_path,
[:read, :append],
opts[:file_permission]
)
{:noreply,
%__MODULE__{
state
| hot: %Segment{path: seg1_path, index: 1},
cold: :array.new(),
first_index: 1,
last_index: 0,
tail_store_handler: h
}}
end
# ----------------- Private -----------------
@spec load_segment(Segment.t(), Store.t()) :: Segment.t()
# index: begin_index,
defp load_segment(%Segment{path: path} = seg, store) do
{:ok, content} = Store.read_all(store, path)
{bc, blocks} = parse_blocks(content, 0, 0, [])
%Segment{seg | buf: content, blocks: :array.from_list(blocks), block_count: bc}
end
@spec parse_segment_filename(String.t()) :: index()
defp parse_segment_filename(filename) do
filename
|> String.trim_leading("0")
|> String.to_integer()
end
@spec segment_filename(index()) :: String.t()
defp segment_filename(index) do
index
|> Integer.to_string()
|> String.pad_leading(20, "0")
end
@spec parse_blocks(
data :: binary(),
since :: non_neg_integer(),
block_count :: non_neg_integer(),
blocks :: [Block.t()]
) ::
{block_count :: non_neg_integer(), blocks :: [Block.t()]}
defp parse_blocks("", _, bc, blocks), do: {bc, Enum.reverse(blocks)}
defp parse_blocks(data, since, bc, acc) do
{size, bytes_read, data} = Uvarint.decode(data)
# check
size == 0 && raise ArgumentError, "invalid block size"
next = since + size + bytes_read
<<_::bytes-size(size), rest::binary>> = data
parse_blocks(rest, next, bc + 1, [
%Block{offset: since, size: size + bytes_read} | acc
])
end
@spec append_entry(Segment.t(), Entry.t()) :: {binary(), Segment.t()}
defp append_entry(%Segment{buf: buf, index: begin_index, block_count: bc, blocks: blocks} = seg, %Entry{
index: index,
data: data
}) do
index == begin_index + bc ||
raise ArgumentError,
"invalid index, begin_index: #{begin_index}, bc: #{bc}, index: #{index}"
data = Uvarint.encode(byte_size(data)) <> data
{data,
%Segment{
seg
| buf: <<buf::binary, data::binary>>,
block_count: bc + 1,
blocks: :array.set(bc, %Block{offset: byte_size(buf), size: byte_size(data)}, blocks)
}}
end
@spec write_entries([Entry.t()], t()) :: t()
defp write_entries([], m), do: m
defp write_entries([entry | t], %__MODULE__{hot: seg, opts: opts, store: store, tail_store_handler: h} = m) do
{data, %Segment{buf: buf, index: begin_index, block_count: bc} = seg} =
append_entry(seg, entry)
:ok = Store.append(store, h, data)
if opts[:nosync] do
:ok = Store.sync(store, h)
end
m = %__MODULE__{m | hot: seg, last_index: begin_index + bc - 1}
if byte_size(buf) < opts[:segment_size] do
write_entries(t, m)
else
m = cycle(m)
write_entries(t, m)
end
end
# Cycle the old segment for a new segment.
@spec cycle(t()) :: t()
defp cycle(
%__MODULE__{
data_path: data_path,
store: store,
tail_store_handler: h,
lru_cache: lru,
last_index: last_index,
hot: %Segment{path: path} = seg,
cold: cold,
opts: opts
} = m
) do
:ok = Store.sync(store, h)
:ok = Store.close(store, h)
:ok = LRU.put(lru, path, seg)
new_seg = %Segment{
path: Path.join(data_path, segment_filename(last_index + 1)),
index: last_index + 1
}
{:ok, h} = Store.open(store, new_seg.path, [:read, :append], opts[:file_permission])
size = :array.size(cold)
%__MODULE__{
m
| tail_store_handler: h,
hot: new_seg,
cold: :array.set(size, %Segment{seg | blocks: nil, buf: ""}, cold),
last_index: last_index
}
end
@spec find_segment(index :: index(), t()) :: Segment.t()
defp find_segment(index, %__MODULE__{hot: %Segment{index: last_begin_index} = seg} = state) do
if index >= last_begin_index do
seg
else
find_cold_segment(index, state)
end
end
@spec find_cold_segment(index(), t()) :: Segment.t()
defp find_cold_segment(index, %__MODULE__{lru_cache: lru, cold: cold, store: store}) do
lru
|> LRU.select(fn %Segment{index: x, block_count: bc} ->
index >= x and index < x + bc
end)
|> case do
nil ->
cold
|> bin_search(index)
|> :array.get(cold)
|> load_segment(store)
|> tap(fn %Segment{path: path} = x -> LRU.put(lru, path, x) end)
seg ->
seg
end
end
@spec bin_search(:array.array(), index()) :: non_neg_integer() | nil
defp bin_search(cold, target_index), do: do_search(cold, target_index, 0, :array.size(cold) - 1)
defp do_search(_cold, _target_index, min, max) when min > max, do: nil
defp do_search(cold, target_index, min, max) when min == max do
%Segment{index: index, block_count: bc} = :array.get(min, cold)
if target_index >= index and target_index < index + bc do
min
end
end
defp do_search(cold, target_index, min, max) do
mid = div(min + max, 2)
%Segment{index: index, block_count: bc} = :array.get(mid, cold)
cond do
target_index < index ->
do_search(cold, target_index, min, mid)
target_index >= index and target_index < index + bc ->
mid
true ->
do_search(cold, target_index, mid + 1, max)
end
end
@spec __truncate_after(index(), t()) :: t()
defp __truncate_after(
index,
%__MODULE__{
hot: %Segment{index: begin_index, blocks: blocks, buf: buf, path: path} = seg,
data_path: data_path,
store: store,
tail_store_handler: h
} = state
)
when index >= begin_index do
# truncate buf and blocks
truncate_idx = index - begin_index + 1
new_seg =
if truncate_idx < :array.size(blocks) do
%Block{offset: offset} = :array.get(truncate_idx, blocks)
buf = binary_part(buf, 0, offset)
blocks = :array.resize(truncate_idx, blocks)
%Segment{
seg
| buf: buf,
block_count: truncate_idx - 1,
blocks: blocks
}
else
seg
end
# truncate file
temp_file = Path.join(data_path, "tmp")
:ok = Store.write_all(store, temp_file, buf)
# swap the tmp file with the old segment file
:ok = Store.close(store, h)
:ok = Store.rename(store, temp_file, path)
# reopen tail handler
{:ok, h} = Store.open(store, path, [:read, :append], state.opts[:file_permission])
# no need to gc blocks, reset block_count is all we need
%__MODULE__{
state
| hot: new_seg,
last_index: index,
tail_store_handler: h
}
end
defp __truncate_after(
index,
%__MODULE__{
cold: cold,
store: store,
data_path: data_path,
tail_store_handler: h,
lru_cache: lru,
hot: %Segment{path: hot_path}
} = state
) do
# find segment by index
idx = bin_search(cold, index)
%Segment{buf: buf, blocks: blocks, index: begin_index, path: path} =
seg =
idx
|> :array.get(cold)
|> load_segment(store)
# truncate buf and blocks
truncate_idx = index - begin_index + 1
new_seg =
if truncate_idx < :array.size(blocks) do
%Block{offset: offset} = :array.get(truncate_idx, blocks)
buf = binary_part(buf, 0, offset)
blocks = :array.resize(truncate_idx, blocks)
%Segment{
seg
| buf: buf,
block_count: truncate_idx - 1,
blocks: blocks
}
else
seg
end
# create tmp file to store new buf
temp_file = Path.join(data_path, "tmp")
:ok = Store.write_all(store, temp_file, buf)
# swap the tmp file with the old segment file
:ok = Store.rename(store, temp_file, path)
# cleanup
# all segments and cache after the new segment should be deleted
:ok = LRU.clear(lru)
if idx + 1 < :array.size(cold) do
Enum.each((idx + 1)..(:array.size(cold) - 1), fn i ->
%Segment{path: path} = :array.get(i, cold)
:ok = Store.rm(store, path)
end)
end
:ok = Store.close(store, h)
:ok = Store.rm(store, hot_path)
# reopen tail handler
{:ok, h} = Store.open(store, path, [:read, :append], state.opts[:file_permission])
%__MODULE__{
state
| hot: new_seg,
cold: :array.resize(idx, cold),
tail_store_handler: h,
last_index: index
}
end
@spec __truncate_before(index(), t()) :: t()
defp __truncate_before(
index,
%__MODULE__{
hot: %Segment{index: begin_index, blocks: blocks, buf: buf, path: path},
cold: cold,
store: store,
data_path: data_path,
tail_store_handler: h,
lru_cache: lru
} = state
)
when index >= begin_index do
%Block{offset: offset} = :array.get(index - begin_index, blocks)
new_buf = binary_part(buf, offset, byte_size(buf) - offset)
{bc, blocks} = parse_blocks(new_buf, 0, 0, [])
# remake last segment
new_seg = %Segment{
index: index,
buf: new_buf,
path: Path.join(data_path, segment_filename(index)),
block_count: bc,
blocks: :array.from_list(blocks)
}
# rewrite file
# make a tmp file to store new buf
temp_file = Path.join(data_path, "tmp")
:ok = Store.write_all(store, temp_file, new_seg.buf)
# swap the tmp file with the old segment file
:ok = Store.close(store, h)
:ok = Store.rename(store, temp_file, new_seg.path)
:ok = Store.rm(store, path)
# delete all segment files after the new segment
cold_size = :array.size(cold)
if cold_size > 0 do
# Task.async_stream(0..(cold_size - 1), fn i ->
# %Segment{path: path} = :array.get(i, cold)
# :ok = Store.rm(store, path)
# :ok = LRU.delete(lru, path)
# end)
# |> Stream.run()
Enum.each(0..(cold_size - 1), fn i ->
%Segment{path: path} = :array.get(i, cold)
:ok = Store.rm(store, path)
:ok = LRU.delete(lru, path)
end)
end
# reopen tail handler
{:ok, h} = Store.open(store, new_seg.path, [:read, :append], state.opts[:file_permission])
%__MODULE__{state | hot: new_seg, first_index: index, cold: :array.new(), tail_store_handler: h}
end
defp __truncate_before(
index,
%__MODULE__{hot: %Segment{index: begin_index}, cold: cold, store: store, data_path: data_path, lru_cache: lru} =
state
)
when index < begin_index do
idx = bin_search(cold, index)
%Segment{buf: buf, blocks: blocks, index: begin_index} =
idx
|> :array.get(cold)
|> load_segment(store)
%Block{offset: offset} = :array.get(index - begin_index, blocks)
# remake segment
new_seg = %Segment{
index: index,
buf: binary_part(buf, offset, byte_size(buf) - offset),
path: Path.join(data_path, segment_filename(index)),
block_count: :array.size(blocks) - (index - begin_index)
}
# clear cache
LRU.clear(lru)
# delete all segment files and cache before the new segment
Enum.each(0..idx, fn i ->
%Segment{path: path} = :array.get(i, cold)
:ok = Store.rm(store, path)
end)
# rewrite file
# make a tmp file to store new buf
temp_file = Path.join(data_path, "tmp")
:ok = Store.write_all(store, temp_file, new_seg.buf)
# swap the tmp file with the old segment file
:ok = Store.rename(store, temp_file, new_seg.path)
cold_size = :array.size(cold)
new_cold =
cold |> array_slice(idx..(cold_size - 1)) |> then(fn x -> :array.set(0, %Segment{new_seg | buf: ""}, x) end)
%__MODULE__{state | first_index: index, cold: new_cold}
end
@spec array_slice(:array.array(), Range.t()) :: :array.array()
defp array_slice(array, range) do
for_result =
for i <- range do
:array.get(i, array)
end
:array.from_list(for_result)
end
end
