summaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/atomic.erl24
-rw-r--r--src/fsyncport.erl88
-rw-r--r--src/hex.erl1
-rw-r--r--src/index.erl87
-rw-r--r--src/perm.erl75
-rw-r--r--src/plop_sup.erl5
-rw-r--r--src/util.erl57
7 files changed, 337 insertions, 0 deletions
diff --git a/src/atomic.erl b/src/atomic.erl
new file mode 100644
index 0000000..5bf5670
--- /dev/null
+++ b/src/atomic.erl
@@ -0,0 +1,24 @@
+%%
+%% Copyright (c) 2014 Kungliga Tekniska Högskolan
+%% (KTH Royal Institute of Technology, Stockholm, Sweden).
+%%
+
+-module(atomic).
+-export([replacefile/2, readfile/1]).
+
+-spec replacefile(string(), binary()) -> ok.
+replacefile(Path, Content) ->
+ TempName = util:tempfilename(Path),
+ util:write_tempfile_and_rename(Path, TempName, Content),
+ util:fsync([Path, filename:dirname(Path)]).
+
+-spec readfile(string()) -> binary().
+readfile(Path) ->
+ case file:read_file(Path) of
+ {ok, Contents} ->
+ Contents;
+ {error, enoent} ->
+ noentry;
+ {error, Error} ->
+ util:exit_with_error(readfile, Error, "Error reading file")
+ end.
diff --git a/src/fsyncport.erl b/src/fsyncport.erl
new file mode 100644
index 0000000..8bc8c60
--- /dev/null
+++ b/src/fsyncport.erl
@@ -0,0 +1,88 @@
+%%
+%% Copyright (c) 2014 Kungliga Tekniska Högskolan
+%% (KTH Royal Institute of Technology, Stockholm, Sweden).
+%%
+
+-module(fsyncport).
+-export([start_link/0, stop/0, init/1]).
+-export([fsync/1]).
+
+start_link() ->
+ Pid = spawn(?MODULE, init, [code:priv_dir(plop) ++ "/fsynchelper"]),
+ {ok, Pid}.
+stop() ->
+ fsyncport ! stop.
+
+fsync(Path) ->
+ call_port({fsync, Path}).
+
+call_port(Msg) ->
+ fsyncport ! {call, self(), Msg},
+ receive
+ {fsyncport, Result} ->
+ Result
+ end.
+
+init(ExtPrg) ->
+ register(fsyncport, self()),
+ process_flag(trap_exit, true),
+ Ports = lists:map(fun(_N) -> open_port({spawn_executable, ExtPrg},
+ [{packet, 2}]) end,
+ lists:seq(1, 32)),
+ loop(Ports).
+
+loop(Ports) ->
+ loop(Ports, dict:new(), queue:new()).
+loop(IdlePorts, BusyPorts, Waiting) ->
+ receive
+ {call, Caller, {fsync, Path}} ->
+ case IdlePorts of
+ [] ->
+ loop(IdlePorts,
+ BusyPorts,
+ queue:in({Caller, Path}, Waiting));
+ [Port | Rest] ->
+ Port ! {self(), {command, Path}},
+ loop(Rest,
+ dict:store(Port, {Caller, os:timestamp()}, BusyPorts),
+ Waiting)
+ end;
+
+ {Port, {data, Data}} when is_port(Port) ->
+ {Caller, Starttime} = dict:fetch(Port, BusyPorts),
+ Stoptime = os:timestamp(),
+ statreport({fsync, Stoptime, Starttime}),
+ Caller ! {fsyncport, list_to_atom(Data)},
+ case queue:out(Waiting) of
+ {empty, _} ->
+ loop([Port | IdlePorts],
+ dict:erase(Port, BusyPorts),
+ Waiting);
+ {{value, {NewCaller, NewPath}}, NewWaiting} ->
+ IdlePorts = [],
+ Port ! {self(), {command, NewPath}},
+ loop(IdlePorts,
+ dict:store(Port, {NewCaller, os:timestamp()},
+ BusyPorts),
+ NewWaiting)
+ end;
+ stop ->
+ lists:foreach(fun (Port) ->
+ Port ! {self(), close}
+ end,
+ IdlePorts),
+ lists:foreach(fun ({Port, {_Caller, _Starttime}}) ->
+ Port ! {self(), close}
+ end,
+ dict:to_list(BusyPorts)),
+ receive
+ {Port, closed} when is_port(Port) ->
+ exit(normal) %% XXX exits when first port is closed
+ end;
+ {'EXIT', Port, _Reason} when is_port(Port) ->
+ %% XXX supervisor doesn't restart fsyncport, why?
+ exit(port_terminated)
+ end.
+
+statreport(_Entry) ->
+ none.
diff --git a/src/hex.erl b/src/hex.erl
index e3c8441..1eb1e6a 100644
--- a/src/hex.erl
+++ b/src/hex.erl
@@ -4,6 +4,7 @@
-module(hex).
-export([bin_to_hexstr/1,hexstr_to_bin/1]).
+-spec bin_to_hexstr(binary()) -> string().
bin_to_hexstr(Bin) ->
lists:flatten([io_lib:format("~2.16.0B", [X]) ||
X <- binary_to_list(Bin)]).
diff --git a/src/index.erl b/src/index.erl
new file mode 100644
index 0000000..5fd468b
--- /dev/null
+++ b/src/index.erl
@@ -0,0 +1,87 @@
+%%
+%% Copyright (c) 2014 Kungliga Tekniska Högskolan
+%% (KTH Royal Institute of Technology, Stockholm, Sweden).
+%%
+
+%% Implements an interface to a file pair (basename and basename.chksum)
+%% that stores an ordered list of fixed-size entries. Entries can be
+%% added at the end and are retrieved by index. The list can also be
+%% truncated.
+%%
+%% Writes(add, truncate, addlast) need to be serialized.
+
+%% TODO: Checksums
+
+-module(index).
+-export([get/2, add/3, addlast/2, truncate/2]).
+
+-define(ENTRYSIZE, 32).
+-define(ENTRYSIZEINFILE, (?ENTRYSIZE*2+1)).
+
+-spec add(string(), integer() | last, binary()) -> ok.
+add(Basepath, Index, Entry) when is_binary(Entry), size(Entry) == ?ENTRYSIZE ->
+ case file:open(Basepath, [read, write, binary]) of
+ {ok, File} ->
+ {ok, Position} = file:position(File, eof),
+ case Index of
+ last when Position rem ?ENTRYSIZEINFILE == 0 ->
+ ok;
+ Index when is_integer(Index),
+ Index * ?ENTRYSIZEINFILE == Position ->
+ ok
+ end,
+ EntryText = hex:bin_to_hexstr(Entry) ++ "\n",
+ ok = file:write(File, EntryText),
+ ok = file:close(File),
+ util:fsync([Basepath, filename:dirname(Basepath)]);
+ {error, Error} ->
+ util:exit_with_error(Error, writefile,
+ "Error opening file for writing")
+ end.
+
+truncate(Basepath, Index) ->
+ case file:open(Basepath, [read, write, binary]) of
+ {ok, File} ->
+ {ok, _Position} = file:position(File, Index * ?ENTRYSIZEINFILE),
+ ok = file:truncate(File),
+ ok = file:close(File),
+ util:fsync([Basepath, filename:dirname(Basepath)]);
+ {error, Error} ->
+ util:exit_with_error(Error, writefile,
+ "Error opening file for writing")
+ end.
+
+
+-spec addlast(string(), integer()) -> ok.
+addlast(Basepath, Entry) ->
+ add(Basepath, last, Entry).
+
+decodedata(EntryText) when length(EntryText) == ?ENTRYSIZEINFILE ->
+ case [lists:last(EntryText)] of
+ "\n" ->
+ hex:hexstr_to_bin(lists:droplast(EntryText));
+ _ ->
+ util:exit_with_error(badformat, readindex,
+ "Index line not ending with linefeed")
+ end.
+
+-spec get(string(), integer()) -> binary().
+get(Basepath, Index) ->
+ case file:open(Basepath, [read, binary]) of
+ {ok, File} ->
+ {ok, Filesize} = file:position(File, eof),
+ if
+ Index * ?ENTRYSIZEINFILE + ?ENTRYSIZEINFILE =< Filesize ->
+ {ok, _Position} = file:position(File,
+ Index * ?ENTRYSIZEINFILE),
+ {ok, EntryText} = file:read(File, ?ENTRYSIZEINFILE),
+ Entry = decodedata(binary_to_list(EntryText)),
+ file:close(File),
+ Entry;
+ true ->
+ noentry
+ end;
+ {error, Error} ->
+ util:exit_with_error(Error, readfile,
+ "Error opening file for reading")
+ end.
diff --git a/src/perm.erl b/src/perm.erl
new file mode 100644
index 0000000..ccb23bc
--- /dev/null
+++ b/src/perm.erl
@@ -0,0 +1,75 @@
+%%
+%% Copyright (c) 2014 Kungliga Tekniska Högskolan
+%% (KTH Royal Institute of Technology, Stockholm, Sweden).
+%%
+
+-module(perm).
+-export([ensurefile/3, readfile/2]).
+
+-spec readfile_and_verify(string(), binary()) -> ok | differ | {error, atom()}.
+readfile_and_verify(Name, Content) ->
+ case file:read_file(Name) of
+ {ok, ContentsRead} when Content == ContentsRead ->
+ ok;
+ {ok, _ContentsRead} ->
+ differ;
+ {error, Error} ->
+ {error, Error}
+ end.
+
+-spec make_dir(string()) -> ok | {error, atom()}.
+make_dir(Name) ->
+ case file:make_dir(Name) of
+ ok ->
+ ok;
+ {error, eexist} ->
+ ok;
+ {error, Error} ->
+ {error, Error}
+ end.
+
+-spec make_dirs([string()]) -> ok | {error, atom()}.
+make_dirs([]) ->
+ ok;
+make_dirs([Name | Rest]) ->
+ case make_dir(Name) of
+ ok ->
+ make_dirs(Rest);
+ {error, Error} ->
+ {error, Error}
+ end.
+
+-spec path_for_key(string(), binary()) -> {[string()], string()}.
+path_for_key(Rootdir, Key) ->
+ Name = hex:bin_to_hexstr(Key),
+ [C1, C2, C3, C4, C5, C6 | _] = Name,
+ Firstlevel = Rootdir ++ [C1, C2],
+ Secondlevel = Firstlevel ++ "/" ++ [C3, C4],
+ Thirdlevel = Secondlevel ++ "/" ++ [C5, C6],
+ Fullpath = Thirdlevel ++ "/" ++ Name,
+ {[Firstlevel, Secondlevel, Thirdlevel], Fullpath}.
+
+-spec ensurefile(string(), binary(), binary()) -> ok | differ.
+ensurefile(Rootdir, Key, Content) ->
+ {Dirs, Path} = path_for_key(Rootdir, Key),
+ case readfile_and_verify(Path, Content) of
+ ok ->
+ util:fsync([Path, Rootdir | Dirs]);
+ differ ->
+ differ;
+ {error, enoent} ->
+ util:check_error(make_dirs([Rootdir, Rootdir ++ "nursery/"]
+ ++ Dirs),
+ makedir, "Error creating directory"),
+ NurseryName = Rootdir ++ "nursery/" ++
+ util:tempfilename(hex:bin_to_hexstr(Key)),
+ util:write_tempfile_and_rename(Path, NurseryName, Content),
+ util:fsync([Path, Rootdir | Dirs]);
+ {error, Error} ->
+ util:exit_with_error(Error, readfile, "Error reading file")
+ end.
+
+-spec readfile(string(), binary()) -> binary().
+readfile(Rootdir, Key) ->
+ {_Dirs, Path} = path_for_key(Rootdir, Key),
+ atomic:readfile(Path).
diff --git a/src/plop_sup.erl b/src/plop_sup.erl
index a5ce905..bcb9756 100644
--- a/src/plop_sup.erl
+++ b/src/plop_sup.erl
@@ -23,6 +23,11 @@ init(Args) ->
permanent,
10000,
worker, [db]},
+ {fsync,
+ {fsyncport, start_link, []},
+ permanent,
+ 10000,
+ worker, [fsyncport]},
{the_ht,
{ht, start_link, []},
permanent,
diff --git a/src/util.erl b/src/util.erl
new file mode 100644
index 0000000..48ebbb0
--- /dev/null
+++ b/src/util.erl
@@ -0,0 +1,57 @@
+%%
+%% Copyright (c) 2014 Kungliga Tekniska Högskolan
+%% (KTH Royal Institute of Technology, Stockholm, Sweden).
+%%
+
+-module(util).
+-export([tempfilename/1, fsync/1, exit_with_error/3,
+ check_error/3, write_tempfile_and_rename/3]).
+
+-spec tempfilename(string()) -> string().
+tempfilename(Base) ->
+ {MegaSecs, Secs, MicroSecs} = now(),
+ Filename = io_lib:format("~s-~s-~p.~p", [Base, os:getpid(),
+ MegaSecs * 1000000 + Secs, MicroSecs]),
+ Filename.
+
+-spec fsync([string()]) -> ok.
+fsync([]) ->
+ ok;
+fsync([Name | Rest]) ->
+ case fsyncport:fsync(Name) of
+ ok ->
+ fsync(Rest);
+ {error, Error} ->
+ exit_with_error(fsync, Error, "Error in fsync")
+ end.
+
+-spec exit_with_error(atom(), atom(), string()) -> no_return().
+exit_with_error(Operation, Error, ErrorMessage) ->
+ io:format("~s(~w): ~w~n", [ErrorMessage, Operation, Error]),
+ exit({fileerror, Operation, Error, ErrorMessage}).
+
+-spec check_error(any(), atom(), string()) -> ok.
+check_error(ReturnValue, Operation, ErrorMessage) ->
+ case ReturnValue of
+ ok ->
+ ok;
+ {error, Error} ->
+ exit_with_error(Operation, Error, ErrorMessage)
+ end.
+
+-spec write_tempfile_and_rename(string(), string(), binary()) -> ok.
+write_tempfile_and_rename(Name, NurseryName, Content) ->
+ case file:open(NurseryName, [write, exclusive]) of
+ {ok, File} ->
+ ok = file:write(File, Content),
+ file:close(File),
+ check_error(file:rename(NurseryName, Name), rename,
+ "Error when renaming tempfile to final file");
+ {error, eexist} ->
+ %% Should not happen, file name should be unique
+ exit_with_error(writefile, eexist,
+ "File existed when creating tempfile");
+ {error, Error} ->
+ exit_with_error(writefile, Error,
+ "Error when creating tempfile")
+ end.
generated by cgit v1.1 at 2024-10-07 05:17:34 +0000