2016-08-02 13:44:48 +01:00
|
|
|
%% -------- Inker ---------
|
2016-09-05 15:01:23 +01:00
|
|
|
%%
|
|
|
|
%% The Inker is responsible for managing access and updates to the Journal.
|
|
|
|
%%
|
|
|
|
%% The Inker maintains a manifest of what files make up the Journal, and which
|
|
|
|
%% file is the current append-only nursery log to accept new PUTs into the
|
|
|
|
%% Journal. The Inker also marshals GET requests to the appropriate database
|
|
|
|
%% file within the Journal (routed by sequence number). The Inker is also
|
|
|
|
%% responsible for scheduling compaction work to be carried out by the Inker's
|
|
|
|
%% clerk.
|
|
|
|
%%
|
2016-10-07 18:07:03 +01:00
|
|
|
%% -------- Journal Files ---------
|
2016-09-05 15:01:23 +01:00
|
|
|
%%
|
2016-10-07 18:07:03 +01:00
|
|
|
%% The Journal is a series of files originally named as <SQN>_<GUID>
|
2016-09-05 15:01:23 +01:00
|
|
|
%% where the sequence number is the first object sequence number (key) within
|
|
|
|
%% the given database file. The files will be named *.cdb at the point they
|
|
|
|
%% have been made immutable (through a rename operation). Prior to this, they
|
|
|
|
%% will originally start out as a *.pnd file.
|
|
|
|
%%
|
|
|
|
%% At some stage in the future compacted versions of old journal cdb files may
|
2016-10-07 18:07:03 +01:00
|
|
|
%% be produced. These files will be named <SQN>-<NewGUID>.cdb, and once
|
|
|
|
%% the manifest is updated the original <SQN>_<GUID>.cdb (or
|
|
|
|
%% <SQN>_<previousGUID>.cdb) files they replace will be erased.
|
2016-08-02 13:44:48 +01:00
|
|
|
%%
|
2016-10-07 18:07:03 +01:00
|
|
|
%% The current Journal is made up of a set of files referenced in the manifest.
|
|
|
|
%% No PUTs are made to files which are not in the manifest.
|
2016-08-02 13:44:48 +01:00
|
|
|
%%
|
2016-09-05 15:01:23 +01:00
|
|
|
%% The Journal is ordered by sequence number from front to back both within
|
|
|
|
%% and across files.
|
|
|
|
%%
|
|
|
|
%% On startup the Inker should open the manifest with the highest sequence
|
|
|
|
%% number, and this will contain the list of filenames that make up the
|
2016-10-07 18:07:03 +01:00
|
|
|
%% non-recent part of the Journal. All the filenames should then be opened.
|
|
|
|
%% How they are opened depends on the file extension:
|
|
|
|
%%
|
|
|
|
%% - If the file extension is *.cdb the file is opened read only
|
|
|
|
%% - If the file extension is *.pnd and the file is not the most recent in the
|
|
|
|
%% manifest, then the file should be completed bfore being opened read-only
|
|
|
|
%% - If the file extension is *.pnd the file is opened for writing
|
|
|
|
%%
|
|
|
|
%% -------- Manifest Files ---------
|
|
|
|
%%
|
|
|
|
%% The manifest is just saved as a straight term_to_binary blob, with a
|
|
|
|
%% filename ordered by the Manifest SQN. The Manifest is first saved with a
|
|
|
|
%% *.pnd extension, and then renamed to one with a *.man extension.
|
|
|
|
%%
|
|
|
|
%% On startup the *.man manifest file with the highest manifest sequence
|
|
|
|
%% number should be used.
|
2016-09-05 15:01:23 +01:00
|
|
|
%%
|
|
|
|
%% -------- Objects ---------
|
|
|
|
%%
|
|
|
|
%% From the perspective of the Inker, objects to store are made up of:
|
2016-10-25 01:57:12 +01:00
|
|
|
%% - An Inker Key formed from
|
|
|
|
%% - A sequence number (assigned by the Inker)
|
|
|
|
%% - An Inker key type (stnd, tomb or keyd)
|
|
|
|
%% - A Ledger Key (as an Erlang term)
|
|
|
|
%% - A value formed from
|
|
|
|
%% - An object (an Erlang term) which should be null for tomb types, and
|
|
|
|
%% maybe null for keyd types
|
|
|
|
%% - A set of Key Deltas associated with the change (which may be an
|
|
|
|
%% empty list )
|
|
|
|
%%
|
|
|
|
%% Note that only the Inker key type of stnd is directly fetchable, other
|
|
|
|
%% key types are to be found only in scans and so can be added without being
|
|
|
|
%% entered into the hashtree
|
2016-09-05 15:01:23 +01:00
|
|
|
%%
|
|
|
|
%% -------- Compaction ---------
|
|
|
|
%%
|
|
|
|
%% Compaction is a process whereby an Inker's clerk will:
|
2016-10-07 18:07:03 +01:00
|
|
|
%% - Request a view of the current Inker manifest and a snaphot of the Ledger
|
2016-10-09 22:33:45 +01:00
|
|
|
%% - Test all files within the Journal to find the approximate comapction
|
2016-10-07 18:07:03 +01:00
|
|
|
%% potential percentage (the volume of the Journal that has been replaced)
|
|
|
|
%% - Attempts to find the optimal "run" of files to compact
|
|
|
|
%% - Compacts those files in the run, by rolling over the files re-writing
|
|
|
|
%% to a new Journal if and only if the Key is still present in the Ledger (or
|
|
|
|
%% the sequence number of the Key is higher than the SQN of the snapshot)
|
|
|
|
%% - Requests the Inker update the manifest with the new changes
|
|
|
|
%% - Instructs the files to destroy themselves when they are next closed
|
2016-08-02 13:44:48 +01:00
|
|
|
%%
|
2016-10-07 18:07:03 +01:00
|
|
|
%% TODO: how to instruct the files to close is tbd
|
2016-08-02 13:44:48 +01:00
|
|
|
%%
|
2016-08-02 17:51:43 +01:00
|
|
|
|
2016-10-07 18:07:03 +01:00
|
|
|
|
2016-08-02 17:51:43 +01:00
|
|
|
-module(leveled_inker).
|
|
|
|
|
2016-09-05 15:01:23 +01:00
|
|
|
-behaviour(gen_server).
|
|
|
|
|
2016-10-18 01:59:03 +01:00
|
|
|
-include("include/leveled.hrl").
|
2016-09-05 15:01:23 +01:00
|
|
|
|
|
|
|
-export([init/1,
|
|
|
|
handle_call/3,
|
|
|
|
handle_cast/2,
|
|
|
|
handle_info/2,
|
|
|
|
terminate/2,
|
|
|
|
code_change/3,
|
|
|
|
ink_start/1,
|
|
|
|
ink_put/4,
|
|
|
|
ink_get/3,
|
2016-09-15 10:53:24 +01:00
|
|
|
ink_fetch/3,
|
|
|
|
ink_loadpcl/4,
|
2016-09-23 18:50:29 +01:00
|
|
|
ink_registersnapshot/2,
|
2016-10-26 20:39:16 +01:00
|
|
|
ink_confirmdelete/2,
|
2016-09-26 10:55:08 +01:00
|
|
|
ink_compactjournal/3,
|
2016-10-03 23:34:28 +01:00
|
|
|
ink_compactioncomplete/1,
|
2016-10-26 20:39:16 +01:00
|
|
|
ink_compactionpending/1,
|
2016-09-27 14:58:26 +01:00
|
|
|
ink_getmanifest/1,
|
2016-10-03 23:34:28 +01:00
|
|
|
ink_updatemanifest/3,
|
2016-09-07 17:58:12 +01:00
|
|
|
ink_print_manifest/1,
|
2016-09-27 14:58:26 +01:00
|
|
|
ink_close/1,
|
2016-09-05 15:01:23 +01:00
|
|
|
build_dummy_journal/0,
|
2016-09-15 10:53:24 +01:00
|
|
|
simple_manifest_reader/2,
|
2016-09-27 14:58:26 +01:00
|
|
|
clean_testdir/1,
|
2016-09-28 11:41:56 +01:00
|
|
|
filepath/2,
|
2016-09-27 14:58:26 +01:00
|
|
|
filepath/3]).
|
2016-09-05 15:01:23 +01:00
|
|
|
|
|
|
|
-include_lib("eunit/include/eunit.hrl").
|
|
|
|
|
|
|
|
-define(MANIFEST_FP, "journal_manifest").
|
|
|
|
-define(FILES_FP, "journal_files").
|
2016-09-27 14:58:26 +01:00
|
|
|
-define(COMPACT_FP, "post_compact").
|
2016-09-05 15:01:23 +01:00
|
|
|
-define(JOURNAL_FILEX, "cdb").
|
|
|
|
-define(MANIFEST_FILEX, "man").
|
|
|
|
-define(PENDING_FILEX, "pnd").
|
2016-09-15 10:53:24 +01:00
|
|
|
-define(LOADING_PAUSE, 5000).
|
|
|
|
-define(LOADING_BATCH, 1000).
|
2016-09-05 15:01:23 +01:00
|
|
|
|
|
|
|
-record(state, {manifest = [] :: list(),
|
|
|
|
manifest_sqn = 0 :: integer(),
|
|
|
|
journal_sqn = 0 :: integer(),
|
|
|
|
active_journaldb :: pid(),
|
2016-09-27 14:58:26 +01:00
|
|
|
pending_removals = [] :: list(),
|
2016-09-26 10:55:08 +01:00
|
|
|
registered_snapshots = [] :: list(),
|
2016-09-07 17:58:12 +01:00
|
|
|
root_path :: string(),
|
2016-09-26 10:55:08 +01:00
|
|
|
cdb_options :: #cdb_options{},
|
2016-10-03 23:34:28 +01:00
|
|
|
clerk :: pid(),
|
|
|
|
compaction_pending = false :: boolean(),
|
2016-10-07 10:04:48 +01:00
|
|
|
is_snapshot = false :: boolean(),
|
|
|
|
source_inker :: pid()}).
|
2016-09-05 15:01:23 +01:00
|
|
|
|
|
|
|
|
|
|
|
%%%============================================================================
|
|
|
|
%%% API
|
|
|
|
%%%============================================================================
|
|
|
|
|
2016-09-07 17:58:12 +01:00
|
|
|
ink_start(InkerOpts) ->
|
|
|
|
gen_server:start(?MODULE, [InkerOpts], []).
|
2016-09-05 15:01:23 +01:00
|
|
|
|
|
|
|
ink_put(Pid, PrimaryKey, Object, KeyChanges) ->
|
|
|
|
gen_server:call(Pid, {put, PrimaryKey, Object, KeyChanges}, infinity).
|
|
|
|
|
|
|
|
ink_get(Pid, PrimaryKey, SQN) ->
|
|
|
|
gen_server:call(Pid, {get, PrimaryKey, SQN}, infinity).
|
|
|
|
|
2016-09-15 10:53:24 +01:00
|
|
|
ink_fetch(Pid, PrimaryKey, SQN) ->
|
|
|
|
gen_server:call(Pid, {fetch, PrimaryKey, SQN}, infinity).
|
2016-09-09 15:58:19 +01:00
|
|
|
|
2016-09-23 18:50:29 +01:00
|
|
|
ink_registersnapshot(Pid, Requestor) ->
|
2016-10-05 09:54:53 +01:00
|
|
|
gen_server:call(Pid, {register_snapshot, Requestor}, infinity).
|
2016-09-05 15:01:23 +01:00
|
|
|
|
2016-10-07 10:04:48 +01:00
|
|
|
ink_releasesnapshot(Pid, Snapshot) ->
|
|
|
|
gen_server:call(Pid, {release_snapshot, Snapshot}, infinity).
|
|
|
|
|
2016-10-26 20:39:16 +01:00
|
|
|
ink_confirmdelete(Pid, ManSQN) ->
|
|
|
|
gen_server:call(Pid, {confirm_delete, ManSQN}, 1000).
|
|
|
|
|
2016-09-06 17:17:31 +01:00
|
|
|
ink_close(Pid) ->
|
2016-10-30 22:06:44 +00:00
|
|
|
gen_server:call(Pid, close, infinity).
|
2016-09-06 17:17:31 +01:00
|
|
|
|
2016-09-15 10:53:24 +01:00
|
|
|
ink_loadpcl(Pid, MinSQN, FilterFun, Penciller) ->
|
|
|
|
gen_server:call(Pid, {load_pcl, MinSQN, FilterFun, Penciller}, infinity).
|
|
|
|
|
2016-10-05 18:28:31 +01:00
|
|
|
ink_compactjournal(Pid, Bookie, Timeout) ->
|
2016-09-28 18:26:52 +01:00
|
|
|
CheckerInitiateFun = fun initiate_penciller_snapshot/1,
|
|
|
|
CheckerFilterFun = fun leveled_penciller:pcl_checksequencenumber/3,
|
|
|
|
gen_server:call(Pid,
|
|
|
|
{compact,
|
2016-10-05 18:28:31 +01:00
|
|
|
Bookie,
|
2016-09-28 18:26:52 +01:00
|
|
|
CheckerInitiateFun,
|
|
|
|
CheckerFilterFun,
|
|
|
|
Timeout},
|
2016-10-03 23:34:28 +01:00
|
|
|
infinity).
|
2016-09-28 18:26:52 +01:00
|
|
|
|
|
|
|
%% Allows the Checker to be overriden in test, use something other than a
|
|
|
|
%% penciller
|
|
|
|
ink_compactjournal(Pid, Checker, InitiateFun, FilterFun, Timeout) ->
|
|
|
|
gen_server:call(Pid,
|
|
|
|
{compact,
|
|
|
|
Checker,
|
|
|
|
InitiateFun,
|
|
|
|
FilterFun,
|
|
|
|
Timeout},
|
|
|
|
infinity).
|
2016-09-26 10:55:08 +01:00
|
|
|
|
2016-10-03 23:34:28 +01:00
|
|
|
ink_compactioncomplete(Pid) ->
|
|
|
|
gen_server:call(Pid, compaction_complete, infinity).
|
|
|
|
|
2016-10-26 20:39:16 +01:00
|
|
|
ink_compactionpending(Pid) ->
|
|
|
|
gen_server:call(Pid, compaction_pending, infinity).
|
|
|
|
|
2016-09-27 14:58:26 +01:00
|
|
|
ink_getmanifest(Pid) ->
|
|
|
|
gen_server:call(Pid, get_manifest, infinity).
|
|
|
|
|
2016-10-03 23:34:28 +01:00
|
|
|
ink_updatemanifest(Pid, ManifestSnippet, DeletedFiles) ->
|
2016-09-27 14:58:26 +01:00
|
|
|
gen_server:call(Pid,
|
|
|
|
{update_manifest,
|
|
|
|
ManifestSnippet,
|
|
|
|
DeletedFiles},
|
|
|
|
infinity).
|
|
|
|
|
2016-09-07 17:58:12 +01:00
|
|
|
ink_print_manifest(Pid) ->
|
|
|
|
gen_server:call(Pid, print_manifest, infinity).
|
|
|
|
|
2016-09-05 15:01:23 +01:00
|
|
|
%%%============================================================================
|
|
|
|
%%% gen_server callbacks
|
|
|
|
%%%============================================================================
|
|
|
|
|
2016-09-07 17:58:12 +01:00
|
|
|
init([InkerOpts]) ->
|
2016-09-23 18:50:29 +01:00
|
|
|
case {InkerOpts#inker_options.root_path,
|
|
|
|
InkerOpts#inker_options.start_snapshot} of
|
|
|
|
{undefined, true} ->
|
|
|
|
SrcInker = InkerOpts#inker_options.source_inker,
|
2016-10-05 09:54:53 +01:00
|
|
|
{Manifest,
|
2016-10-07 18:07:03 +01:00
|
|
|
ActiveJournalDB} = ink_registersnapshot(SrcInker, self()),
|
2016-09-23 18:50:29 +01:00
|
|
|
{ok, #state{manifest=Manifest,
|
2016-10-03 23:34:28 +01:00
|
|
|
active_journaldb=ActiveJournalDB,
|
2016-10-07 10:04:48 +01:00
|
|
|
source_inker=SrcInker,
|
2016-10-03 23:34:28 +01:00
|
|
|
is_snapshot=true}};
|
2016-09-23 18:50:29 +01:00
|
|
|
%% Need to do something about timeout
|
|
|
|
{_RootPath, false} ->
|
|
|
|
start_from_file(InkerOpts)
|
|
|
|
end.
|
2016-09-05 15:01:23 +01:00
|
|
|
|
|
|
|
|
2016-10-07 18:07:03 +01:00
|
|
|
handle_call({put, Key, Object, KeyChanges}, _From, State) ->
|
2016-09-05 15:01:23 +01:00
|
|
|
case put_object(Key, Object, KeyChanges, State) of
|
2016-09-15 10:53:24 +01:00
|
|
|
{ok, UpdState, ObjSize} ->
|
|
|
|
{reply, {ok, UpdState#state.journal_sqn, ObjSize}, UpdState};
|
|
|
|
{rolling, UpdState, ObjSize} ->
|
2016-10-07 18:07:03 +01:00
|
|
|
ok = leveled_cdb:cdb_roll(State#state.active_journaldb),
|
|
|
|
{reply, {ok, UpdState#state.journal_sqn, ObjSize}, UpdState}
|
2016-09-05 15:01:23 +01:00
|
|
|
end;
|
2016-09-15 10:53:24 +01:00
|
|
|
handle_call({fetch, Key, SQN}, _From, State) ->
|
2016-10-07 18:07:03 +01:00
|
|
|
case get_object(Key, SQN, State#state.manifest) of
|
2016-09-15 10:53:24 +01:00
|
|
|
{{SQN, Key}, {Value, _IndexSpecs}} ->
|
|
|
|
{reply, {ok, Value}, State};
|
|
|
|
Other ->
|
|
|
|
io:format("Unexpected failure to fetch value for" ++
|
2016-09-19 15:56:35 +01:00
|
|
|
"Key=~w SQN=~w with reason ~w", [Key, SQN, Other]),
|
2016-09-15 10:53:24 +01:00
|
|
|
{reply, not_present, State}
|
|
|
|
end;
|
2016-09-05 15:01:23 +01:00
|
|
|
handle_call({get, Key, SQN}, _From, State) ->
|
2016-10-07 18:07:03 +01:00
|
|
|
{reply, get_object(Key, SQN, State#state.manifest), State};
|
2016-09-15 10:53:24 +01:00
|
|
|
handle_call({load_pcl, StartSQN, FilterFun, Penciller}, _From, State) ->
|
2016-10-07 18:07:03 +01:00
|
|
|
Manifest = lists:reverse(State#state.manifest),
|
2016-09-15 10:53:24 +01:00
|
|
|
Reply = load_from_sequence(StartSQN, FilterFun, Penciller, Manifest),
|
|
|
|
{reply, Reply, State};
|
2016-09-26 10:55:08 +01:00
|
|
|
handle_call({register_snapshot, Requestor}, _From , State) ->
|
|
|
|
Rs = [{Requestor,
|
|
|
|
State#state.manifest_sqn}|State#state.registered_snapshots],
|
2016-10-27 20:56:18 +01:00
|
|
|
io:format("Journal snapshot ~w registered at SQN ~w~n",
|
2016-10-07 10:04:48 +01:00
|
|
|
[Requestor, State#state.manifest_sqn]),
|
2016-09-06 17:17:31 +01:00
|
|
|
{reply, {State#state.manifest,
|
2016-10-07 18:07:03 +01:00
|
|
|
State#state.active_journaldb},
|
2016-09-26 10:55:08 +01:00
|
|
|
State#state{registered_snapshots=Rs}};
|
2016-10-07 10:04:48 +01:00
|
|
|
handle_call({release_snapshot, Snapshot}, _From , State) ->
|
|
|
|
Rs = lists:keydelete(Snapshot, 1, State#state.registered_snapshots),
|
2016-10-27 20:56:18 +01:00
|
|
|
io:format("Journal snapshot ~w released~n", [Snapshot]),
|
|
|
|
io:format("Remaining journal snapshots are ~w~n", [Rs]),
|
2016-10-07 10:04:48 +01:00
|
|
|
{reply, ok, State#state{registered_snapshots=Rs}};
|
2016-10-26 20:39:16 +01:00
|
|
|
handle_call({confirm_delete, ManSQN}, _From, State) ->
|
|
|
|
Reply = lists:foldl(fun({_R, SnapSQN}, Bool) ->
|
|
|
|
case SnapSQN < ManSQN of
|
|
|
|
true ->
|
|
|
|
Bool;
|
|
|
|
false ->
|
|
|
|
false
|
|
|
|
end end,
|
|
|
|
true,
|
|
|
|
State#state.registered_snapshots),
|
|
|
|
{reply, Reply, State};
|
2016-09-27 14:58:26 +01:00
|
|
|
handle_call(get_manifest, _From, State) ->
|
|
|
|
{reply, State#state.manifest, State};
|
|
|
|
handle_call({update_manifest,
|
|
|
|
ManifestSnippet,
|
|
|
|
DeletedFiles}, _From, State) ->
|
|
|
|
Man0 = lists:foldl(fun(ManEntry, AccMan) ->
|
2016-10-07 18:07:03 +01:00
|
|
|
remove_from_manifest(AccMan, ManEntry)
|
2016-09-27 14:58:26 +01:00
|
|
|
end,
|
2016-10-07 18:07:03 +01:00
|
|
|
State#state.manifest,
|
|
|
|
DeletedFiles),
|
2016-09-27 14:58:26 +01:00
|
|
|
Man1 = lists:foldl(fun(ManEntry, AccMan) ->
|
2016-09-28 18:26:52 +01:00
|
|
|
add_to_manifest(AccMan, ManEntry) end,
|
|
|
|
Man0,
|
|
|
|
ManifestSnippet),
|
2016-09-27 14:58:26 +01:00
|
|
|
NewManifestSQN = State#state.manifest_sqn + 1,
|
2016-10-07 18:07:03 +01:00
|
|
|
manifest_printer(Man1),
|
2016-09-27 14:58:26 +01:00
|
|
|
ok = simple_manifest_writer(Man1, NewManifestSQN, State#state.root_path),
|
2016-10-26 20:39:16 +01:00
|
|
|
{reply,
|
|
|
|
{ok, NewManifestSQN},
|
|
|
|
State#state{manifest=Man1,
|
|
|
|
manifest_sqn=NewManifestSQN,
|
|
|
|
pending_removals=DeletedFiles}};
|
2016-09-07 17:58:12 +01:00
|
|
|
handle_call(print_manifest, _From, State) ->
|
|
|
|
manifest_printer(State#state.manifest),
|
|
|
|
{reply, ok, State};
|
2016-09-28 18:26:52 +01:00
|
|
|
handle_call({compact,
|
|
|
|
Checker,
|
|
|
|
InitiateFun,
|
|
|
|
FilterFun,
|
|
|
|
Timeout},
|
|
|
|
_From, State) ->
|
2016-09-27 14:58:26 +01:00
|
|
|
leveled_iclerk:clerk_compact(State#state.clerk,
|
2016-09-28 18:26:52 +01:00
|
|
|
Checker,
|
|
|
|
InitiateFun,
|
|
|
|
FilterFun,
|
2016-09-27 14:58:26 +01:00
|
|
|
self(),
|
|
|
|
Timeout),
|
2016-10-03 23:34:28 +01:00
|
|
|
{reply, ok, State#state{compaction_pending=true}};
|
|
|
|
handle_call(compaction_complete, _From, State) ->
|
|
|
|
{reply, ok, State#state{compaction_pending=false}};
|
2016-10-26 20:39:16 +01:00
|
|
|
handle_call(compaction_pending, _From, State) ->
|
|
|
|
{reply, State#state.compaction_pending, State};
|
2016-10-30 22:06:44 +00:00
|
|
|
handle_call(close, _From, State) ->
|
|
|
|
case State#state.compaction_pending of
|
|
|
|
true ->
|
2016-10-03 23:34:28 +01:00
|
|
|
{reply, pause, State};
|
2016-10-30 22:06:44 +00:00
|
|
|
false ->
|
2016-10-03 23:34:28 +01:00
|
|
|
{stop, normal, ok, State}
|
2016-10-16 16:58:55 +01:00
|
|
|
end.
|
2016-09-05 15:01:23 +01:00
|
|
|
|
|
|
|
handle_cast(_Msg, State) ->
|
|
|
|
{noreply, State}.
|
|
|
|
|
|
|
|
handle_info(_Info, State) ->
|
|
|
|
{noreply, State}.
|
|
|
|
|
2016-09-06 17:17:31 +01:00
|
|
|
terminate(Reason, State) ->
|
2016-10-03 23:34:28 +01:00
|
|
|
case State#state.is_snapshot of
|
|
|
|
true ->
|
2016-10-07 10:04:48 +01:00
|
|
|
ok = ink_releasesnapshot(State#state.source_inker, self());
|
2016-10-03 23:34:28 +01:00
|
|
|
false ->
|
|
|
|
io:format("Inker closing journal for reason ~w~n", [Reason]),
|
|
|
|
io:format("Close triggered with journal_sqn=~w and manifest_sqn=~w~n",
|
|
|
|
[State#state.journal_sqn, State#state.manifest_sqn]),
|
|
|
|
io:format("Manifest when closing is: ~n"),
|
|
|
|
leveled_iclerk:clerk_stop(State#state.clerk),
|
|
|
|
lists:foreach(fun({Snap, _SQN}) -> ok = ink_close(Snap) end,
|
|
|
|
State#state.registered_snapshots),
|
|
|
|
manifest_printer(State#state.manifest),
|
2016-10-26 20:39:16 +01:00
|
|
|
ok = close_allmanifest(State#state.manifest)
|
2016-10-03 23:34:28 +01:00
|
|
|
end.
|
2016-09-05 15:01:23 +01:00
|
|
|
|
|
|
|
code_change(_OldVsn, State, _Extra) ->
|
|
|
|
{ok, State}.
|
|
|
|
|
|
|
|
|
|
|
|
%%%============================================================================
|
|
|
|
%%% Internal functions
|
|
|
|
%%%============================================================================
|
|
|
|
|
2016-10-25 23:13:14 +01:00
|
|
|
start_from_file(InkOpts) ->
|
|
|
|
RootPath = InkOpts#inker_options.root_path,
|
|
|
|
CDBopts = InkOpts#inker_options.cdb_options,
|
2016-09-23 18:50:29 +01:00
|
|
|
JournalFP = filepath(RootPath, journal_dir),
|
2016-10-07 18:07:03 +01:00
|
|
|
filelib:ensure_dir(JournalFP),
|
|
|
|
CompactFP = filepath(RootPath, journal_compact_dir),
|
|
|
|
filelib:ensure_dir(CompactFP),
|
|
|
|
|
2016-09-23 18:50:29 +01:00
|
|
|
ManifestFP = filepath(RootPath, manifest_dir),
|
2016-10-29 00:52:49 +01:00
|
|
|
ok = filelib:ensure_dir(ManifestFP),
|
|
|
|
{ok, ManifestFilenames} = file:list_dir(ManifestFP),
|
2016-09-27 14:58:26 +01:00
|
|
|
|
|
|
|
IClerkCDBOpts = CDBopts#cdb_options{file_path = CompactFP},
|
2016-10-25 23:13:14 +01:00
|
|
|
ReloadStrategy = InkOpts#inker_options.reload_strategy,
|
2016-10-29 00:52:49 +01:00
|
|
|
MRL = InkOpts#inker_options.max_run_length,
|
2016-09-27 14:58:26 +01:00
|
|
|
IClerkOpts = #iclerk_options{inker = self(),
|
2016-10-25 23:13:14 +01:00
|
|
|
cdb_options=IClerkCDBOpts,
|
2016-10-29 00:52:49 +01:00
|
|
|
reload_strategy = ReloadStrategy,
|
|
|
|
max_run_length = MRL},
|
2016-09-27 14:58:26 +01:00
|
|
|
{ok, Clerk} = leveled_iclerk:clerk_new(IClerkOpts),
|
|
|
|
|
2016-09-23 18:50:29 +01:00
|
|
|
{Manifest,
|
2016-10-07 18:07:03 +01:00
|
|
|
ManifestSQN,
|
2016-09-23 18:50:29 +01:00
|
|
|
JournalSQN,
|
2016-10-07 18:07:03 +01:00
|
|
|
ActiveJournal} = build_manifest(ManifestFilenames,
|
2016-09-23 18:50:29 +01:00
|
|
|
RootPath,
|
|
|
|
CDBopts),
|
2016-10-07 18:07:03 +01:00
|
|
|
{ok, #state{manifest = Manifest,
|
2016-09-23 18:50:29 +01:00
|
|
|
manifest_sqn = ManifestSQN,
|
|
|
|
journal_sqn = JournalSQN,
|
|
|
|
active_journaldb = ActiveJournal,
|
|
|
|
root_path = RootPath,
|
2016-09-26 10:55:08 +01:00
|
|
|
cdb_options = CDBopts,
|
|
|
|
clerk = Clerk}}.
|
2016-09-23 18:50:29 +01:00
|
|
|
|
|
|
|
|
2016-10-25 23:13:14 +01:00
|
|
|
put_object(LedgerKey, Object, KeyChanges, State) ->
|
2016-09-05 15:01:23 +01:00
|
|
|
NewSQN = State#state.journal_sqn + 1,
|
2016-10-26 11:39:27 +01:00
|
|
|
{JournalKey, JournalBin} = leveled_codec:to_inkerkv(LedgerKey,
|
|
|
|
NewSQN,
|
|
|
|
Object,
|
|
|
|
KeyChanges),
|
2016-09-05 15:01:23 +01:00
|
|
|
case leveled_cdb:cdb_put(State#state.active_journaldb,
|
2016-10-25 23:13:14 +01:00
|
|
|
JournalKey,
|
2016-10-26 11:39:27 +01:00
|
|
|
JournalBin) of
|
2016-09-05 15:01:23 +01:00
|
|
|
ok ->
|
2016-10-25 23:13:14 +01:00
|
|
|
{ok, State#state{journal_sqn=NewSQN}, byte_size(JournalBin)};
|
2016-09-05 15:01:23 +01:00
|
|
|
roll ->
|
2016-10-07 18:07:03 +01:00
|
|
|
SW = os:timestamp(),
|
2016-09-07 17:58:12 +01:00
|
|
|
CDBopts = State#state.cdb_options,
|
2016-10-07 18:07:03 +01:00
|
|
|
ManEntry = start_new_activejournal(NewSQN,
|
|
|
|
State#state.root_path,
|
|
|
|
CDBopts),
|
|
|
|
{_, _, NewJournalP} = ManEntry,
|
|
|
|
NewManifest = add_to_manifest(State#state.manifest, ManEntry),
|
|
|
|
ok = simple_manifest_writer(NewManifest,
|
|
|
|
State#state.manifest_sqn + 1,
|
|
|
|
State#state.root_path),
|
2016-10-25 01:57:12 +01:00
|
|
|
ok = leveled_cdb:cdb_put(NewJournalP,
|
2016-10-25 23:13:14 +01:00
|
|
|
JournalKey,
|
2016-10-26 11:39:27 +01:00
|
|
|
JournalBin),
|
2016-10-07 18:07:03 +01:00
|
|
|
io:format("Put to new active journal " ++
|
|
|
|
"with manifest write took ~w microseconds~n",
|
|
|
|
[timer:now_diff(os:timestamp(),SW)]),
|
|
|
|
{rolling,
|
|
|
|
State#state{journal_sqn=NewSQN,
|
|
|
|
manifest=NewManifest,
|
|
|
|
manifest_sqn = State#state.manifest_sqn + 1,
|
|
|
|
active_journaldb=NewJournalP},
|
2016-10-25 23:13:14 +01:00
|
|
|
byte_size(JournalBin)}
|
2016-10-23 22:45:43 +01:00
|
|
|
end.
|
|
|
|
|
|
|
|
|
2016-10-25 23:13:14 +01:00
|
|
|
get_object(LedgerKey, SQN, Manifest) ->
|
2016-10-07 18:07:03 +01:00
|
|
|
JournalP = find_in_manifest(SQN, Manifest),
|
2016-10-25 23:13:14 +01:00
|
|
|
{InkerKey, _V, true} = leveled_codec:to_inkerkv(LedgerKey,
|
|
|
|
SQN,
|
|
|
|
to_fetch,
|
|
|
|
null),
|
|
|
|
Obj = leveled_cdb:cdb_get(JournalP, InkerKey),
|
|
|
|
leveled_codec:from_inkerkv(Obj).
|
2016-09-07 17:58:12 +01:00
|
|
|
|
2016-09-05 15:01:23 +01:00
|
|
|
|
|
|
|
build_manifest(ManifestFilenames,
|
2016-09-07 17:58:12 +01:00
|
|
|
RootPath,
|
|
|
|
CDBopts) ->
|
2016-10-07 18:07:03 +01:00
|
|
|
% Find the manifest with a highest Manifest sequence number
|
|
|
|
% Open it and read it to get the current Confirmed Manifest
|
2016-09-05 15:01:23 +01:00
|
|
|
ManifestRegex = "(?<MSQN>[0-9]+)\\." ++ ?MANIFEST_FILEX,
|
|
|
|
ValidManSQNs = sequencenumbers_fromfilenames(ManifestFilenames,
|
|
|
|
ManifestRegex,
|
|
|
|
'MSQN'),
|
2016-10-07 18:07:03 +01:00
|
|
|
{Manifest,
|
2016-09-05 15:01:23 +01:00
|
|
|
ManifestSQN} = case length(ValidManSQNs) of
|
|
|
|
0 ->
|
2016-10-07 18:07:03 +01:00
|
|
|
{[], 1};
|
2016-09-05 15:01:23 +01:00
|
|
|
_ ->
|
|
|
|
PersistedManSQN = lists:max(ValidManSQNs),
|
2016-10-07 18:07:03 +01:00
|
|
|
M1 = simple_manifest_reader(PersistedManSQN,
|
|
|
|
RootPath),
|
|
|
|
{M1, PersistedManSQN}
|
2016-09-05 15:01:23 +01:00
|
|
|
end,
|
|
|
|
|
2016-10-07 18:07:03 +01:00
|
|
|
% Open the manifest files, completing if necessary and ensure there is
|
|
|
|
% a valid active journal at the head of the manifest
|
|
|
|
OpenManifest = open_all_manifest(Manifest, RootPath, CDBopts),
|
|
|
|
{ActiveLowSQN, _FN, ActiveJournal} = lists:nth(1, OpenManifest),
|
|
|
|
JournalSQN = case leveled_cdb:cdb_lastkey(ActiveJournal) of
|
|
|
|
empty ->
|
|
|
|
ActiveLowSQN;
|
2016-10-25 01:57:12 +01:00
|
|
|
{JSQN, _Type, _LastKey} ->
|
2016-10-07 18:07:03 +01:00
|
|
|
JSQN
|
|
|
|
end,
|
2016-09-05 15:01:23 +01:00
|
|
|
|
2016-10-07 18:07:03 +01:00
|
|
|
% Update the manifest if it has been changed by the process of laoding
|
|
|
|
% the manifest (must also increment the manifest SQN).
|
|
|
|
UpdManifestSQN = if
|
|
|
|
length(OpenManifest) > length(Manifest) ->
|
|
|
|
io:format("Updated manifest on startup: ~n"),
|
|
|
|
manifest_printer(OpenManifest),
|
|
|
|
simple_manifest_writer(OpenManifest,
|
|
|
|
ManifestSQN + 1,
|
|
|
|
RootPath),
|
|
|
|
ManifestSQN + 1;
|
|
|
|
true ->
|
|
|
|
io:format("Unchanged manifest on startup: ~n"),
|
|
|
|
manifest_printer(OpenManifest),
|
|
|
|
ManifestSQN
|
|
|
|
end,
|
|
|
|
{OpenManifest, UpdManifestSQN, JournalSQN, ActiveJournal}.
|
|
|
|
|
2016-09-05 15:01:23 +01:00
|
|
|
|
2016-10-07 18:07:03 +01:00
|
|
|
close_allmanifest([]) ->
|
|
|
|
ok;
|
|
|
|
close_allmanifest([H|ManifestT]) ->
|
2016-09-05 15:01:23 +01:00
|
|
|
{_, _, Pid} = H,
|
2016-09-06 17:17:31 +01:00
|
|
|
ok = leveled_cdb:cdb_close(Pid),
|
2016-10-07 18:07:03 +01:00
|
|
|
close_allmanifest(ManifestT).
|
|
|
|
|
|
|
|
|
|
|
|
open_all_manifest([], RootPath, CDBOpts) ->
|
|
|
|
io:format("Manifest is empty, starting from manifest SQN 1~n"),
|
|
|
|
add_to_manifest([], start_new_activejournal(1, RootPath, CDBOpts));
|
|
|
|
open_all_manifest(Man0, RootPath, CDBOpts) ->
|
|
|
|
Man1 = lists:reverse(lists:sort(Man0)),
|
|
|
|
[{HeadSQN, HeadFN}|ManifestTail] = Man1,
|
|
|
|
CompleteHeadFN = HeadFN ++ "." ++ ?JOURNAL_FILEX,
|
|
|
|
PendingHeadFN = HeadFN ++ "." ++ ?PENDING_FILEX,
|
|
|
|
Man2 = case filelib:is_file(CompleteHeadFN) of
|
|
|
|
true ->
|
|
|
|
io:format("Head manifest entry ~s is complete~n",
|
|
|
|
[HeadFN]),
|
|
|
|
{ok, HeadR} = leveled_cdb:cdb_open_reader(CompleteHeadFN),
|
2016-10-25 01:57:12 +01:00
|
|
|
{LastSQN, _Type, _PK} = leveled_cdb:cdb_lastkey(HeadR),
|
2016-10-07 18:07:03 +01:00
|
|
|
add_to_manifest(add_to_manifest(ManifestTail,
|
|
|
|
{HeadSQN, HeadFN, HeadR}),
|
|
|
|
start_new_activejournal(LastSQN + 1,
|
|
|
|
RootPath,
|
|
|
|
CDBOpts));
|
|
|
|
false ->
|
|
|
|
{ok, HeadW} = leveled_cdb:cdb_open_writer(PendingHeadFN,
|
|
|
|
CDBOpts),
|
|
|
|
add_to_manifest(ManifestTail, {HeadSQN, HeadFN, HeadW})
|
|
|
|
end,
|
|
|
|
lists:map(fun(ManEntry) ->
|
|
|
|
case ManEntry of
|
|
|
|
{LowSQN, FN} ->
|
|
|
|
CFN = FN ++ "." ++ ?JOURNAL_FILEX,
|
|
|
|
PFN = FN ++ "." ++ ?PENDING_FILEX,
|
|
|
|
case filelib:is_file(CFN) of
|
|
|
|
true ->
|
|
|
|
{ok,
|
|
|
|
Pid} = leveled_cdb:cdb_open_reader(CFN),
|
|
|
|
{LowSQN, FN, Pid};
|
|
|
|
false ->
|
|
|
|
{ok,
|
|
|
|
Pid} = leveled_cdb:cdb_open_reader(PFN),
|
|
|
|
{LowSQN, FN, Pid}
|
|
|
|
end;
|
|
|
|
_ ->
|
|
|
|
ManEntry
|
|
|
|
end end,
|
|
|
|
Man2).
|
|
|
|
|
|
|
|
|
|
|
|
start_new_activejournal(SQN, RootPath, CDBOpts) ->
|
|
|
|
Filename = filepath(RootPath, SQN, new_journal),
|
|
|
|
{ok, PidW} = leveled_cdb:cdb_open_writer(Filename, CDBOpts),
|
|
|
|
{SQN, Filename, PidW}.
|
|
|
|
|
|
|
|
add_to_manifest(Manifest, Entry) ->
|
|
|
|
{SQN, FN, PidR} = Entry,
|
|
|
|
StrippedName = filename:rootname(FN),
|
|
|
|
lists:reverse(lists:sort([{SQN, StrippedName, PidR}|Manifest])).
|
|
|
|
|
|
|
|
remove_from_manifest(Manifest, Entry) ->
|
|
|
|
{SQN, FN, _PidR} = Entry,
|
|
|
|
io:format("File ~s to be removed from manifest~n", [FN]),
|
|
|
|
lists:keydelete(SQN, 1, Manifest).
|
|
|
|
|
|
|
|
find_in_manifest(SQN, [{LowSQN, _FN, Pid}|_Tail]) when SQN >= LowSQN ->
|
|
|
|
Pid;
|
|
|
|
find_in_manifest(SQN, [_Head|Tail]) ->
|
|
|
|
find_in_manifest(SQN, Tail).
|
|
|
|
|
2016-09-05 15:01:23 +01:00
|
|
|
|
|
|
|
|
2016-09-15 10:53:24 +01:00
|
|
|
%% Scan between sequence numbers applying FilterFun to each entry where
|
|
|
|
%% FilterFun{K, V, Acc} -> Penciller Key List
|
|
|
|
%% Load the output for the CDB file into the Penciller.
|
|
|
|
|
|
|
|
load_from_sequence(_MinSQN, _FilterFun, _Penciller, []) ->
|
|
|
|
ok;
|
2016-10-05 09:54:53 +01:00
|
|
|
load_from_sequence(MinSQN, FilterFun, Penciller, [{LowSQN, FN, Pid}|Rest])
|
2016-09-19 15:31:26 +01:00
|
|
|
when LowSQN >= MinSQN ->
|
2016-10-05 09:54:53 +01:00
|
|
|
load_between_sequence(MinSQN,
|
|
|
|
MinSQN + ?LOADING_BATCH,
|
|
|
|
FilterFun,
|
|
|
|
Penciller,
|
|
|
|
Pid,
|
|
|
|
undefined,
|
|
|
|
FN,
|
|
|
|
Rest);
|
|
|
|
load_from_sequence(MinSQN, FilterFun, Penciller, [{_LowSQN, FN, Pid}|Rest]) ->
|
|
|
|
case Rest of
|
|
|
|
[] ->
|
|
|
|
load_between_sequence(MinSQN,
|
|
|
|
MinSQN + ?LOADING_BATCH,
|
2016-09-15 10:53:24 +01:00
|
|
|
FilterFun,
|
|
|
|
Penciller,
|
2016-10-05 09:54:53 +01:00
|
|
|
Pid,
|
|
|
|
undefined,
|
|
|
|
FN,
|
|
|
|
Rest);
|
2016-10-13 17:51:47 +01:00
|
|
|
[{NextSQN, _NxtFN, _NxtPid}|_Rest] when NextSQN > MinSQN ->
|
2016-10-05 09:54:53 +01:00
|
|
|
load_between_sequence(MinSQN,
|
|
|
|
MinSQN + ?LOADING_BATCH,
|
|
|
|
FilterFun,
|
|
|
|
Penciller,
|
|
|
|
Pid,
|
|
|
|
undefined,
|
|
|
|
FN,
|
|
|
|
Rest);
|
|
|
|
_ ->
|
|
|
|
load_from_sequence(MinSQN, FilterFun, Penciller, Rest)
|
|
|
|
end.
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
load_between_sequence(MinSQN, MaxSQN, FilterFun, Penciller,
|
|
|
|
CDBpid, StartPos, FN, Rest) ->
|
|
|
|
io:format("Loading from filename ~s from SQN ~w~n", [FN, MinSQN]),
|
2016-10-20 02:23:45 +01:00
|
|
|
InitAcc = {MinSQN, MaxSQN, gb_trees:empty()},
|
2016-10-05 09:54:53 +01:00
|
|
|
Res = case leveled_cdb:cdb_scan(CDBpid, FilterFun, InitAcc, StartPos) of
|
|
|
|
{eof, {AccMinSQN, _AccMaxSQN, AccKL}} ->
|
|
|
|
ok = push_to_penciller(Penciller, AccKL),
|
|
|
|
{ok, AccMinSQN};
|
|
|
|
{LastPosition, {_AccMinSQN, _AccMaxSQN, AccKL}} ->
|
|
|
|
ok = push_to_penciller(Penciller, AccKL),
|
|
|
|
NextSQN = MaxSQN + 1,
|
|
|
|
load_between_sequence(NextSQN,
|
|
|
|
NextSQN + ?LOADING_BATCH,
|
|
|
|
FilterFun,
|
|
|
|
Penciller,
|
|
|
|
CDBpid,
|
|
|
|
LastPosition,
|
|
|
|
FN,
|
|
|
|
Rest)
|
|
|
|
end,
|
|
|
|
case Res of
|
|
|
|
{ok, LMSQN} ->
|
|
|
|
load_from_sequence(LMSQN, FilterFun, Penciller, Rest);
|
|
|
|
ok ->
|
|
|
|
ok
|
2016-09-15 10:53:24 +01:00
|
|
|
end.
|
2016-09-09 15:58:19 +01:00
|
|
|
|
2016-10-20 02:23:45 +01:00
|
|
|
push_to_penciller(Penciller, KeyTree) ->
|
|
|
|
% The push to penciller must start as a tree to correctly de-duplicate
|
|
|
|
% the list by order before becoming a de-duplicated list for loading
|
2016-10-27 20:56:18 +01:00
|
|
|
R = leveled_penciller:pcl_pushmem(Penciller, KeyTree),
|
|
|
|
case R of
|
2016-10-30 18:25:30 +00:00
|
|
|
returned ->
|
2016-10-20 02:23:45 +01:00
|
|
|
timer:sleep(?LOADING_PAUSE),
|
|
|
|
push_to_penciller(Penciller, KeyTree);
|
2016-10-27 20:56:18 +01:00
|
|
|
ok ->
|
2016-09-15 10:53:24 +01:00
|
|
|
ok
|
|
|
|
end.
|
|
|
|
|
2016-09-05 15:01:23 +01:00
|
|
|
|
|
|
|
sequencenumbers_fromfilenames(Filenames, Regex, IntName) ->
|
|
|
|
lists:foldl(fun(FN, Acc) ->
|
|
|
|
case re:run(FN,
|
|
|
|
Regex,
|
|
|
|
[{capture, [IntName], list}]) of
|
|
|
|
nomatch ->
|
|
|
|
Acc;
|
|
|
|
{match, [Int]} when is_list(Int) ->
|
|
|
|
Acc ++ [list_to_integer(Int)];
|
|
|
|
_ ->
|
|
|
|
Acc
|
|
|
|
end end,
|
|
|
|
[],
|
|
|
|
Filenames).
|
|
|
|
|
|
|
|
|
|
|
|
filepath(RootPath, journal_dir) ->
|
|
|
|
RootPath ++ "/" ++ ?FILES_FP ++ "/";
|
|
|
|
filepath(RootPath, manifest_dir) ->
|
2016-09-27 14:58:26 +01:00
|
|
|
RootPath ++ "/" ++ ?MANIFEST_FP ++ "/";
|
|
|
|
filepath(RootPath, journal_compact_dir) ->
|
|
|
|
filepath(RootPath, journal_dir) ++ "/" ++ ?COMPACT_FP ++ "/".
|
2016-09-05 15:01:23 +01:00
|
|
|
|
|
|
|
filepath(RootPath, NewSQN, new_journal) ->
|
|
|
|
filename:join(filepath(RootPath, journal_dir),
|
2016-10-07 18:07:03 +01:00
|
|
|
integer_to_list(NewSQN) ++ "_"
|
2016-10-18 01:59:03 +01:00
|
|
|
++ leveled_codec:generate_uuid()
|
2016-09-28 11:41:56 +01:00
|
|
|
++ "." ++ ?PENDING_FILEX);
|
|
|
|
filepath(CompactFilePath, NewSQN, compact_journal) ->
|
|
|
|
filename:join(CompactFilePath,
|
2016-10-07 18:07:03 +01:00
|
|
|
integer_to_list(NewSQN) ++ "_"
|
2016-10-18 01:59:03 +01:00
|
|
|
++ leveled_codec:generate_uuid()
|
2016-09-05 15:01:23 +01:00
|
|
|
++ "." ++ ?PENDING_FILEX).
|
|
|
|
|
2016-10-07 18:07:03 +01:00
|
|
|
|
2016-09-05 15:01:23 +01:00
|
|
|
simple_manifest_reader(SQN, RootPath) ->
|
|
|
|
ManifestPath = filepath(RootPath, manifest_dir),
|
2016-09-19 15:31:26 +01:00
|
|
|
io:format("Opening manifest file at ~s with SQN ~w~n",
|
|
|
|
[ManifestPath, SQN]),
|
2016-09-05 15:01:23 +01:00
|
|
|
{ok, MBin} = file:read_file(filename:join(ManifestPath,
|
|
|
|
integer_to_list(SQN)
|
|
|
|
++ ".man")),
|
|
|
|
binary_to_term(MBin).
|
|
|
|
|
|
|
|
|
|
|
|
simple_manifest_writer(Manifest, ManSQN, RootPath) ->
|
|
|
|
ManPath = filepath(RootPath, manifest_dir),
|
2016-09-19 15:31:26 +01:00
|
|
|
NewFN = filename:join(ManPath,
|
|
|
|
integer_to_list(ManSQN) ++ "." ++ ?MANIFEST_FILEX),
|
|
|
|
TmpFN = filename:join(ManPath,
|
|
|
|
integer_to_list(ManSQN) ++ "." ++ ?PENDING_FILEX),
|
|
|
|
MBin = term_to_binary(lists:map(fun({SQN, FN, _PID}) -> {SQN, FN} end,
|
2016-10-19 17:34:58 +01:00
|
|
|
Manifest), [compressed]),
|
2016-09-06 17:17:31 +01:00
|
|
|
case filelib:is_file(NewFN) of
|
2016-09-05 15:01:23 +01:00
|
|
|
true ->
|
|
|
|
io:format("Error - trying to write manifest for"
|
|
|
|
++ " ManifestSQN=~w which already exists~n", [ManSQN]),
|
|
|
|
error;
|
|
|
|
false ->
|
|
|
|
io:format("Writing new version of manifest for "
|
|
|
|
++ " manifestSQN=~w~n", [ManSQN]),
|
|
|
|
ok = file:write_file(TmpFN, MBin),
|
|
|
|
ok = file:rename(TmpFN, NewFN),
|
|
|
|
ok
|
|
|
|
end.
|
|
|
|
|
2016-09-07 17:58:12 +01:00
|
|
|
manifest_printer(Manifest) ->
|
2016-10-30 22:06:44 +00:00
|
|
|
lists:foreach(fun({SQN, FN, _PID}) ->
|
|
|
|
io:format("At SQN=~w journal has filename ~s~n",
|
2016-09-07 17:58:12 +01:00
|
|
|
[SQN, FN]) end,
|
|
|
|
Manifest).
|
2016-09-05 15:01:23 +01:00
|
|
|
|
2016-10-05 18:28:31 +01:00
|
|
|
initiate_penciller_snapshot(Bookie) ->
|
|
|
|
{ok,
|
2016-10-19 00:10:48 +01:00
|
|
|
{LedgerSnap, LedgerCache},
|
2016-10-05 18:28:31 +01:00
|
|
|
_} = leveled_bookie:book_snapshotledger(Bookie, self(), undefined),
|
2016-10-27 20:56:18 +01:00
|
|
|
ok = leveled_penciller:pcl_loadsnapshot(LedgerSnap, LedgerCache),
|
2016-10-05 18:28:31 +01:00
|
|
|
MaxSQN = leveled_penciller:pcl_getstartupsequencenumber(LedgerSnap),
|
|
|
|
{LedgerSnap, MaxSQN}.
|
2016-09-28 18:26:52 +01:00
|
|
|
|
2016-09-05 15:01:23 +01:00
|
|
|
%%%============================================================================
|
|
|
|
%%% Test
|
|
|
|
%%%============================================================================
|
|
|
|
|
|
|
|
-ifdef(TEST).
|
|
|
|
|
|
|
|
build_dummy_journal() ->
|
2016-10-25 23:13:14 +01:00
|
|
|
F = fun(X) -> X end,
|
|
|
|
build_dummy_journal(F).
|
|
|
|
|
|
|
|
build_dummy_journal(KeyConvertF) ->
|
2016-09-08 14:21:30 +01:00
|
|
|
RootPath = "../test/journal",
|
2016-09-28 18:26:52 +01:00
|
|
|
clean_testdir(RootPath),
|
2016-09-05 15:01:23 +01:00
|
|
|
JournalFP = filepath(RootPath, journal_dir),
|
|
|
|
ManifestFP = filepath(RootPath, manifest_dir),
|
|
|
|
ok = filelib:ensure_dir(RootPath),
|
|
|
|
ok = filelib:ensure_dir(JournalFP),
|
|
|
|
ok = filelib:ensure_dir(ManifestFP),
|
|
|
|
F1 = filename:join(JournalFP, "nursery_1.pnd"),
|
|
|
|
{ok, J1} = leveled_cdb:cdb_open_writer(F1),
|
2016-10-25 23:13:14 +01:00
|
|
|
{K1, V1} = {KeyConvertF("Key1"), "TestValue1"},
|
|
|
|
{K2, V2} = {KeyConvertF("Key2"), "TestValue2"},
|
2016-10-25 01:57:12 +01:00
|
|
|
ok = leveled_cdb:cdb_put(J1, {1, stnd, K1}, term_to_binary({V1, []})),
|
|
|
|
ok = leveled_cdb:cdb_put(J1, {2, stnd, K2}, term_to_binary({V2, []})),
|
2016-10-07 18:07:03 +01:00
|
|
|
ok = leveled_cdb:cdb_roll(J1),
|
|
|
|
_LK = leveled_cdb:cdb_lastkey(J1),
|
|
|
|
ok = leveled_cdb:cdb_close(J1),
|
2016-09-05 15:01:23 +01:00
|
|
|
F2 = filename:join(JournalFP, "nursery_3.pnd"),
|
|
|
|
{ok, J2} = leveled_cdb:cdb_open_writer(F2),
|
2016-10-25 23:13:14 +01:00
|
|
|
{K1, V3} = {KeyConvertF("Key1"), "TestValue3"},
|
|
|
|
{K4, V4} = {KeyConvertF("Key4"), "TestValue4"},
|
2016-10-25 01:57:12 +01:00
|
|
|
ok = leveled_cdb:cdb_put(J2, {3, stnd, K1}, term_to_binary({V3, []})),
|
|
|
|
ok = leveled_cdb:cdb_put(J2, {4, stnd, K4}, term_to_binary({V4, []})),
|
2016-09-05 15:01:23 +01:00
|
|
|
ok = leveled_cdb:cdb_close(J2),
|
2016-10-07 18:07:03 +01:00
|
|
|
Manifest = [{1, "../test/journal/journal_files/nursery_1"},
|
|
|
|
{3, "../test/journal/journal_files/nursery_3"}],
|
2016-09-05 15:01:23 +01:00
|
|
|
ManifestBin = term_to_binary(Manifest),
|
|
|
|
{ok, MF1} = file:open(filename:join(ManifestFP, "1.man"),
|
|
|
|
[binary, raw, read, write]),
|
|
|
|
ok = file:write(MF1, ManifestBin),
|
|
|
|
ok = file:close(MF1).
|
|
|
|
|
|
|
|
|
|
|
|
clean_testdir(RootPath) ->
|
|
|
|
clean_subdir(filepath(RootPath, journal_dir)),
|
2016-10-05 18:28:31 +01:00
|
|
|
clean_subdir(filepath(RootPath, journal_compact_dir)),
|
2016-09-05 15:01:23 +01:00
|
|
|
clean_subdir(filepath(RootPath, manifest_dir)).
|
|
|
|
|
|
|
|
clean_subdir(DirPath) ->
|
2016-10-03 23:34:28 +01:00
|
|
|
ok = filelib:ensure_dir(DirPath),
|
2016-09-05 15:01:23 +01:00
|
|
|
{ok, Files} = file:list_dir(DirPath),
|
2016-09-28 18:26:52 +01:00
|
|
|
lists:foreach(fun(FN) ->
|
|
|
|
File = filename:join(DirPath, FN),
|
|
|
|
case file:delete(File) of
|
|
|
|
ok -> io:format("Success deleting ~s~n", [File]);
|
|
|
|
_ -> io:format("Error deleting ~s~n", [File])
|
|
|
|
end
|
|
|
|
end,
|
2016-09-05 15:01:23 +01:00
|
|
|
Files).
|
|
|
|
|
|
|
|
|
2016-10-07 18:07:03 +01:00
|
|
|
simple_inker_test() ->
|
2016-09-08 14:21:30 +01:00
|
|
|
RootPath = "../test/journal",
|
2016-09-07 17:58:12 +01:00
|
|
|
build_dummy_journal(),
|
|
|
|
CDBopts = #cdb_options{max_size=300000},
|
|
|
|
{ok, Ink1} = ink_start(#inker_options{root_path=RootPath,
|
|
|
|
cdb_options=CDBopts}),
|
2016-10-07 18:07:03 +01:00
|
|
|
Obj1 = ink_get(Ink1, "Key1", 1),
|
|
|
|
?assertMatch({{1, "Key1"}, {"TestValue1", []}}, Obj1),
|
|
|
|
Obj2 = ink_get(Ink1, "Key4", 4),
|
|
|
|
?assertMatch({{4, "Key4"}, {"TestValue4", []}}, Obj2),
|
2016-09-07 17:58:12 +01:00
|
|
|
ink_close(Ink1),
|
|
|
|
clean_testdir(RootPath).
|
2016-10-08 22:15:48 +01:00
|
|
|
|
|
|
|
simple_inker_completeactivejournal_test() ->
|
|
|
|
RootPath = "../test/journal",
|
|
|
|
build_dummy_journal(),
|
|
|
|
CDBopts = #cdb_options{max_size=300000},
|
2016-10-16 16:58:55 +01:00
|
|
|
JournalFP = filepath(RootPath, journal_dir),
|
|
|
|
F2 = filename:join(JournalFP, "nursery_3.pnd"),
|
|
|
|
{ok, PidW} = leveled_cdb:cdb_open_writer(F2),
|
|
|
|
{ok, _F2} = leveled_cdb:cdb_complete(PidW),
|
2016-10-18 01:59:03 +01:00
|
|
|
F1 = filename:join(JournalFP, "nursery_1.cdb"),
|
|
|
|
F1r = filename:join(JournalFP, "nursery_1.pnd"),
|
|
|
|
ok = file:rename(F1, F1r),
|
2016-10-08 22:15:48 +01:00
|
|
|
{ok, Ink1} = ink_start(#inker_options{root_path=RootPath,
|
|
|
|
cdb_options=CDBopts}),
|
|
|
|
Obj1 = ink_get(Ink1, "Key1", 1),
|
|
|
|
?assertMatch({{1, "Key1"}, {"TestValue1", []}}, Obj1),
|
|
|
|
Obj2 = ink_get(Ink1, "Key4", 4),
|
|
|
|
?assertMatch({{4, "Key4"}, {"TestValue4", []}}, Obj2),
|
|
|
|
ink_close(Ink1),
|
|
|
|
clean_testdir(RootPath).
|
2016-09-07 17:58:12 +01:00
|
|
|
|
2016-10-25 23:13:14 +01:00
|
|
|
test_ledgerkey(Key) ->
|
|
|
|
{o, "Bucket", Key, null}.
|
|
|
|
|
2016-09-28 18:26:52 +01:00
|
|
|
compact_journal_test() ->
|
|
|
|
RootPath = "../test/journal",
|
2016-10-25 23:13:14 +01:00
|
|
|
build_dummy_journal(fun test_ledgerkey/1),
|
2016-09-28 18:26:52 +01:00
|
|
|
CDBopts = #cdb_options{max_size=300000},
|
2016-10-25 23:13:14 +01:00
|
|
|
RStrategy = [{?STD_TAG, recovr}],
|
2016-09-28 18:26:52 +01:00
|
|
|
{ok, Ink1} = ink_start(#inker_options{root_path=RootPath,
|
2016-10-25 23:13:14 +01:00
|
|
|
cdb_options=CDBopts,
|
|
|
|
reload_strategy=RStrategy}),
|
|
|
|
{ok, NewSQN1, _ObjSize} = ink_put(Ink1,
|
|
|
|
test_ledgerkey("KeyAA"),
|
|
|
|
"TestValueAA", []),
|
2016-09-28 18:26:52 +01:00
|
|
|
?assertMatch(NewSQN1, 5),
|
|
|
|
ok = ink_print_manifest(Ink1),
|
2016-10-25 23:13:14 +01:00
|
|
|
R0 = ink_get(Ink1, test_ledgerkey("KeyAA"), 5),
|
|
|
|
?assertMatch(R0, {{5, test_ledgerkey("KeyAA")}, {"TestValueAA", []}}),
|
2016-10-07 18:07:03 +01:00
|
|
|
FunnyLoop = lists:seq(1, 48),
|
2016-09-28 18:26:52 +01:00
|
|
|
Checker = lists:map(fun(X) ->
|
|
|
|
PK = "KeyZ" ++ integer_to_list(X),
|
|
|
|
{ok, SQN, _} = ink_put(Ink1,
|
2016-10-25 23:13:14 +01:00
|
|
|
test_ledgerkey(PK),
|
2016-09-28 18:26:52 +01:00
|
|
|
crypto:rand_bytes(10000),
|
|
|
|
[]),
|
2016-10-25 23:13:14 +01:00
|
|
|
{SQN, test_ledgerkey(PK)}
|
2016-09-28 18:26:52 +01:00
|
|
|
end,
|
|
|
|
FunnyLoop),
|
2016-10-25 23:13:14 +01:00
|
|
|
{ok, NewSQN2, _ObjSize} = ink_put(Ink1,
|
|
|
|
test_ledgerkey("KeyBB"),
|
|
|
|
"TestValueBB", []),
|
2016-09-28 18:26:52 +01:00
|
|
|
?assertMatch(NewSQN2, 54),
|
|
|
|
ActualManifest = ink_getmanifest(Ink1),
|
2016-10-07 18:07:03 +01:00
|
|
|
ok = ink_print_manifest(Ink1),
|
|
|
|
?assertMatch(3, length(ActualManifest)),
|
2016-09-28 18:26:52 +01:00
|
|
|
ok = ink_compactjournal(Ink1,
|
|
|
|
Checker,
|
2016-10-05 18:28:31 +01:00
|
|
|
fun(X) -> {X, 55} end,
|
2016-09-28 18:26:52 +01:00
|
|
|
fun(L, K, SQN) -> lists:member({SQN, K}, L) end,
|
|
|
|
5000),
|
|
|
|
timer:sleep(1000),
|
2016-10-07 18:07:03 +01:00
|
|
|
CompactedManifest1 = ink_getmanifest(Ink1),
|
|
|
|
?assertMatch(2, length(CompactedManifest1)),
|
|
|
|
Checker2 = lists:sublist(Checker, 16),
|
|
|
|
ok = ink_compactjournal(Ink1,
|
|
|
|
Checker2,
|
|
|
|
fun(X) -> {X, 55} end,
|
|
|
|
fun(L, K, SQN) -> lists:member({SQN, K}, L) end,
|
|
|
|
5000),
|
|
|
|
timer:sleep(1000),
|
|
|
|
CompactedManifest2 = ink_getmanifest(Ink1),
|
|
|
|
R = lists:foldl(fun({_SQN, FN, _P}, Acc) ->
|
|
|
|
case string:str(FN, "post_compact") of
|
|
|
|
N when N > 0 ->
|
|
|
|
true;
|
|
|
|
0 ->
|
|
|
|
Acc
|
|
|
|
end end,
|
|
|
|
false,
|
|
|
|
CompactedManifest2),
|
|
|
|
?assertMatch(true, R),
|
|
|
|
?assertMatch(2, length(CompactedManifest2)),
|
2016-09-28 18:26:52 +01:00
|
|
|
ink_close(Ink1),
|
|
|
|
clean_testdir(RootPath).
|
2016-09-05 15:01:23 +01:00
|
|
|
|
2016-10-07 18:07:03 +01:00
|
|
|
empty_manifest_test() ->
|
|
|
|
RootPath = "../test/journal",
|
|
|
|
clean_testdir(RootPath),
|
|
|
|
CDBopts = #cdb_options{max_size=300000},
|
|
|
|
{ok, Ink1} = ink_start(#inker_options{root_path=RootPath,
|
|
|
|
cdb_options=CDBopts}),
|
|
|
|
?assertMatch(not_present, ink_fetch(Ink1, "Key1", 1)),
|
|
|
|
ok = ink_compactjournal(Ink1,
|
|
|
|
[],
|
|
|
|
fun(X) -> {X, 55} end,
|
|
|
|
fun(L, K, SQN) -> lists:member({SQN, K}, L) end,
|
|
|
|
5000),
|
|
|
|
timer:sleep(1000),
|
|
|
|
?assertMatch(1, length(ink_getmanifest(Ink1))),
|
|
|
|
ok = ink_close(Ink1),
|
|
|
|
{ok, Ink2} = ink_start(#inker_options{root_path=RootPath,
|
|
|
|
cdb_options=CDBopts}),
|
|
|
|
?assertMatch(not_present, ink_fetch(Ink2, "Key1", 1)),
|
|
|
|
{ok, SQN, Size} = ink_put(Ink2, "Key1", "Value1", []),
|
|
|
|
?assertMatch(2, SQN),
|
|
|
|
?assertMatch(true, Size > 0),
|
|
|
|
{ok, V} = ink_fetch(Ink2, "Key1", 2),
|
|
|
|
?assertMatch("Value1", V),
|
|
|
|
ink_close(Ink2),
|
|
|
|
clean_testdir(RootPath).
|
|
|
|
|
|
|
|
|
2016-09-05 15:01:23 +01:00
|
|
|
-endif.
|