%%%------------------------------------------------------------------- %%% @author benoitc %%% @copyright (C) 2017, %%% @doc %%% %%% @end %%% Created : 03. Nov 2017 12:11 %%%------------------------------------------------------------------- -module(memstore). -author("benoitc"). %% API -export([ start_link/2, open/2, close/1, put/3, delete/2, write_batch/2, get/2, get/3, new_snapshot/1, release_snapshot/1, get_snapshot_sequence/1, iterator/2, iterator_close/1, iterator_move/2, get_last_sequence/1 ]). -export([ init/1 ]). %% default write buffer size: 64 MB -define(WRITER_BUFFER_SIZE, 64 bsl 20). %% default compaction interval: 1s -define(COMPACTION_INTERVAL, 1000). -type db() :: atom(). -type key() :: term(). -type value() :: term(). -type snapshot() :: term(). -type iterator() :: pid(). -type batch() :: [ {put, Key :: key(), Value :: value()} | {delete, Key :: key()} ]. -type db_opts() :: [ {write_buffer_size, non_neg_integer()} | {compaction_interval, non_neg_integer()} ]. -type read_options() :: [{snapshot, snapshot()}]. -type iterator_action() :: {seek, Key :: term()} | {seek_for_prev, term()} | first | last | next | prev. -type iterator_result() :: {ok, key(), value()} | false. -export_types([ db/0, key/0, value/0, batch/0, snapshot/0, db_opts/0, read_options/0, iterator/0, iterator_action/0, iterator_result/0 ]). %% @doc open a database -spec open(Db :: db(), Options :: db_opts()) -> ok | {error, any()}. open(Name, Options) -> case supervisor:start_child(memstore_sup, [Name, Options]) of {ok, _Pid} -> ok; {error, {already_started, _Pid}} -> ok; Error -> Error end. %% %% @doc close a database -spec close(Db :: db()) -> ok. close(Name) -> supervisor:terminate_child(memstore_sup, whereis(Name)). %% @doc update a key Key with value Value -spec put(db(), key(), value()) -> ok | {error, any()}. put(Name, Key, Value) -> write_batch(Name, [{put, Key, Value}]). %% @doc delete a key -spec delete(db(), key()) -> ok |{error, any()}. delete(Name, Key) -> write_batch(Name, [{delete, Key}]). %% @doc atomically do some update/delete operations -spec write_batch(db(), batch()) -> ok | {error, term()}. write_batch(Name, Batch) when is_list(Batch) -> Writer = ets:lookup_element(Name, writer, 2), memstore_writer:write_batch(Writer, Batch). %% @doc fetch a key -spec get(db(), key()) -> {ok, value()} | not_found | {error, term()}. get(Name, Key) -> get(Name, Key, []). -spec get(db(), key(), read_options()) ->{ok, value()} | not_found | {error, term()}. get(Name, Key, Options) -> Tab = ets:lookup_element(Name, tab, 2), case proplists:get_value(snapshot, Options) of #{ seq := Seq } -> memstore_iterator:lookup(Tab, Key, Seq); _ -> memstore_iterator:lookup_last(Tab, Key) end. %% @doc get a new snapshot of the database. Snapshots provide consistent %% read-only views over the entire state of the key-value store. -spec new_snapshot(db()) -> snapshot(). new_snapshot(Name) -> Mgr = ets:lookup_element(Name, version_mgr, 2), gen_server:call(Mgr, new_snapshot). %% @doc release a snapshot -spec release_snapshot(snapshot()) -> ok. release_snapshot(undefined) -> ok; release_snapshot(#{ name := Name, snapshot_id := _Id} = Snapshot) -> Mgr = ets:lookup_element(Name, version_mgr, 2), gen_server:call(Mgr, {release_snapshot, Snapshot}); release_snapshot(_) -> erlang:error(badarg). %% @doc get the database version linked to this snapshots -spec get_snapshot_sequence(snapshot()) -> non_neg_integer(). get_snapshot_sequence(#{ snapshot_id := _Id, seq := Seq}) -> Seq; get_snapshot_sequence(_) -> erlang:error(badarg). %% @doc create an iterator -spec iterator(db(), read_options()) -> {ok, iterator()} |{error, any()}. iterator(Name, Options) -> Sup = ets:lookup_element(Name, iterator_sup, 2), supervisor:start_child(Sup, [Name, Options]). %% @doc close an iteratror -spec iterator_close(iterator()) -> ok. iterator_close(Itr) -> memstore_iterator:iterator_close(Itr). %% @doc move in the iterator -spec iterator_move(iterator(), iterator_action()) -> iterator_result(). iterator_move(Itr, Action) -> memstore_iterator:iterator_move(Itr, Action). %% @doc return the last sequence -spec get_last_sequence(db()) -> non_neg_integer(). get_last_sequence(Name) -> Mgr = ets:lookup_element(Name, version_mgr, 2), gen_server:call(Mgr, get_last_seq). start_link(Name, Options) -> supervisor:start_link({local, Name}, ?MODULE, [Name, Options]). init([Name, Options]) -> VersionMgrName = version_mgr_name(Name), WriterName = writer_name(Name), DataTab = table_name(Name), IteratorSup = iterator_sup(Name), WriteBufferSize = proplists:get_value(write_buffer_size, Options,?WRITER_BUFFER_SIZE), Interval = proplists:get_value(compaction_interval, Options, ?COMPACTION_INTERVAL), %% init metadata table _ = ets:new(Name, [ordered_set, named_table, public, {read_concurrency, true}]), ets:insert( Name, [ {sup, self()}, {tab, DataTab}, {writer, WriterName}, {version_mgr, VersionMgrName}, {iterator_sup, IteratorSup}, {last_seq, 0} ] ), Specs = [ %% versions manager #{ id => versions_mgr, start => {memstore_versions, start_link, [VersionMgrName, Name]}, restart => transient, shutdown => 2000, type => worker, modules => [memstore_versions] }, %% writer #{ id => writer, start => {memstore_writer, start_link, [WriterName, Name, DataTab, WriteBufferSize]}, restart => transient, shutdown => 2000, type => worker, modules => [memstore_writer] }, %% iterator supervisor #{ id => iterator_sup, start => {memstore_iterator_sup, start_link, [IteratorSup]}, restart => transient, shutdown => 2000, type => worker, modules => [memstore_writer] }, %% background compaction #{ id => gc, start => {memstore_gc, start_link, [Name, DataTab, Interval]}, restart => transient, shutdown => 2000, type => worker, modules => [memstore_gc] } ], %% finally start the supervisor SupFlags = #{ strategy => one_for_all, intensity => 4, period => 3600}, {ok, {SupFlags, Specs}}. version_mgr_name(Name) -> list_to_atom(atom_to_list(Name) ++ "_versions_mgr"). writer_name(Name) -> list_to_atom(atom_to_list(Name) ++ "_writer"). table_name(Name) -> list_to_atom(atom_to_list(Name) ++ "_data"). iterator_sup(Name) -> list_to_atom(atom_to_list(Name) ++ "_iterator_sup").