Current section
Files
Jump to
Current section
Files
src/concuerror_scheduler.erl
%%% @private
%%% @doc Concuerror's scheduler component
%%%
%%% concuerror_scheduler is the main driver of interleaving
%%% exploration. A rough trace through it is the following:
%%% The entry point is `concuerror_scheduler:run/1` which takes the
%%% options and initializes the exploration, spawning the main
%%% process. There are plenty of state info that are kept in the
%%% `#scheduler_state` record the most important of which being a list
%%% of `#trace_state` records, recording events in the exploration.
%%% This list corresponds more or less to "E" in the various DPOR
%%% papers (representing the execution trace).
%%% The logic of the exploration goes through `explore_scheduling/1`,
%%% which in turn calls 'explore/1'. Both functions are fairly clean:
%%% as long as there are more processes that can be executed and yield
%%% events, `get_next_event/1` will be returning `ok`, after executing
%%% one of them and doing all necessary updates to the state (adding
%%% new `#trace_state`s, etc). If `get_next_event/1` returns `none`,
%%% we are at the end of an interleaving (either due to no more
%%% enabled processes or due to "sleep set blocking") and can do race
%%% analysis and report any errors found in the interleaving. Race
%%% analysis is contained in `plan_more_interleavings/1`, reporting
%%% whether the current interleaving was buggy is contained in
%%% `log_trace/1` and resetting most parts to continue exploration is
%%% contained in `has_more_to_explore/1`.
%%% Focusing on `plan_more_interleavings`, it is composed out of two
%%% steps: first (`assign_happens_before/3`) we assign a
%%% happens-before relation to all events to be able to detect when
%%% races are reversible or not (if two events are dependent not only
%%% directly but also via a chain of dependent events then the race is
%%% not reversible) and then (`plan_more_interleavings/3`) for each
%%% event (`more_interleavings_for_event/6`) we do an actual race
%%% analysis, adding initials or wakeup sequences in appropriate
%%% places in the list of `#trace_state`s.
-module(concuerror_scheduler).
%% User interface
-export([run/1, explain_error/1]).
-export_type([ interleaving_error/0,
interleaving_error_tag/0,
interleaving_result/0,
unique_id/0
]).
%% =============================================================================
%% DATA STRUCTURES & TYPES
%% =============================================================================
-include("concuerror.hrl").
%%------------------------------------------------------------------------------
-type interleaving_id() :: pos_integer().
-ifdef(BEFORE_OTP_17).
-type clock_map() :: dict().
-type message_event_queue() :: queue().
-else.
-type vector_clock() :: #{actor() => index()}.
-type clock_map() :: #{actor() => vector_clock()}.
-type message_event_queue() :: queue:queue(#message_event{}).
-endif.
-record(backtrack_entry, {
conservative = false :: boolean(),
event :: event(),
origin = 1 :: interleaving_id(),
wakeup_tree = [] :: event_tree()
}).
-type event_tree() :: [#backtrack_entry{}].
-type channel_actor() :: {channel(), message_event_queue()}.
-type unique_id() :: {interleaving_id(), index()}.
-record(trace_state, {
actors :: [pid() | channel_actor()],
clock_map = empty_map():: clock_map(),
done = [] :: [event()],
enabled = [] :: [pid() | channel_actor()],
index :: index(),
unique_id :: unique_id(),
previous_actor = 'none' :: 'none' | actor(),
scheduling_bound :: concuerror_options:bound(),
sleep_set = [] :: [event()],
wakeup_tree = [] :: event_tree()
}).
-type trace_state() :: #trace_state{}.
-type interleaving_result() ::
'ok' | 'sleep_set_block' | {[interleaving_error()], [event()]}.
-type interleaving_error_tag() ::
'abnormal_exit' |
'abnormal_halt' |
'deadlock' |
'depth_bound'.
-type interleaving_error() ::
{'abnormal_exit', {index(), pid(), term(), [term()]}} |
{'abnormal_halt', {index(), pid(), term()}} |
{'deadlock', [pid()]} |
{'depth_bound', concuerror_options:bound()} |
'fatal'.
-type scope() :: 'all' | [pid()].
%% DO NOT ADD A DEFAULT VALUE IF IT WILL ALWAYS BE OVERWRITTEN.
%% Default values for fields should be specified in ONLY ONE PLACE.
%% For e.g., user options this is normally in the _options module.
-record(scheduler_state, {
assertions_only :: boolean(),
assume_racing :: assume_racing_opt(),
depth_bound :: pos_integer(),
dpor :: concuerror_options:dpor(),
entry_point :: mfargs(),
estimator :: concuerror_estimator:estimator(),
first_process :: pid(),
ignore_error :: [{interleaving_error_tag(), scope()}],
interleaving_bound :: concuerror_options:bound(),
interleaving_errors :: [interleaving_error()],
interleaving_id :: interleaving_id(),
keep_going :: boolean(),
logger :: pid(),
last_scheduled :: pid(),
need_to_replay :: boolean(),
non_racing_system :: [atom()],
origin :: interleaving_id(),
print_depth :: pos_integer(),
processes :: processes(),
receive_timeout_total :: non_neg_integer(),
report_error :: [{interleaving_error_tag(), scope()}],
scheduling :: concuerror_options:scheduling(),
scheduling_bound_type :: concuerror_options:scheduling_bound_type(),
show_races :: boolean(),
strict_scheduling :: boolean(),
timeout :: timeout(),
trace :: [trace_state()],
treat_as_normal :: [atom()],
use_receive_patterns :: boolean(),
use_sleep_sets :: boolean(),
use_unsound_bpor :: boolean()
}).
%% =============================================================================
%% LOGIC (high level description of the exploration algorithm)
%% =============================================================================
-spec run(concuerror_options:options()) -> ok.
run(Options) ->
process_flag(trap_exit, true),
put(bound_exceeded, false),
FirstProcess = concuerror_callback:spawn_first_process(Options),
EntryPoint = ?opt(entry_point, Options),
Timeout = ?opt(timeout, Options),
ok =
concuerror_callback:start_first_process(FirstProcess, EntryPoint, Timeout),
SchedulingBound = ?opt(scheduling_bound, Options, infinity),
InitialTrace =
#trace_state{
actors = [FirstProcess],
enabled = [E || E <- [FirstProcess], enabled(E)],
index = 1,
scheduling_bound = SchedulingBound,
unique_id = {1, 1}
},
Logger = ?opt(logger, Options),
{SchedulingBoundType, UnsoundBPOR} =
case ?opt(scheduling_bound_type, Options) of
ubpor -> {bpor, true};
Else -> {Else, false}
end,
{IgnoreError, ReportError} =
generate_filtering_rules(Options, FirstProcess),
InitialState =
#scheduler_state{
assertions_only = ?opt(assertions_only, Options),
assume_racing = {?opt(assume_racing, Options), Logger},
depth_bound = ?opt(depth_bound, Options) + 1,
dpor = ?opt(dpor, Options),
entry_point = EntryPoint,
estimator = ?opt(estimator, Options),
first_process = FirstProcess,
ignore_error = IgnoreError,
interleaving_bound = ?opt(interleaving_bound, Options),
interleaving_errors = [],
interleaving_id = 1,
keep_going = ?opt(keep_going, Options),
last_scheduled = FirstProcess,
logger = Logger,
need_to_replay = false,
non_racing_system = ?opt(non_racing_system, Options),
origin = 1,
print_depth = ?opt(print_depth, Options),
processes = Processes = ?opt(processes, Options),
receive_timeout_total = 0,
report_error = ReportError,
scheduling = ?opt(scheduling, Options),
scheduling_bound_type = SchedulingBoundType,
show_races = ?opt(show_races, Options),
strict_scheduling = ?opt(strict_scheduling, Options),
trace = [InitialTrace],
treat_as_normal = ?opt(treat_as_normal, Options),
timeout = Timeout,
use_receive_patterns = ?opt(use_receive_patterns, Options),
use_sleep_sets = not ?opt(disable_sleep_sets, Options),
use_unsound_bpor = UnsoundBPOR
},
case SchedulingBound =:= infinity of
true ->
?unique(Logger, ?ltip, msg(scheduling_bound_tip), []);
false ->
ok
end,
concuerror_logger:plan(Logger),
?time(Logger, "Exploration start"),
Ret = explore_scheduling(InitialState),
concuerror_callback:cleanup_processes(Processes),
Ret.
%%------------------------------------------------------------------------------
explore_scheduling(State) ->
UpdatedState = explore(State),
LogState = log_trace(UpdatedState),
RacesDetectedState = plan_more_interleavings(LogState),
{HasMore, NewState} = has_more_to_explore(RacesDetectedState),
case HasMore of
true -> explore_scheduling(NewState);
false -> ok
end.
explore(State) ->
{Status, UpdatedState} =
try
get_next_event(State)
catch
C:R ->
S = [],
{{crash, C, R, S}, State}
end,
case Status of
ok -> explore(UpdatedState);
none -> UpdatedState;
{crash, Class, Reason, Stack} ->
FatalCrashState =
add_error(fatal, discard_last_trace_state(UpdatedState)),
catch log_trace(FatalCrashState),
erlang:raise(Class, Reason, Stack)
end.
%%------------------------------------------------------------------------------
log_trace(#scheduler_state{logger = Logger} = State) ->
Log =
case filter_errors(State) of
[] -> none;
Errors ->
case proplists:get_value(sleep_set_block, Errors) of
{Origin, Sleep} ->
case State#scheduler_state.dpor =:= optimal of
true -> ?crash({optimal_sleep_set_block, Origin, Sleep});
false -> ok
end,
sleep_set_block;
undefined ->
#scheduler_state{trace = Trace} = State,
Fold =
fun(#trace_state{done = [A|_], index = I}, Acc) ->
[{I, A}|Acc]
end,
TraceInfo = lists:foldl(Fold, [], Trace),
{lists:reverse(Errors), TraceInfo}
end
end,
concuerror_logger:complete(Logger, Log),
case Log =/= none andalso Log =/= sleep_set_block of
true when not State#scheduler_state.keep_going ->
?unique(Logger, ?lerror, msg(stop_first_error), []),
State#scheduler_state{trace = []};
Other ->
case Other of
true ->
?unique(Logger, ?linfo, "Continuing after error (-k)~n", []);
false ->
ok
end,
InterleavingId = State#scheduler_state.interleaving_id,
NextInterleavingId = InterleavingId + 1,
NextState =
State#scheduler_state{
interleaving_errors = [],
interleaving_id = NextInterleavingId,
receive_timeout_total = 0
},
case NextInterleavingId =< State#scheduler_state.interleaving_bound of
true -> NextState;
false ->
UniqueMsg = "Reached interleaving bound (~p)~n",
?unique(Logger, ?lwarning, UniqueMsg, [InterleavingId]),
NextState#scheduler_state{trace = []}
end
end.
%%------------------------------------------------------------------------------
generate_filtering_rules(Options, FirstProcess) ->
IgnoreErrors = ?opt(ignore_error, Options),
OnlyFirstProcessErrors = ?opt(first_process_errors_only, Options),
case OnlyFirstProcessErrors of
false -> {[{IE, all} || IE <- IgnoreErrors], []};
true ->
AllCategories = [abnormal_exit, abnormal_halt, deadlock],
Ignored = [{IE, all} || IE <- AllCategories],
Reported = [{IE, [FirstProcess]} || IE <- AllCategories -- IgnoreErrors],
{Ignored, Reported}
end.
filter_errors(State) ->
#scheduler_state{
ignore_error = Ignored,
interleaving_errors = UnfilteredErrors,
logger = Logger,
report_error = Reported
} = State,
TaggedErrors = [{true, E} || E <- UnfilteredErrors],
IgnoredErrors = update_all_tags(TaggedErrors, Ignored, false),
ReportedErrors = update_all_tags(IgnoredErrors, Reported, true),
FinalErrors = [E || {true, E} <- ReportedErrors],
case FinalErrors =/= UnfilteredErrors of
true ->
UniqueMsg = "Some errors were ignored ('--ignore_error').~n",
?unique(Logger, ?lwarning, UniqueMsg, []);
false -> ok
end,
FinalErrors.
update_all_tags([], _, _) -> [];
update_all_tags(TaggedErrors, [], _) -> TaggedErrors;
update_all_tags(TaggedErrors, Rules, Value) ->
[update_tag(E, Rules, Value) || E <- TaggedErrors].
update_tag({OldTag, Error}, Rules, NewTag) ->
RuleAppliesPred = fun(Rule) -> rule_applies(Rule, Error) end,
case lists:any(RuleAppliesPred, Rules) of
true -> {NewTag, Error};
false -> {OldTag, Error}
end.
rule_applies({Tag, Scope}, {Tag, _} = Error) ->
scope_applies(Scope, Error);
rule_applies(_, _) -> false.
scope_applies(all, _) -> true;
scope_applies(Pids, ErrorInfo) ->
case ErrorInfo of
{deadlock, Deadlocked} ->
DPids = [element(1, D) || D <- Deadlocked],
DPids -- Pids =/= DPids;
{abnormal_exit, {_, Pid, _, _}} -> lists:member(Pid, Pids);
{abnormal_halt, {_, Pid, _}} -> lists:member(Pid, Pids);
_ -> false
end.
discard_last_trace_state(State) ->
#scheduler_state{trace = [_|Trace]} = State,
State#scheduler_state{trace = Trace}.
add_error(Error, State) ->
add_errors([Error], State).
add_errors(Errors, State) ->
#scheduler_state{interleaving_errors = OldErrors} = State,
State#scheduler_state{interleaving_errors = Errors ++ OldErrors}.
%%------------------------------------------------------------------------------
get_next_event(
#scheduler_state{
depth_bound = Bound,
logger = Logger,
trace = [#trace_state{index = Bound}|_]} = State) ->
?unique(Logger, ?lwarning, msg(depth_bound_reached), []),
NewState =
add_error({depth_bound, Bound - 1}, discard_last_trace_state(State)),
{none, NewState};
get_next_event(#scheduler_state{logger = _Logger, trace = [Last|_]} = State) ->
#trace_state{index = _I, wakeup_tree = WakeupTree} = Last,
case WakeupTree of
[] ->
Event = #event{label = make_ref()},
get_next_event(Event, State);
[#backtrack_entry{event = Event, origin = N}|_] ->
?debug(
_Logger,
"New interleaving detected in ~p (diverge @ ~p)~n",
[N, _I]),
get_next_event(Event, State#scheduler_state{origin = N})
end.
get_next_event(Event, MaybeNeedsReplayState) ->
State = replay(MaybeNeedsReplayState),
#scheduler_state{trace = [Last|_]} = State,
#trace_state{actors = Actors, sleep_set = SleepSet} = Last,
SortedActors = schedule_sort(Actors, State),
#event{actor = Actor, label = Label} = Event,
case Actor =:= undefined of
true ->
AvailableActors = filter_sleep_set(SleepSet, SortedActors),
free_schedule(Event, AvailableActors, State);
false ->
#scheduler_state{print_depth = PrintDepth} = State,
#trace_state{index = I} = Last,
false = lists:member(Actor, SleepSet),
OkUpdatedEvent =
case Label =/= undefined of
true ->
NewEvent = get_next_event_backend(Event, State),
try {ok, Event} = NewEvent
catch
_:_ ->
New =
case NewEvent of
{ok, E} -> E;
_ -> NewEvent
end,
Reason = {replay_mismatch, I, Event, New, PrintDepth},
?crash(Reason)
end;
false ->
%% Last event = Previously racing event = Result may differ.
ResetEvent = reset_event(Event),
get_next_event_backend(ResetEvent, State)
end,
case OkUpdatedEvent of
{ok, UpdatedEvent} ->
update_state(UpdatedEvent, State);
retry ->
BReason = {blocked_mismatch, I, Event, PrintDepth},
?crash(BReason)
end
end.
filter_sleep_set([], AvailableActors) -> AvailableActors;
filter_sleep_set([#event{actor = Actor}|SleepSet], AvailableActors) ->
NewAvailableActors =
case ?is_channel(Actor) of
true -> lists:keydelete(Actor, 1, AvailableActors);
false -> lists:delete(Actor, AvailableActors)
end,
filter_sleep_set(SleepSet, NewAvailableActors).
schedule_sort([], _State) -> [];
schedule_sort(Actors, State) ->
#scheduler_state{
last_scheduled = LastScheduled,
scheduling = Scheduling,
strict_scheduling = StrictScheduling
} = State,
Sorted =
case Scheduling of
oldest -> Actors;
newest -> lists:reverse(Actors);
round_robin ->
Split = fun(E) -> E =/= LastScheduled end,
{Pre, Post} = lists:splitwith(Split, Actors),
Post ++ Pre
end,
case StrictScheduling of
true when Scheduling =:= round_robin ->
[LastScheduled|Rest] = Sorted,
Rest ++ [LastScheduled];
false when Scheduling =/= round_robin ->
[LastScheduled|lists:delete(LastScheduled, Sorted)];
_ -> Sorted
end.
free_schedule(Event, Actors, State) ->
#scheduler_state{
dpor = DPOR,
estimator = Estimator,
logger = Logger,
scheduling_bound_type = SchedulingBoundType,
trace = [Last|Prev]
} = State,
case DPOR =/= none of
true -> free_schedule_1(Event, Actors, State);
false ->
Enabled = [A || A <- Actors, enabled(A)],
ToBeExplored =
case SchedulingBoundType =:= delay of
false -> Enabled;
true ->
#trace_state{scheduling_bound = SchedulingBound} = Last,
?debug(Logger, "Select ~p of ~p~n", [SchedulingBound, Enabled]),
lists:sublist(Enabled, SchedulingBound + 1)
end,
case ToBeExplored < Enabled of
true -> bound_reached(Logger);
false -> ok
end,
Eventify = [maybe_prepare_channel_event(E, #event{}) || E <- ToBeExplored],
FullBacktrack = [#backtrack_entry{event = Ev} || Ev <- Eventify],
case FullBacktrack of
[] -> ok;
[_|L] ->
_ = [concuerror_logger:plan(Logger) || _ <- L],
Index = Last#trace_state.index,
_ = [concuerror_estimator:plan(Estimator, Index) || _ <- L],
ok
end,
NewLast = Last#trace_state{wakeup_tree = FullBacktrack},
NewTrace = [NewLast|Prev],
NewState = State#scheduler_state{trace = NewTrace},
free_schedule_1(Event, Actors, NewState)
end.
enabled({_,_}) -> true;
enabled(P) -> concuerror_callback:enabled(P).
free_schedule_1(Event, [Actor|_], State) when ?is_channel(Actor) ->
%% Pending messages can always be sent
PrepEvent = maybe_prepare_channel_event(Actor, Event),
{ok, FinalEvent} = get_next_event_backend(PrepEvent, State),
update_state(FinalEvent, State);
free_schedule_1(Event, [P|ActiveProcesses], State) ->
case get_next_event_backend(Event#event{actor = P}, State) of
retry -> free_schedule_1(Event, ActiveProcesses, State);
{ok, UpdatedEvent} -> update_state(UpdatedEvent, State)
end;
free_schedule_1(_Event, [], State) ->
%% Nothing to do, trace is completely explored
#scheduler_state{logger = _Logger, trace = [Last|_]} = State,
#trace_state{actors = Actors, sleep_set = SleepSet} = Last,
NewErrors =
case SleepSet =/= [] of
true ->
?debug(_Logger, "Sleep set block:~n ~p~n", [SleepSet]),
[{sleep_set_block, {State#scheduler_state.origin, SleepSet}}];
false ->
case concuerror_callback:collect_deadlock_info(Actors) of
[] -> [];
Info ->
?debug(_Logger, "Deadlock: ~p~n", [[element(1, I) || I <- Info]]),
[{deadlock, Info}]
end
end,
{none, add_errors(NewErrors, discard_last_trace_state(State))}.
maybe_prepare_channel_event(Actor, Event) ->
case ?is_channel(Actor) of
false -> Event#event{actor = Actor};
true ->
{Channel, Queue} = Actor,
MessageEvent = queue:get(Queue),
Event#event{actor = Channel, event_info = MessageEvent}
end.
reset_event(#event{actor = Actor, event_info = EventInfo}) ->
ResetEventInfo =
case ?is_channel(Actor) of
true -> EventInfo;
false -> undefined
end,
#event{
actor = Actor,
event_info = ResetEventInfo,
label = make_ref()
}.
%%------------------------------------------------------------------------------
update_state(#event{actor = Actor} = Event, State) ->
#scheduler_state{
estimator = Estimator,
logger = Logger,
scheduling_bound_type = SchedulingBoundType,
trace = [Last|Prev],
use_sleep_sets = UseSleepSets
} = State,
#trace_state{
actors = Actors,
done = RawDone,
index = Index,
previous_actor = PreviousActor,
scheduling_bound = SchedulingBound,
sleep_set = SleepSet,
unique_id = {InterleavingId, Index} = UID,
wakeup_tree = WakeupTree
} = Last,
?debug(Logger, "~s~n", [?pretty_s(Index, Event)]),
concuerror_logger:graph_new_node(Logger, UID, Index, Event),
Done = reset_receive_done(RawDone, State),
NextSleepSet =
case UseSleepSets of
true ->
AllSleepSet =
case WakeupTree of
[#backtrack_entry{conservative = true}|_] ->
concuerror_logger:plan(Logger),
concuerror_estimator:plan(Estimator, Index),
SleepSet;
_ -> ordsets:union(ordsets:from_list(Done), SleepSet)
end,
update_sleep_set(Event, AllSleepSet, State);
false -> []
end,
{NewLastWakeupTree, NextWakeupTree} =
case WakeupTree of
[] -> {[], []};
[#backtrack_entry{wakeup_tree = NWT}|Rest] -> {Rest, NWT}
end,
NewSchedulingBound =
next_bound(SchedulingBoundType, Done, PreviousActor, SchedulingBound),
?trace(Logger, " Next bound: ~p~n", [NewSchedulingBound]),
NewLastDone = [Event|Done],
NextIndex = Index + 1,
InitNextTrace =
#trace_state{
actors = Actors,
index = NextIndex,
previous_actor = Actor,
scheduling_bound = NewSchedulingBound,
sleep_set = NextSleepSet,
unique_id = {InterleavingId, NextIndex},
wakeup_tree = NextWakeupTree
},
NewLastTrace =
Last#trace_state{
done = NewLastDone,
wakeup_tree = NewLastWakeupTree
},
UpdatedSpecialNextTrace =
update_special(Event#event.special, InitNextTrace),
NextTrace =
maybe_update_enabled(SchedulingBoundType, UpdatedSpecialNextTrace),
InitNewState =
State#scheduler_state{trace = [NextTrace, NewLastTrace|Prev]},
NewState = maybe_log(Event, InitNewState, Index),
{ok, NewState}.
maybe_log(#event{actor = P} = Event, State0, Index) ->
#scheduler_state{
assertions_only = AssertionsOnly,
logger = Logger,
receive_timeout_total = ReceiveTimeoutTotal,
treat_as_normal = Normal
} = State0,
State =
case is_pid(P) of
true -> State0#scheduler_state{last_scheduled = P};
false -> State0
end,
case Event#event.event_info of
#builtin_event{mfargs = {erlang, halt, [Status|_]}}
when Status =/= 0 ->
#event{actor = Actor} = Event,
add_error({abnormal_halt, {Index, Actor, Status}}, State);
#exit_event{reason = Reason} = Exit when Reason =/= normal ->
{Tag, WasTimeout} =
case is_tuple(Reason) andalso (tuple_size(Reason) > 0) of
true ->
T = element(1, Reason),
{T, T =:= timeout};
false -> {Reason, false}
end,
case is_atom(Tag) andalso lists:member(Tag, Normal) of
true ->
?unique(Logger, ?lwarning, msg(treat_as_normal), []),
State;
false ->
case {WasTimeout, Tag} of
{true, _} -> ?unique(Logger, ?ltip, msg(timeout), []);
{_, shutdown} -> ?unique(Logger, ?ltip, msg(shutdown), []);
_ -> ok
end,
IsAssertLike =
case Tag of
{MaybeAssert, _} when is_atom(MaybeAssert) ->
case atom_to_list(MaybeAssert) of
"assert" ++ _ -> true;
_ -> false
end;
_ -> false
end,
Report =
case {IsAssertLike, AssertionsOnly} of
{false, true} ->
?unique(Logger, ?lwarning, msg(assertions_only_filter), []),
false;
{true, false} ->
?unique(Logger, ?ltip, msg(assertions_only_use), []),
true;
_ -> true
end,
case Report of
true ->
#event{actor = Actor} = Event,
Stacktrace = Exit#exit_event.stacktrace,
add_error({abnormal_exit, {Index, Actor, Reason, Stacktrace}}, State);
false -> State
end
end;
#receive_event{message = 'after'} ->
NewReceiveTimeoutTotal = ReceiveTimeoutTotal + 1,
Threshold = 50,
case NewReceiveTimeoutTotal =:= Threshold of
true ->
?unique(Logger, ?ltip, msg(maybe_receive_loop), [Threshold]);
false -> ok
end,
State#scheduler_state{receive_timeout_total = NewReceiveTimeoutTotal};
_ -> State
end.
update_sleep_set(NewEvent, SleepSet, State) ->
#scheduler_state{logger = _Logger} = State,
Pred =
fun(OldEvent) ->
V = concuerror_dependencies:dependent_safe(NewEvent, OldEvent),
?debug(_Logger, " Awaking (~p): ~s~n", [V,?pretty_s(OldEvent)]),
V =:= false
end,
lists:filter(Pred, SleepSet).
update_special(List, TraceState) when is_list(List) ->
lists:foldl(fun update_special/2, TraceState, List);
update_special(Special, #trace_state{actors = Actors} = TraceState) ->
NewActors =
case Special of
halt -> [];
{message, Message} ->
add_message(Message, Actors);
{message_delivered, MessageEvent} ->
remove_message(MessageEvent, Actors);
{message_received, _Message} ->
Actors;
{new, SpawnedPid} ->
Actors ++ [SpawnedPid];
{system_communication, _} ->
Actors
end,
TraceState#trace_state{actors = NewActors}.
add_message(MessageEvent, Actors) ->
#message_event{recipient = Recipient, sender = Sender} = MessageEvent,
Channel = {Sender, Recipient},
Update = fun(Queue) -> queue:in(MessageEvent, Queue) end,
Initial = queue:from_list([MessageEvent]),
insert_message(Channel, Update, Initial, Actors).
insert_message(Channel, Update, Initial, Actors) ->
insert_message(Channel, Update, Initial, Actors, false, []).
insert_message(Channel, _Update, Initial, [], Found, Acc) ->
case Found of
true -> lists:reverse(Acc, [{Channel, Initial}]);
false -> [{Channel, Initial}|lists:reverse(Acc)]
end;
insert_message(Channel, Update, _Initial, [{Channel, Queue}|Rest], true, Acc) ->
NewQueue = Update(Queue),
lists:reverse(Acc, [{Channel, NewQueue}|Rest]);
insert_message({From, _} = Channel, Update, Initial, [Other|Rest], Found, Acc) ->
case Other of
{{_,_},_} ->
insert_message(Channel, Update, Initial, Rest, Found, [Other|Acc]);
From ->
insert_message(Channel, Update, Initial, Rest, true, [Other|Acc]);
_ ->
case Found of
false ->
insert_message(Channel, Update, Initial, Rest, Found, [Other|Acc]);
true ->
lists:reverse(Acc, [{Channel, Initial},Other|Rest])
end
end.
remove_message(#message_event{recipient = Recipient, sender = Sender}, Actors) ->
Channel = {Sender, Recipient},
remove_message(Channel, Actors, []).
remove_message(Channel, [{Channel, Queue}|Rest], Acc) ->
NewQueue = queue:drop(Queue),
case queue:is_empty(NewQueue) of
true -> lists:reverse(Acc, Rest);
false -> lists:reverse(Acc, [{Channel, NewQueue}|Rest])
end;
remove_message(Channel, [Other|Rest], Acc) ->
remove_message(Channel, Rest, [Other|Acc]).
maybe_update_enabled(bpor, TraceState) ->
#trace_state{actors = Actors} = TraceState,
Enabled = [E || E <- Actors, enabled(E)],
TraceState#trace_state{enabled = Enabled};
maybe_update_enabled(_, TraceState) ->
TraceState.
%%------------------------------------------------------------------------------
plan_more_interleavings(#scheduler_state{dpor = none} = State) ->
#scheduler_state{logger = _Logger} = State,
?debug(_Logger, "Skipping race detection~n", []),
State;
plan_more_interleavings(State) ->
#scheduler_state{
dpor = DPOR,
logger = Logger,
trace = RevTrace,
use_receive_patterns = UseReceivePatterns
} = State,
?time(Logger, "Assigning happens-before..."),
{RE, UntimedLate} = split_trace(RevTrace),
{RevEarly, Late} =
case UseReceivePatterns of
false ->
{RE, lists:reverse(assign_happens_before(UntimedLate, RE, State))};
true ->
RevUntimedLate = lists:reverse(UntimedLate),
{ObsLate, Dict} = fix_receive_info(RevUntimedLate),
{ObsEarly, _} = fix_receive_info(RE, Dict),
case lists:reverse(ObsEarly) =:= RE of
true ->
{RE, lists:reverse(assign_happens_before(ObsLate, RE, State))};
false ->
RevHBEarly = assign_happens_before(ObsEarly, [], State),
RevHBLate = assign_happens_before(ObsLate, RevHBEarly, State),
{[], lists:reverse(RevHBLate ++ RevHBEarly)}
end
end,
?time(Logger, "Planning more interleavings..."),
NewRevTrace =
case DPOR =:= optimal of
true ->
plan_more_interleavings(lists:reverse(RevEarly, Late), [], State);
false ->
plan_more_interleavings(Late, RevEarly, State)
end,
State#scheduler_state{trace = NewRevTrace}.
%%------------------------------------------------------------------------------
split_trace(RevTrace) ->
split_trace(RevTrace, []).
split_trace([], UntimedLate) ->
{[], UntimedLate};
split_trace([#trace_state{clock_map = ClockMap} = State|RevEarlier] = RevEarly,
UntimedLate) ->
case is_empty_map(ClockMap) of
true -> split_trace(RevEarlier, [State|UntimedLate]);
false -> {RevEarly, UntimedLate}
end.
%%------------------------------------------------------------------------------
assign_happens_before(UntimedLate, RevEarly, State) ->
assign_happens_before(UntimedLate, [], RevEarly, State).
assign_happens_before([], RevLate, _RevEarly, _State) ->
RevLate;
assign_happens_before([TraceState|Later], RevLate, RevEarly, State) ->
%% We will calculate two separate clocks for each state:
%% - ops unavoidably needed to reach the state will make up the 'state' clock
%% - ops that happened before the state, will make up the 'Actor' clock
#trace_state{done = [Event|_], index = Index} = TraceState,
#scheduler_state{logger = _Logger} = State,
#event{actor = Actor, special = Special} = Event,
?debug(_Logger, "HB: ~s~n", [?pretty_s(Index, Event)]),
%% Start from the latest vector clock of the actor itself
ClockMap = get_base_clock_map(RevLate, RevEarly),
ActorLastClock = lookup_clock(Actor, ClockMap),
%% Add the step itself:
ActorNewClock = clock_store(Actor, Index, ActorLastClock),
%% And add all irreversible edges
IrreversibleClock =
add_pre_message_clocks(Special, ClockMap, ActorNewClock),
%% Apart from those, for the Actor clock we need all the ops that
%% affect the state. That is, anything dependent with the step:
HappenedBeforeClock =
update_clock(RevLate ++ RevEarly, Event, IrreversibleClock, State),
%% The 'state' clock contains the irreversible clock or
%% 'independent' if no other dependencies were found
StateClock =
case IrreversibleClock =:= HappenedBeforeClock of
true -> independent;
false -> IrreversibleClock
end,
BaseNewClockMap = map_store(state, StateClock, ClockMap),
NewClockMap = map_store(Actor, HappenedBeforeClock, BaseNewClockMap),
%% The HB clock should also be added to anything else stemming
%% from the step (spawns, sends and deliveries)
FinalClockMap =
add_new_and_messages(Special, HappenedBeforeClock, NewClockMap),
?trace(_Logger, " SC: ~w~n", [StateClock]),
?trace(_Logger, " AC: ~w~n", [HappenedBeforeClock]),
NewTraceState = TraceState#trace_state{clock_map = FinalClockMap},
assign_happens_before(Later, [NewTraceState|RevLate], RevEarly, State).
get_base_clock_map(RevLate, RevEarly) ->
case get_base_clock_map(RevLate) of
{ok, V} -> V;
none ->
case get_base_clock_map(RevEarly) of
{ok, V} -> V;
none -> empty_map()
end
end.
get_base_clock_map([#trace_state{clock_map = ClockMap}|_]) -> {ok, ClockMap};
get_base_clock_map([]) -> none.
add_pre_message_clocks([], _, Clock) -> Clock;
add_pre_message_clocks([Special|Specials], ClockMap, Clock) ->
NewClock =
case Special of
{message_delivered, #message_event{message = #message{id = Id}}} ->
max_cv(Clock, lookup_clock({Id, sent}, ClockMap));
_ -> Clock
end,
add_pre_message_clocks(Specials, ClockMap, NewClock).
add_new_and_messages([], _Clock, ClockMap) ->
ClockMap;
add_new_and_messages([Special|Rest], Clock, ClockMap) ->
NewClockMap =
case Special of
{new, SpawnedPid} ->
map_store(SpawnedPid, Clock, ClockMap);
{message, #message_event{message = #message{id = Id}}} ->
map_store({Id, sent}, Clock, ClockMap);
_ -> ClockMap
end,
add_new_and_messages(Rest, Clock, NewClockMap).
update_clock([], _Event, Clock, _State) ->
Clock;
update_clock([TraceState|Rest], Event, Clock, State) ->
#trace_state{
done = [#event{actor = EarlyActor} = EarlyEvent|_],
index = EarlyIndex
} = TraceState,
EarlyClock = lookup_clock_value(EarlyActor, Clock),
NewClock =
case EarlyIndex > EarlyClock of
false -> Clock;
true ->
#scheduler_state{assume_racing = AssumeRacing} = State,
Dependent =
concuerror_dependencies:dependent(EarlyEvent, Event, AssumeRacing),
?debug(State#scheduler_state.logger,
" ~s ~s~n",
begin
Star = fun(false) -> " "; (_) -> "->" end,
[Star(Dependent), ?pretty_s(EarlyIndex, EarlyEvent)]
end),
case Dependent =:= false of
true -> Clock;
false ->
#trace_state{clock_map = ClockMap} = TraceState,
EarlyActorClock = lookup_clock(EarlyActor, ClockMap),
max_cv(Clock, clock_store(EarlyActor, EarlyIndex, EarlyActorClock))
end
end,
update_clock(Rest, Event, NewClock, State).
%%------------------------------------------------------------------------------
plan_more_interleavings([], RevEarly, _SchedulerState) ->
RevEarly;
plan_more_interleavings([TraceState|Later], RevEarly, State) ->
case skip_planning(TraceState, State) of
true ->
plan_more_interleavings(Later, [TraceState|RevEarly], State);
false ->
#scheduler_state{logger = _Logger} = State,
#trace_state{
clock_map = ClockMap,
done = [#event{actor = Actor} = _Event|_],
index = _Index
} = TraceState,
StateClock = lookup_clock(state, ClockMap),
%% If no dependencies were found skip this altogether
case StateClock =:= independent of
true ->
plan_more_interleavings(Later, [TraceState|RevEarly], State);
false ->
?debug(_Logger, "~s~n", [?pretty_s(_Index, _Event)]),
ActorClock = lookup_clock(Actor, ClockMap),
%% Otherwise we zero-down to the latest op that happened before
LatestHBIndex = find_latest_hb_index(ActorClock, StateClock),
?trace(_Logger, " SC: ~w~n", [StateClock]),
?trace(_Logger, " AC: ~w~n", [ActorClock]),
?debug(_Logger, " Nearest race @ ~w~n", [LatestHBIndex]),
NewRevEarly =
more_interleavings_for_event(
TraceState, RevEarly, LatestHBIndex, StateClock, Later, State),
plan_more_interleavings(Later, NewRevEarly, State)
end
end.
skip_planning(TraceState, State) ->
#scheduler_state{non_racing_system = NonRacingSystem} = State,
#trace_state{done = [Event|_]} = TraceState,
#event{special = Special} = Event,
case proplists:lookup(system_communication, Special) of
{system_communication, System} -> lists:member(System, NonRacingSystem);
none -> false
end.
more_interleavings_for_event(TraceState, RevEarly, NextIndex, Clock, Later, State) ->
more_interleavings_for_event(TraceState, RevEarly, NextIndex, Clock, Later, State, []).
more_interleavings_for_event(TraceState, RevEarly, -1, _Clock, _Later,
_State, UpdEarly) ->
?trace(_State#scheduler_state.logger, " Finished checking races for event~n", []),
[TraceState|lists:reverse(UpdEarly, RevEarly)];
more_interleavings_for_event(TraceState, [], _NextIndex, _Clock, _Later,
_State, UpdEarly) ->
?trace(
_State#scheduler_state.logger,
" Finished checking races for event (NOT FAST)~n",
[]),
[TraceState|lists:reverse(UpdEarly)];
more_interleavings_for_event(TraceState, [EarlyTraceState|RevEarly], NextIndex,
Clock, Later, State, UpdEarly) ->
#trace_state{
clock_map = ClockMap,
done = [#event{actor = Actor} = Event|_]
} = TraceState,
#trace_state{
clock_map = EarlyClockMap,
done = [#event{actor = EarlyActor} = EarlyEvent|_],
index = EarlyIndex
} = EarlyTraceState,
Action =
case NextIndex =:= EarlyIndex of
false -> none;
true ->
Dependent =
case concuerror_dependencies:dependent_safe(EarlyEvent, Event) of
true -> {true, no_observer};
Other -> Other
end,
case Dependent of
false -> none;
irreversible -> update_clock;
{true, ObserverInfo} ->
?debug(State#scheduler_state.logger,
" races with ~s~n",
[?pretty_s(EarlyIndex, EarlyEvent)]),
case
update_trace(
EarlyEvent, Event, Clock, EarlyTraceState,
Later, UpdEarly, RevEarly, ObserverInfo, State)
of
skip -> update_clock;
{UpdatedNewEarly, ConservativeCandidates} ->
{update, UpdatedNewEarly, ConservativeCandidates}
end
end
end,
{NewClock, NewNextIndex} =
case Action =:= none of
true -> {Clock, NextIndex};
false ->
NC = max_cv(lookup_clock(EarlyActor, EarlyClockMap), Clock),
ActorClock = lookup_clock(Actor, ClockMap),
NI = find_latest_hb_index(ActorClock, NC),
?debug(State#scheduler_state.logger, " Next nearest race @ ~w~n", [NI]),
{NC, NI}
end,
{NewUpdEarly, NewRevEarly} =
case Action of
none -> {[EarlyTraceState|UpdEarly], RevEarly};
update_clock -> {[EarlyTraceState|UpdEarly], RevEarly};
{update, S, CC} ->
maybe_log_race(EarlyTraceState, TraceState, State),
EarlyClock = lookup_clock_value(EarlyActor, Clock),
NRE = add_conservative(RevEarly, EarlyActor, EarlyClock, CC, State),
{S, NRE}
end,
more_interleavings_for_event(TraceState, NewRevEarly, NewNextIndex, NewClock,
Later, State, NewUpdEarly).
update_trace(
EarlyEvent, Event, Clock, TraceState,
Later, NewOldTrace, Rest, ObserverInfo, State
) ->
#scheduler_state{
dpor = DPOR,
estimator = Estimator,
interleaving_id = Origin,
logger = Logger,
scheduling_bound_type = SchedulingBoundType,
use_unsound_bpor = UseUnsoundBPOR,
use_receive_patterns = UseReceivePatterns
} = State,
#trace_state{
done = [#event{actor = EarlyActor} = EarlyEvent|Done] = AllDone,
index = EarlyIndex,
previous_actor = PreviousActor,
scheduling_bound = BaseBound,
sleep_set = BaseSleepSet,
wakeup_tree = Wakeup
} = TraceState,
Bound = next_bound(SchedulingBoundType, AllDone, PreviousActor, BaseBound),
DPORInfo =
{DPOR,
case DPOR =:= persistent of
true -> Clock;
false -> {EarlyActor, EarlyIndex}
end},
RevEvent = update_context(Event, EarlyEvent),
FastSkip =
case Bound < 0 of
true ->
CI =
case SchedulingBoundType =:= bpor of
true ->
ND = not_dep(NewOldTrace, Later, DPORInfo, RevEvent),
get_initials(ND);
false ->
false
end,
{true, {over_bound, [], CI}};
false -> false
end,
{MaybeNewWakeup, VSeq, ConservativeInfo} =
case FastSkip of
{true, FastSkipReason} -> FastSkipReason;
false ->
SleepSet = BaseSleepSet ++ Done,
NotDep = not_dep(NewOldTrace, Later, DPORInfo, RevEvent),
case DPOR =:= optimal of
true ->
case UseReceivePatterns of
false ->
V = NotDep,
NW = insert_wakeup_optimal(SleepSet, Wakeup, V, Bound, Origin),
{NW, V, false};
true ->
ExtV =
case ObserverInfo =:= no_observer of
true -> NotDep;
false ->
NotObsRaw =
not_obs_raw(NewOldTrace, Later, ObserverInfo, Event),
NotObs = NotObsRaw -- NotDep,
ResetEvent = EarlyEvent#event{label = undefined},
NotDep ++ [ResetEvent] ++ NotObs
end,
RevExtV = lists:reverse(ExtV),
{V, ReceiveInfoDict} = fix_receive_info(RevExtV),
{FixedRest, _} = fix_receive_info(Rest, ReceiveInfoDict),
debug_show_sequence("v sequence", Logger, 1, V),
RevFixedRest = lists:reverse(FixedRest),
case has_weak_initial_before(RevFixedRest, V, Logger) of
true -> {skip, V, false};
false ->
NW = insert_wakeup_optimal(Done, Wakeup, V, Bound, Origin),
{NW, V, false}
end
end;
false ->
Initials = get_initials(NotDep),
V = Initials,
AddCons =
case SchedulingBoundType =:= bpor of
true -> Initials;
false -> false
end,
NW = insert_wakeup_non_optimal(SleepSet, Wakeup, V, false, Origin),
{NW, V, AddCons}
end
end,
case MaybeNewWakeup of
skip ->
?debug(Logger, " SKIP~n",[]),
skip;
over_bound ->
bound_reached(Logger),
case UseUnsoundBPOR of
true -> ok;
false -> put(bound_exceeded, true)
end,
{[TraceState|NewOldTrace], ConservativeInfo};
NewWakeup ->
debug_show_sequence("PLAN", Logger, EarlyIndex, VSeq),
NS = TraceState#trace_state{wakeup_tree = NewWakeup},
concuerror_logger:plan(Logger),
concuerror_estimator:plan(Estimator, EarlyIndex),
{[NS|NewOldTrace], ConservativeInfo}
end.
not_dep(Trace, Later, DPORInfo, RevEvent) ->
NotDep = not_dep1(Trace, Later, DPORInfo, []),
lists:reverse([RevEvent|NotDep]).
not_dep1([], [], _DPORInfo, NotDep) ->
NotDep;
not_dep1([], T, {DPOR, _} = DPORInfo, NotDep) ->
KeepLooking =
case DPOR =:= optimal of
true -> T;
false -> []
end,
not_dep1(KeepLooking, [], DPORInfo, NotDep);
not_dep1([TraceState|Rest], Later, {DPOR, Info} = DPORInfo, NotDep) ->
#trace_state{
clock_map = ClockMap,
done = [#event{actor = LaterActor} = LaterEvent|_],
index = LateIndex
} = TraceState,
NewNotDep =
case DPOR =:= persistent of
true ->
Clock = Info,
LaterActorClock = lookup_clock_value(LaterActor, Clock),
case LateIndex > LaterActorClock of
true -> NotDep;
false -> [LaterEvent|NotDep]
end;
false ->
{Actor, Index} = Info,
LaterClock = lookup_clock(LaterActor, ClockMap),
ActorLaterClock = lookup_clock_value(Actor, LaterClock),
case Index > ActorLaterClock of
false -> NotDep;
true -> [LaterEvent|NotDep]
end
end,
not_dep1(Rest, Later, DPORInfo, NewNotDep).
update_context(Event, EarlyEvent) ->
NewEventInfo =
case Event#event.event_info of
%% A receive statement...
#receive_event{message = Msg} = Info when Msg =/= 'after' ->
%% ... in race with a message.
case is_process_info_related(EarlyEvent) of
true -> Info;
false -> Info#receive_event{message = 'after'}
end;
Info -> Info
end,
%% The racing event's effect should differ, so new label.
Event#event{
event_info = NewEventInfo,
label = undefined
}.
is_process_info_related(Event) ->
case Event#event.event_info of
#builtin_event{mfargs = {erlang, process_info, _}} -> true;
_ -> false
end.
not_obs_raw(NewOldTrace, Later, ObserverInfo, Event) ->
lists:reverse(not_obs_raw(NewOldTrace, Later, ObserverInfo, Event, [])).
not_obs_raw([], [], _ObserverInfo, _Event, _NotObs) ->
[];
not_obs_raw([], Later, ObserverInfo, Event, NotObs) ->
not_obs_raw(Later, [], ObserverInfo, Event, NotObs);
not_obs_raw([TraceState|Rest], Later, ObserverInfo, Event, NotObs) ->
#trace_state{done = [#event{special = Special} = E|_]} = TraceState,
case [Id || {message_received, Id} <- Special, Id =:= ObserverInfo] =:= [] of
true ->
not_obs_raw(Rest, Later, ObserverInfo, Event, [E|NotObs]);
false ->
#event{special = NewSpecial} = Event,
ObsNewSpecial =
case lists:keyfind(message_delivered, 1, NewSpecial) of
{message_delivered, #message_event{message = #message{id = NewId}}} ->
lists:keyreplace(ObserverInfo, 2, Special, {message_received, NewId});
_ -> exit(impossible)
end,
[E#event{label = undefined, special = ObsNewSpecial}|NotObs]
end.
has_weak_initial_before([], _, _Logger) ->
?debug(_Logger, " No earlier weak initials found~n",[]),
false;
has_weak_initial_before([TraceState|Rest], V, Logger) ->
#trace_state{done = [EarlyEvent|Done]} = TraceState,
case has_initial(Done, [EarlyEvent|V]) of
true ->
?debug(
Logger,
" Has weak initial in: ~s~n",
[?join([?pretty_s(0, D) || D <- Done], "~n")]
),
debug_show_sequence("if seen as", Logger, 1, [EarlyEvent|V]),
true;
false ->
has_weak_initial_before(Rest, [EarlyEvent|V], Logger)
end.
debug_show_sequence(_Type, _Logger, _Index, _NotDep) ->
?debug(
_Logger, " ~s:~n~s",
begin
Indices = lists:seq(_Index, _Index + length(_NotDep) - 1),
IndexedNotDep = lists:zip(Indices, _NotDep),
Format = " ~s~n",
[_Type] ++
[lists:append(
[io_lib:format(Format, [?pretty_s(I,S)])
|| {I,S} <- IndexedNotDep])]
end).
maybe_log_race(EarlyTraceState, TraceState, State) ->
#scheduler_state{logger = Logger} = State,
case State#scheduler_state.show_races of
true ->
#trace_state{
done = [EarlyEvent|_],
index = EarlyIndex,
unique_id = EarlyUID
} = EarlyTraceState,
#trace_state{
done = [Event|_],
index = Index,
unique_id = UID
} = TraceState,
concuerror_logger:graph_race(Logger, EarlyUID, UID),
IndexedEarly = {EarlyIndex, EarlyEvent#event{location = []}},
IndexedLate = {Index, Event#event{location = []}},
concuerror_logger:race(Logger, IndexedEarly, IndexedLate);
false ->
?unique(Logger, ?linfo, msg(show_races), [])
end.
insert_wakeup_non_optimal(SleepSet, Wakeup, Initials, Conservative, Origin) ->
case existing(SleepSet, Initials) of
true -> skip;
false -> add_or_make_compulsory(Wakeup, Initials, Conservative, Origin)
end.
add_or_make_compulsory(Wakeup, Initials, Conservative, Origin) ->
add_or_make_compulsory(Wakeup, Initials, Conservative, Origin, []).
add_or_make_compulsory([], [E|_], Conservative, Origin, Acc) ->
Entry =
#backtrack_entry{
conservative = Conservative,
event = E, origin = Origin,
wakeup_tree = []
},
lists:reverse([Entry|Acc]);
add_or_make_compulsory([Entry|Rest], Initials, Conservative, Origin, Acc) ->
#backtrack_entry{conservative = C, event = E, wakeup_tree = []} = Entry,
#event{actor = A} = E,
Pred = fun(#event{actor = B}) -> A =:= B end,
case lists:any(Pred, Initials) of
true ->
case C andalso not Conservative of
true ->
NewEntry = Entry#backtrack_entry{conservative = false},
lists:reverse(Acc, [NewEntry|Rest]);
false -> skip
end;
false ->
NewAcc = [Entry|Acc],
add_or_make_compulsory(Rest, Initials, Conservative, Origin, NewAcc)
end.
insert_wakeup_optimal(SleepSet, Wakeup, V, Bound, Origin) ->
case has_initial(SleepSet, V) of
true -> skip;
false -> insert_wakeup(Wakeup, V, Bound, Origin)
end.
has_initial([Event|Rest], V) ->
case check_initial(Event, V) =:= false of
true -> has_initial(Rest, V);
false -> true
end;
has_initial([], _) -> false.
insert_wakeup( _, _NotDep, Bound, _Origin) when Bound < 0 ->
over_bound;
insert_wakeup( [], NotDep, _Bound, Origin) ->
backtrackify(NotDep, Origin);
insert_wakeup([Node|Rest], NotDep, Bound, Origin) ->
#backtrack_entry{event = Event, origin = M, wakeup_tree = Deeper} = Node,
case check_initial(Event, NotDep) of
false ->
NewBound =
case is_integer(Bound) of
true -> Bound - 1;
false -> Bound
end,
case insert_wakeup(Rest, NotDep, NewBound, Origin) of
Special
when
Special =:= skip;
Special =:= over_bound -> Special;
NewTree -> [Node|NewTree]
end;
NewNotDep ->
case Deeper =:= [] of
true -> skip;
false ->
case insert_wakeup(Deeper, NewNotDep, Bound, Origin) of
Special
when
Special =:= skip;
Special =:= over_bound -> Special;
NewTree ->
Entry =
#backtrack_entry{
event = Event,
origin = M,
wakeup_tree = NewTree},
[Entry|Rest]
end
end
end.
backtrackify(Seq, Cause) ->
Fold =
fun(Event, Acc) ->
[#backtrack_entry{event = Event, origin = Cause, wakeup_tree = Acc}]
end,
lists:foldr(Fold, [], Seq).
check_initial(Event, NotDep) ->
check_initial(Event, NotDep, []).
check_initial(_Event, [], Acc) ->
lists:reverse(Acc);
check_initial(Event, [E|NotDep], Acc) ->
#event{actor = EventActor} = Event,
#event{actor = EActor} = E,
case EventActor =:= EActor of
true -> lists:reverse(Acc,NotDep);
false ->
case concuerror_dependencies:dependent_safe(E, Event) =:= false of
true -> check_initial(Event, NotDep, [E|Acc]);
false -> false
end
end.
get_initials(NotDeps) ->
get_initials(NotDeps, [], []).
get_initials([], Initials, _) -> lists:reverse(Initials);
get_initials([Event|Rest], Initials, All) ->
Fold =
fun(Initial, Acc) ->
Acc andalso
concuerror_dependencies:dependent_safe(Initial, Event) =:= false
end,
NewInitials =
case lists:foldr(Fold, true, All) of
true -> [Event|Initials];
false -> Initials
end,
get_initials(Rest, NewInitials, [Event|All]).
existing([], _) -> false;
existing([#event{actor = A}|Rest], Initials) ->
Pred = fun(#event{actor = B}) -> A =:= B end,
lists:any(Pred, Initials) orelse existing(Rest, Initials).
add_conservative(Rest, _Actor, _Clock, false, _State) ->
Rest;
add_conservative(Rest, Actor, Clock, Candidates, State) ->
case add_conservative(Rest, Actor, Clock, Candidates, State, []) of
abort ->
?debug(State#scheduler_state.logger, " aborted~n",[]),
Rest;
NewRest -> NewRest
end.
add_conservative([], _Actor, _Clock, _Candidates, _State, _Acc) ->
abort;
add_conservative([TraceState|Rest], Actor, Clock, Candidates, State, Acc) ->
#scheduler_state{
interleaving_id = Origin,
logger = _Logger
} = State,
#trace_state{
done = [#event{actor = EarlyActor} = _EarlyEvent|Done],
enabled = Enabled,
index = EarlyIndex,
previous_actor = PreviousActor,
sleep_set = BaseSleepSet,
wakeup_tree = Wakeup
} = TraceState,
?debug(_Logger,
" conservative check with ~s~n",
[?pretty_s(EarlyIndex, _EarlyEvent)]),
case (EarlyActor =/= Actor) orelse (EarlyIndex < Clock) of
true -> abort;
false ->
case PreviousActor =:= Actor of
true ->
NewAcc = [TraceState|Acc],
add_conservative(Rest, Actor, Clock, Candidates, State, NewAcc);
false ->
EnabledCandidates =
[C ||
#event{actor = A} = C <- Candidates,
lists:member(A, Enabled)],
case EnabledCandidates =:= [] of
true -> abort;
false ->
SleepSet = BaseSleepSet ++ Done,
case
insert_wakeup_non_optimal(
SleepSet, Wakeup, EnabledCandidates, true, Origin
)
of
skip -> abort;
NewWakeup ->
NS = TraceState#trace_state{wakeup_tree = NewWakeup},
lists:reverse(Acc, [NS|Rest])
end
end
end
end.
%%------------------------------------------------------------------------------
has_more_to_explore(State) ->
#scheduler_state{
estimator = Estimator,
scheduling_bound_type = SchedulingBoundType,
trace = Trace
} = State,
TracePrefix = find_prefix(Trace, SchedulingBoundType),
case TracePrefix =:= [] of
true -> {false, State#scheduler_state{trace = []}};
false ->
NewState =
State#scheduler_state{need_to_replay = true, trace = TracePrefix},
[Last|_] = TracePrefix,
TopIndex = Last#trace_state.index,
concuerror_estimator:restart(Estimator, TopIndex),
{true, NewState}
end.
find_prefix([], _SchedulingBoundType) -> [];
find_prefix(Trace, SchedulingBoundType) ->
[#trace_state{wakeup_tree = Tree} = TraceState|Rest] = Trace,
case SchedulingBoundType =/= 'bpor' orelse get(bound_exceeded) of
false ->
case [B || #backtrack_entry{conservative = false} = B <- Tree] of
[] -> find_prefix(Rest, SchedulingBoundType);
WUT -> [TraceState#trace_state{wakeup_tree = WUT}|Rest]
end;
true ->
case Tree =:= [] of
true -> find_prefix(Rest, SchedulingBoundType);
false -> Trace
end
end.
replay(#scheduler_state{need_to_replay = false} = State) ->
State;
replay(State) ->
#scheduler_state{interleaving_id = N, logger = Logger, trace = Trace} = State,
[#trace_state{index = I, unique_id = Sibling} = Last|
[#trace_state{unique_id = Parent}|_] = Rest] = Trace,
concuerror_logger:graph_set_node(Logger, Parent, Sibling),
NewTrace =
[Last#trace_state{unique_id = {N, I}, clock_map = empty_map()}|Rest],
S = io_lib:format("New interleaving ~p. Replaying...", [N]),
?time(Logger, S),
NewState = replay_prefix(NewTrace, State#scheduler_state{trace = NewTrace}),
?debug(Logger, "~s~n",["Replay done."]),
NewState#scheduler_state{need_to_replay = false}.
%% =============================================================================
reset_receive_done([Event|Rest], #scheduler_state{use_receive_patterns = true}) ->
NewSpecial =
[patch_message_delivery(S, empty_map()) || S <- Event#event.special],
[Event#event{special = NewSpecial}|Rest];
reset_receive_done(Done, _) ->
Done.
fix_receive_info(RevTraceOrEvents) ->
fix_receive_info(RevTraceOrEvents, empty_map()).
fix_receive_info(RevTraceOrEvents, ReceiveInfoDict) ->
fix_receive_info(RevTraceOrEvents, ReceiveInfoDict, []).
fix_receive_info([], ReceiveInfoDict, TraceOrEvents) ->
{TraceOrEvents, ReceiveInfoDict};
fix_receive_info([#trace_state{} = TraceState|RevTrace], ReceiveInfoDict, Trace) ->
[Event|Rest] = TraceState#trace_state.done,
{[NewEvent], NewDict} = fix_receive_info([Event], ReceiveInfoDict, []),
NewTraceState = TraceState#trace_state{done = [NewEvent|Rest]},
fix_receive_info(RevTrace, NewDict, [NewTraceState|Trace]);
fix_receive_info([#event{} = Event|RevEvents], ReceiveInfoDict, Events) ->
case has_delivery_or_receive(Event#event.special) of
true ->
#event{event_info = EventInfo, special = Special} = Event,
NewReceiveInfoDict = store_receive_info(EventInfo, Special, ReceiveInfoDict),
NewSpecial = [patch_message_delivery(S, NewReceiveInfoDict) || S <- Special],
NewEventInfo =
case EventInfo of
#message_event{} ->
{_, NI} =
patch_message_delivery({message_delivered, EventInfo}, NewReceiveInfoDict),
NI;
_ -> EventInfo
end,
NewEvent = Event#event{event_info = NewEventInfo, special = NewSpecial},
fix_receive_info(RevEvents, NewReceiveInfoDict, [NewEvent|Events]);
false ->
fix_receive_info(RevEvents, ReceiveInfoDict, [Event|Events])
end.
has_delivery_or_receive([]) -> false;
has_delivery_or_receive([{M,_}|_])
when M =:= message_delivered; M =:= message_received ->
true;
has_delivery_or_receive([_|R]) -> has_delivery_or_receive(R).
store_receive_info(EventInfo, Special, ReceiveInfoDict) ->
case [ID || {message_received, ID} <- Special] of
[] -> ReceiveInfoDict;
IDs ->
ReceiveInfo =
case EventInfo of
#receive_event{receive_info = RI} -> RI;
_ -> {system, fun(_) -> true end}
end,
Fold = fun(ID,Dict) -> map_store(ID, ReceiveInfo, Dict) end,
lists:foldl(Fold, ReceiveInfoDict, IDs)
end.
patch_message_delivery({message_delivered, MessageEvent}, ReceiveInfoDict) ->
#message_event{message = #message{id = Id}} = MessageEvent,
ReceiveInfo =
case map_find(Id, ReceiveInfoDict) of
{ok, RI} -> RI;
error -> not_received
end,
{message_delivered, MessageEvent#message_event{receive_info = ReceiveInfo}};
patch_message_delivery(Other, _ReceiveInfoDict) ->
Other.
%% =============================================================================
%% ENGINE (manipulation of the Erlang processes under the scheduler)
%% =============================================================================
replay_prefix(Trace, State) ->
#scheduler_state{
entry_point = EntryPoint,
first_process = FirstProcess,
processes = Processes,
timeout = Timeout
} = State,
concuerror_callback:reset_processes(Processes),
ok =
concuerror_callback:start_first_process(FirstProcess, EntryPoint, Timeout),
NewState = State#scheduler_state{last_scheduled = FirstProcess},
replay_prefix_aux(lists:reverse(Trace), NewState).
replay_prefix_aux([_], State) ->
%% Last state has to be properly replayed.
State;
replay_prefix_aux([#trace_state{done = [Event|_], index = I}|Rest], State) ->
#scheduler_state{logger = _Logger, print_depth = PrintDepth} = State,
?debug(_Logger, "~s~n", [?pretty_s(I, Event)]),
{ok, #event{actor = Actor} = NewEvent} = get_next_event_backend(Event, State),
try
true = Event =:= NewEvent
catch
_:_ ->
#scheduler_state{print_depth = PrintDepth} = State,
?crash({replay_mismatch, I, Event, NewEvent, PrintDepth})
end,
NewLastScheduled =
case is_pid(Actor) of
true -> Actor;
false -> State#scheduler_state.last_scheduled
end,
NewState = State#scheduler_state{last_scheduled = NewLastScheduled},
replay_prefix_aux(Rest, maybe_log(Event, NewState, I)).
%% =============================================================================
%% INTERNAL INTERFACES
%% =============================================================================
%% Between scheduler and an instrumented process
%%------------------------------------------------------------------------------
get_next_event_backend(#event{actor = Channel} = Event, State)
when ?is_channel(Channel) ->
#scheduler_state{timeout = Timeout} = State,
#event{event_info = MessageEvent} = Event,
assert_no_messages(),
UpdatedEvent =
concuerror_callback:deliver_message(Event, MessageEvent, Timeout),
{ok, UpdatedEvent};
get_next_event_backend(#event{actor = Pid} = Event, State) when is_pid(Pid) ->
#scheduler_state{timeout = Timeout} = State,
assert_no_messages(),
Pid ! Event,
concuerror_callback:wait_actor_reply(Event, Timeout).
assert_no_messages() ->
receive
Msg -> error({pending_message, Msg})
after
0 -> ok
end.
%%%----------------------------------------------------------------------
%%% Helper functions
%%%----------------------------------------------------------------------
-ifdef(BEFORE_OTP_17).
empty_map() ->
dict:new().
map_store(K, V, Map) ->
dict:store(K, V, Map).
map_find(K, Map) ->
dict:find(K, Map).
is_empty_map(Map) ->
dict:size(Map) =:= 0.
lookup_clock(P, ClockMap) ->
case dict:find(P, ClockMap) of
{ok, Clock} -> Clock;
error -> clock_new()
end.
clock_new() ->
orddict:new().
clock_store(_, 0, VectorClock) ->
VectorClock;
clock_store(Actor, Index, VectorClock) ->
orddict:store(Actor, Index, VectorClock).
lookup_clock_value(Actor, VectorClock) ->
case orddict:find(Actor, VectorClock) of
{ok, Value} -> Value;
error -> 0
end.
max_cv(D1, D2) ->
Merger = fun(_Key, V1, V2) -> max(V1, V2) end,
orddict:merge(Merger, D1, D2).
find_latest_hb_index(ActorClock, StateClock) ->
%% This is the max index that is in the Actor clock but not in the
%% corresponding state clock.
Fold =
fun(K, V, Next) ->
case orddict:find(K, StateClock) =:= {ok, V} of
true -> Next;
false -> max(V, Next)
end
end,
orddict:fold(Fold, -1, ActorClock).
-else.
empty_map() ->
#{}.
map_store(K, V, Map) ->
maps:put(K, V, Map).
map_find(K, Map) ->
maps:find(K, Map).
is_empty_map(Map) ->
maps:size(Map) =:= 0.
lookup_clock(P, ClockMap) ->
maps:get(P, ClockMap, clock_new()).
clock_new() ->
#{}.
clock_store(_, 0, VectorClock) ->
VectorClock;
clock_store(Actor, Index, VectorClock) ->
maps:put(Actor, Index, VectorClock).
lookup_clock_value(Actor, VectorClock) ->
maps:get(Actor, VectorClock, 0).
max_cv(VC1, VC2) ->
ODVC1 = orddict:from_list(maps:to_list(VC1)),
ODVC2 = orddict:from_list(maps:to_list(VC2)),
Merger = fun(_Key, V1, V2) -> max(V1, V2) end,
MaxVC = orddict:merge(Merger, ODVC1, ODVC2),
maps:from_list(MaxVC).
find_latest_hb_index(ActorClock, StateClock) ->
%% This is the max index that is in the Actor clock but not in the
%% corresponding state clock.
Fold =
fun(K, V, Next) ->
case maps:find(K, StateClock) =:= {ok, V} of
true -> Next;
false -> max(V, Next)
end
end,
maps:fold(Fold, -1, ActorClock).
-endif.
next_bound(SchedulingBoundType, Done, PreviousActor, Bound) ->
case SchedulingBoundType of
none -> Bound;
bpor ->
NonPreemptExplored =
[E || #event{actor = PA} = E <- Done, PA =:= PreviousActor] =/= [],
case NonPreemptExplored of
true -> Bound - 1;
false -> Bound
end;
delay ->
%% Every reschedule costs.
Bound - length(Done)
end.
bound_reached(Logger) ->
?unique(Logger, ?lwarning, msg(scheduling_bound_warning), []),
?debug(Logger, "OVER BOUND~n",[]),
concuerror_logger:bound_reached(Logger).
%% =============================================================================
-spec explain_error(term()) -> string().
explain_error({blocked_mismatch, I, Event, Depth}) ->
EString = concuerror_io_lib:pretty_s(Event, Depth),
io_lib:format(
"On step ~p, replaying a built-in returned a different result than"
" expected:~n"
" original:~n"
" ~s~n"
" new:~n"
" blocked~n"
?notify_us_msg,
[I,EString]
);
explain_error({optimal_sleep_set_block, Origin, Who}) ->
io_lib:format(
"During a run of the optimal algorithm, the following events were left in~n"
"a sleep set (the race was detected at interleaving #~p)~n~n"
" ~p~n"
?notify_us_msg,
[Origin, Who]
);
explain_error({replay_mismatch, I, Event, NewEvent, Depth}) ->
[EString, NEString] =
[concuerror_io_lib:pretty_s(E, Depth) || E <- [Event, NewEvent]],
[Original, New] =
case EString =/= NEString of
true -> [EString, NEString];
false ->
[io_lib:format("~p",[E]) || E <- [Event, NewEvent]]
end,
io_lib:format(
"On step ~p, replaying a built-in returned a different result than"
" expected:~n"
" original:~n"
" ~s~n"
" new:~n"
" ~s~n"
?notify_us_msg,
[I,Original,New]
).
%%==============================================================================
msg(after_timeout_tip) ->
"You can use e.g. '--after_timeout 5000' to treat after timeouts that exceed"
" some threshold (here 4999ms) as 'infinity'.~n";
msg(assertions_only_filter) ->
"Only assertion failures are considered abnormal exits ('--assertions_only').~n";
msg(assertions_only_use) ->
"A process exited with reason '{{assert*,_}, _}'. If you want to see only"
" this kind of error you can use the '--assertions_only' option.~n";
msg(depth_bound_reached) ->
"An interleaving reached the depth bound. This can happen if a test has an"
" infinite execution. Concuerror is not sound for testing programs with"
" infinite executions. Consider limiting the size of the test or increasing"
" the bound ('-h depth_bound').~n";
msg(maybe_receive_loop) ->
"The trace contained more than ~w receive timeout events"
" (receive statements that executed their 'after' clause). Concuerror by"
" default treats 'after' clauses as always possible, so a 'receive loop'"
" using a timeout can lead to an infinite execution. "
++ msg(after_timeout_tip);
msg(scheduling_bound_tip) ->
"Running without a scheduling_bound corresponds to verification and"
" may take a long time.~n";
msg(scheduling_bound_warning) ->
"Some interleavings will not be explored because they exceed the scheduling"
" bound.~n";
msg(show_races) ->
"You can see pairs of racing instructions (in the report and"
" '--graph') with '--show_races true'~n";
msg(shutdown) ->
"A process exited with reason 'shutdown'. This may happen when a"
" supervisor is terminating its children. You can use '--treat_as_normal"
" shutdown' if this is expected behaviour.~n";
msg(stop_first_error) ->
"Stop testing on first error. (Check '-h keep_going').~n";
msg(timeout) ->
"A process exited with reason '{timeout, ...}'. This may happen when a"
" call to a gen_server (or similar) does not receive a reply within some"
" timeout (5000ms by default). "
++ msg(after_timeout_tip);
msg(treat_as_normal) ->
"Some abnormal exit reasons were treated as normal ('--treat_as_normal').~n".