From 6487aac5977cf470bc6a2cd0964da2850ee38717 Mon Sep 17 00:00:00 2001 From: Rickard Green Date: Thu, 30 Oct 2014 23:57:01 +0100 Subject: Introduce a new time API The old time API is based on erlang:now/0. The major issue with erlang:now/0 is that it was intended to be used for so many unrelated things. This tied these unrelated operations together and unnecessarily caused performance, scalability as well as accuracy, and precision issues for operations that do not need to have such issues. The new API spreads different functionality over multiple functions in order to improve on this. The new API consists of a number of new BIFs: - erlang:convert_time_unit/3 - erlang:monotonic_time/0 - erlang:monotonic_time/1 - erlang:system_time/0 - erlang:system_time/1 - erlang:time_offset/0 - erlang:time_offset/1 - erlang:timestamp/0 - erlang:unique_integer/0 - erlang:unique_integer/1 - os:system_time/0 - os:system_time/1 and a number of extensions of existing BIFs: - erlang:monitor(time_offset, clock_service) - erlang:system_flag(time_offset, finalize) - erlang:system_info(os_monotonic_time_source) - erlang:system_info(time_offset) - erlang:system_info(time_warp_mode) - erlang:system_info(time_correction) - erlang:system_info(start_time) See the "Time and Time Correction in Erlang" chapter of the ERTS User's Guide for more information. --- erts/preloaded/ebin/erlang.beam | Bin 97956 -> 100456 bytes erts/preloaded/ebin/erts_internal.beam | Bin 4164 -> 4688 bytes erts/preloaded/ebin/init.beam | Bin 48808 -> 48800 bytes erts/preloaded/src/erlang.erl | 136 ++++++++++++++++++++++++++++----- erts/preloaded/src/erts_internal.erl | 30 ++++++++ 5 files changed, 148 insertions(+), 18 deletions(-) (limited to 'erts/preloaded') diff --git a/erts/preloaded/ebin/erlang.beam b/erts/preloaded/ebin/erlang.beam index 7a3d94dd16..4ec388a7b9 100644 Binary files a/erts/preloaded/ebin/erlang.beam and b/erts/preloaded/ebin/erlang.beam differ diff --git a/erts/preloaded/ebin/erts_internal.beam b/erts/preloaded/ebin/erts_internal.beam index 9b364d8b21..705bb61247 100644 Binary files a/erts/preloaded/ebin/erts_internal.beam and b/erts/preloaded/ebin/erts_internal.beam differ diff --git a/erts/preloaded/ebin/init.beam b/erts/preloaded/ebin/init.beam index a190129b2c..aadc9797cb 100644 Binary files a/erts/preloaded/ebin/init.beam and b/erts/preloaded/ebin/init.beam differ diff --git a/erts/preloaded/src/erlang.erl b/erts/preloaded/src/erlang.erl index 611a568014..79eb60f362 100644 --- a/erts/preloaded/src/erlang.erl +++ b/erts/preloaded/src/erlang.erl @@ -38,7 +38,6 @@ -export([integer_to_list/2]). -export([integer_to_binary/2]). --export([flush_monitor_message/2]). -export([set_cpu_topology/1, format_cpu_topology/1]). -export([await_proc_exit/3]). -export([memory/0, memory/1]). @@ -48,7 +47,7 @@ await_sched_wall_time_modifications/2, gather_gc_info_result/1]). --deprecated([hash/2]). +-deprecated([hash/2, now/0]). %% Get rid of autoimports of spawn to avoid clashes with ourselves. -compile({no_auto_import,[spawn_link/1]}). @@ -58,12 +57,21 @@ -compile({no_auto_import,[spawn_opt/5]}). -export_type([timestamp/0]). +-export_type([time_unit/0]). -type ext_binary() :: binary(). -type timestamp() :: {MegaSecs :: non_neg_integer(), Secs :: non_neg_integer(), MicroSecs :: non_neg_integer()}. +-type time_unit() :: + pos_integer() + | 'seconds' + | 'milli_seconds' + | 'micro_seconds' + | 'nano_seconds' + | 'native'. + %%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%%% %% Native code BIF stubs and their types %% (BIF's actually implemented in this module goes last in the file) @@ -104,7 +112,8 @@ -export([list_to_bitstring/1, list_to_existing_atom/1, list_to_float/1]). -export([list_to_integer/1, list_to_integer/2]). -export([list_to_pid/1, list_to_tuple/1, loaded/0]). --export([localtime/0, make_ref/0, map_size/1, match_spec_test/3, md5/1, md5_final/1]). +-export([localtime/0, make_ref/0]). +-export([map_size/1, match_spec_test/3, md5/1, md5_final/1]). -export([md5_init/0, md5_update/2, module_loaded/1, monitor/2]). -export([monitor_node/2, monitor_node/3, nif_error/1, nif_error/2]). -export([node/0, node/1, now/0, phash/2, phash2/1, phash2/2]). @@ -112,6 +121,11 @@ -export([port_connect/2, port_control/3, port_get_data/1]). -export([port_set_data/2, port_to_list/1, ports/0]). -export([posixtime_to_universaltime/1, pre_loaded/0, prepare_loading/2]). +-export([monotonic_time/0, monotonic_time/1]). +-export([system_time/0, system_time/1]). +-export([convert_time_unit/3]). +-export([unique_integer/0, unique_integer/1]). +-export([time_offset/0, time_offset/1, timestamp/0]). -export([process_display/2]). -export([process_flag/3, process_info/1, processes/0, purge_module/1]). -export([put/2, raise/3, read_timer/1, ref_to_list/1, register/2]). @@ -1184,13 +1198,18 @@ md5_update(_Context, _Data) -> module_loaded(_Module) -> erlang:nif_error(undefined). +-type registered_name() :: atom(). + +-type registered_process_identifier() :: registered_name() | {registered_name(), node()}. + +-type monitor_process_identifier() :: pid() | registered_process_identifier(). + %% monitor/2 --spec monitor(Type, Item) -> MonitorRef when - Type :: process, - Item :: pid() | RegName | {RegName, Node}, - RegName :: module(), - Node :: node(), +-spec monitor(process, monitor_process_identifier()) -> MonitorRef when + MonitorRef :: reference(); + (time_offset, clock_service) -> MonitorRef when MonitorRef :: reference(). + monitor(_Type, _Item) -> erlang:nif_error(undefined). @@ -1292,6 +1311,90 @@ ports() -> posixtime_to_universaltime(_P1) -> erlang:nif_error(undefined). +-spec erlang:unique_integer(ModifierList) -> integer() when + ModifierList :: [Modifier], + Modifier :: positive | monotonic. + +unique_integer(_ModifierList) -> + erlang:nif_error(undefined). + +-spec erlang:unique_integer() -> integer(). + +unique_integer() -> + erlang:nif_error(undefined). + +-spec erlang:monotonic_time() -> integer(). + +monotonic_time() -> + erlang:nif_error(undefined). + +-spec erlang:monotonic_time(Unit) -> integer() when + Unit :: time_unit(). + +monotonic_time(_Unit) -> + erlang:nif_error(undefined). + +-spec erlang:system_time() -> integer(). + +system_time() -> + erlang:nif_error(undefined). + +-spec erlang:system_time(Unit) -> integer() when + Unit :: time_unit(). + +system_time(_Unit) -> + erlang:nif_error(undefined). + +-spec erlang:convert_time_unit(Time, FromUnit, ToUnit) -> ConvertedTime when + Time :: integer(), + ConvertedTime :: integer(), + FromUnit :: time_unit(), + ToUnit :: time_unit(). + +convert_time_unit(Time, FromUnit, ToUnit) -> + try + FU = case FromUnit of + native -> erts_internal:time_unit(); + nano_seconds -> 1000*1000*1000; + micro_seconds -> 1000*1000; + milli_seconds -> 1000; + seconds -> 1; + _ when FromUnit > 0 -> FromUnit + end, + TU = case ToUnit of + native -> erts_internal:time_unit(); + nano_seconds -> 1000*1000*1000; + micro_seconds -> 1000*1000; + milli_seconds -> 1000; + seconds -> 1; + _ when ToUnit > 0 -> ToUnit + end, + case Time < 0 of + true -> TU*Time - (FU - 1); + false -> TU*Time + end div FU + catch + _ : _ -> + erlang:error(badarg, [Time, FromUnit, ToUnit]) + end. + +-spec erlang:time_offset() -> integer(). + +time_offset() -> + erlang:nif_error(undefined). + +-spec erlang:time_offset(Unit) -> integer() when + Unit :: time_unit(). + +time_offset(_Unit) -> + erlang:nif_error(undefined). + +-spec erlang:timestamp() -> Timestamp when + Timestamp :: timestamp(). + +timestamp() -> + erlang:nif_error(undefined). + %% prepare_loading/2 -spec erlang:prepare_loading(Module, Code) -> PreparedCode | {error, Reason} when Module :: module(), @@ -2118,6 +2221,8 @@ subtract(_,_) -> (trace_control_word, TCW) -> OldTCW when TCW :: non_neg_integer(), OldTCW :: non_neg_integer(); + (time_offset, finalize) -> OldState when + OldState :: preliminary | final | volatile; %% These are deliberately not documented (internal_cpu_topology, term()) -> term(); (sequential_tracer, pid() | port() | false) -> pid() | port() | false; @@ -2254,6 +2359,7 @@ tuple_to_list(_Tuple) -> (multi_scheduling_blockers) -> [PID :: pid()]; (nif_version) -> string(); (otp_release) -> string(); + (os_monotonic_time_source) -> [{atom(),term()}]; (port_count) -> non_neg_integer(); (port_limit) -> pos_integer(); (process_count) -> pos_integer(); @@ -2271,10 +2377,14 @@ tuple_to_list(_Tuple) -> (scheduler_id) -> SchedulerId :: pos_integer(); (schedulers | schedulers_online) -> pos_integer(); (smp_support) -> boolean(); + (start_time) -> integer(); (system_version) -> string(); (system_architecture) -> string(); (threads) -> boolean(); (thread_pool_size) -> non_neg_integer(); + (time_correction) -> true | false; + (time_offset) -> preliminary | final | volatile; + (time_warp_mode) -> no_time_warp | single_time_warp | multi_time_warp; (tolerant_timeofday) -> enabled | disabled; (trace_control_word) -> non_neg_integer(); (update_cpu_info) -> changed | unchanged; @@ -3041,16 +3151,6 @@ integer_to_binary(I0, Base, R0) -> true -> integer_to_binary(I1, Base, R1) end. -%% erlang:flush_monitor_message/2 is for internal use only! -%% -%% erlang:demonitor(Ref, [flush]) traps to -%% erlang:flush_monitor_message(Ref, Res) when -%% it needs to flush a monitor message. -flush_monitor_message(Ref, Res) when erlang:is_reference(Ref), - erlang:is_atom(Res) -> - receive {_, Ref, _, _, _} -> ok after 0 -> ok end, - Res. - -record(cpu, {node = -1, processor = -1, processor_node = -1, diff --git a/erts/preloaded/src/erts_internal.erl b/erts/preloaded/src/erts_internal.erl index 2c5bd82cf0..30df75b406 100644 --- a/erts/preloaded/src/erts_internal.erl +++ b/erts/preloaded/src/erts_internal.erl @@ -38,6 +38,10 @@ -export([check_process_code/2]). +-export([flush_monitor_messages/3]). + +-export([time_unit/0]). + %% %% Await result of send to port %% @@ -178,3 +182,29 @@ cmp_term(_A,_B) -> map_to_tuple_keys(_M) -> erlang:nif_error(undefined). + +-spec erts_internal:flush_monitor_messages(Ref, Multi, Res) -> term() when + Ref :: reference(), + Multi :: boolean(), + Res :: term(). + +%% erlang:demonitor(Ref, [flush]) traps to +%% erts_internal:flush_monitor_messages(Ref, Res) when +%% it needs to flush monitor messages. +flush_monitor_messages(Ref, Multi, Res) when is_reference(Ref) -> + receive + {_, Ref, _, _, _} -> + case Multi of + false -> + Res; + _ -> + flush_monitor_messages(Ref, Multi, Res) + end + after 0 -> + Res + end. + +-spec erts_internal:time_unit() -> pos_integer(). + +time_unit() -> + erlang:nif_error(undefined). -- cgit v1.2.3 From 6b1921d767de5cd1a980234f83b36dbfa13d9fc7 Mon Sep 17 00:00:00 2001 From: Rickard Green Date: Fri, 13 Feb 2015 00:25:57 +0100 Subject: Erlang based BIF timer implementation for scalability --- erts/preloaded/ebin/erlang.beam | Bin 100456 -> 105832 bytes erts/preloaded/ebin/erts_internal.beam | Bin 4688 -> 12368 bytes erts/preloaded/ebin/init.beam | Bin 48800 -> 49744 bytes erts/preloaded/src/erlang.erl | 191 +++++++++++++++++++++++-- erts/preloaded/src/erts_internal.erl | 251 +++++++++++++++++++++++++++++++++ erts/preloaded/src/init.erl | 32 ++++- 6 files changed, 459 insertions(+), 15 deletions(-) (limited to 'erts/preloaded') diff --git a/erts/preloaded/ebin/erlang.beam b/erts/preloaded/ebin/erlang.beam index 4ec388a7b9..303f3f47b6 100644 Binary files a/erts/preloaded/ebin/erlang.beam and b/erts/preloaded/ebin/erlang.beam differ diff --git a/erts/preloaded/ebin/erts_internal.beam b/erts/preloaded/ebin/erts_internal.beam index 705bb61247..68fb357eb4 100644 Binary files a/erts/preloaded/ebin/erts_internal.beam and b/erts/preloaded/ebin/erts_internal.beam differ diff --git a/erts/preloaded/ebin/init.beam b/erts/preloaded/ebin/init.beam index aadc9797cb..7f79493b4d 100644 Binary files a/erts/preloaded/ebin/init.beam and b/erts/preloaded/ebin/init.beam differ diff --git a/erts/preloaded/src/erlang.erl b/erts/preloaded/src/erlang.erl index 79eb60f362..83010b17d2 100644 --- a/erts/preloaded/src/erlang.erl +++ b/erts/preloaded/src/erlang.erl @@ -89,7 +89,7 @@ -export([binary_to_list/3, binary_to_term/1, binary_to_term/2]). -export([bit_size/1, bitsize/1, bitstring_to_list/1]). -export([bump_reductions/1, byte_size/1, call_on_load_function/1]). --export([cancel_timer/1, check_old_code/1, check_process_code/2, +-export([cancel_timer/1, cancel_timer/2, check_old_code/1, check_process_code/2, check_process_code/3, crc32/1]). -export([crc32/2, crc32_combine/3, date/0, decode_packet/3]). -export([delete_element/2]). @@ -128,7 +128,7 @@ -export([time_offset/0, time_offset/1, timestamp/0]). -export([process_display/2]). -export([process_flag/3, process_info/1, processes/0, purge_module/1]). --export([put/2, raise/3, read_timer/1, ref_to_list/1, register/2]). +-export([put/2, raise/3, read_timer/1, read_timer/2, ref_to_list/1, register/2]). -export([registered/0, resume_process/1, round/1, self/0, send_after/3]). -export([seq_trace/2, seq_trace_print/1, seq_trace_print/2, setnode/2]). -export([setnode/3, size/1, spawn/3, spawn_link/3, split_binary/2]). @@ -427,8 +427,77 @@ call_on_load_function(_P1) -> -spec erlang:cancel_timer(TimerRef) -> Time | false when TimerRef :: reference(), Time :: non_neg_integer(). -cancel_timer(_TimerRef) -> - erlang:nif_error(undefined). +cancel_timer(TimerRef) -> + try + case erts_internal:access_bif_timer(TimerRef) of + undefined -> + false; + {BTR, TSrv} -> + Req = erlang:make_ref(), + TSrv ! {cancel_timeout, BTR, erlang:self(), + true, Req, TimerRef}, + receive + {cancel_timer, Req, Result} -> + Result + end + end + catch + _:_ -> erlang:error(badarg, [TimerRef]) + end. + +%% cancel_timer/2 +-spec erlang:cancel_timer(TimerRef, Options) -> Time | false | ok when + TimerRef :: reference(), + Option :: {async, boolean()} | {info, boolean()}, + Options :: [Option], + Time :: non_neg_integer(). +cancel_timer(TimerRef, Options) -> + try + {Async, Info} = get_cancel_timer_options(Options, false, true), + case erts_internal:access_bif_timer(TimerRef) of + undefined -> + case {Async, Info} of + {true, true} -> + erlang:self() ! {cancel_timer, TimerRef, false}, ok; + {false, true} -> + false; + _ -> + ok + end; + {BTR, TSrv} -> + case Async of + true -> + TSrv ! {cancel_timeout, BTR, erlang:self(), + Info, TimerRef, TimerRef}, + ok; + false -> + Req = erlang:make_ref(), + TSrv ! {cancel_timeout, BTR, erlang:self(), + true, Req, TimerRef}, + receive + {cancel_timer, Req, Result} -> + case Info of + true -> Result; + false -> ok + end + end + end + end + catch + _:_ -> erlang:error(badarg, [TimerRef, Options]) + end. + +get_cancel_timer_options([], Async, Info) -> + {Async, Info}; +get_cancel_timer_options([{async, Bool} | Opts], + _Async, Info) when Bool == true; + Bool == false -> + get_cancel_timer_options(Opts, Bool, Info); +get_cancel_timer_options([{info, Bool} | Opts], + Async, _Info) when Bool == true; + Bool == false -> + get_cancel_timer_options(Opts, Async, Bool). + %% check_old_code/1 -spec check_old_code(Module) -> boolean() when @@ -1462,8 +1531,53 @@ raise(_Class, _Reason, _Stacktrace) -> %% read_timer/1 -spec erlang:read_timer(TimerRef) -> non_neg_integer() | false when TimerRef :: reference(). -read_timer(_TimerRef) -> - erlang:nif_error(undefined). + +read_timer(TimerRef) -> + read_timer(TimerRef, []). + +%% read_timer/2 +-spec erlang:read_timer(TimerRef, Options) -> non_neg_integer() | false | ok when + TimerRef :: reference(), + Option :: {async, boolean()}, + Options :: [Option]. + +read_timer(TimerRef, Options) -> + try + Async = get_read_timer_options(Options, false), + case erts_internal:access_bif_timer(TimerRef) of + undefined -> + case Async of + true -> + erlang:self() ! {read_timer, TimerRef, false}, + ok; + false -> + false + end; + {BTR, TSrv} -> + case Async of + true -> + TSrv ! {read_timeout, BTR, erlang:self(), + TimerRef, TimerRef}, + ok; + false -> + Req = erlang:make_ref(), + TSrv ! {read_timeout, BTR, erlang:self(), + Req, TimerRef}, + receive + {read_timer, Req, Result} -> + Result + end + end + end + catch + _:_ -> erlang:error(badarg, [TimerRef]) + end. + +get_read_timer_options([], Async) -> + Async; +get_read_timer_options([{async, Bool} | Opts], + _Async) when Bool == true; Bool == false -> + get_read_timer_options(Opts, Bool). %% ref_to_list/1 -spec erlang:ref_to_list(Ref) -> string() when @@ -1509,8 +1623,36 @@ self() -> Dest :: pid() | atom(), Msg :: term(), TimerRef :: reference(). -send_after(_Time, _Dest, _Msg) -> - erlang:nif_error(undefined). + +send_after(0, Dest, Msg) -> + try + true = ((erlang:is_pid(Dest) + andalso erlang:node(Dest) == erlang:node()) + orelse (erlang:is_atom(Dest) + andalso Dest /= undefined)), + try Dest ! Msg catch _:_ -> ok end, + erlang:make_ref() + catch + _:_ -> + erlang:error(badarg, [0, Dest, Msg]) + end; +send_after(Time, Dest, Msg) -> + Now = erlang:monotonic_time(), + try + true = ((erlang:is_pid(Dest) + andalso erlang:node(Dest) == erlang:node()) + orelse (erlang:is_atom(Dest) + andalso Dest /= undefined)), + true = Time > 0, + true = Time < (1 bsl 32), % Maybe lift this restriction... + TO = Now + (erts_internal:time_unit()*Time) div 1000, + {BTR, TSrv, TRef} = erts_internal:create_bif_timer(), + TSrv ! {set_timeout, BTR, Dest, TO, TRef, Msg}, + TRef + catch + _:_ -> + erlang:error(badarg, [Time, Dest, Msg]) + end. %% seq_trace/2 -spec erlang:seq_trace(P1, P2) -> seq_trace_info_returns() | {term(), term(), term(), term(), term()} when @@ -1583,8 +1725,37 @@ split_binary(_Bin, _Pos) -> Dest :: pid() | atom(), Msg :: term(), TimerRef :: reference(). -start_timer(_Time, _Dest, _Msg) -> - erlang:nif_error(undefined). +start_timer(0, Dest, Msg) -> + try + true = ((erlang:is_pid(Dest) + andalso erlang:node(Dest) == erlang:node()) + orelse (erlang:is_atom(Dest) + andalso Dest /= undefined)), + TimerRef = erlang:make_ref(), + try Dest ! {timeout, TimerRef, Msg} catch _:_ -> ok end, + TimerRef + catch + _:_ -> + erlang:error(badarg, [0, Dest, Msg]) + end; +start_timer(Time, Dest, Msg) -> + Now = erlang:monotonic_time(), + try + true = ((erlang:is_pid(Dest) + andalso erlang:node(Dest) == erlang:node()) + orelse (erlang:is_atom(Dest) + andalso Dest /= undefined)), + true = Time > 0, + true = Time < (1 bsl 32), % Maybe lift this restriction... + TO = Now + (erts_internal:time_unit()*Time) div 1000, + {BTR, TSrv, TimerRef} = erts_internal:create_bif_timer(), + TSrv ! {set_timeout, BTR, Dest, TO, TimerRef, + {timeout, TimerRef, Msg}}, + TimerRef + catch + _:_ -> + erlang:error(badarg, [Time, Dest, Msg]) + end. %% suspend_process/2 -spec erlang:suspend_process(Suspendee, OptList) -> boolean() when diff --git a/erts/preloaded/src/erts_internal.erl b/erts/preloaded/src/erts_internal.erl index 30df75b406..2c701d75e4 100644 --- a/erts/preloaded/src/erts_internal.erl +++ b/erts/preloaded/src/erts_internal.erl @@ -42,6 +42,14 @@ -export([time_unit/0]). +-export([bif_timer_server/2]). + +-export([get_bif_timer_servers/0, create_bif_timer/0, access_bif_timer/1]). + +-export([monitor_process/2]). + +-export([is_system_process/1]). + %% %% Await result of send to port %% @@ -208,3 +216,246 @@ flush_monitor_messages(Ref, Multi, Res) when is_reference(Ref) -> time_unit() -> erlang:nif_error(undefined). + +-spec erts_internal:get_bif_timer_servers() -> Pids when + Pid :: pid(), + Pids :: [Pid]. + +get_bif_timer_servers() -> + erlang:nif_error(undefined). + +-spec erts_internal:create_bif_timer() -> Res when + Res :: {reference(), pid(), reference()}. + +create_bif_timer() -> + erlang:nif_error(undefined). + +-spec erts_internal:access_bif_timer(Ref) -> Res when + Ref :: reference(), + Res :: {reference(), pid(), reference()}. + +access_bif_timer(_Ref) -> + erlang:nif_error(undefined). + +-spec erts_internal:monitor_process(Pid, Ref) -> boolean() when + Pid :: pid(), + Ref :: reference(). + +monitor_process(_Pid, _Ref) -> + erlang:nif_error(undefined). + +-spec erts_internal:is_system_process(Pid) -> boolean() when + Pid :: pid(). + +is_system_process(_Pid) -> + erlang:nif_error(undefined). + +%% +%% BIF timer servers +%% + +-record(tsrv_state, {rtab, + ttab, + btr, + unit, + next}). + +bif_timer_server(N, BTR) -> + try + tsrv_loop(tsrv_init_static_state(N, BTR), infinity) + catch + Type:Reason -> + erlang:display({'BIF_timer_server', + {Type, Reason}, + erlang:get_stacktrace()}), + exit(Reason) + end. + +tsrv_init_static_state(N, BTR) -> + process_flag(trap_exit, true), + NList = integer_to_list(N), + RTabName = list_to_atom("BIF_timer_reference_table_" ++ NList), + TTabName = list_to_atom("BIF_timer_time_table_" ++ NList), + #tsrv_state{rtab = ets:new(RTabName, + [set, private, {keypos, 2}]), + ttab = ets:new(TTabName, + [ordered_set, private, {keypos, 1}]), + btr = BTR, + unit = erts_internal:time_unit(), + next = infinity}. + + +tsrv_loop(#tsrv_state{unit = Unit} = StaticState, Nxt) -> + CallTime = erlang:monotonic_time(), + %% 'infinity' is greater than all integers... + NewNxt = case CallTime >= Nxt of + true -> + tsrv_handle_timeout(CallTime, StaticState); + false -> + TMO = try + (1000*(Nxt - CallTime - 1)) div Unit + 1 + catch + error:badarith when Nxt == infinity -> infinity + end, + receive + Msg -> + tsrv_handle_msg(Msg, StaticState, Nxt) + after TMO -> + Nxt + end + end, + tsrv_loop(StaticState, NewNxt). + +tsrv_handle_msg({set_timeout, BTR, Proc, Time, TRef, Msg}, + #tsrv_state{rtab = RTab, + ttab = TTab, + btr = BTR}, + Nxt) when erlang:is_integer(Time) -> + RcvTime = erlang:monotonic_time(), + case Time =< RcvTime of + true -> + try Proc ! Msg catch _:_ -> ok end, + Nxt; + false -> + Ins = case erlang:is_atom(Proc) of + true -> + true; + false -> + try + erts_internal:monitor_process(Proc, TRef) + catch + _:_ -> false + end + end, + case Ins of + false -> + Nxt; + true -> + TKey = {Time, TRef}, + true = ets:insert(RTab, TKey), + true = ets:insert(TTab, {TKey, Proc, Msg}), + case Time < Nxt of + true -> Time; + false -> Nxt + end + end + end; +tsrv_handle_msg({cancel_timeout, BTR, From, Reply, Req, TRef}, + #tsrv_state{rtab = RTab, + ttab = TTab, + unit = Unit, + btr = BTR}, + Nxt) -> + case ets:lookup(RTab, TRef) of + [] -> + case Reply of + false -> + ok; + _ -> + try From ! {cancel_timer, Req, false} catch _:_ -> ok end + end, + Nxt; + [{Time, TRef} = TKey] -> + ets:delete(RTab, TRef), + ets:delete(TTab, TKey), + erlang:demonitor(TRef), + case Reply of + false -> + ok; + _ -> + RcvTime = erlang:monotonic_time(), + RT = case Time =< RcvTime of + true -> + 0; + false -> + ((1000*(Time - RcvTime)) div Unit) + end, + try From ! {cancel_timer, Req, RT} catch _:_ -> ok end + end, + case Time =:= Nxt of + false -> + Nxt; + true -> + case ets:first(TTab) of + '$end_of_table' -> infinity; + {NextTime, _TRef} -> NextTime + end + end + end; +tsrv_handle_msg({read_timeout, BTR, From, Req, TRef}, + #tsrv_state{rtab = RTab, + unit = Unit, + btr = BTR}, + Nxt) -> + case ets:lookup(RTab, TRef) of + [] -> + try From ! {read_timer, Req, false} catch _:_ -> ok end; + [{Time, TRef}] -> + RcvTime = erlang:monotonic_time(), + RT = case Time =< RcvTime of + true -> 0; + false -> (1000*(Time - RcvTime)) div Unit + end, + try From ! {read_timer, Req, RT} catch _:_ -> ok end + end, + Nxt; +tsrv_handle_msg({'DOWN', TRef, process, _, _}, + #tsrv_state{rtab = RTab, + ttab = TTab}, + Nxt) -> + case ets:lookup(RTab, TRef) of + [] -> + Nxt; + [{Time, TRef} = TKey] -> + ets:delete(RTab, TRef), + ets:delete(TTab, TKey), + case Time =:= Nxt of + false -> + Nxt; + true -> + case ets:first(TTab) of + '$end_of_table' -> infinity; + {NextTime, _} -> NextTime + end + end + end; +tsrv_handle_msg({cancel_all_timeouts, BTR, From, Ref}, + #tsrv_state{rtab = RTab, + ttab = TTab, + btr = BTR}, + _Nxt) -> + tsrv_delete_monitor_objects(RTab), + ets:delete_all_objects(TTab), + try From ! {canceled_all_timeouts, Ref} catch _:_ -> ok end, + infinity; +tsrv_handle_msg(_GarbageMsg, _StaticState, Nxt) -> + Nxt. + +tsrv_delete_monitor_objects(RTab) -> + case ets:first(RTab) of + '$end_of_table' -> + ok; + TRef -> + erlang:demonitor(TRef), + ets:delete(RTab, TRef), + tsrv_delete_monitor_objects(RTab) + end. + +tsrv_handle_timeout(CallTime, #tsrv_state{rtab = RTab, + ttab = TTab} = S) -> + case ets:first(TTab) of + '$end_of_table' -> + infinity; + {Time, _TRef} when Time > CallTime -> + Time; + {_Time, TRef} = TKey -> + [{TKey, Proc, Msg}] = ets:lookup(TTab, TKey), + case erlang:is_pid(Proc) of + false -> ok; + _ -> erlang:demonitor(TRef) + end, + ets:delete(TTab, TKey), + ets:delete(RTab, TRef), + try Proc ! Msg catch _:_ -> ok end, + tsrv_handle_timeout(CallTime, S) + end. diff --git a/erts/preloaded/src/init.erl b/erts/preloaded/src/init.erl index e95e11b3e6..48c5c37717 100644 --- a/erts/preloaded/src/init.erl +++ b/erts/preloaded/src/init.erl @@ -522,6 +522,7 @@ shutdown_pids(Heart,BootPid,State) -> Timer = shutdown_timer(State#state.flags), catch shutdown(State#state.kernel,BootPid,Timer,State), kill_all_pids(Heart), % Even the shutdown timer. + cancel_all_bif_timeouts(), kill_all_ports(Heart), flush_timout(Timer). @@ -580,6 +581,30 @@ resend([ExitMsg|Exits]) -> resend(_) -> ok. + +cancel_all_bif_timeouts() -> + TSrvs = erts_internal:get_bif_timer_servers(), + Ref = make_ref(), + {BTR, _TSrv} = erts_internal:access_bif_timer(Ref), %% Cheat... + request_cancel_all_bif_timeouts(Ref, BTR, TSrvs), + wait_response_cancel_all_bif_timeouts(Ref, BTR, TSrvs), + ok. + +request_cancel_all_bif_timeouts(_Ref, _BTR, []) -> + ok; +request_cancel_all_bif_timeouts(Ref, BTR, [TSrv|TSrvs]) -> + TSrv ! {cancel_all_timeouts, BTR, self(), {Ref, TSrv}}, + request_cancel_all_bif_timeouts(Ref, BTR, TSrvs). + +wait_response_cancel_all_bif_timeouts(_Ref, _BTR, []) -> + ok; +wait_response_cancel_all_bif_timeouts(Ref, BTR, [TSrv|TSrvs]) -> + receive + {canceled_all_timeouts, {Ref, TSrv}} -> + wait_response_cancel_all_bif_timeouts(Ref, BTR, TSrvs) + end. + + %% %% Kill all existing pids in the system (except init and heart). kill_all_pids(Heart) -> @@ -591,12 +616,9 @@ kill_all_pids(Heart) -> kill_all_pids(Heart) % Continue until all are really killed. end. -%% All except zombies. -alive_processes() -> - [P || P <- processes(), erlang:is_process_alive(P)]. - +%% All except system processes. get_pids(Heart) -> - Pids = alive_processes(), + Pids = [P || P <- processes(), not erts_internal:is_system_process(P)], delete(Heart,self(),Pids). delete(Heart,Init,[Heart|Pids]) -> delete(Heart,Init,Pids); -- cgit v1.2.3