diff options
Diffstat (limited to 'lib/kernel/src')
29 files changed, 2073 insertions, 437 deletions
diff --git a/lib/kernel/src/Makefile b/lib/kernel/src/Makefile index 5946620f0f..0bc9f121a0 100644 --- a/lib/kernel/src/Makefile +++ b/lib/kernel/src/Makefile @@ -106,6 +106,7 @@ MODULES = \ inet_sctp \ kernel \ kernel_config \ + kernel_refc \ local_udp \ local_tcp \ net \ @@ -120,6 +121,13 @@ MODULES = \ user \ user_drv \ user_sup \ + raw_file_io \ + raw_file_io_compressed \ + raw_file_io_inflate \ + raw_file_io_deflate \ + raw_file_io_delayed \ + raw_file_io_list \ + raw_file_io_raw \ wrap_log_reader HRL_FILES= ../include/file.hrl ../include/inet.hrl ../include/inet_sctp.hrl \ @@ -226,7 +234,8 @@ $(EBIN)/disk_log_server.beam: disk_log.hrl $(EBIN)/dist_util.beam: ../include/dist_util.hrl ../include/dist.hrl $(EBIN)/erl_boot_server.beam: inet_boot.hrl $(EBIN)/erl_epmd.beam: inet_int.hrl erl_epmd.hrl -$(EBIN)/file.beam: ../include/file.hrl +$(EBIN)/file.beam: ../include/file.hrl file_int.hrl +$(EBIN)/file_io_server.beam: ../include/file.hrl file_int.hrl $(EBIN)/gen_tcp.beam: inet_int.hrl $(EBIN)/gen_udp.beam: inet_int.hrl $(EBIN)/gen_sctp.beam: ../include/inet_sctp.hrl @@ -254,3 +263,10 @@ $(EBIN)/net_kernel.beam: ../include/net_address.hrl $(EBIN)/os.beam: ../include/file.hrl $(EBIN)/ram_file.beam: ../include/file.hrl $(EBIN)/wrap_log_reader.beam: disk_log.hrl ../include/file.hrl +$(EBIN)/raw_file_io.beam: ../include/file.hrl file_int.hrl +$(EBIN)/raw_file_io_compressed.beam: ../include/file.hrl file_int.hrl +$(EBIN)/raw_file_io_inflate.beam: ../include/file.hrl file_int.hrl +$(EBIN)/raw_file_io_deflate.beam: ../include/file.hrl file_int.hrl +$(EBIN)/raw_file_io_delayed.beam: ../include/file.hrl file_int.hrl +$(EBIN)/raw_file_io_list.beam: ../include/file.hrl file_int.hrl +$(EBIN)/raw_file_io_raw.beam: ../include/file.hrl file_int.hrl diff --git a/lib/kernel/src/dist_util.erl b/lib/kernel/src/dist_util.erl index b3507e5d13..fb9f7fd7eb 100644 --- a/lib/kernel/src/dist_util.erl +++ b/lib/kernel/src/dist_util.erl @@ -74,6 +74,48 @@ ticked = 0 }). +dflag2str(?DFLAG_PUBLISHED) -> + "PUBLISHED"; +dflag2str(?DFLAG_ATOM_CACHE) -> + "ATOM_CACHE"; +dflag2str(?DFLAG_EXTENDED_REFERENCES) -> + "EXTENDED_REFERENCES"; +dflag2str(?DFLAG_DIST_MONITOR) -> + "DIST_MONITOR"; +dflag2str(?DFLAG_FUN_TAGS) -> + "FUN_TAGS"; +dflag2str(?DFLAG_DIST_MONITOR_NAME) -> + "DIST_MONITOR_NAME"; +dflag2str(?DFLAG_HIDDEN_ATOM_CACHE) -> + "HIDDEN_ATOM_CACHE"; +dflag2str(?DFLAG_NEW_FUN_TAGS) -> + "NEW_FUN_TAGS"; +dflag2str(?DFLAG_EXTENDED_PIDS_PORTS) -> + "EXTENDED_PIDS_PORTS"; +dflag2str(?DFLAG_EXPORT_PTR_TAG) -> + "EXPORT_PTR_TAG"; +dflag2str(?DFLAG_BIT_BINARIES) -> + "BIT_BINARIES"; +dflag2str(?DFLAG_NEW_FLOATS) -> + "NEW_FLOATS"; +dflag2str(?DFLAG_UNICODE_IO) -> + "UNICODE_IO"; +dflag2str(?DFLAG_DIST_HDR_ATOM_CACHE) -> + "DIST_HDR_ATOM_CACHE"; +dflag2str(?DFLAG_SMALL_ATOM_TAGS) -> + "SMALL_ATOM_TAGS"; +dflag2str(?DFLAG_UTF8_ATOMS) -> + "UTF8_ATOMS"; +dflag2str(?DFLAG_MAP_TAG) -> + "MAP_TAG"; +dflag2str(?DFLAG_BIG_CREATION) -> + "BIG_CREATION"; +dflag2str(?DFLAG_SEND_SENDER) -> + "SEND_SENDER"; +dflag2str(_) -> + "UNKNOWN". + + remove_flag(Flag, Flags) -> case Flags band Flag of 0 -> @@ -82,13 +124,13 @@ remove_flag(Flag, Flags) -> Flags - Flag end. -adjust_flags(ThisFlags, OtherFlags) -> +adjust_flags(ThisFlags, OtherFlags, RejectFlags) -> case (?DFLAG_PUBLISHED band ThisFlags) band OtherFlags of 0 -> {remove_flag(?DFLAG_PUBLISHED, ThisFlags), remove_flag(?DFLAG_PUBLISHED, OtherFlags)}; _ -> - {ThisFlags, OtherFlags} + {ThisFlags, OtherFlags band (bnot RejectFlags)} end. publish_flag(hidden, _) -> @@ -101,36 +143,71 @@ publish_flag(_, OtherNode) -> 0 end. -make_this_flags(RequestType, OtherNode) -> - publish_flag(RequestType, OtherNode) bor - %% The parenthesis below makes the compiler generate better code. - (?DFLAG_EXPORT_PTR_TAG bor - ?DFLAG_EXTENDED_PIDS_PORTS bor - ?DFLAG_EXTENDED_REFERENCES bor - ?DFLAG_DIST_MONITOR bor - ?DFLAG_FUN_TAGS bor - ?DFLAG_DIST_MONITOR_NAME bor - ?DFLAG_HIDDEN_ATOM_CACHE bor - ?DFLAG_NEW_FUN_TAGS bor - ?DFLAG_BIT_BINARIES bor - ?DFLAG_NEW_FLOATS bor - ?DFLAG_UNICODE_IO bor - ?DFLAG_DIST_HDR_ATOM_CACHE bor - ?DFLAG_SMALL_ATOM_TAGS bor - ?DFLAG_UTF8_ATOMS bor - ?DFLAG_MAP_TAG bor - ?DFLAG_BIG_CREATION). - -handshake_other_started(#hs_data{request_type=ReqType}=HSData0) -> +-define(DFLAGS_REMOVABLE, + (?DFLAG_DIST_HDR_ATOM_CACHE + bor ?DFLAG_HIDDEN_ATOM_CACHE + bor ?DFLAG_ATOM_CACHE)). + +-define(DFLAGS_ADDABLE, + (?DFLAGS_ALL + band (bnot (?DFLAG_PUBLISHED + bor ?DFLAG_HIDDEN_ATOM_CACHE + bor ?DFLAG_ATOM_CACHE)))). + +-define(DFLAGS_THIS_DEFAULT, + (?DFLAG_EXPORT_PTR_TAG + bor ?DFLAG_EXTENDED_PIDS_PORTS + bor ?DFLAG_EXTENDED_REFERENCES + bor ?DFLAG_DIST_MONITOR + bor ?DFLAG_FUN_TAGS + bor ?DFLAG_DIST_MONITOR_NAME + bor ?DFLAG_NEW_FUN_TAGS + bor ?DFLAG_BIT_BINARIES + bor ?DFLAG_NEW_FLOATS + bor ?DFLAG_UNICODE_IO + bor ?DFLAG_DIST_HDR_ATOM_CACHE + bor ?DFLAG_SMALL_ATOM_TAGS + bor ?DFLAG_UTF8_ATOMS + bor ?DFLAG_MAP_TAG + bor ?DFLAG_BIG_CREATION + bor ?DFLAG_SEND_SENDER)). + +make_this_flags(RequestType, AddFlags, RemoveFlags, OtherNode) -> + case RemoveFlags band (bnot ?DFLAGS_REMOVABLE) of + 0 -> ok; + Rerror -> exit({"Rejecting non rejectable flags", Rerror}) + end, + case AddFlags band (bnot ?DFLAGS_ADDABLE) of + 0 -> ok; + Aerror -> exit({"Adding non addable flags", Aerror}) + end, + Flgs0 = ?DFLAGS_THIS_DEFAULT, + Flgs1 = Flgs0 bor publish_flag(RequestType, OtherNode), + Flgs2 = Flgs1 bor AddFlags, + Flgs3 = Flgs2 band (bnot (?DFLAG_HIDDEN_ATOM_CACHE + bor ?DFLAG_ATOM_CACHE)), + Flgs3 band (bnot RemoveFlags). + +handshake_other_started(#hs_data{request_type=ReqType, + add_flags=AddFlgs0, + reject_flags=RejFlgs0, + require_flags=ReqFlgs0}=HSData0) -> + AddFlgs = convert_flags(AddFlgs0), + RejFlgs = convert_flags(RejFlgs0), + ReqFlgs = convert_flags(ReqFlgs0), {PreOtherFlags,Node,Version} = recv_name(HSData0), - PreThisFlags = make_this_flags(ReqType, Node), + PreThisFlags = make_this_flags(ReqType, AddFlgs, RejFlgs, Node), {ThisFlags, OtherFlags} = adjust_flags(PreThisFlags, - PreOtherFlags), + PreOtherFlags, + RejFlgs), HSData = HSData0#hs_data{this_flags=ThisFlags, other_flags=OtherFlags, other_version=Version, other_node=Node, - other_started=true}, + other_started=true, + add_flags=AddFlgs, + reject_flags=RejFlgs, + require_flags=ReqFlgs}, check_dflags(HSData), is_allowed(HSData), ?debug({"MD5 connection from ~p (V~p)~n", @@ -165,23 +242,19 @@ is_allowed(#hs_data{other_node = Node, end. %% -%% Check that both nodes can handle the same types of extended -%% node containers. If they can not, abort the connection. +%% Check mandatory flags... %% check_dflags(#hs_data{other_node = Node, other_flags = OtherFlags, - other_started = OtherStarted} = HSData) -> - - Mandatory = [{?DFLAG_EXTENDED_REFERENCES, "EXTENDED_REFERENCES"}, - {?DFLAG_EXTENDED_PIDS_PORTS, "EXTENDED_PIDS_PORTS"}, - {?DFLAG_UTF8_ATOMS, "UTF8_ATOMS"}], - Missing = lists:filtermap(fun({Bit, Str}) -> - case Bit band OtherFlags of - Bit -> false; - 0 -> {true, Str} - end - end, - Mandatory), + other_started = OtherStarted, + require_flags = RequiredFlags} = HSData) -> + Mandatory = ((?DFLAG_EXTENDED_REFERENCES + bor ?DFLAG_EXTENDED_PIDS_PORTS + bor ?DFLAG_UTF8_ATOMS + bor ?DFLAG_NEW_FUN_TAGS) + bor RequiredFlags), + Missing = check_mandatory(0, ?DFLAGS_ALL, Mandatory, + OtherFlags, []), case Missing of [] -> ok; @@ -201,6 +274,22 @@ check_dflags(#hs_data{other_node = Node, ?shutdown2(Node, {check_dflags_failed, Missing}) end. +check_mandatory(_Bit, 0, _Mandatory, _OtherFlags, Missing) -> + Missing; +check_mandatory(Bit, Left, Mandatory, OtherFlags, Missing) -> + DFlag = (1 bsl Bit), + NewLeft = Left band (bnot DFlag), + NewMissing = case {DFlag band Mandatory, + DFlag band OtherFlags} of + {DFlag, 0} -> + %% Mandatory and missing... + [dflag2str(DFlag) | Missing]; + _ -> + %% Not mandatory or present... + Missing + end, + check_mandatory(Bit+1, NewLeft, Mandatory, OtherFlags, NewMissing). + %% No nodedown will be sent if we fail before this process has %% succeeded to mark the node as pending. @@ -314,13 +403,24 @@ flush_down() -> end. handshake_we_started(#hs_data{request_type=ReqType, - other_node=Node}=PreHSData) -> - PreThisFlags = make_this_flags(ReqType, Node), - HSData = PreHSData#hs_data{this_flags=PreThisFlags}, + other_node=Node, + add_flags=AddFlgs0, + reject_flags=RejFlgs0, + require_flags=ReqFlgs0}=PreHSData) -> + AddFlgs = convert_flags(AddFlgs0), + RejFlgs = convert_flags(RejFlgs0), + ReqFlgs = convert_flags(ReqFlgs0), + PreThisFlags = make_this_flags(ReqType, AddFlgs, RejFlgs, Node), + HSData = PreHSData#hs_data{this_flags = PreThisFlags, + add_flags = AddFlgs, + reject_flags = RejFlgs, + require_flags = ReqFlgs}, send_name(HSData), recv_status(HSData), {PreOtherFlags,ChallengeA} = recv_challenge(HSData), - {ThisFlags,OtherFlags} = adjust_flags(PreThisFlags, PreOtherFlags), + {ThisFlags,OtherFlags} = adjust_flags(PreThisFlags, + PreOtherFlags, + RejFlgs), NewHSData = HSData#hs_data{this_flags = ThisFlags, other_flags = OtherFlags, other_started = false}, @@ -336,15 +436,16 @@ handshake_we_started(#hs_data{request_type=ReqType, handshake_we_started(OldHsData) when element(1,OldHsData) =:= hs_data -> handshake_we_started(convert_old_hsdata(OldHsData)). -convert_old_hsdata({hs_data, KP, ON, TN, S, T, TF, A, OV, OF, OS, FS, FR, - FS_PRE, FS_POST, FG, FA, MFT, MFG, RT}) -> - #hs_data{ - kernel_pid = KP, other_node = ON, this_node = TN, socket = S, timer = T, - this_flags = TF, allowed = A, other_version = OV, other_flags = OF, - other_started = OS, f_send = FS, f_recv = FR, f_setopts_pre_nodeup = FS_PRE, - f_setopts_post_nodeup = FS_POST, f_getll = FG, f_address = FA, - mf_tick = MFT, mf_getstat = MFG, request_type = RT}. +convert_old_hsdata(OldHsData) -> + OHSDL = tuple_to_list(OldHsData), + NoMissing = tuple_size(#hs_data{}) - tuple_size(OldHsData), + true = NoMissing > 0, + list_to_tuple(OHSDL ++ lists:duplicate(NoMissing, undefined)). +convert_flags(Flags) when is_integer(Flags) -> + Flags; +convert_flags(_Undefined) -> + 0. %% -------------------------------------------------------------- %% The connection has been established. @@ -359,15 +460,20 @@ connection(#hs_data{other_node = Node, PType = publish_type(HSData#hs_data.other_flags), case FPreNodeup(Socket) of ok -> - do_setnode(HSData), % Succeeds or exits the process. + DHandle = do_setnode(HSData), % Succeeds or exits the process. Address = FAddress(Socket,Node), mark_nodeup(HSData,Address), case FPostNodeup(Socket) of ok -> + case HSData#hs_data.f_handshake_complete of + undefined -> ok; + HsComplete -> HsComplete(Socket, Node, DHandle) + end, con_loop({HSData#hs_data.kernel_pid, Node, Socket, PType, + DHandle, HSData#hs_data.mf_tick, HSData#hs_data.mf_getstat, HSData#hs_data.mf_setopts, @@ -425,18 +531,16 @@ do_setnode(#hs_data{other_node = Node, socket = Socket, [Node, Port, {publish_type(Flags), '(', Flags, ')', Version}]), - case (catch - erlang:setnode(Node, Port, - {Flags, Version, '', ''})) of - {'EXIT', {system_limit, _}} -> + try + erlang:setnode(Node, Port, {Flags, Version, '', ''}) + catch + error:system_limit -> error_msg("** Distribution system limit reached, " "no table space left for node ~w ** ~n", [Node]), ?shutdown(Node); - {'EXIT', Other} -> - exit(Other); - _Else -> - ok + error:Other -> + exit({Other, erlang:get_stacktrace()}) end; _ -> error_msg("** Distribution connection error, " @@ -468,7 +572,13 @@ mark_nodeup(#hs_data{kernel_pid = Kernel, ?shutdown(Node) end. -con_loop({Kernel, Node, Socket, Type, MFTick, MFGetstat, MFSetOpts, MFGetOpts}=ConData, +getstat(DHandle, _Socket, undefined) -> + erlang:dist_get_stat(DHandle); +getstat(_DHandle, Socket, MFGetstat) -> + MFGetstat(Socket). + +con_loop({Kernel, Node, Socket, Type, DHandle, MFTick, MFGetstat, + MFSetOpts, MFGetOpts}=ConData, Tick) -> receive {tcp_closed, Socket} -> @@ -476,7 +586,7 @@ con_loop({Kernel, Node, Socket, Type, MFTick, MFGetstat, MFSetOpts, MFGetOpts}=C {Kernel, disconnect} -> ?shutdown2(Node, disconnected); {Kernel, aux_tick} -> - case MFGetstat(Socket) of + case getstat(DHandle, Socket, MFGetstat) of {ok, _, _, PendWrite} -> send_tick(Socket, PendWrite, MFTick); _ -> @@ -484,7 +594,7 @@ con_loop({Kernel, Node, Socket, Type, MFTick, MFGetstat, MFSetOpts, MFGetOpts}=C end, con_loop(ConData, Tick); {Kernel, tick} -> - case send_tick(Socket, Tick, Type, + case send_tick(DHandle, Socket, Tick, Type, MFTick, MFGetstat) of {ok, NewTick} -> con_loop(ConData, NewTick); @@ -497,7 +607,7 @@ con_loop({Kernel, Node, Socket, Type, MFTick, MFGetstat, MFSetOpts, MFGetOpts}=C ?shutdown2(Node, send_net_tick_failed) end; {From, get_status} -> - case MFGetstat(Socket) of + case getstat(DHandle, Socket, MFGetstat) of {ok, Read, Write, _} -> From ! {self(), get_status, {ok, Read, Write}}, con_loop(ConData, Tick); @@ -735,14 +845,14 @@ send_status(#hs_data{socket = Socket, other_node = Node, %% we haven't read anything as a hidden node only ticks when it receives %% a TICK !! -send_tick(Socket, Tick, Type, MFTick, MFGetstat) -> +send_tick(DHandle, Socket, Tick, Type, MFTick, MFGetstat) -> #tick{tick = T0, read = Read, write = Write, ticked = Ticked} = Tick, T = T0 + 1, T1 = T rem 4, - case MFGetstat(Socket) of + case getstat(DHandle, Socket, MFGetstat) of {ok, Read, _, _} when Ticked =:= T -> {error, not_responding}; {ok, Read, W, Pend} when Type =:= hidden -> @@ -771,11 +881,10 @@ send_tick(Socket, Tick, Type, MFTick, MFGetstat) -> Error end. -send_tick(Socket, 0, MFTick) -> - MFTick(Socket); -send_tick(_, _Pend, _) -> - %% Dont send tick if pending write. - ok. +send_tick(_, Pend, _) when Pend /= false, Pend /= 0 -> + ok; %% Dont send tick if pending write. +send_tick(Socket, _Pend, MFTick) -> + MFTick(Socket). %% ------------------------------------------------------------ %% Connection setup timeout timer. diff --git a/lib/kernel/src/erl_boot_server.erl b/lib/kernel/src/erl_boot_server.erl index 2578b74428..4ac945ce01 100644 --- a/lib/kernel/src/erl_boot_server.erl +++ b/lib/kernel/src/erl_boot_server.erl @@ -252,9 +252,9 @@ handle_info({udp, U, IP, Port, Data}, S0) -> "~w is not a valid address ** ~n", [IP]), {noreply,S0}; {true,_,_} -> - case catch string:substr(Data, 1, length(?EBOOT_REQUEST)) of + case catch string:slice(Data, 0, length(?EBOOT_REQUEST)) of ?EBOOT_REQUEST -> - Vsn = string:substr(Data, length(?EBOOT_REQUEST)+1, length(Data)), + Vsn = string:slice(Data, length(?EBOOT_REQUEST), length(Data)), error_logger:error_msg("** Illegal boot server connection attempt: " "client version is ~s ** ~n", [Vsn]); _ -> diff --git a/lib/kernel/src/erl_epmd.erl b/lib/kernel/src/erl_epmd.erl index 7bc9e2ede3..f96bc88913 100644 --- a/lib/kernel/src/erl_epmd.erl +++ b/lib/kernel/src/erl_epmd.erl @@ -79,7 +79,13 @@ port_please(Node, EpmdAddr, Timeout) -> port_please1(Node,HostName, Timeout) -> - case inet:gethostbyname(HostName, inet, Timeout) of + Family = case inet_db:res_option(inet6) of + true -> + inet6; + false -> + inet + end, + case inet:gethostbyname(HostName, Family, Timeout) of {ok,{hostent, _Name, _ , _Af, _Size, [EpmdAddr | _]}} -> get_port(Node, EpmdAddr, Timeout); Else -> diff --git a/lib/kernel/src/erl_reply.erl b/lib/kernel/src/erl_reply.erl index e1e046cbb4..e1c4ffe839 100644 --- a/lib/kernel/src/erl_reply.erl +++ b/lib/kernel/src/erl_reply.erl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 1997-2016. All Rights Reserved. +%% Copyright Ericsson AB 1997-2017. All Rights Reserved. %% %% Licensed under the Apache License, Version 2.0 (the "License"); %% you may not use this file except in compliance with the License. @@ -42,7 +42,7 @@ reply(_) -> %% convert ip number to tuple ip_string_to_tuple(Ip) -> - [Ip1,Ip2,Ip3,Ip4] = string:tokens(Ip,"."), + [Ip1,Ip2,Ip3,Ip4] = string:lexemes(Ip,"."), {list_to_integer(Ip1), list_to_integer(Ip2), list_to_integer(Ip3), diff --git a/lib/kernel/src/erts_debug.erl b/lib/kernel/src/erts_debug.erl index 9662f8fa90..3456c8511e 100644 --- a/lib/kernel/src/erts_debug.erl +++ b/lib/kernel/src/erts_debug.erl @@ -21,7 +21,7 @@ %% Low-level debugging support. EXPERIMENTAL! --export([size/1,df/1,df/2,df/3,df/4,ic/1]). +-export([size/1,df/1,df/2,df/3,dis_to_file/2,ic/1]). %% This module contains the following *experimental* BIFs: %% disassemble/1 @@ -347,55 +347,52 @@ is_term_seen(_, []) -> false. -spec df(module()) -> df_ret(). df(Mod) when is_atom(Mod) -> - df(lists:concat([Mod, ".dis"]), Mod). - --spec df(module(), atom()) -> df_ret(); - (file:io_device() | file:filename(), module()) -> df_ret(). - -df(Mod, Func) when is_atom(Mod), is_atom(Func) -> - df(lists:concat([Mod, "_", Func, ".dis"]), Mod, Func); -df(Name, Mod) when is_atom(Mod) -> try Mod:module_info(functions) of Fs0 when is_list(Fs0) -> + Name = lists:concat([Mod, ".dis"]), Fs = [{Mod,Func,Arity} || {Func,Arity} <- Fs0], dff(Name, Fs) catch _:_ -> {undef,Mod} end. +-spec df(module(), atom()) -> df_ret(). --spec df(module(), atom(), arity()) -> df_ret(); - (file:io_device() | file:filename(), module(), atom()) -> df_ret(). - -df(Mod, Func, Arity) when is_atom(Mod), is_atom(Func), is_integer(Arity) -> - df(lists:concat([Mod, "_", Func, "_", Arity, ".dis"]), Mod, Func, Arity); -df(Name, Mod, Func) when is_atom(Mod), is_atom(Func) -> +df(Mod, Func) when is_atom(Mod), is_atom(Func) -> try Mod:module_info(functions) of Fs0 when is_list(Fs0) -> + Name = lists:concat([Mod, "_", Func, ".dis"]), Fs = [{Mod,Func1,Arity} || {Func1,Arity} <- Fs0, Func1 =:= Func], dff(Name, Fs) catch _:_ -> {undef,Mod} end. --spec df(file:io_device() | file:filename(), module(), atom(), arity()) -> df_ret(). -df(Name, Mod, Func, Arity) when is_atom(Mod), is_atom(Func), is_integer(Arity) -> +-spec df(module(), atom(), arity()) -> df_ret(). + +df(Mod, Func, Arity) when is_atom(Mod), is_atom(Func) -> try Mod:module_info(functions) of Fs0 when is_list(Fs0) -> + Name = lists:concat([Mod, "_", Func, "_", Arity, ".dis"]), Fs = [{Mod,Func1,Arity1} || {Func1,Arity1} <- Fs0, Func1 =:= Func, Arity1 =:= Arity], dff(Name, Fs) catch _:_ -> {undef,Mod} end. -dff(File, Fs) when is_pid(File), is_list(Fs) -> - lists:foreach(fun(Mfa) -> - disassemble_function(File, Mfa), - io:nl(File) - end, Fs); -dff(Name, Fs) when is_list(Name) -> - case file:open(Name, [write]) of +-spec dis_to_file(module(), file:filename()) -> df_ret(). + +dis_to_file(Mod, Name) when is_atom(Mod) -> + try Mod:module_info(functions) of + Fs0 when is_list(Fs0) -> + Fs = [{Mod,Func,Arity} || {Func,Arity} <- Fs0], + dff(Name, Fs) + catch _:_ -> {undef,Mod} + end. + +dff(Name, Fs) -> + case file:open(Name, [write,raw,delayed_write]) of {ok,F} -> try - dff(F, Fs) + dff_1(F, Fs) after _ = file:close(F) end; @@ -403,12 +400,18 @@ dff(Name, Fs) when is_list(Name) -> {error,{badopen,Reason}} end. +dff_1(File, Fs) -> + lists:foreach(fun(Mfa) -> + disassemble_function(File, Mfa), + file:write(File, "\n") + end, Fs). + disassemble_function(File, {_,_,_}=MFA) -> cont_dis(File, erts_debug:disassemble(MFA), MFA). cont_dis(_, false, _) -> ok; cont_dis(File, {Addr,Str,MFA}, MFA) -> - io:put_chars(File, binary_to_list(Str)), + ok = file:write(File, Str), cont_dis(File, erts_debug:disassemble(Addr), MFA); cont_dis(_, {_,_,_}, _) -> ok. diff --git a/lib/kernel/src/file.erl b/lib/kernel/src/file.erl index 933f2d5f65..d05199897f 100644 --- a/lib/kernel/src/file.erl +++ b/lib/kernel/src/file.erl @@ -72,7 +72,7 @@ io_device/0, name/0, name_all/0, posix/0]). %%% Includes and defines --include("file.hrl"). +-include("file_int.hrl"). -define(FILE_IO_SERVER_TABLE, file_io_servers). @@ -454,41 +454,23 @@ raw_write_file_info(Name, #file_info{} = Info) -> Reason :: posix() | badarg | system_limit. open(Item, ModeList) when is_list(ModeList) -> - case lists:member(raw, ModeList) of - %% Raw file, use ?PRIM_FILE to handle this file - true -> + case {lists:member(raw, ModeList), lists:member(ram, ModeList)} of + {false, false} -> + %% File server file Args = [file_name(Item) | ModeList], case check_args(Args) of ok -> [FileName | _] = Args, - %% We rely on the returned Handle (in {ok, Handle}) - %% being a pid() or a #file_descriptor{} - ?PRIM_FILE:open(FileName, ModeList); + call(open, [FileName, ModeList]); Error -> Error - end; - false -> - case lists:member(ram, ModeList) of - %% RAM file, use ?RAM_FILE to handle this file - true -> - case check_args(ModeList) of - ok -> - ?RAM_FILE:open(Item, ModeList); - Error -> - Error - end; - %% File server file - false -> - Args = [file_name(Item) | ModeList], - case check_args(Args) of - ok -> - [FileName | _] = Args, - call(open, [FileName, ModeList]); - Error -> - Error - end - end + end; + {true, _Either} -> + raw_file_io:open(file_name(Item), ModeList); + {false, true} -> + ram_file:open(Item, ModeList) end; + %% Old obsolete mode specification in atom or 2-tuple format open(Item, Mode) -> open(Item, mode_list(Mode)). @@ -1254,15 +1236,18 @@ sendfile(File, _Sock, _Offet, _Bytes, _Opts) when is_pid(File) -> sendfile(File, Sock, Offset, Bytes, []) -> sendfile(File, Sock, Offset, Bytes, ?MAX_CHUNK_SIZE, [], [], []); sendfile(File, Sock, Offset, Bytes, Opts) -> - ChunkSize0 = proplists:get_value(chunk_size, Opts, ?MAX_CHUNK_SIZE), - ChunkSize = if ChunkSize0 > ?MAX_CHUNK_SIZE -> - ?MAX_CHUNK_SIZE; - true -> ChunkSize0 - end, - %% Support for headers, trailers and options has been removed because the - %% Darwin and BSD API for using it does not play nice with - %% non-blocking sockets. See unix_efile.c for more info. - sendfile(File, Sock, Offset, Bytes, ChunkSize, [], [], Opts). + try proplists:get_value(chunk_size, Opts, ?MAX_CHUNK_SIZE) of + ChunkSize0 when is_integer(ChunkSize0) -> + ChunkSize = erlang:min(ChunkSize0, ?MAX_CHUNK_SIZE), + %% Support for headers, trailers and options has been removed + %% because the Darwin and BSD API for using it does not play nice + %% with non-blocking sockets. See unix_efile.c for more info. + sendfile(File, Sock, Offset, Bytes, ChunkSize, [], [], Opts); + _Other -> + {error, badarg} + catch + error:_ -> {error, badarg} + end. %% sendfile/2 -spec sendfile(Filename, Socket) -> diff --git a/lib/kernel/src/file_int.hrl b/lib/kernel/src/file_int.hrl new file mode 100644 index 0000000000..bafc330c04 --- /dev/null +++ b/lib/kernel/src/file_int.hrl @@ -0,0 +1,33 @@ +%% +%% %CopyrightBegin% +%% +%% Copyright Ericsson AB 2017. All Rights Reserved. +%% +%% Licensed under the Apache License, Version 2.0 (the "License"); +%% you may not use this file except in compliance with the License. +%% You may obtain a copy of the License at +%% +%% http://www.apache.org/licenses/LICENSE-2.0 +%% +%% Unless required by applicable law or agreed to in writing, software +%% distributed under the License is distributed on an "AS IS" BASIS, +%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +%% See the License for the specific language governing permissions and +%% limitations under the License. +%% +%% %CopyrightEnd% +%% + +%% +%% Internal definitions for the 'file' module and friends. +%% + +-ifndef(FILE_INTERNAL_HRL_). +-define(FILE_INTERNAL_HRL_, 1). + +-include("file.hrl"). + +-define(CALL_FD(Fd, Method, Args), + apply(Fd#file_descriptor.module, Method, [Fd | Args])). + +-endif. diff --git a/lib/kernel/src/file_io_server.erl b/lib/kernel/src/file_io_server.erl index deb7b315b1..34d5497a4a 100644 --- a/lib/kernel/src/file_io_server.erl +++ b/lib/kernel/src/file_io_server.erl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 2000-2015. All Rights Reserved. +%% Copyright Ericsson AB 2000-2017. All Rights Reserved. %% %% Licensed under the Apache License, Version 2.0 (the "License"); %% you may not use this file except in compliance with the License. @@ -28,7 +28,8 @@ -record(state, {handle,owner,mref,buf,read_mode,unic}). --define(PRIM_FILE, prim_file). +-include("file_int.hrl"). + -define(READ_SIZE_LIST, 128). -define(READ_SIZE_BINARY, (8*1024)). @@ -67,8 +68,9 @@ do_start(Spawn, Owner, FileName, ModeList) -> erlang:dt_restore_tag(Utag), %% process_flag(trap_exit, true), case parse_options(ModeList) of - {ReadMode, UnicodeMode, Opts} -> - case ?PRIM_FILE:open(FileName, Opts) of + {ReadMode, UnicodeMode, Opts0} -> + Opts = maybe_add_read_ahead(ReadMode, Opts0), + case raw_file_io:open(FileName, [raw | Opts]) of {error, Reason} = Error -> Self ! {Ref, Error}, exit(Reason); @@ -157,6 +159,24 @@ valid_enc({utf32,little}) -> valid_enc(_Other) -> {error,badarg}. +%% Add a small read_ahead buffer if the file is opened for reading +%% only in list mode and no read_ahead is already given. +maybe_add_read_ahead(binary, Opts) -> + Opts; +maybe_add_read_ahead(list, Opts) -> + P = fun(read_ahead) -> true; + ({read_ahead,_}) -> true; + (append) -> true; + (exclusive) -> true; + (write) -> true; + (_) -> false + end, + case lists:any(P, Opts) of + false -> + [{read_ahead, 4096}|Opts]; + true -> + Opts + end. server_loop(#state{mref = Mref} = State) -> receive @@ -205,7 +225,7 @@ io_reply(From, ReplyAs, Reply) -> file_request({advise,Offset,Length,Advise}, #state{handle=Handle}=State) -> - case ?PRIM_FILE:advise(Handle, Offset, Length, Advise) of + case ?CALL_FD(Handle, advise, [Offset, Length, Advise]) of {error,Reason}=Reply -> {stop,Reason,Reply,State}; Reply -> @@ -213,7 +233,7 @@ file_request({advise,Offset,Length,Advise}, end; file_request({allocate, Offset, Length}, #state{handle = Handle} = State) -> - Reply = ?PRIM_FILE:allocate(Handle, Offset, Length), + Reply = ?CALL_FD(Handle, allocate, [Offset, Length]), {reply, Reply, State}; file_request({pread,At,Sz}, State) when At =:= cur; @@ -256,7 +276,7 @@ file_request({pwrite,At,Data}, end; file_request(datasync, #state{handle=Handle}=State) -> - case ?PRIM_FILE:datasync(Handle) of + case ?CALL_FD(Handle, datasync, []) of {error,Reason}=Reply -> {stop,Reason,Reply,State}; Reply -> @@ -264,7 +284,7 @@ file_request(datasync, end; file_request(sync, #state{handle=Handle}=State) -> - case ?PRIM_FILE:sync(Handle) of + case ?CALL_FD(Handle, sync, []) of {error,Reason}=Reply -> {stop,Reason,Reply,State}; Reply -> @@ -272,7 +292,7 @@ file_request(sync, end; file_request(close, #state{handle=Handle}=State) -> - case ?PRIM_FILE:close(Handle) of + case ?CALL_FD(Handle, close, []) of {error,Reason}=Reply -> {stop,Reason,Reply,State#state{buf= <<>>}}; Reply -> @@ -288,7 +308,7 @@ file_request({position,At}, end; file_request(truncate, #state{handle=Handle}=State) -> - case ?PRIM_FILE:truncate(Handle) of + case ?CALL_FD(Handle, truncate, []) of {error,Reason}=Reply -> {stop,Reason,Reply,State#state{buf= <<>>}}; Reply -> @@ -398,7 +418,7 @@ io_request_loop([Request|Tail], %% put_chars(Chars, latin1, #state{handle=Handle, unic=latin1}=State) -> NewState = State#state{buf = <<>>}, - case ?PRIM_FILE:write(Handle, Chars) of + case ?CALL_FD(Handle, write, [Chars]) of {error,Reason}=Reply -> {stop,Reason,Reply,NewState}; Reply -> @@ -408,7 +428,7 @@ put_chars(Chars, InEncoding, #state{handle=Handle, unic=OutEncoding}=State) -> NewState = State#state{buf = <<>>}, case unicode:characters_to_binary(Chars,InEncoding,OutEncoding) of Bin when is_binary(Bin) -> - case ?PRIM_FILE:write(Handle, Bin) of + case ?CALL_FD(Handle, write, [Bin]) of {error,Reason}=Reply -> {stop,Reason,Reply,NewState}; Reply -> @@ -422,7 +442,7 @@ put_chars(Chars, InEncoding, #state{handle=Handle, unic=OutEncoding}=State) -> get_line(S, {<<>>, Cont}, OutEnc, #state{handle=Handle, read_mode=Mode, unic=InEnc}=State) -> - case ?PRIM_FILE:read(Handle, read_size(Mode)) of + case ?CALL_FD(Handle, read, [read_size(Mode)]) of {ok,Bin} -> get_line(S, convert_enc([Cont, Bin], InEnc, OutEnc), OutEnc, State); eof -> @@ -472,7 +492,7 @@ get_chars(N, OutEnc,#state{handle=Handle,buf=Buf,read_mode=ReadMode,unic=latin1} BufSize = byte_size(Buf), NeedSize = N-BufSize, Size = erlang:max(NeedSize, ?READ_SIZE_BINARY), - case ?PRIM_FILE:read(Handle, Size) of + case ?CALL_FD(Handle, read, [Size]) of {ok, B} -> if BufSize+byte_size(B) < N -> std_reply(cat(Buf, B, ReadMode,latin1,OutEnc), State); @@ -504,7 +524,7 @@ get_chars(N, OutEnc,#state{handle=Handle,buf=Buf,read_mode=ReadMode,unic=InEncod %% Need more, Try to read 4*needed in bytes... NeedSize = (N - BufCount) * 4, Size = erlang:max(NeedSize, ?READ_SIZE_BINARY), - case ?PRIM_FILE:read(Handle, Size) of + case ?CALL_FD(Handle, read, [Size]) of {ok, B} -> NewBuf = list_to_binary([Buf,B]), {NewCount,NewSplit} = count_and_find(NewBuf,N,InEncoding), @@ -544,7 +564,7 @@ get_chars(Mod, Func, XtraArg, OutEnc, #state{buf=Buf}=State) -> get_chars_empty(Mod, Func, XtraArg, S, latin1, #state{handle=Handle,read_mode=ReadMode, unic=latin1}=State) -> - case ?PRIM_FILE:read(Handle, read_size(ReadMode)) of + case ?CALL_FD(Handle, read, [read_size(ReadMode)]) of {ok,Bin} -> get_chars_apply(Mod, Func, XtraArg, S, latin1, State, Bin); eof -> @@ -554,7 +574,7 @@ get_chars_empty(Mod, Func, XtraArg, S, latin1, end; get_chars_empty(Mod, Func, XtraArg, S, OutEnc, #state{handle=Handle,read_mode=ReadMode}=State) -> - case ?PRIM_FILE:read(Handle, read_size(ReadMode)) of + case ?CALL_FD(Handle, read, [read_size(ReadMode)]) of {ok,Bin} -> get_chars_apply(Mod, Func, XtraArg, S, OutEnc, State, Bin); eof -> @@ -564,7 +584,7 @@ get_chars_empty(Mod, Func, XtraArg, S, OutEnc, end. get_chars_notempty(Mod, Func, XtraArg, S, OutEnc, #state{handle=Handle,read_mode=ReadMode,buf = B}=State) -> - case ?PRIM_FILE:read(Handle, read_size(ReadMode)) of + case ?CALL_FD(Handle, read, [read_size(ReadMode)]) of {ok,Bin} -> get_chars_apply(Mod, Func, XtraArg, S, OutEnc, State, list_to_binary([B,Bin])); eof -> @@ -918,13 +938,10 @@ cbv({utf32,little},_) -> %% Compensates ?PRIM_FILE:position/2 for the number of bytes %% we have buffered position(Handle, At, Buf) -> - ?PRIM_FILE:position( - Handle, - case At of - cur -> - {cur, -byte_size(Buf)}; - {cur, Offs} -> - {cur, Offs-byte_size(Buf)}; - _ -> - At - end). + SeekTo = + case At of + {cur, Offs} -> {cur, Offs-byte_size(Buf)}; + cur -> {cur, -byte_size(Buf)}; + _ -> At + end, + ?CALL_FD(Handle, position, [SeekTo]). diff --git a/lib/kernel/src/file_server.erl b/lib/kernel/src/file_server.erl index 6e8f64d932..ecc1ffbdd6 100644 --- a/lib/kernel/src/file_server.erl +++ b/lib/kernel/src/file_server.erl @@ -63,7 +63,7 @@ stop() -> %%% Callback functions from gen_server %%%---------------------------------------------------------------------- --type state() :: port(). % Internal type +-type state() :: term(). % Internal type %%---------------------------------------------------------------------- %% Func: init/1 @@ -77,14 +77,8 @@ stop() -> init([]) -> process_flag(trap_exit, true), - case ?PRIM_FILE:start() of - {ok, Handle} -> - ?FILE_IO_SERVER_TABLE = - ets:new(?FILE_IO_SERVER_TABLE, [named_table]), - {ok, Handle}; - {error, Reason} -> - {stop, Reason} - end. + ?FILE_IO_SERVER_TABLE = ets:new(?FILE_IO_SERVER_TABLE, [named_table]), + {ok, undefined}. %%---------------------------------------------------------------------- %% Func: handle_call/3 @@ -101,7 +95,7 @@ init([]) -> {'reply', 'eof' | 'ok' | {'error', term()} | {'ok', term()}, state()} | {'stop', 'normal', 'stopped', state()}. -handle_call({open, Name, ModeList}, {Pid, _Tag} = _From, Handle) +handle_call({open, Name, ModeList}, {Pid, _Tag} = _From, State) when is_list(ModeList) -> Child = ?FILE_IO_SERVER:start_link(Pid, Name, ModeList), case Child of @@ -110,78 +104,78 @@ handle_call({open, Name, ModeList}, {Pid, _Tag} = _From, Handle) _ -> ok end, - {reply, Child, Handle}; + {reply, Child, State}; -handle_call({open, _Name, _Mode}, _From, Handle) -> - {reply, {error, einval}, Handle}; +handle_call({open, _Name, _Mode}, _From, State) -> + {reply, {error, einval}, State}; -handle_call({read_file, Name}, _From, Handle) -> - {reply, ?PRIM_FILE:read_file(Name), Handle}; +handle_call({read_file, Name}, _From, State) -> + {reply, ?PRIM_FILE:read_file(Name), State}; -handle_call({write_file, Name, Bin}, _From, Handle) -> - {reply, ?PRIM_FILE:write_file(Name, Bin), Handle}; +handle_call({write_file, Name, Bin}, _From, State) -> + {reply, ?PRIM_FILE:write_file(Name, Bin), State}; -handle_call({set_cwd, Name}, _From, Handle) -> - {reply, ?PRIM_FILE:set_cwd(Handle, Name), Handle}; +handle_call({set_cwd, Name}, _From, State) -> + {reply, ?PRIM_FILE:set_cwd(Name), State}; -handle_call({delete, Name}, _From, Handle) -> - {reply, ?PRIM_FILE:delete(Handle, Name), Handle}; +handle_call({delete, Name}, _From, State) -> + {reply, ?PRIM_FILE:delete(Name), State}; -handle_call({rename, Fr, To}, _From, Handle) -> - {reply, ?PRIM_FILE:rename(Handle, Fr, To), Handle}; +handle_call({rename, Fr, To}, _From, State) -> + {reply, ?PRIM_FILE:rename(Fr, To), State}; -handle_call({make_dir, Name}, _From, Handle) -> - {reply, ?PRIM_FILE:make_dir(Handle, Name), Handle}; +handle_call({make_dir, Name}, _From, State) -> + {reply, ?PRIM_FILE:make_dir(Name), State}; -handle_call({del_dir, Name}, _From, Handle) -> - {reply, ?PRIM_FILE:del_dir(Handle, Name), Handle}; +handle_call({del_dir, Name}, _From, State) -> + {reply, ?PRIM_FILE:del_dir(Name), State}; -handle_call({list_dir, Name}, _From, Handle) -> - {reply, ?PRIM_FILE:list_dir(Handle, Name), Handle}; -handle_call({list_dir_all, Name}, _From, Handle) -> - {reply, ?PRIM_FILE:list_dir_all(Handle, Name), Handle}; +handle_call({list_dir, Name}, _From, State) -> + {reply, ?PRIM_FILE:list_dir(Name), State}; +handle_call({list_dir_all, Name}, _From, State) -> + {reply, ?PRIM_FILE:list_dir_all(Name), State}; -handle_call(get_cwd, _From, Handle) -> - {reply, ?PRIM_FILE:get_cwd(Handle), Handle}; -handle_call({get_cwd}, _From, Handle) -> - {reply, ?PRIM_FILE:get_cwd(Handle), Handle}; -handle_call({get_cwd, Name}, _From, Handle) -> - {reply, ?PRIM_FILE:get_cwd(Handle, Name), Handle}; +handle_call(get_cwd, _From, State) -> + {reply, ?PRIM_FILE:get_cwd(), State}; +handle_call({get_cwd}, _From, State) -> + {reply, ?PRIM_FILE:get_cwd(), State}; +handle_call({get_cwd, Name}, _From, State) -> + {reply, ?PRIM_FILE:get_cwd(Name), State}; -handle_call({read_file_info, Name}, _From, Handle) -> - {reply, ?PRIM_FILE:read_file_info(Handle, Name), Handle}; +handle_call({read_file_info, Name}, _From, State) -> + {reply, ?PRIM_FILE:read_file_info(Name), State}; -handle_call({read_file_info, Name, Opts}, _From, Handle) -> - {reply, ?PRIM_FILE:read_file_info(Handle, Name, Opts), Handle}; +handle_call({read_file_info, Name, Opts}, _From, State) -> + {reply, ?PRIM_FILE:read_file_info(Name, Opts), State}; -handle_call({altname, Name}, _From, Handle) -> - {reply, ?PRIM_FILE:altname(Handle, Name), Handle}; +handle_call({altname, Name}, _From, State) -> + {reply, ?PRIM_FILE:altname(Name), State}; -handle_call({write_file_info, Name, Info}, _From, Handle) -> - {reply, ?PRIM_FILE:write_file_info(Handle, Name, Info), Handle}; +handle_call({write_file_info, Name, Info}, _From, State) -> + {reply, ?PRIM_FILE:write_file_info(Name, Info), State}; -handle_call({write_file_info, Name, Info, Opts}, _From, Handle) -> - {reply, ?PRIM_FILE:write_file_info(Handle, Name, Info, Opts), Handle}; +handle_call({write_file_info, Name, Info, Opts}, _From, State) -> + {reply, ?PRIM_FILE:write_file_info(Name, Info, Opts), State}; -handle_call({read_link_info, Name}, _From, Handle) -> - {reply, ?PRIM_FILE:read_link_info(Handle, Name), Handle}; +handle_call({read_link_info, Name}, _From, State) -> + {reply, ?PRIM_FILE:read_link_info(Name), State}; -handle_call({read_link_info, Name, Opts}, _From, Handle) -> - {reply, ?PRIM_FILE:read_link_info(Handle, Name, Opts), Handle}; +handle_call({read_link_info, Name, Opts}, _From, State) -> + {reply, ?PRIM_FILE:read_link_info(Name, Opts), State}; -handle_call({read_link, Name}, _From, Handle) -> - {reply, ?PRIM_FILE:read_link(Handle, Name), Handle}; -handle_call({read_link_all, Name}, _From, Handle) -> - {reply, ?PRIM_FILE:read_link_all(Handle, Name), Handle}; +handle_call({read_link, Name}, _From, State) -> + {reply, ?PRIM_FILE:read_link(Name), State}; +handle_call({read_link_all, Name}, _From, State) -> + {reply, ?PRIM_FILE:read_link_all(Name), State}; -handle_call({make_link, Old, New}, _From, Handle) -> - {reply, ?PRIM_FILE:make_link(Handle, Old, New), Handle}; +handle_call({make_link, Old, New}, _From, State) -> + {reply, ?PRIM_FILE:make_link(Old, New), State}; -handle_call({make_symlink, Old, New}, _From, Handle) -> - {reply, ?PRIM_FILE:make_symlink(Handle, Old, New), Handle}; +handle_call({make_symlink, Old, New}, _From, State) -> + {reply, ?PRIM_FILE:make_symlink(Old, New), State}; handle_call({copy, SourceName, SourceOpts, DestName, DestOpts, Length}, - _From, Handle) -> + _From, State) -> Reply = case ?PRIM_FILE:open(SourceName, [read, binary | SourceOpts]) of {ok, Source} -> @@ -201,14 +195,14 @@ handle_call({copy, SourceName, SourceOpts, DestName, DestOpts, Length}, {error, _} = Error -> Error end, - {reply, Reply, Handle}; + {reply, Reply, State}; -handle_call(stop, _From, Handle) -> - {stop, normal, stopped, Handle}; +handle_call(stop, _From, State) -> + {stop, normal, stopped, State}; -handle_call(Request, From, Handle) -> +handle_call(Request, From, State) -> error_logger:error_msg("handle_call(~tp, ~tp, _)", [Request, From]), - {noreply, Handle}. + {noreply, State}. %%---------------------------------------------------------------------- %% Func: handle_cast/2 @@ -233,14 +227,9 @@ handle_cast(Msg, State) -> -spec handle_info(term(), state()) -> {'noreply', state()} | {'stop', 'normal', state()}. -handle_info({'EXIT', Pid, _Reason}, Handle) when is_pid(Pid) -> +handle_info({'EXIT', Pid, _Reason}, State) when is_pid(Pid) -> ets:delete(?FILE_IO_SERVER_TABLE, Pid), - {noreply, Handle}; - -handle_info({'EXIT', Handle, _Reason}, Handle) -> - error_logger:error_msg("Port controlling ~w terminated in ~w", - [?FILE_SERVER, ?MODULE]), - {stop, normal, Handle}; + {noreply, State}; handle_info(Info, State) -> error_logger:error_msg("handle_Info(~tp, _)", [Info]), @@ -254,8 +243,8 @@ handle_info(Info, State) -> -spec terminate(term(), state()) -> 'ok'. -terminate(_Reason, Handle) -> - ?PRIM_FILE:stop(Handle). +terminate(_Reason, _State) -> + ok. %%---------------------------------------------------------------------- %% Func: code_change/3 diff --git a/lib/kernel/src/group.erl b/lib/kernel/src/group.erl index bf785959ff..e1198d2587 100644 --- a/lib/kernel/src/group.erl +++ b/lib/kernel/src/group.erl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 1996-2017. All Rights Reserved. +%% Copyright Ericsson AB 1996-2016. All Rights Reserved. %% %% Licensed under the Apache License, Version 2.0 (the "License"); %% you may not use this file except in compliance with the License. @@ -261,11 +261,11 @@ io_request({put_chars,latin1,M,F,As}, Drv, From, Buf) -> end; io_request({get_chars,Encoding,Prompt,N}, Drv, _From, Buf) -> - get_chars(Prompt, io_lib, collect_chars, N, Drv, Buf, Encoding); + get_chars_n(Prompt, io_lib, collect_chars, N, Drv, Buf, Encoding); io_request({get_line,Encoding,Prompt}, Drv, _From, Buf) -> - get_chars(Prompt, io_lib, collect_line, [], Drv, Buf, Encoding); + get_chars_line(Prompt, io_lib, collect_line, [], Drv, Buf, Encoding); io_request({get_until,Encoding, Prompt,M,F,As}, Drv, _From, Buf) -> - get_chars(Prompt, io_lib, get_until, {M,F,As}, Drv, Buf, Encoding); + get_chars_line(Prompt, io_lib, get_until, {M,F,As}, Drv, Buf, Encoding); io_request({get_password,_Encoding},Drv,_From,Buf) -> get_password_chars(Drv, Buf); io_request({setopts,Opts}, Drv, _From, Buf) when is_list(Opts) -> @@ -434,7 +434,7 @@ getopts(Drv,Buf) -> {ok,[Exp,Echo,Bin,Uni],Buf}. -%% get_chars(Prompt, Module, Function, XtraArgument, Drv, Buffer) +%% get_chars_*(Prompt, Module, Function, XtraArgument, Drv, Buffer) %% Gets characters from the input Drv until as the applied function %% returns {stop,Result,Rest}. Does not block output until input has been %% received. @@ -452,12 +452,21 @@ get_password_chars(Drv,Buf) -> {exit, terminated} end. -get_chars(Prompt, M, F, Xa, Drv, Buf, Encoding) -> +get_chars_n(Prompt, M, F, Xa, Drv, Buf, Encoding) -> + Pbs = prompt_bytes(Prompt, Encoding), + case get(echo) of + true -> + get_chars_loop(Pbs, M, F, Xa, Drv, Buf, start, Encoding); + false -> + get_chars_n_loop(Pbs, M, F, Xa, Drv, Buf, start, Encoding) + end. + +get_chars_line(Prompt, M, F, Xa, Drv, Buf, Encoding) -> Pbs = prompt_bytes(Prompt, Encoding), get_chars_loop(Pbs, M, F, Xa, Drv, Buf, start, Encoding). get_chars_loop(Pbs, M, F, Xa, Drv, Buf0, State, Encoding) -> - Result = case get(echo) of + Result = case get(echo) of true -> get_line(Buf0, Pbs, Drv, Encoding); false -> @@ -466,8 +475,8 @@ get_chars_loop(Pbs, M, F, Xa, Drv, Buf0, State, Encoding) -> get_line_echo_off(Buf0, Pbs, Drv) end, case Result of - {done,Line,Buf1} -> - get_chars_apply(Pbs, M, F, Xa, Drv, Buf1, State, Line, Encoding); + {done,Line,Buf} -> + get_chars_apply(Pbs, M, F, Xa, Drv, Buf, State, Line, Encoding); interrupted -> {error,{error,interrupted},[]}; terminated -> @@ -476,12 +485,29 @@ get_chars_loop(Pbs, M, F, Xa, Drv, Buf0, State, Encoding) -> get_chars_apply(Pbs, M, F, Xa, Drv, Buf, State0, Line, Encoding) -> case catch M:F(State0, cast(Line,get(read_mode), Encoding), Encoding, Xa) of - {stop,Result,Rest} -> - {ok,Result,append(Rest, Buf, Encoding)}; - {'EXIT',_} -> - {error,{error,err_func(M, F, Xa)},[]}; - State1 -> - get_chars_loop(Pbs, M, F, Xa, Drv, Buf, State1, Encoding) + {stop,Result,Rest} -> + {ok,Result,append(Rest, Buf, Encoding)}; + {'EXIT',_} -> + {error,{error,err_func(M, F, Xa)},[]}; + State1 -> + get_chars_loop(Pbs, M, F, Xa, Drv, Buf, State1, Encoding) + end. + +get_chars_n_loop(Pbs, M, F, Xa, Drv, Buf0, State, Encoding) -> + try M:F(State, cast(Buf0, get(read_mode), Encoding), Encoding, Xa) of + {stop,Result,Rest} -> + {ok, Result, Rest}; + State1 -> + case get_chars_echo_off(Pbs, Drv) of + interrupted -> + {error,{error,interrupted},[]}; + terminated -> + {exit,terminated}; + Buf -> + get_chars_n_loop(Pbs, M, F, Xa, Drv, Buf, State1, Encoding) + end + catch _:_ -> + {error,{error,err_func(M, F, Xa)},[]} end. %% Convert error code to make it look as before @@ -684,6 +710,29 @@ get_line_echo_off1({Chars,[]}, Drv) -> get_line_echo_off1({Chars,Rest}, _Drv) -> {done,lists:reverse(Chars),case Rest of done -> []; _ -> Rest end}. +get_chars_echo_off(Pbs, Drv) -> + send_drv_reqs(Drv, [{put_chars, unicode,Pbs}]), + get_chars_echo_off1(Drv). + +get_chars_echo_off1(Drv) -> + receive + {Drv, {data, Cs}} -> + Cs; + {Drv, eof} -> + eof; + {io_request,From,ReplyAs,Req} when is_pid(From) -> + io_request(Req, From, ReplyAs, Drv, []), + get_chars_echo_off1(Drv); + {reply,{{From,ReplyAs},Reply}} when From =/= undefined -> + %% We take care of replies from puts here as well + io_reply(From, ReplyAs, Reply), + get_chars_echo_off1(Drv); + {'EXIT',Drv,interrupt} -> + interrupted; + {'EXIT',Drv,_} -> + terminated + end. + %% We support line editing for the ICANON mode except the following %% line editing characters, which already has another meaning in %% echo-on mode (See Advanced Programming in the Unix Environment, 2nd ed, @@ -793,9 +842,9 @@ search_up_stack(Stack, Substr) -> case up_stack(Stack) of {none,NewStack} -> {none,NewStack}; {L, NewStack} -> - case string:str(L, Substr) of - 0 -> search_up_stack(NewStack, Substr); - _ -> {string:strip(L,right,$\n), NewStack} + case string:find(L, Substr) of + nomatch -> search_up_stack(NewStack, Substr); + _ -> {string:trim(L, trailing, "$\n"), NewStack} end end. @@ -803,9 +852,9 @@ search_down_stack(Stack, Substr) -> case down_stack(Stack) of {none,NewStack} -> {none,NewStack}; {L, NewStack} -> - case string:str(L, Substr) of - 0 -> search_down_stack(NewStack, Substr); - _ -> {string:strip(L,right,$\n), NewStack} + case string:find(L, Substr) of + nomatch -> search_down_stack(NewStack, Substr); + _ -> {string:trim(L, trailing, "$\n"), NewStack} end end. diff --git a/lib/kernel/src/inet.erl b/lib/kernel/src/inet.erl index dc20c21c77..fe91b0d33e 100644 --- a/lib/kernel/src/inet.erl +++ b/lib/kernel/src/inet.erl @@ -72,7 +72,7 @@ %% timer interface -export([start_timer/1, timeout/1, timeout/2, stop_timer/1]). --export_type([address_family/0, hostent/0, hostname/0, ip4_address/0, +-export_type([address_family/0, socket_protocol/0, hostent/0, hostname/0, ip4_address/0, ip6_address/0, ip_address/0, port_number/0, local_address/0, socket_address/0, returned_non_ip_address/0, socket_setopt/0, socket_getopt/0, @@ -1452,11 +1452,14 @@ fdopen(Fd, Addr, Port, Opts, Protocol, Family, Type, Module) -> %% socket stat %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% +-spec i() -> ok. i() -> i(tcp), i(udp), i(sctp). +-spec i(socket_protocol()) -> ok. i(Proto) -> i(Proto, [port, module, recv, sent, owner, local_address, foreign_address, state, type]). +-spec i(socket_protocol(), [atom()]) -> ok. i(tcp, Fs) -> ii(tcp_sockets(), Fs, tcp); i(udp, Fs) -> diff --git a/lib/kernel/src/inet_config.erl b/lib/kernel/src/inet_config.erl index 4bbc520449..9f76360b8b 100644 --- a/lib/kernel/src/inet_config.erl +++ b/lib/kernel/src/inet_config.erl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 1997-2016. All Rights Reserved. +%% Copyright Ericsson AB 1997-2017. All Rights Reserved. %% %% Licensed under the Apache License, Version 2.0 (the "License"); %% you may not use this file except in compliance with the License. @@ -369,7 +369,7 @@ win32_load1(Reg,Type,HFileKey) -> end. win32_split_line(Line,nt) -> inet_parse:split_line(Line); -win32_split_line(Line,windows) -> string:tokens(Line, ","). +win32_split_line(Line,windows) -> string:lexemes(Line, ","). win32_get_strings(Reg, Names) -> win32_get_strings(Reg, Names, []). diff --git a/lib/kernel/src/inet_dns.erl b/lib/kernel/src/inet_dns.erl index d5f982cc51..f1f58bc872 100644 --- a/lib/kernel/src/inet_dns.erl +++ b/lib/kernel/src/inet_dns.erl @@ -1,7 +1,7 @@ %% %% %CopyrightBegin% %% -%% Copyright Ericsson AB 1997-2016. All Rights Reserved. +%% Copyright Ericsson AB 1997-2017. All Rights Reserved. %% %% Licensed under the Apache License, Version 2.0 (the "License"); %% you may not use this file except in compliance with the License. @@ -29,7 +29,7 @@ -export([decode/1, encode/1]). --import(lists, [reverse/1, reverse/2, nthtail/2]). +-import(lists, [reverse/1]). -include("inet_int.hrl"). -include("inet_dns.hrl"). @@ -473,7 +473,7 @@ decode_data(<<Order:16,Preference:16,Data0/binary>>, _, ?S_NAPTR, Buffer) -> {Data2,Services} = decode_string(Data1), {Data,Regexp} = decode_characters(Data2, utf8), Replacement = decode_domain(Data, Buffer), - {Order,Preference,string:to_lower(Flags),string:to_lower(Services), + {Order,Preference,string:lowercase(Flags),string:lowercase(Services), Regexp,Replacement}; %% ?S_OPT falls through to default decode_data(Data, _, ?S_TXT, _) -> diff --git a/lib/kernel/src/inet_int.hrl b/lib/kernel/src/inet_int.hrl index bc5b67f7bf..357e27826c 100644 --- a/lib/kernel/src/inet_int.hrl +++ b/lib/kernel/src/inet_int.hrl @@ -100,6 +100,8 @@ -define(TCP_REQ_RECV, 42). -define(TCP_REQ_UNRECV, 43). -define(TCP_REQ_SHUTDOWN, 44). +-define(TCP_REQ_SENDFILE, 45). + %% UDP and SCTP requests -define(PACKET_REQ_RECV, 60). %%-define(SCTP_REQ_LISTEN, 61). MERGED @@ -319,6 +321,12 @@ [((X) bsr 24) band 16#ff, ((X) bsr 16) band 16#ff, ((X) bsr 8) band 16#ff, (X) band 16#ff]). +-define(int64(X), + [((X) bsr 56) band 16#ff, ((X) bsr 48) band 16#ff, + ((X) bsr 40) band 16#ff, ((X) bsr 32) band 16#ff, + ((X) bsr 24) band 16#ff, ((X) bsr 16) band 16#ff, + ((X) bsr 8) band 16#ff, (X) band 16#ff]). + -define(intAID(X), % For SCTP AssocID ?int32(X)). diff --git a/lib/kernel/src/inet_parse.erl b/lib/kernel/src/inet_parse.erl index 29804dc50b..e9685c6554 100644 --- a/lib/kernel/src/inet_parse.erl +++ b/lib/kernel/src/inet_parse.erl @@ -95,7 +95,7 @@ hosts(Fname,File) -> %% interface with a %if suffix. These kind of %% addresses maybe need to be gracefully handled %% throughout inet* and inet_drv. - case string:tokens(Address, "%") of + case string:lexemes(Address, "%") of [Addr,_] -> {ok,_} = address(Addr), skip; @@ -407,7 +407,7 @@ is_dom1([C | Cs]) when C >= $a, C =< $z -> is_dom_ldh(Cs); is_dom1([C | Cs]) when C >= $A, C =< $Z -> is_dom_ldh(Cs); is_dom1([C | Cs]) when C >= $0, C =< $9 -> case is_dom_ldh(Cs) of - true -> is_dom2(string:tokens([C | Cs],".")); + true -> is_dom2(string:lexemes([C | Cs],".")); false -> false end; is_dom1(_) -> false. diff --git a/lib/kernel/src/inet_res.erl b/lib/kernel/src/inet_res.erl index 90e49ddfdf..49aa5f8bda 100644 --- a/lib/kernel/src/inet_res.erl +++ b/lib/kernel/src/inet_res.erl @@ -859,15 +859,17 @@ query_ns(S0, Id, Buffer, IP, Port, Timer, Retry, I, {ok,S} -> Timeout = inet:timeout( (Tm * (1 bsl I)) div Retry, Timer), - {S, case query_udp( S, Id, Buffer, IP, Port, Timeout, Verbose) of {ok,#dns_rec{header=H}} when H#dns_header.tc -> TcpTimeout = inet:timeout(Tm*5, Timer), - query_tcp( - TcpTimeout, Id, Buffer, IP, Port, Verbose); - Reply -> Reply - end}; + {S, query_tcp( + TcpTimeout, Id, Buffer, IP, Port, Verbose)}; + {error, econnrefused} = Err -> + ok = udp_close(S), + {#sock{}, Err}; + Reply -> {S, Reply} + end; Error -> {S0,Error} end diff --git a/lib/kernel/src/kernel.app.src b/lib/kernel/src/kernel.app.src index 2a88cc7e26..82a3571da9 100644 --- a/lib/kernel/src/kernel.app.src +++ b/lib/kernel/src/kernel.app.src @@ -57,6 +57,7 @@ inet_tcp_dist, kernel, kernel_config, + kernel_refc, local_tcp, local_udp, net, @@ -88,6 +89,13 @@ inet_udp, inet_sctp, pg2, + raw_file_io, + raw_file_io_compressed, + raw_file_io_deflate, + raw_file_io_delayed, + raw_file_io_inflate, + raw_file_io_list, + raw_file_io_raw, seq_trace, standard_error, wrap_log_reader]}, @@ -107,6 +115,7 @@ heart, init, kernel_config, + kernel_refc, kernel_sup, net_kernel, net_sup, @@ -120,6 +129,6 @@ {applications, []}, {env, [{error_logger, tty}]}, {mod, {kernel, []}}, - {runtime_dependencies, ["erts-9.1", "stdlib-3.4", "sasl-3.0"]} + {runtime_dependencies, ["erts-10.0", "stdlib-3.5", "sasl-3.0"]} ] }. diff --git a/lib/kernel/src/kernel.erl b/lib/kernel/src/kernel.erl index cba57088ec..0382764b39 100644 --- a/lib/kernel/src/kernel.erl +++ b/lib/kernel/src/kernel.erl @@ -111,6 +111,13 @@ init([]) -> type => worker, modules => [kernel_config]}, + RefC = #{id => kernel_refc, + start => {kernel_refc, start_link, []}, + restart => permanent, + shutdown => 2000, + type => worker, + modules => [kernel_refc]}, + Code = #{id => code_server, start => {code, start_link, []}, restart => permanent, @@ -148,7 +155,7 @@ init([]) -> case init:get_argument(mode) of {ok, [["minimal"]]} -> - {ok, {SupFlags, [Code, File, StdError, User, Config, SafeSup]}}; + {ok, {SupFlags, [Code, File, StdError, User, Config, RefC, SafeSup]}}; _ -> Rpc = #{id => rex, start => {rpc, start_link, []}, @@ -199,7 +206,7 @@ init([]) -> {ok, {SupFlags, [Code, Rpc, Global, InetDb | DistAC] ++ [NetSup, GlGroup, File, SigSrv, - StdError, User, Config, SafeSup] ++ Timer}} + StdError, User, Config, RefC, SafeSup] ++ Timer}} end; init(safe) -> SupFlags = #{strategy => one_for_one, diff --git a/lib/kernel/src/kernel_refc.erl b/lib/kernel/src/kernel_refc.erl new file mode 100644 index 0000000000..05076dc885 --- /dev/null +++ b/lib/kernel/src/kernel_refc.erl @@ -0,0 +1,139 @@ +%% +%% %CopyrightBegin% +%% +%% Copyright Ericsson AB 2017. All Rights Reserved. +%% +%% Licensed under the Apache License, Version 2.0 (the "License"); +%% you may not use this file except in compliance with the License. +%% You may obtain a copy of the License at +%% +%% http://www.apache.org/licenses/LICENSE-2.0 +%% +%% Unless required by applicable law or agreed to in writing, software +%% distributed under the License is distributed on an "AS IS" BASIS, +%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +%% See the License for the specific language governing permissions and +%% limitations under the License. +%% +%% %CopyrightEnd% +%% +-module(kernel_refc). + +-behaviour(gen_server). + +%% External exports +-export([start_link/0, scheduler_wall_time/1]). +%% Internal exports +-export([init/1, handle_info/2, terminate/2]). +-export([handle_call/3, handle_cast/2, code_change/3]). + +%%%----------------------------------------------------------------- +%%% This module implements a process that handles reference counters for +%%% various erts or other kernel resources which needs reference counting. +%%% +%%% Should not be documented nor used directly by user applications. +%%%----------------------------------------------------------------- +start_link() -> + gen_server:start_link({local,kernel_refc}, kernel_refc, [], []). + +-spec scheduler_wall_time(boolean()) -> boolean(). +scheduler_wall_time(Bool) -> + gen_server:call(kernel_refc, {scheduler_wall_time, self(), Bool}, infinity). + +%%----------------------------------------------------------------- +%% Callback functions from gen_server +%%----------------------------------------------------------------- + +-spec init([]) -> {'ok', map()} | {'stop', term()}. + +init([]) -> + resource(scheduler_wall_time, false), + {ok, #{scheduler_wall_time=>#{}}}. + +-spec handle_call(term(), term(), State) -> {'reply', term(), State}. +handle_call({What, Who, false}, _From, State) -> + {Reply, Cnt} = do_stop(What, maps:get(What, State), Who), + {reply, Reply, State#{What:=Cnt}}; +handle_call({What, Who, true}, _From, State) -> + {Reply, Cnt} = do_start(What, maps:get(What, State), Who), + {reply, Reply, State#{What:=Cnt}}; +handle_call(_, _From, State) -> + {reply, badarg, State}. + +-spec handle_cast(term(), State) -> {'noreply', State}. +handle_cast(_, State) -> + {noreply, State}. + +-spec handle_info(term(), State) -> {'noreply', State}. +handle_info({'DOWN', _Ref, process, Pid, _}, State) -> + Cleanup = fun(Resource, Cnts) -> + cleanup(Resource, Cnts, Pid) + end, + {noreply, maps:map(Cleanup, State)}; +handle_info(_, State) -> + {noreply, State}. + +-spec terminate(term(), term()) -> 'ok'. +terminate(_Reason, _State) -> + ok. + +-spec code_change(term(), State, term()) -> {'ok', State}. +code_change(_OldVsn, State, _Extra) -> + {ok, State}. + +%%----------------------------------------------------------------- +%% Internal functions +%%----------------------------------------------------------------- + +do_start(Resource, Cnt, Pid) -> + case maps:get(Pid, Cnt, undefined) of + undefined -> + Ref = erlang:monitor(process, Pid), + case any(Cnt) of + true -> + {true, Cnt#{Pid=>{1, Ref}}}; + false -> + resource(Resource, true), + {false, Cnt#{Pid=>{1, Ref}}} + end; + {N, Ref} -> + {true, Cnt#{Pid=>{N+1, Ref}}} + end. + +do_stop(Resource, Cnt0, Pid) -> + case maps:get(Pid, Cnt0, undefined) of + undefined -> + {any(Cnt0), Cnt0}; + {1, Ref} -> + erlang:demonitor(Ref, [flush]), + Cnt = maps:remove(Pid, Cnt0), + case any(Cnt) of + true -> + {true, Cnt}; + false -> + resource(Resource, false), + {true, Cnt} + end; + {N, Ref} -> + {true, Cnt0#{Pid=>{N-1, Ref}}} + end. + +cleanup(Resource, Cnt0, Pid) -> + case maps:is_key(Pid, Cnt0) of + true -> + Cnt = maps:remove(Pid, Cnt0), + case any(Cnt) of + true -> + Cnt; + false -> + resource(Resource, false), + Cnt + end; + false -> + Cnt0 + end. + +any(Cnt) -> maps:size(Cnt) > 0. + +resource(scheduler_wall_time, Enable) -> + _ = erts_internal:scheduler_wall_time(Enable). diff --git a/lib/kernel/src/net_kernel.erl b/lib/kernel/src/net_kernel.erl index 7da89dd7cb..cdb10a7b12 100644 --- a/lib/kernel/src/net_kernel.erl +++ b/lib/kernel/src/net_kernel.erl @@ -70,8 +70,8 @@ protocol_childspecs/0, epmd_module/0]). --export([connect/1, disconnect/1, hidden_connect/1, passive_cnct/1]). --export([hidden_connect_node/1]). %% explicit connect +-export([disconnect/1, passive_cnct/1]). +-export([hidden_connect_node/1]). -export([set_net_ticktime/1, set_net_ticktime/2, get_net_ticktime/0]). -export([node_info/1, node_info/2, nodes_info/0, @@ -122,6 +122,7 @@ -record(connection, { node, %% remote node name + conn_id, %% Connection identity state, %% pending | up | up_pending owner, %% owner pid pending_owner, %% possible new owner @@ -247,14 +248,15 @@ ticktime_res(A) when is_atom(A) -> A. %% Called though BIF's -connect(Node) -> do_connect(Node, normal, false). %%% Long timeout if blocked (== barred), only affects nodes with %%% {dist_auto_connect, once} set. -passive_cnct(Node) -> do_connect(Node, normal, true). -disconnect(Node) -> request({disconnect, Node}). +passive_cnct(Node) -> + case request({passive_cnct, Node}) of + ignored -> false; + Other -> Other + end. -%% connect but not seen -hidden_connect(Node) -> do_connect(Node, hidden, false). +disconnect(Node) -> request({disconnect, Node}). %% Should this node publish itself on Node? publish_on_node(Node) when is_atom(Node) -> @@ -272,67 +274,30 @@ connect_node(Node) when is_atom(Node) -> hidden_connect_node(Node) when is_atom(Node) -> request({connect, hidden, Node}). -do_connect(Node, Type, WaitForBarred) -> %% Type = normal | hidden - case catch ets:lookup(sys_dist, Node) of - {'EXIT', _} -> - ?connect_failure(Node,{table_missing, sys_dist}), - false; - [#barred_connection{}] -> - case WaitForBarred of - false -> - false; - true -> - Pid = spawn(?MODULE,passive_connect_monitor,[self(),Node]), - receive - {Pid, true} -> - %%io:format("Net Kernel: barred connection (~p) " - %% "connected from other end.~n",[Node]), - true; - {Pid, false} -> - ?connect_failure(Node,{barred_connection, - ets:lookup(sys_dist, Node)}), - %%io:format("Net Kernel: barred connection (~p) " - %% "- failure.~n",[Node]), - false - end - end; - Else -> - case application:get_env(kernel, dist_auto_connect) of - {ok, never} -> - ?connect_failure(Node,{dist_auto_connect,never}), - false; - % This might happen due to connection close - % not beeing propagated to user space yet. - % Save the day by just not connecting... - {ok, once} when Else =/= [], - (hd(Else))#connection.state =:= up -> - ?connect_failure(Node,{barred_connection, - ets:lookup(sys_dist, Node)}), - false; - _ -> - request({connect, Type, Node}) - end - end. -passive_connect_monitor(Parent, Node) -> +passive_connect_monitor(From, Node) -> ok = monitor_nodes(true,[{node_type,all}]), - case lists:member(Node,nodes([connected])) of - true -> - ok = monitor_nodes(false,[{node_type,all}]), - Parent ! {self(),true}; - _ -> - Ref = make_ref(), - Tref = erlang:send_after(connecttime(),self(),Ref), - receive - Ref -> - ok = monitor_nodes(false,[{node_type,all}]), - Parent ! {self(), false}; - {nodeup,Node,_} -> - ok = monitor_nodes(false,[{node_type,all}]), - _ = erlang:cancel_timer(Tref), - Parent ! {self(),true} - end - end. + Reply = case lists:member(Node,nodes([connected])) of + true -> + io:format("~p: passive_connect_monitor ~p\n", [self(), ?LINE]), + true; + _ -> + receive + {nodeup,Node,_} -> + io:format("~p: passive_connect_monitor ~p\n", [self(), ?LINE]), + true + after connecttime() -> + io:format("~p: passive_connect_monitor ~p\n", [self(), ?LINE]), + false + end + end, + ok = monitor_nodes(false,[{node_type,all}]), + io:format("~p: passive_connect_monitor ~p\n", [self(), ?LINE]), + {Pid, Tag} = From, + io:format("~p: passive_connect_monitor ~p\n", [self(), ?LINE]), + erlang:send(Pid, {Tag, Reply}), + io:format("~p: passive_connect_monitor ~p\n", [self(), ?LINE]). + %% If the net_kernel isn't running we ignore all requests to the %% kernel, thus basically accepting them :-) @@ -394,40 +359,135 @@ init({Name, LongOrShortNames, TickT, CleanHalt}) -> end. +do_auto_connect(Type, Node, ConnId, WaitForBarred, From, State) -> + ConnLookup = ets:lookup(sys_dist, Node), + + case ConnLookup of + [#barred_connection{}] -> + case WaitForBarred of + false -> + {reply, false, State}; + true -> + spawn(?MODULE,passive_connect_monitor,[From,Node]), + {noreply, State} + end; + + [#connection{conn_id=ConnId, state = up}] -> + {reply, true, State}; + [#connection{conn_id=ConnId, waiting=Waiting}=Conn] -> + case From of + noreply -> ok; + _ -> ets:insert(sys_dist, Conn#connection{waiting = [From|Waiting]}) + end, + {noreply, State}; + + _ -> + case application:get_env(kernel, dist_auto_connect) of + {ok, never} -> + ?connect_failure(Node,{dist_auto_connect,never}), + {reply, false, State}; + + %% This might happen due to connection close + %% not beeing propagated to user space yet. + %% Save the day by just not connecting... + {ok, once} when ConnLookup =/= [], + (hd(ConnLookup))#connection.state =:= up -> + ?connect_failure(Node,{barred_connection, + ets:lookup(sys_dist, Node)}), + {reply, false, State}; + _ -> + case setup(ConnLookup, Node,ConnId,Type,From,State) of + {ok, SetupPid} -> + Owners = [{SetupPid, Node} | State#state.conn_owners], + {noreply,State#state{conn_owners=Owners}}; + _Error -> + ?connect_failure(Node, {setup_call, failed, _Error}), + {reply, false, State} + end + end + end. + + +do_explicit_connect([#connection{conn_id = ConnId, state = up}], _, _, ConnId, _From, State) -> + {reply, true, State}; +do_explicit_connect([#connection{conn_id = ConnId}=Conn], _, _, ConnId, From, State) + when Conn#connection.state =:= pending; + Conn#connection.state =:= up_pending -> + Waiting = Conn#connection.waiting, + ets:insert(sys_dist, Conn#connection{waiting = [From|Waiting]}), + {noreply, State}; +do_explicit_connect([#barred_connection{}], Type, Node, ConnId, From , State) -> + %% Barred connection only affects auto_connect, ignore it. + do_explicit_connect([], Type, Node, ConnId, From , State); +do_explicit_connect(ConnLookup, Type, Node, ConnId, From , State) -> + case setup(ConnLookup, Node,ConnId,Type,From,State) of + {ok, SetupPid} -> + Owners = [{SetupPid, Node} | State#state.conn_owners], + {noreply,State#state{conn_owners=Owners}}; + _Error -> + ?connect_failure(Node, {setup_call, failed, _Error}), + {reply, false, State} + end. + +-define(ERTS_DIST_CON_ID_MASK, 16#ffffff). % also in external.h + +verify_new_conn_id([], {Nr,_DHandle}) + when (Nr band (bnot ?ERTS_DIST_CON_ID_MASK)) =:= 0 -> + true; +verify_new_conn_id([#connection{conn_id = {Old,_}}], {New,_}) + when New =:= ((Old+1) band ?ERTS_DIST_CON_ID_MASK) -> + true; +verify_new_conn_id(_, _) -> + false. + + + %% ------------------------------------------------------------ %% handle_call. %% ------------------------------------------------------------ %% -%% Set up a connection to Node. -%% The response is delayed until the connection is up and -%% running. +%% Passive auto-connect to Node. +%% The response is delayed until the connection is up and running. %% -handle_call({connect, _, Node}, From, State) when Node =:= node() -> +handle_call({passive_cnct, Node}, From, State) when Node =:= node() -> + async_reply({reply, true, State}, From); +handle_call({passive_cnct, Node}, From, State) -> + verbose({passive_cnct, Node}, 1, State), + Type = normal, + WaitForBarred = true, + R = case (catch erts_internal:new_connection(Node)) of + {Nr,_DHandle}=ConnId when is_integer(Nr) -> + do_auto_connect(Type, Node, ConnId, WaitForBarred, From, State); + + _Error -> + error_logger:error_msg("~n** Cannot get connection id for node ~w~n", + [Node]), + {reply, false, State} + end, + + return_call(R, From); + +%% +%% Explicit connect +%% The response is delayed until the connection is up and running. +%% +handle_call({connect, _, Node, _, _}, From, State) when Node =:= node() -> async_reply({reply, true, State}, From); handle_call({connect, Type, Node}, From, State) -> verbose({connect, Type, Node}, 1, State), - case ets:lookup(sys_dist, Node) of - [Conn] when Conn#connection.state =:= up -> - async_reply({reply, true, State}, From); - [Conn] when Conn#connection.state =:= pending -> - Waiting = Conn#connection.waiting, - ets:insert(sys_dist, Conn#connection{waiting = [From|Waiting]}), - {noreply, State}; - [Conn] when Conn#connection.state =:= up_pending -> - Waiting = Conn#connection.waiting, - ets:insert(sys_dist, Conn#connection{waiting = [From|Waiting]}), - {noreply, State}; - _ -> - case setup(Node,Type,From,State) of - {ok, SetupPid} -> - Owners = [{SetupPid, Node} | State#state.conn_owners], - {noreply,State#state{conn_owners=Owners}}; - _ -> - ?connect_failure(Node, {setup_call, failed}), - async_reply({reply, false, State}, From) - end - end; + ConnLookup = ets:lookup(sys_dist, Node), + R = case (catch erts_internal:new_connection(Node)) of + {Nr,_DHandle}=ConnId when is_integer(Nr) -> + do_explicit_connect(ConnLookup, Type, Node, ConnId, From, State); + + _Error -> + error_logger:error_msg("~n** Cannot get connection id for node ~w~n", + [Node]), + {reply, false, State} + end, + return_call(R, From); + %% %% Close the connection to Node. @@ -634,6 +694,26 @@ terminate(_Reason, State) -> %% ------------------------------------------------------------ %% +%% Asynchronous auto connect request +%% +handle_info({auto_connect,Node, Nr, DHandle}, State) -> + verbose({auto_connect, Node, Nr, DHandle}, 1, State), + ConnId = {Nr, DHandle}, + NewState = + case do_auto_connect(normal, Node, ConnId, false, noreply, State) of + {noreply, S} -> %% Pending connection + S; + + {reply, true, S} -> %% Already connected + S; + + {reply, false, S} -> %% Connection refused + erts_internal:abort_connection(Node, ConnId), + S + end, + {noreply, NewState}; + +%% %% accept a new connection. %% handle_info({accept,AcceptPid,Socket,Family,Proto}, State) -> @@ -713,14 +793,23 @@ handle_info({AcceptPid, {accept_pending,MyNode,Node,Address,Type}}, State) -> AcceptPid ! {self(), {accept_pending, already_pending}}, {noreply, State}; _ -> - ets:insert(sys_dist, #connection{node = Node, - state = pending, - owner = AcceptPid, - address = Address, - type = Type}), - AcceptPid ! {self(),{accept_pending,ok}}, - Owners = [{AcceptPid,Node} | State#state.conn_owners], - {noreply, State#state{conn_owners = Owners}} + case (catch erts_internal:new_connection(Node)) of + {Nr,_DHandle}=ConnId when is_integer(Nr) -> + ets:insert(sys_dist, #connection{node = Node, + conn_id = ConnId, + state = pending, + owner = AcceptPid, + address = Address, + type = Type}), + AcceptPid ! {self(),{accept_pending,ok}}, + Owners = [{AcceptPid,Node} | State#state.conn_owners], + {noreply, State#state{conn_owners = Owners}}; + + _ -> + error_logger:error_msg("~n** Cannot get connection id for node ~w~n", + [Node]), + AcceptPid ! {self(),{accept_pending,nok_pending}} + end end; handle_info({SetupPid, {is_pending, Node}}, State) -> @@ -906,6 +995,7 @@ pending_nodedown(Conn, Node, Type, State) -> % Don't bar connections that have never been alive %mark_sys_dist_nodedown(Node), % - instead just delete the node: + erts_internal:abort_connection(Node, Conn#connection.conn_id), ets:delete(sys_dist, Node), reply_waiting(Node,Conn#connection.waiting, false), case Type of @@ -920,7 +1010,9 @@ up_pending_nodedown(Conn, Node, _Reason, _Type, State) -> AcceptPid = Conn#connection.pending_owner, Owners = State#state.conn_owners, Pend = lists:keydelete(AcceptPid, 1, State#state.pend_owners), + erts_internal:abort_connection(Node, Conn#connection.conn_id), Conn1 = Conn#connection { owner = AcceptPid, + conn_id = erts_internal:new_connection(Node), pending_owner = undefined, state = pending }, ets:insert(sys_dist, Conn1), @@ -928,15 +1020,16 @@ up_pending_nodedown(Conn, Node, _Reason, _Type, State) -> State#state{conn_owners = [{AcceptPid,Node}|Owners], pend_owners = Pend}. -up_nodedown(_Conn, Node, _Reason, Type, State) -> - mark_sys_dist_nodedown(Node), +up_nodedown(Conn, Node, _Reason, Type, State) -> + mark_sys_dist_nodedown(Conn, Node), case Type of normal -> ?nodedown(Node, State); _ -> ok end, State. -mark_sys_dist_nodedown(Node) -> +mark_sys_dist_nodedown(Conn, Node) -> + erts_internal:abort_connection(Node, Conn#connection.conn_id), case application:get_env(kernel, dist_auto_connect) of {ok, once} -> ets:insert(sys_dist, #barred_connection{node = Node}); @@ -1179,15 +1272,8 @@ spawn_func(_,{From,Tag},M,F,A,Gleader) -> %% Set up connection to a new node. %% ----------------------------------------------------------- -setup(Node,Type,From,State) -> - Allowed = State#state.allowed, - case lists:member(Node, Allowed) of - false when Allowed =/= [] -> - error_msg("** Connection attempt with " - "disallowed node ~w ** ~n", [Node]), - {error, bad_node}; - _ -> - case select_mod(Node, State#state.listen) of +setup(ConnLookup, Node,ConnId,Type,From,State) -> + case setup_check(ConnLookup, Node, ConnId, State) of {ok, L} -> Mod = L#listen.module, LAddr = L#listen.address, @@ -1200,18 +1286,45 @@ setup(Node,Type,From,State) -> Addr = LAddr#net_address { address = undefined, host = undefined }, + Waiting = case From of + noreply -> []; + _ -> [From] + end, ets:insert(sys_dist, #connection{node = Node, + conn_id = ConnId, state = pending, owner = Pid, - waiting = [From], + waiting = Waiting, address = Addr, type = normal}), {ok, Pid}; Error -> Error - end end. +setup_check(ConnLookup, Node, ConnId, State) -> + Allowed = State#state.allowed, + case lists:member(Node, Allowed) of + false when Allowed =/= [] -> + error_msg("** Connection attempt with " + "disallowed node ~w ** ~n", [Node]), + {error, bad_node}; + _ -> + case verify_new_conn_id(ConnLookup, ConnId) of + false -> + error_msg("** Connection attempt to ~w with " + "bad connection id ~w ** ~n", [Node, ConnId]), + {error, bad_conn_id}; + true -> + case select_mod(Node, State#state.listen) of + {ok, _L}=OK -> OK; + Error -> Error + end + end + end. + + + %% %% Find a module that is willing to handle connection setup to Node %% @@ -1652,6 +1765,11 @@ verbose(_, _, _) -> getnode(P) when is_pid(P) -> node(P); getnode(P) -> P. +return_call({noreply, _State}=R, _From) -> + R; +return_call(R, From) -> + async_reply(R, From). + async_reply({reply, Msg, State}, From) -> async_gen_server_reply(From, Msg), {noreply, State}. diff --git a/lib/kernel/src/os.erl b/lib/kernel/src/os.erl index 0250783632..fbc046c8f9 100644 --- a/lib/kernel/src/os.erl +++ b/lib/kernel/src/os.erl @@ -25,36 +25,33 @@ -include("file.hrl"). +-export_type([env_var_name/0, env_var_value/0, env_var_name_value/0, command_input/0]). + +-export([getenv/0, getenv/1, getenv/2, putenv/2, unsetenv/1]). + %%% BIFs --export([getenv/0, getenv/1, getenv/2, getpid/0, - perf_counter/0, perf_counter/1, - putenv/2, set_signal/2, system_time/0, system_time/1, - timestamp/0, unsetenv/1]). +-export([get_env_var/1, getpid/0, list_env_vars/0, perf_counter/0, + perf_counter/1, set_env_var/2, set_signal/2, system_time/0, + system_time/1, timestamp/0, unset_env_var/1]). --spec getenv() -> [string()]. +-type env_var_name() :: nonempty_string(). -getenv() -> erlang:nif_error(undef). +-type env_var_value() :: string(). --spec getenv(VarName) -> Value | false when - VarName :: string(), - Value :: string(). +-type env_var_name_value() :: nonempty_string(). -getenv(_) -> - erlang:nif_error(undef). +-type command_input() :: atom() | io_lib:chars(). --spec getenv(VarName, DefaultValue) -> Value when - VarName :: string(), - DefaultValue :: string(), - Value :: string(). +-spec list_env_vars() -> [{env_var_name(), env_var_value()}]. +list_env_vars() -> + erlang:nif_error(undef). -getenv(VarName, DefaultValue) -> - case os:getenv(VarName) of - false -> - DefaultValue; - Value -> - Value - end. +-spec get_env_var(VarName) -> Value | false when + VarName :: env_var_name(), + Value :: env_var_value(). +get_env_var(_VarName) -> + erlang:nif_error(undef). -spec getpid() -> Value when Value :: string(). @@ -74,11 +71,10 @@ perf_counter() -> perf_counter(Unit) -> erlang:convert_time_unit(os:perf_counter(), perf_counter, Unit). --spec putenv(VarName, Value) -> true when - VarName :: string(), - Value :: string(). - -putenv(_, _) -> +-spec set_env_var(VarName, Value) -> true when + VarName :: env_var_name(), + Value :: env_var_value(). +set_env_var(_, _) -> erlang:nif_error(undef). -spec system_time() -> integer(). @@ -98,10 +94,9 @@ system_time(_Unit) -> timestamp() -> erlang:nif_error(undef). --spec unsetenv(VarName) -> true when - VarName :: string(). - -unsetenv(_) -> +-spec unset_env_var(VarName) -> true when + VarName :: env_var_name(). +unset_env_var(_) -> erlang:nif_error(undef). -spec set_signal(Signal, Option) -> 'ok' when @@ -115,6 +110,39 @@ set_signal(_Signal, _Option) -> %%% End of BIFs +-spec getenv() -> [env_var_name_value()]. +getenv() -> + [lists:flatten([Key, $=, Value]) || {Key, Value} <- os:list_env_vars() ]. + +-spec getenv(VarName) -> Value | false when + VarName :: env_var_name(), + Value :: env_var_value(). +getenv(VarName) -> + os:get_env_var(VarName). + +-spec getenv(VarName, DefaultValue) -> Value when + VarName :: env_var_name(), + DefaultValue :: env_var_value(), + Value :: env_var_value(). +getenv(VarName, DefaultValue) -> + case os:getenv(VarName) of + false -> + DefaultValue; + Value -> + Value + end. + +-spec putenv(VarName, Value) -> true when + VarName :: env_var_name(), + Value :: env_var_value(). +putenv(VarName, Value) -> + os:set_env_var(VarName, Value). + +-spec unsetenv(VarName) -> true when + VarName :: env_var_name(). +unsetenv(VarName) -> + os:unset_env_var(VarName). + -spec type() -> {Osfamily, Osname} when Osfamily :: unix | win32, Osname :: atom(). @@ -178,7 +206,7 @@ verify_executable(Name0, [Ext|Rest], OrigExtensions) -> end; verify_executable(Name, [], OrigExtensions) when OrigExtensions =/= [""] -> %% Windows %% Will only happen on windows, hence case insensitivity - case can_be_full_name(string:to_lower(Name),OrigExtensions) of + case can_be_full_name(string:lowercase(Name),OrigExtensions) of true -> verify_executable(Name,[""],[""]); _ -> @@ -232,10 +260,9 @@ extensions() -> %% Executes the given command in the default shell for the operating system. -spec cmd(Command) -> string() when - Command :: atom() | io_lib:chars(). + Command :: os:command_input(). cmd(Cmd) -> - validate(Cmd), - {SpawnCmd, SpawnOpts, SpawnInput, Eot} = mk_cmd(os:type(), Cmd), + {SpawnCmd, SpawnOpts, SpawnInput, Eot} = mk_cmd(os:type(), validate(Cmd)), Port = open_port({spawn, SpawnCmd}, [binary, stderr_to_stdout, stream, in, hide | SpawnOpts]), MonRef = erlang:monitor(port, Port), @@ -255,8 +282,6 @@ mk_cmd({win32,Wtype}, Cmd) -> {Cspec,_} -> lists:concat([Cspec," /c",Cmd]) end, {Command, [], [], <<>>}; -mk_cmd(OsType,Cmd) when is_atom(Cmd) -> - mk_cmd(OsType, atom_to_list(Cmd)); mk_cmd(_,Cmd) -> %% Have to send command in like this in order to make sh commands like %% cd and ulimit available @@ -279,17 +304,33 @@ mk_cmd(_,Cmd) -> <<$\^D>>}. validate(Atom) when is_atom(Atom) -> - ok; + validate(atom_to_list(Atom)); validate(List) when is_list(List) -> - validate1(List). + case validate1(List) of + false -> + List; + true -> + %% Had zeros at end; remove them... + string:trim(List, trailing, [0]) + end. -validate1([C|Rest]) when is_integer(C) -> +validate1([0|Rest]) -> + validate2(Rest); +validate1([C|Rest]) when is_integer(C), C > 0 -> validate1(Rest); validate1([List|Rest]) when is_list(List) -> - validate1(List), - validate1(Rest); + validate1(List) or validate1(Rest); validate1([]) -> - ok. + false. + +%% Ensure that the rest is zero only... +validate2([]) -> + true; +validate2([0|Rest]) -> + validate2(Rest); +validate2([List|Rest]) when is_list(List) -> + validate2(List), + validate2(Rest). get_data(Port, MonRef, Eot, Sofar) -> receive diff --git a/lib/kernel/src/raw_file_io.erl b/lib/kernel/src/raw_file_io.erl new file mode 100644 index 0000000000..e3c07c8f78 --- /dev/null +++ b/lib/kernel/src/raw_file_io.erl @@ -0,0 +1,75 @@ +%% +%% %CopyrightBegin% +%% +%% Copyright Ericsson AB 2017. All Rights Reserved. +%% +%% Licensed under the Apache License, Version 2.0 (the "License"); +%% you may not use this file except in compliance with the License. +%% You may obtain a copy of the License at +%% +%% http://www.apache.org/licenses/LICENSE-2.0 +%% +%% Unless required by applicable law or agreed to in writing, software +%% distributed under the License is distributed on an "AS IS" BASIS, +%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +%% See the License for the specific language governing permissions and +%% limitations under the License. +%% +%% %CopyrightEnd% +%% +-module(raw_file_io). + +-export([open/2]). + +open(Filename, Modes) -> + %% Layers are applied in this order, and the listed modules will call this + %% function again as necessary. eg. a raw compressed delayed file in list + %% mode will walk through [_list -> _compressed -> _delayed -> _raw]. + ModuleOrder = [{raw_file_io_list, fun match_list/1}, + {raw_file_io_compressed, fun match_compressed/1}, + {raw_file_io_delayed, fun match_delayed/1}, + {raw_file_io_raw, fun match_raw/1}], + open_1(ModuleOrder, Filename, add_implicit_modes(Modes)). +open_1([], _Filename, _Modes) -> + error(badarg); +open_1([{Module, Match} | Rest], Filename, Modes) -> + case lists:any(Match, Modes) of + true -> + {Options, ChildModes} = + lists:partition(fun(Mode) -> Match(Mode) end, Modes), + Module:open_layer(Filename, ChildModes, Options); + false -> + open_1(Rest, Filename, Modes) + end. + +%% 'read' and 'list' mode are enabled unless disabled by another option, so +%% we'll explicitly add them to avoid duplicating this logic in child layers. +add_implicit_modes(Modes0) -> + Modes1 = add_unless_matched(Modes0, fun match_writable/1, read), + add_unless_matched(Modes1, fun match_binary/1, list). +add_unless_matched(Modes, Match, Default) -> + case lists:any(Match, Modes) of + false -> [Default | Modes]; + true -> Modes + end. + +match_list(list) -> true; +match_list(_Other) -> false. + +match_compressed(compressed) -> true; +match_compressed(_Other) -> false. + +match_delayed({delayed_write, _Size, _Timeout}) -> true; +match_delayed(delayed_write) -> true; +match_delayed(_Other) -> false. + +match_raw(raw) -> true; +match_raw(_Other) -> false. + +match_writable(write) -> true; +match_writable(append) -> true; +match_writable(exclusive) -> true; +match_writable(_Other) -> false. + +match_binary(binary) -> true; +match_binary(_Other) -> false. diff --git a/lib/kernel/src/raw_file_io_compressed.erl b/lib/kernel/src/raw_file_io_compressed.erl new file mode 100644 index 0000000000..d5ab042d25 --- /dev/null +++ b/lib/kernel/src/raw_file_io_compressed.erl @@ -0,0 +1,134 @@ +%% +%% %CopyrightBegin% +%% +%% Copyright Ericsson AB 2017. All Rights Reserved. +%% +%% Licensed under the Apache License, Version 2.0 (the "License"); +%% you may not use this file except in compliance with the License. +%% You may obtain a copy of the License at +%% +%% http://www.apache.org/licenses/LICENSE-2.0 +%% +%% Unless required by applicable law or agreed to in writing, software +%% distributed under the License is distributed on an "AS IS" BASIS, +%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +%% See the License for the specific language governing permissions and +%% limitations under the License. +%% +%% %CopyrightEnd% +%% +-module(raw_file_io_compressed). + +-export([close/1, sync/1, datasync/1, truncate/1, advise/4, allocate/3, + position/2, write/2, pwrite/2, pwrite/3, + read_line/1, read/2, pread/2, pread/3]). + +%% OTP internal. +-export([ipread_s32bu_p32bu/3, sendfile/8]). + +-export([open_layer/3]). + +-include("file_int.hrl"). + +open_layer(Filename, Modes, Options) -> + IsAppend = lists:member(append, Modes), + IsDeflate = lists:member(write, Modes), + IsInflate = lists:member(read, Modes), + if + IsDeflate, IsInflate; IsAppend -> + {error, einval}; + IsDeflate, not IsInflate -> + start_server_module(raw_file_io_deflate, Filename, Modes, Options); + IsInflate -> + start_server_module(raw_file_io_inflate, Filename, Modes, Options) + end. + +start_server_module(Module, Filename, Modes, Options) -> + Secret = make_ref(), + case gen_statem:start(Module, {self(), Secret, Options}, []) of + {ok, Pid} -> open_next_layer(Pid, Secret, Filename, Modes); + Other -> Other + end. + +open_next_layer(Pid, Secret, Filename, Modes) -> + case gen_statem:call(Pid, {'$open', Secret, Filename, Modes}, infinity) of + ok -> + PublicFd = #file_descriptor{ + module = raw_file_io_compressed, data = {self(), Pid} }, + {ok, PublicFd}; + Other -> Other + end. + +close(Fd) -> + wrap_call(Fd, [close]). + +sync(Fd) -> + wrap_call(Fd, [sync]). +datasync(Fd) -> + wrap_call(Fd, [datasync]). + +truncate(Fd) -> + wrap_call(Fd, [truncate]). + +advise(Fd, Offset, Length, Advise) -> + wrap_call(Fd, [advise, Offset, Length, Advise]). +allocate(Fd, Offset, Length) -> + wrap_call(Fd, [allocate, Offset, Length]). + +position(Fd, Mark) -> + wrap_call(Fd, [position, Mark]). + +write(Fd, IOData) -> + try + CompactedData = erlang:iolist_to_iovec(IOData), + wrap_call(Fd, [write, CompactedData]) + catch + error:badarg -> {error, badarg} + end. + +pwrite(Fd, Offset, IOData) -> + try + CompactedData = erlang:iolist_to_iovec(IOData), + wrap_call(Fd, [pwrite, Offset, CompactedData]) + catch + error:badarg -> {error, badarg} + end. +pwrite(Fd, LocBytes) -> + try + CompactedLocBytes = + [ {Offset, erlang:iolist_to_iovec(IOData)} || + {Offset, IOData} <- LocBytes ], + wrap_call(Fd, [pwrite, CompactedLocBytes]) + catch + error:badarg -> {error, badarg} + end. + +read_line(Fd) -> + wrap_call(Fd, [read_line]). +read(Fd, Size) -> + wrap_call(Fd, [read, Size]). +pread(Fd, Offset, Size) -> + wrap_call(Fd, [pread, Offset, Size]). +pread(Fd, LocNums) -> + wrap_call(Fd, [pread, LocNums]). + +ipread_s32bu_p32bu(Fd, Offset, MaxSize) -> + wrap_call(Fd, [ipread_s32bu_p32bu, Offset, MaxSize]). + +sendfile(_,_,_,_,_,_,_,_) -> + {error, enotsup}. + +wrap_call(Fd, Command) -> + {_Owner, Pid} = get_fd_data(Fd), + try gen_statem:call(Pid, Command, infinity) of + Result -> Result + catch + exit:{noproc, _StackTrace} -> {error, einval} + end. + +get_fd_data(#file_descriptor{ data = Data }) -> + {Owner, _ServerPid} = Data, + case self() of + Owner -> Data; + _ -> error(not_on_controlling_process) + end. diff --git a/lib/kernel/src/raw_file_io_deflate.erl b/lib/kernel/src/raw_file_io_deflate.erl new file mode 100644 index 0000000000..acfc546743 --- /dev/null +++ b/lib/kernel/src/raw_file_io_deflate.erl @@ -0,0 +1,159 @@ +%% +%% %CopyrightBegin% +%% +%% Copyright Ericsson AB 2017. All Rights Reserved. +%% +%% Licensed under the Apache License, Version 2.0 (the "License"); +%% you may not use this file except in compliance with the License. +%% You may obtain a copy of the License at +%% +%% http://www.apache.org/licenses/LICENSE-2.0 +%% +%% Unless required by applicable law or agreed to in writing, software +%% distributed under the License is distributed on an "AS IS" BASIS, +%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +%% See the License for the specific language governing permissions and +%% limitations under the License. +%% +%% %CopyrightEnd% +%% +-module(raw_file_io_deflate). + +-behavior(gen_statem). + +-export([init/1, callback_mode/0, terminate/3]). +-export([opening/3, opened/3]). + +-include("file_int.hrl"). + +-define(GZIP_WBITS, 16 + 15). + +callback_mode() -> state_functions. + +init({Owner, Secret, [compressed]}) -> + Monitor = monitor(process, Owner), + Z = zlib:open(), + ok = zlib:deflateInit(Z, default, deflated, ?GZIP_WBITS, 8, default), + Data = + #{ owner => Owner, + monitor => Monitor, + secret => Secret, + position => 0, + zlib => Z }, + {ok, opening, Data}. + +opening({call, From}, {'$open', Secret, Filename, Modes}, #{ secret := Secret } = Data) -> + case raw_file_io:open(Filename, Modes) of + {ok, PrivateFd} -> + NewData = Data#{ handle => PrivateFd }, + {next_state, opened, NewData, [{reply, From, ok}]}; + Other -> + {stop_and_reply, normal, [{reply, From, Other}]} + end; +opening(_Event, _Contents, _Data) -> + {keep_state_and_data, [postpone]}. + +%% + +opened(info, {'DOWN', Monitor, process, _Owner, Reason}, #{ monitor := Monitor } = Data) -> + if + Reason =/= kill -> flush_deflate_state(Data); + Reason =:= kill -> ignored + end, + {stop, shutdown}; + +opened(info, _Message, _Data) -> + keep_state_and_data; + +opened({call, {Owner, _Tag} = From}, [close], #{ owner := Owner } = Data) -> + #{ handle := PrivateFd } = Data, + Response = + case flush_deflate_state(Data) of + ok -> ?CALL_FD(PrivateFd, close, []); + Other -> Other + end, + {stop_and_reply, normal, [{reply, From, Response}]}; + +opened({call, {Owner, _Tag} = From}, [position, Mark], #{ owner := Owner } = Data) -> + case position(Data, Mark) of + {ok, NewData, Result} -> + Response = {ok, Result}, + {keep_state, NewData, [{reply, From, Response}]}; + Other -> + {keep_state_and_data, [{reply, From, Other}]} + end; + +opened({call, {Owner, _Tag} = From}, [write, IOVec], #{ owner := Owner } = Data) -> + case write(Data, IOVec) of + {ok, NewData} -> {keep_state, NewData, [{reply, From, ok}]}; + Other -> {keep_state_and_data, [{reply, From, Other}]} + end; + +opened({call, {Owner, _Tag} = From}, [read, _Size], #{ owner := Owner }) -> + Response = {error, ebadf}, + {keep_state_and_data, [{reply, From, Response}]}; + +opened({call, {Owner, _Tag} = From}, [read_line], #{ owner := Owner }) -> + Response = {error, ebadf}, + {keep_state_and_data, [{reply, From, Response}]}; + +opened({call, {Owner, _Tag} = From}, _Command, #{ owner := Owner }) -> + Response = {error, enotsup}, + {keep_state_and_data, [{reply, From, Response}]}; + +opened({call, _From}, _Command, _Data) -> + %% The client functions filter this out, so we'll crash if the user does + %% anything stupid on purpose. + {shutdown, protocol_violation}; + +opened(_Event, _Request, _Data) -> + keep_state_and_data. + +write(Data, IOVec) -> + #{ handle := PrivateFd, position := Position, zlib := Z } = Data, + UncompressedSize = iolist_size(IOVec), + case ?CALL_FD(PrivateFd, write, [zlib:deflate(Z, IOVec)]) of + ok -> {ok, Data#{ position := (Position + UncompressedSize) }}; + Other -> Other + end. + +%% +%% We support "seeking" forward as long as it isn't relative to EOF. +%% +%% Seeking is a bit of a misnomer as it's really just compressing zeroes until +%% we reach the desired point, but it has always behaved like this. +%% + +position(Data, Mark) when is_atom(Mark) -> + position(Data, {Mark, 0}); +position(Data, Offset) when is_integer(Offset) -> + position(Data, {bof, Offset}); +position(Data, {bof, Offset}) when is_integer(Offset) -> + position_1(Data, Offset); +position(Data, {cur, Offset}) when is_integer(Offset) -> + #{ position := Position } = Data, + position_1(Data, Position + Offset); +position(_Data, {eof, Offset}) when is_integer(Offset) -> + {error, einval}; +position(_Data, _Any) -> + {error, badarg}. + +position_1(#{ position := Desired } = Data, Desired) -> + {ok, Data, Desired}; +position_1(#{ position := Current } = Data, Desired) when Current < Desired -> + BytesToWrite = min(Desired - Current, 4 bsl 20), + case write(Data, <<0:(BytesToWrite)/unit:8>>) of + {ok, NewData} -> position_1(NewData, Desired); + Other -> Other + end; +position_1(#{ position := Current }, Desired) when Current > Desired -> + {error, einval}. + +flush_deflate_state(#{ handle := PrivateFd, zlib := Z }) -> + case ?CALL_FD(PrivateFd, write, [zlib:deflate(Z, [], finish)]) of + ok -> ok; + Other -> Other + end. + +terminate(_Reason, _State, _Data) -> + ok. diff --git a/lib/kernel/src/raw_file_io_delayed.erl b/lib/kernel/src/raw_file_io_delayed.erl new file mode 100644 index 0000000000..d2ad7550a1 --- /dev/null +++ b/lib/kernel/src/raw_file_io_delayed.erl @@ -0,0 +1,320 @@ +%% +%% %CopyrightBegin% +%% +%% Copyright Ericsson AB 2017. All Rights Reserved. +%% +%% Licensed under the Apache License, Version 2.0 (the "License"); +%% you may not use this file except in compliance with the License. +%% You may obtain a copy of the License at +%% +%% http://www.apache.org/licenses/LICENSE-2.0 +%% +%% Unless required by applicable law or agreed to in writing, software +%% distributed under the License is distributed on an "AS IS" BASIS, +%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +%% See the License for the specific language governing permissions and +%% limitations under the License. +%% +%% %CopyrightEnd% +%% +-module(raw_file_io_delayed). + +-behavior(gen_statem). + +-export([close/1, sync/1, datasync/1, truncate/1, advise/4, allocate/3, + position/2, write/2, pwrite/2, pwrite/3, + read_line/1, read/2, pread/2, pread/3]). + +%% OTP internal. +-export([ipread_s32bu_p32bu/3, sendfile/8]). + +-export([open_layer/3]). + +-export([init/1, callback_mode/0, terminate/3]). +-export([opening/3, opened/3]). + +-include("file_int.hrl"). + +open_layer(Filename, Modes, Options) -> + Secret = make_ref(), + case gen_statem:start(?MODULE, {self(), Secret, Options}, []) of + {ok, Pid} -> + gen_statem:call(Pid, {'$open', Secret, Filename, Modes}, infinity); + Other -> + Other + end. + +callback_mode() -> state_functions. + +init({Owner, Secret, Options}) -> + Monitor = monitor(process, Owner), + Defaults = + #{ owner => Owner, + monitor => Monitor, + secret => Secret, + timer => none, + pid => self(), + buffer => prim_buffer:new(), + delay_size => 64 bsl 10, + delay_time => 2000 }, + Data = fill_delay_values(Defaults, Options), + {ok, opening, Data}. + +fill_delay_values(Data, []) -> + Data; +fill_delay_values(Data, [{delayed_write, Size, Time} | Options]) -> + fill_delay_values(Data#{ delay_size => Size, delay_time => Time }, Options); +fill_delay_values(Data, [_ | Options]) -> + fill_delay_values(Data, Options). + +opening({call, From}, {'$open', Secret, Filename, Modes}, #{ secret := Secret } = Data) -> + case raw_file_io:open(Filename, Modes) of + {ok, PrivateFd} -> + PublicData = maps:with([owner, buffer, delay_size, pid], Data), + PublicFd = #file_descriptor{ module = ?MODULE, data = PublicData }, + + NewData = Data#{ handle => PrivateFd }, + Response = {ok, PublicFd}, + {next_state, opened, NewData, [{reply, From, Response}]}; + Other -> + {stop_and_reply, normal, [{reply, From, Other}]} + end; +opening(_Event, _Contents, _Data) -> + {keep_state_and_data, [postpone]}. + +%% + +opened(info, {'$timed_out', Secret}, #{ secret := Secret } = Data) -> + %% If the user writes something at this exact moment, the flush will fail + %% and the timer won't reset on the next write since the buffer won't be + %% empty (Unless we collided on a flush). We therefore reset the timeout to + %% ensure that data won't sit idle for extended periods of time. + case try_flush_write_buffer(Data) of + busy -> gen_statem:cast(self(), '$reset_timeout'); + ok -> ok + end, + {keep_state, Data#{ timer => none }, []}; + +opened(info, {'DOWN', Monitor, process, _Owner, Reason}, #{ monitor := Monitor } = Data) -> + if + Reason =/= kill -> try_flush_write_buffer(Data); + Reason =:= kill -> ignored + end, + {stop, shutdown}; + +opened(info, _Message, _Data) -> + keep_state_and_data; + +opened({call, {Owner, _Tag} = From}, [close], #{ owner := Owner } = Data) -> + case flush_write_buffer(Data) of + ok -> + #{ handle := PrivateFd } = Data, + Response = ?CALL_FD(PrivateFd, close, []), + {stop_and_reply, normal, [{reply, From, Response}]}; + Other -> + {stop_and_reply, normal, [{reply, From, Other}]} + end; + +opened({call, {Owner, _Tag} = From}, '$wait', #{ owner := Owner }) -> + %% Used in write/2 to synchronize writes on lock conflicts. + {keep_state_and_data, [{reply, From, ok}]}; + +opened({call, {Owner, _Tag} = From}, '$synchronous_flush', #{ owner := Owner } = Data) -> + cancel_flush_timeout(Data), + Response = flush_write_buffer(Data), + {keep_state_and_data, [{reply, From, Response}]}; + +opened({call, {Owner, _Tag} = From}, Command, #{ owner := Owner } = Data) -> + Response = + case flush_write_buffer(Data) of + ok -> dispatch_command(Data, Command); + Other -> Other + end, + {keep_state_and_data, [{reply, From, Response}]}; + +opened({call, _From}, _Command, _Data) -> + %% The client functions filter this out, so we'll crash if the user does + %% anything stupid on purpose. + {shutdown, protocol_violation}; + +opened(cast, '$reset_timeout', #{ delay_time := Timeout, secret := Secret } = Data) -> + cancel_flush_timeout(Data), + Timer = erlang:send_after(Timeout, self(), {'$timed_out', Secret}), + {keep_state, Data#{ timer => Timer }, []}; + +opened(cast, _Message, _Data) -> + {keep_state_and_data, []}. + +dispatch_command(Data, [Function | Args]) -> + #{ handle := Handle } = Data, + Module = Handle#file_descriptor.module, + apply(Module, Function, [Handle | Args]). + +cancel_flush_timeout(#{ timer := none }) -> + ok; +cancel_flush_timeout(#{ timer := Timer }) -> + _ = erlang:cancel_timer(Timer, [{async, true}]), + ok. + +try_flush_write_buffer(#{ buffer := Buffer, handle := PrivateFd }) -> + case prim_buffer:try_lock(Buffer) of + acquired -> + flush_write_buffer_1(Buffer, PrivateFd), + prim_buffer:unlock(Buffer), + ok; + busy -> + busy + end. + +%% This is only safe to use when there is no chance of conflict with the owner +%% process, or in other words, "during synchronous calls outside of the locked +%% section of write/2" +flush_write_buffer(#{ buffer := Buffer, handle := PrivateFd }) -> + acquired = prim_buffer:try_lock(Buffer), + Result = flush_write_buffer_1(Buffer, PrivateFd), + prim_buffer:unlock(Buffer), + Result. + +flush_write_buffer_1(Buffer, PrivateFd) -> + case prim_buffer:size(Buffer) of + Size when Size > 0 -> + ?CALL_FD(PrivateFd, write, [prim_buffer:read_iovec(Buffer, Size)]); + 0 -> + ok + end. + +terminate(_Reason, _State, _Data) -> + ok. + +%% Client functions + +write(Fd, IOData) -> + try + enqueue_write(Fd, erlang:iolist_to_iovec(IOData)) + catch + error:badarg -> {error, badarg} + end. +enqueue_write(_Fd, []) -> + ok; +enqueue_write(Fd, IOVec) -> + %% get_fd_data will reject everyone except the process that opened the Fd, + %% so we can't race with anyone except the wrapper process. + #{ delay_size := DelaySize, + buffer := Buffer, + pid := Pid } = get_fd_data(Fd), + case prim_buffer:try_lock(Buffer) of + acquired -> + %% (The wrapper process will exit without flushing if we're killed + %% while holding the lock). + enqueue_write_locked(Pid, Buffer, DelaySize, IOVec); + busy -> + %% This can only happen while we're processing a timeout in the + %% wrapper process, so we perform a bogus call to get a completion + %% notification before trying again. + gen_statem:call(Pid, '$wait'), + enqueue_write(Fd, IOVec) + end. +enqueue_write_locked(Pid, Buffer, DelaySize, IOVec) -> + %% The synchronous operations (write, forced flush) are safe since we're + %% running on the only process that can fill the buffer; a timeout being + %% processed just before $synchronous_flush will cause the flush to nop, + %% and a timeout sneaking in just before a synchronous write won't do + %% anything since the buffer is guaranteed to be empty at that point. + BufSize = prim_buffer:size(Buffer), + case is_iovec_smaller_than(IOVec, DelaySize - BufSize) of + true when BufSize > 0 -> + prim_buffer:write(Buffer, IOVec), + prim_buffer:unlock(Buffer); + true -> + prim_buffer:write(Buffer, IOVec), + prim_buffer:unlock(Buffer), + gen_statem:cast(Pid, '$reset_timeout'); + false when BufSize > 0 -> + prim_buffer:write(Buffer, IOVec), + prim_buffer:unlock(Buffer), + gen_statem:call(Pid, '$synchronous_flush'); + false -> + prim_buffer:unlock(Buffer), + gen_statem:call(Pid, [write, IOVec]) + end. + +%% iolist_size/1 will always look through the entire list to get a precise +%% amount, which is pretty inefficient since we only need to know whether we've +%% hit the buffer threshold or not. +%% +%% We only handle the binary case since write/2 forcibly translates input to +%% erlang:iovec(). +is_iovec_smaller_than(IOVec, Max) -> + is_iovec_smaller_than_1(IOVec, Max, 0). +is_iovec_smaller_than_1(_IOVec, Max, Acc) when Acc >= Max -> + false; +is_iovec_smaller_than_1([], _Max, _Acc) -> + true; +is_iovec_smaller_than_1([Binary | Rest], Max, Acc) when is_binary(Binary) -> + is_iovec_smaller_than_1(Rest, Max, Acc + byte_size(Binary)). + +close(Fd) -> + wrap_call(Fd, [close]). + +sync(Fd) -> + wrap_call(Fd, [sync]). +datasync(Fd) -> + wrap_call(Fd, [datasync]). + +truncate(Fd) -> + wrap_call(Fd, [truncate]). + +advise(Fd, Offset, Length, Advise) -> + wrap_call(Fd, [advise, Offset, Length, Advise]). +allocate(Fd, Offset, Length) -> + wrap_call(Fd, [allocate, Offset, Length]). + +position(Fd, Mark) -> + wrap_call(Fd, [position, Mark]). + +pwrite(Fd, Offset, IOData) -> + try + CompactedData = erlang:iolist_to_iovec(IOData), + wrap_call(Fd, [pwrite, Offset, CompactedData]) + catch + error:badarg -> {error, badarg} + end. +pwrite(Fd, LocBytes) -> + try + CompactedLocBytes = + [ {Offset, erlang:iolist_to_iovec(IOData)} || + {Offset, IOData} <- LocBytes ], + wrap_call(Fd, [pwrite, CompactedLocBytes]) + catch + error:badarg -> {error, badarg} + end. + +read_line(Fd) -> + wrap_call(Fd, [read_line]). +read(Fd, Size) -> + wrap_call(Fd, [read, Size]). +pread(Fd, Offset, Size) -> + wrap_call(Fd, [pread, Offset, Size]). +pread(Fd, LocNums) -> + wrap_call(Fd, [pread, LocNums]). + +ipread_s32bu_p32bu(Fd, Offset, MaxSize) -> + wrap_call(Fd, [ipread_s32bu_p32bu, Offset, MaxSize]). + +sendfile(_,_,_,_,_,_,_,_) -> + {error, enotsup}. + +wrap_call(Fd, Command) -> + #{ pid := Pid } = get_fd_data(Fd), + try gen_statem:call(Pid, Command, infinity) of + Result -> Result + catch + exit:{noproc, _StackTrace} -> {error, einval} + end. + +get_fd_data(#file_descriptor{ data = Data }) -> + #{ owner := Owner } = Data, + case self() of + Owner -> Data; + _ -> error(not_on_controlling_process) + end. diff --git a/lib/kernel/src/raw_file_io_inflate.erl b/lib/kernel/src/raw_file_io_inflate.erl new file mode 100644 index 0000000000..7e9780310c --- /dev/null +++ b/lib/kernel/src/raw_file_io_inflate.erl @@ -0,0 +1,261 @@ +%% +%% %CopyrightBegin% +%% +%% Copyright Ericsson AB 2017. All Rights Reserved. +%% +%% Licensed under the Apache License, Version 2.0 (the "License"); +%% you may not use this file except in compliance with the License. +%% You may obtain a copy of the License at +%% +%% http://www.apache.org/licenses/LICENSE-2.0 +%% +%% Unless required by applicable law or agreed to in writing, software +%% distributed under the License is distributed on an "AS IS" BASIS, +%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +%% See the License for the specific language governing permissions and +%% limitations under the License. +%% +%% %CopyrightEnd% +%% +-module(raw_file_io_inflate). + +-behavior(gen_statem). + +-export([init/1, callback_mode/0, terminate/3]). +-export([opening/3, opened_gzip/3, opened_passthrough/3]). + +-include("file_int.hrl"). + +-define(INFLATE_CHUNK_SIZE, (1 bsl 10)). +-define(GZIP_WBITS, (16 + 15)). + +callback_mode() -> state_functions. + +init({Owner, Secret, [compressed]}) -> + Monitor = monitor(process, Owner), + %% We're using the undocumented inflateInit/3 to open the stream in + %% 'reset mode', which resets the inflate state at the end of every stream, + %% allowing us to read concatenated gzip files. + Z = zlib:open(), + ok = zlib:inflateInit(Z, ?GZIP_WBITS, reset), + Data = + #{ owner => Owner, + monitor => Monitor, + secret => Secret, + position => 0, + buffer => prim_buffer:new(), + zlib => Z }, + {ok, opening, Data}. + +%% The old driver fell back to plain reads if the file didn't start with the +%% magic gzip bytes. +choose_decompression_state(PrivateFd) -> + State = + case ?CALL_FD(PrivateFd, read, [2]) of + {ok, <<16#1F, 16#8B>>} -> opened_gzip; + _Other -> opened_passthrough + end, + {ok, 0} = ?CALL_FD(PrivateFd, position, [0]), + State. + +opening({call, From}, {'$open', Secret, Filename, Modes}, #{ secret := Secret } = Data) -> + case raw_file_io:open(Filename, Modes) of + {ok, PrivateFd} -> + NextState = choose_decompression_state(PrivateFd), + NewData = Data#{ handle => PrivateFd }, + {next_state, NextState, NewData, [{reply, From, ok}]}; + Other -> + {stop_and_reply, normal, [{reply, From, Other}]} + end; +opening(_Event, _Contents, _Data) -> + {keep_state_and_data, [postpone]}. + +internal_close(From, Data) -> + #{ handle := PrivateFd } = Data, + Response = ?CALL_FD(PrivateFd, close, []), + {stop_and_reply, normal, [{reply, From, Response}]}. + +opened_passthrough(info, {'DOWN', Monitor, process, _Owner, _Reason}, #{ monitor := Monitor }) -> + {stop, shutdown}; + +opened_passthrough(info, _Message, _Data) -> + keep_state_and_data; + +opened_passthrough({call, {Owner, _Tag} = From}, [close], #{ owner := Owner } = Data) -> + internal_close(From, Data); + +opened_passthrough({call, {Owner, _Tag} = From}, [Method | Args], #{ owner := Owner } = Data) -> + #{ handle := PrivateFd } = Data, + Response = ?CALL_FD(PrivateFd, Method, Args), + {keep_state_and_data, [{reply, From, Response}]}; + +opened_passthrough({call, _From}, _Command, _Data) -> + %% The client functions filter this out, so we'll crash if the user does + %% anything stupid on purpose. + {shutdown, protocol_violation}; + +opened_passthrough(_Event, _Request, _Data) -> + keep_state_and_data. + +%% + +opened_gzip(info, {'DOWN', Monitor, process, _Owner, _Reason}, #{ monitor := Monitor }) -> + {stop, shutdown}; + +opened_gzip(info, _Message, _Data) -> + keep_state_and_data; + +opened_gzip({call, {Owner, _Tag} = From}, [close], #{ owner := Owner } = Data) -> + internal_close(From, Data); + +opened_gzip({call, {Owner, _Tag} = From}, [position, Mark], #{ owner := Owner } = Data) -> + case position(Data, Mark) of + {ok, NewData, Result} -> + Response = {ok, Result}, + {keep_state, NewData, [{reply, From, Response}]}; + Other -> + {keep_state_and_data, [{reply, From, Other}]} + end; + +opened_gzip({call, {Owner, _Tag} = From}, [read, Size], #{ owner := Owner } = Data) -> + case read(Data, Size) of + {ok, NewData, Result} -> + Response = {ok, Result}, + {keep_state, NewData, [{reply, From, Response}]}; + Other -> + {keep_state_and_data, [{reply, From, Other}]} + end; + +opened_gzip({call, {Owner, _Tag} = From}, [read_line], #{ owner := Owner } = Data) -> + case read_line(Data) of + {ok, NewData, Result} -> + Response = {ok, Result}, + {keep_state, NewData, [{reply, From, Response}]}; + Other -> + {keep_state_and_data, [{reply, From, Other}]} + end; + +opened_gzip({call, {Owner, _Tag} = From}, [write, _IOData], #{ owner := Owner }) -> + Response = {error, ebadf}, + {keep_state_and_data, [{reply, From, Response}]}; + +opened_gzip({call, {Owner, _Tag} = From}, _Request, #{ owner := Owner }) -> + Response = {error, enotsup}, + {keep_state_and_data, [{reply, From, Response}]}; + +opened_gzip({call, _From}, _Request, _Data) -> + %% The client functions filter this out, so we'll crash if the user does + %% anything stupid on purpose. + {shutdown, protocol_violation}; + +opened_gzip(_Event, _Request, _Data) -> + keep_state_and_data. + +%% + +read(#{ buffer := Buffer } = Data, Size) -> + try read_1(Data, Buffer, prim_buffer:size(Buffer), Size) of + Result -> Result + catch + error:badarg -> {error, badarg}; + error:_ -> {error, eio} + end. +read_1(Data, Buffer, BufferSize, ReadSize) when BufferSize >= ReadSize -> + #{ position := Position } = Data, + Decompressed = prim_buffer:read(Buffer, ReadSize), + {ok, Data#{ position => (Position + ReadSize) }, Decompressed}; +read_1(Data, Buffer, BufferSize, ReadSize) when BufferSize < ReadSize -> + #{ handle := PrivateFd } = Data, + case ?CALL_FD(PrivateFd, read, [?INFLATE_CHUNK_SIZE]) of + {ok, Compressed} -> + #{ zlib := Z } = Data, + Uncompressed = erlang:iolist_to_iovec(zlib:inflate(Z, Compressed)), + prim_buffer:write(Buffer, Uncompressed), + read_1(Data, Buffer, prim_buffer:size(Buffer), ReadSize); + eof when BufferSize > 0 -> + read_1(Data, Buffer, BufferSize, BufferSize); + Other -> + Other + end. + +read_line(#{ buffer := Buffer } = Data) -> + try read_line_1(Data, Buffer, prim_buffer:find_byte_index(Buffer, $\n)) of + {ok, NewData, Decompressed} -> {ok, NewData, Decompressed}; + Other -> Other + catch + error:badarg -> {error, badarg}; + error:_ -> {error, eio} + end. + +read_line_1(Data, Buffer, not_found) -> + #{ handle := PrivateFd, zlib := Z } = Data, + case ?CALL_FD(PrivateFd, read, [?INFLATE_CHUNK_SIZE]) of + {ok, Compressed} -> + Uncompressed = erlang:iolist_to_iovec(zlib:inflate(Z, Compressed)), + prim_buffer:write(Buffer, Uncompressed), + read_line_1(Data, Buffer, prim_buffer:find_byte_index(Buffer, $\n)); + eof -> + case prim_buffer:size(Buffer) of + Size when Size > 0 -> {ok, prim_buffer:read(Buffer, Size)}; + Size when Size =:= 0 -> eof + end; + Error -> + Error + end; +read_line_1(Data, Buffer, {ok, LFIndex}) -> + %% Translate CRLF into just LF, completely ignoring which encoding is used, + %% but treat the file position as including CR. + #{ position := Position } = Data, + NewData = Data#{ position => (Position + LFIndex + 1) }, + CRIndex = (LFIndex - 1), + TranslatedLine = + case prim_buffer:read(Buffer, LFIndex + 1) of + <<Line:CRIndex/binary, "\r\n">> -> <<Line/binary, "\n">>; + Line -> Line + end, + {ok, NewData, TranslatedLine}. + +%% +%% We support seeking in both directions as long as it isn't relative to EOF. +%% +%% Seeking backwards is extremely inefficient since we have to seek to the very +%% beginning and then decompress up to the desired point. +%% + +position(Data, Mark) when is_atom(Mark) -> + position(Data, {Mark, 0}); +position(Data, Offset) when is_integer(Offset) -> + position(Data, {bof, Offset}); +position(Data, {bof, Offset}) when is_integer(Offset) -> + position_1(Data, Offset); +position(Data, {cur, Offset}) when is_integer(Offset) -> + #{ position := Position } = Data, + position_1(Data, Position + Offset); +position(_Data, {eof, Offset}) when is_integer(Offset) -> + {error, einval}; +position(_Data, _Other) -> + {error, badarg}. + +position_1(_Data, Desired) when Desired < 0 -> + {error, einval}; +position_1(#{ position := Desired } = Data, Desired) -> + {ok, Data, Desired}; +position_1(#{ position := Current } = Data, Desired) when Current < Desired -> + case read(Data, min(Desired - Current, ?INFLATE_CHUNK_SIZE)) of + {ok, NewData, _Data} -> position_1(NewData, Desired); + eof -> {ok, Data, Current}; + Other -> Other + end; +position_1(#{ position := Current } = Data, Desired) when Current > Desired -> + #{ handle := PrivateFd, buffer := Buffer, zlib := Z } = Data, + case ?CALL_FD(PrivateFd, position, [bof]) of + {ok, 0} -> + ok = zlib:inflateReset(Z), + prim_buffer:wipe(Buffer), + position_1(Data#{ position => 0 }, Desired); + Other -> + Other + end. + +terminate(_Reason, _State, _Data) -> + ok. diff --git a/lib/kernel/src/raw_file_io_list.erl b/lib/kernel/src/raw_file_io_list.erl new file mode 100644 index 0000000000..2e16e63f0e --- /dev/null +++ b/lib/kernel/src/raw_file_io_list.erl @@ -0,0 +1,128 @@ +%% +%% %CopyrightBegin% +%% +%% Copyright Ericsson AB 2017. All Rights Reserved. +%% +%% Licensed under the Apache License, Version 2.0 (the "License"); +%% you may not use this file except in compliance with the License. +%% You may obtain a copy of the License at +%% +%% http://www.apache.org/licenses/LICENSE-2.0 +%% +%% Unless required by applicable law or agreed to in writing, software +%% distributed under the License is distributed on an "AS IS" BASIS, +%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +%% See the License for the specific language governing permissions and +%% limitations under the License. +%% +%% %CopyrightEnd% +%% +-module(raw_file_io_list). + +-export([close/1, sync/1, datasync/1, truncate/1, advise/4, allocate/3, + position/2, write/2, pwrite/2, pwrite/3, + read_line/1, read/2, pread/2, pread/3]). + +%% OTP internal. +-export([ipread_s32bu_p32bu/3, sendfile/8]). + +-export([open_layer/3]). + +-include("file_int.hrl"). + +open_layer(Filename, Modes, [list]) -> + case raw_file_io:open(Filename, [binary | Modes]) of + {ok, PrivateFd} -> {ok, make_public_fd(PrivateFd, Modes)}; + Other -> Other + end. + +%% We can skip wrapping the file if it's write-only since only read operations +%% are affected by list mode. Since raw_file_io fills in all implicit options +%% for us, all we need to do is check whether 'read' is among them. +make_public_fd(PrivateFd, Modes) -> + case lists:member(read, Modes) of + true -> #file_descriptor{ module = ?MODULE, data = PrivateFd }; + false -> PrivateFd + end. + +close(Fd) -> + PrivateFd = Fd#file_descriptor.data, + ?CALL_FD(PrivateFd, close, []). + +sync(Fd) -> + PrivateFd = Fd#file_descriptor.data, + ?CALL_FD(PrivateFd, sync, []). +datasync(Fd) -> + PrivateFd = Fd#file_descriptor.data, + ?CALL_FD(PrivateFd, datasync, []). + +truncate(Fd) -> + PrivateFd = Fd#file_descriptor.data, + ?CALL_FD(PrivateFd, truncate, []). + +advise(Fd, Offset, Length, Advise) -> + PrivateFd = Fd#file_descriptor.data, + ?CALL_FD(PrivateFd, advise, [Offset, Length, Advise]). +allocate(Fd, Offset, Length) -> + PrivateFd = Fd#file_descriptor.data, + ?CALL_FD(PrivateFd, allocate, [Offset, Length]). + +position(Fd, Mark) -> + PrivateFd = Fd#file_descriptor.data, + ?CALL_FD(PrivateFd, position, [Mark]). + +write(Fd, IOData) -> + PrivateFd = Fd#file_descriptor.data, + ?CALL_FD(PrivateFd, write, [IOData]). + +pwrite(Fd, Offset, IOData) -> + PrivateFd = Fd#file_descriptor.data, + ?CALL_FD(PrivateFd, pwrite, [Offset, IOData]). +pwrite(Fd, LocBytes) -> + PrivateFd = Fd#file_descriptor.data, + ?CALL_FD(PrivateFd, pwrite, [LocBytes]). + +read_line(Fd) -> + PrivateFd = Fd#file_descriptor.data, + case ?CALL_FD(PrivateFd, read_line, []) of + {ok, Binary} -> {ok, binary_to_list(Binary)}; + Other -> Other + end. +read(Fd, Size) -> + PrivateFd = Fd#file_descriptor.data, + case ?CALL_FD(PrivateFd, read, [Size]) of + {ok, Binary} -> {ok, binary_to_list(Binary)}; + Other -> Other + end. +pread(Fd, Offset, Size) -> + PrivateFd = Fd#file_descriptor.data, + case ?CALL_FD(PrivateFd, pread, [Offset, Size]) of + {ok, Binary} -> {ok, binary_to_list(Binary)}; + Other -> Other + end. +pread(Fd, LocNums) -> + PrivateFd = Fd#file_descriptor.data, + case ?CALL_FD(PrivateFd, pread, [LocNums]) of + {ok, LocResults} -> + TranslatedResults = + [ case Result of + Result when is_binary(Result) -> binary_to_list(Result); + eof -> eof + end || Result <- LocResults ], + {ok, TranslatedResults}; + Other -> Other + end. + +ipread_s32bu_p32bu(Fd, Offset, MaxSize) -> + PrivateFd = Fd#file_descriptor.data, + case ?CALL_FD(PrivateFd, ipread_s32bu_p32bu, [Offset, MaxSize]) of + {ok, {Size, Pointer, Binary}} when is_binary(Binary) -> + {ok, {Size, Pointer, binary_to_list(Binary)}}; + Other -> + Other + end. + +sendfile(Fd, Dest, Offset, Bytes, ChunkSize, Headers, Trailers, Flags) -> + Args = [Dest, Offset, Bytes, ChunkSize, Headers, Trailers, Flags], + PrivateFd = Fd#file_descriptor.data, + ?CALL_FD(PrivateFd, sendfile, Args). diff --git a/lib/kernel/src/raw_file_io_raw.erl b/lib/kernel/src/raw_file_io_raw.erl new file mode 100644 index 0000000000..9a9fe78eb1 --- /dev/null +++ b/lib/kernel/src/raw_file_io_raw.erl @@ -0,0 +1,25 @@ +%% +%% %CopyrightBegin% +%% +%% Copyright Ericsson AB 2017. All Rights Reserved. +%% +%% Licensed under the Apache License, Version 2.0 (the "License"); +%% you may not use this file except in compliance with the License. +%% You may obtain a copy of the License at +%% +%% http://www.apache.org/licenses/LICENSE-2.0 +%% +%% Unless required by applicable law or agreed to in writing, software +%% distributed under the License is distributed on an "AS IS" BASIS, +%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +%% See the License for the specific language governing permissions and +%% limitations under the License. +%% +%% %CopyrightEnd% +%% +-module(raw_file_io_raw). + +-export([open_layer/3]). + +open_layer(Filename, Modes, [raw]) -> + prim_file:open(Filename, [raw | Modes]). |