Self Stabilizing Systems in Spite of Distributed Control

ss.erl

%% @doc
%% This Erlang module implements the ideas illustrated in Edsger Dijkstra's
%% famous paper "Self Stabilizing Systems in Spite of Distributed Control"
%% http://courses.csail.mit.edu/6.852/05/papers/p643-Dijkstra.pdf
%% @end
-module(ss).
-export([stop/1, show/1, start/1]).

-define(DELAY, 1000).
-define(THRESHOLD, 0.90).

-record(state, {
    id :: non_neg_integer(),
    k :: pos_integer(),
    left :: pid(),
    right :: pid(),
    value :: integer()
}).

%% public API
show(L) ->
    [ Pid ! show || {_, Pid} <- L ].

stop(L) ->
    [ Pid ! stop || {_, Pid} <- L ].

start(N) when N > 1 ->
    Bottom = spawn_node(0, N+1, undefined),
    spawn_nodes(lists:seq(1, N-1), N+1, Bottom, Bottom, [{0, Bottom}]);
start(_Other) ->
    io:format(user, "N must be bigger than 1", []),
    ok.

%% private
spawn_nodes([], _K, Top, Bottom, Acc) ->
    Top ! {left, Bottom},
    Bottom ! {right, Top},
    lists:reverse(Acc);
spawn_nodes([H|T], K, Right, Bottom, Acc) ->
    Pid = spawn_node(H, K, Right),
    Right ! {left, Pid},
    spawn_nodes(T, K, Pid, Bottom, [{H, Pid}|Acc]).

spawn_node(Id, K, Right) ->
    Pid = spawn(fun() -> machine(#state{id = Id, k = K, value = 0, right = Right}) end),
    erlang:send_after(?DELAY, Pid, tick),
    Pid.

%% central machine loop
machine(#state{ id = Id, left = L, k = K, value = V } = State) ->
    receive
        stop ->
            io:format(user, "ok, bye~n", []),
            ok;
        tick ->
            io:format(user, "Machine ~p: state: ~p~n", [Id, V]),
            erlang:send_after(?DELAY, self(), tick),
            NewState = maybe_corrupt(State),
            L ! {get_state, self()},
            machine(NewState);
        {left, Pid} ->
            machine(State#state{left = Pid});
        {right, Pid} ->
            machine(State#state{right = Pid});
        show ->
            io:format(user, "Id: ~p, State: ~p~n", [Id, State]),
            machine(State);
        {get_state, CallerPid} ->
            CallerPid ! {state, Id, V},
            machine(State);
        %% Bottom node stabilization
        {state, LeftId, V} when Id == 0 ->
            io:format(user, "Bottom machine: my left neighbor (id ~p) has the same state value as me!~n",
                      [LeftId]),
            machine(State#state{ value = (V+1) rem K });
        {state, LeftId, _LeftState} when Id == 0 ->
            io:format(user, "Bottom machine: my left neighbor (id ~p) has a different state value than me!~n",
                      [LeftId]),
            machine(State);
        %% every other node stabilization
        {state, LeftId, V} ->
            io:format(user, "Machine ~p: my left neighbor (id ~p) has same state value~n",
                      [Id, LeftId]),
            machine(State);
        {state, LeftId, LeftState} ->
            io:format(user, "Machine ~p: my left neighbor (id ~p) has state value ~p~n",
                      [Id, LeftId, LeftState]),
            machine(State#state{ value = LeftState });
        Other ->
            io:format(user, "Unexpected call ~p", [Other]),
            machine(State)
    after 5000 ->
        erlang:error(timeout)
    end.

maybe_corrupt(State) ->
    case check_state(State) of
        invalid -> State;
        valid ->
            case rand:uniform() of
                X when X > ?THRESHOLD ->
                    State#state{ value = rand:uniform(100) };
                _ ->
                    State
            end
    end.

check_state(#state{ k = K, value = V }) when V >= 0 andalso V < K -> valid;
check_state(_) -> invalid.