aboutsummaryrefslogtreecommitdiffstats
path: root/erts/preloaded/src
diff options
context:
space:
mode:
authorRickard Green <[email protected]>2015-02-13 00:25:57 +0100
committerRickard Green <[email protected]>2015-03-20 15:28:52 +0100
commit6b1921d767de5cd1a980234f83b36dbfa13d9fc7 (patch)
tree0b7a016c2b41f30f1adf81559e5fc6a4567099ea /erts/preloaded/src
parentfa7b2c00cbf9212c4a3551980939b92fc6606510 (diff)
downloadotp-6b1921d767de5cd1a980234f83b36dbfa13d9fc7.tar.gz
otp-6b1921d767de5cd1a980234f83b36dbfa13d9fc7.tar.bz2
otp-6b1921d767de5cd1a980234f83b36dbfa13d9fc7.zip
Erlang based BIF timer implementation for scalability
Diffstat (limited to 'erts/preloaded/src')
-rw-r--r--erts/preloaded/src/erlang.erl191
-rw-r--r--erts/preloaded/src/erts_internal.erl251
-rw-r--r--erts/preloaded/src/init.erl32
3 files changed, 459 insertions, 15 deletions
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);