diff options
Diffstat (limited to 'src')
| -rw-r--r-- | src/fsyncport.erl | 88 | ||||
| -rw-r--r-- | src/perm.erl | 95 | ||||
| -rw-r--r-- | src/plop_sup.erl | 5 |
3 files changed, 188 insertions, 0 deletions
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/perm.erl b/src/perm.erl new file mode 100644 index 0000000..2ce5b46 --- /dev/null +++ b/src/perm.erl @@ -0,0 +1,95 @@ +%% +%% Copyright (c) 2014 Kungliga Tekniska Högskolan +%% (KTH Royal Institute of Technology, Stockholm, Sweden). +%% + +-module(perm). +-export([ensurefile/3]). + +fsync(Name) -> + fsyncport:fsync(Name). + +readfile_and_verify(Name, Content) -> + case file:read_file(Name) of + {ok, ContentsReadBinary} -> + ContentsRead = binary_to_list(ContentsReadBinary), + if Content == ContentsRead -> + ok; + true -> + {error, "File contents differ"} + end; + {error, Error} -> + {error, Error} + end. + +writefile(Name, NurseryName, Content) -> + case file:open(NurseryName, [write, exclusive]) of + {ok, File} -> + %io:format("Write file: ~p~n", [Name]), + ok = file:write(File, Content), + file:close(File), + Result = file:rename(NurseryName, Name), + Result; + {error, eexist} -> + %% Should not happen, file name should be unique + {error, eexist}; + {error, Error} -> + {error, Error} + end. + +make_dir(Name) -> + case file:make_dir(Name) of + ok -> + ok; + {error, eexist} -> + ok; + {error, Error} -> + {error, Error} + end. + +make_dirs([]) -> + ok; +make_dirs([Name | Rest]) -> + case make_dir(Name) of + ok -> + make_dirs(Rest); + {error, Error} -> + {error, Error} + end. + +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}. + +tempfilename(Base) -> + {MegaSecs, Secs, MicroSecs} = now(), + Filename = io_lib:format("~s-~s-~p.~p", [Base, os:getpid(), + MegaSecs * 1000000 + Secs, MicroSecs]), + Filename. + +ensurefile(Rootdir, Key, Content) -> + {Dirs, Path} = path_for_key(Rootdir, Key), + case readfile_and_verify(Path, Content) of + ok -> + lists:foreach(fun (Dir) -> fsync(Dir) end, [Path, Rootdir | Dirs]); + {error, enoent} -> + case make_dirs([Rootdir, Rootdir ++ "nursery/"] ++ Dirs) of + ok -> + NurseryName = Rootdir ++ "nursery/" ++ + tempfilename(hex:bin_to_hexstr(Key)), + _Result = writefile(Path, NurseryName, Content), + lists:foreach(fun (Dir) -> + fsync(Dir) + end, + [Path, Rootdir | Dirs]); %% XXX check results + {error, Error} -> + io:format("Error creating directory: ~w~n", [Error]) + end; + {error, Error} -> + exit({perm, fileerror, "Error reading file", Error}) + end. 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, |