IoT City: gossip, gossip e ancora gossip

Ad IoT City le notizie volano, non solo quando si tratta di vincite alla lotteria o tradimenti. L’amministrazione comunale è interessata a studiare protocolli decentralizzati che consentano ai cittadini di trasferire informazioni vitali anche in caso di emergenza, quando i canali di comunicazione principali o gli uffici pubblici possono risultare inutilizzabili. Una famiglia di protocolli utilizzata per allineare un insieme distribuito di nodi in modo decentralizzato è costituita dagli algoritmi anti-entropia. Si chiamano così proprio in virtù della loro finalità: ridurre l’entropia fra nodi diversi, riconciliando eventuali differenze. I protocolli di questo tipo non si limitano alla diffusione delle informazioni: possono essere utilizzati, a diversi livelli di granularità, anche per conciliare versioni diverse delle medesime informazioni e identificare eventuali dati mancanti o danneggiati.

Il funzionamento dei protocolli anti-entropia è semplice e consente l’adozione di più varianti. Iniziamo dalla variante PUSH:

• ciascun nodo mantiene un proprio stato e un insieme di nodi a esso collegati;
• a intervalli regolari il nodo, se ha un aggiornamento, sceglie casualmente uno dei nodi collegati e propaga il proprio stato;
• se il nodo ricevente non è in possesso dell’informazione (o se l’informazione ricevuta ne modifica lo stato), si attiva e propaga a sua volta l’informazione.

Questa modalità è particolarmente efficace durante le prime fasi del processo di diffusione: è infatti molto probabile trovare altri nodi ancora all’oscuro dell’informazione (come avveniva anche nel protocollo rumor mongering). Con il tempo, tuttavia, diventa inefficiente, poiché la probabilità di incontrare nodi non aggiornati diminuisce: la convergenza rallenta e l’overhead aumenta. In sintesi, si tratta di una strategia efficace in scenari write-intensive (molte variazioni da propagare) ma poco adatta a piccoli aggiornamenti sporadici.

Implementiamo il nostro attore CitizenActor.

 public enum CitizenState
 {
     IGNORANT,
     NOT_INGORANT
 }
 public record SetCitizens(List<IActorRef> Citizens);
 public record Rumor(String Sender = null);
public record MakePhoneCall();

 public record GetState();
 public record StateResponse(string Name, CitizenState State);

 public class CitizenActor : ReceiveActor
 {
     private List<IActorRef> _citizens;
     private String _name;

     private CitizenState _state;

     private readonly Random _random = new Random(); 
     private ICancelable gossipTask;

     public CitizenActor(string name)
     {
         _name = name;
         _state = CitizenState.IGNORANT;
         _citizens = new List<IActorRef>();

         Receive<SetCitizens>(msg =>
         {
             _citizens = new List<IActorRef>();
             foreach (var a in msg.Citizens)
             {
                 if (a != Self)
                 {
                     _citizens.Add(a);
                 }
             }
         });

         Receive<Rumor>(msg =>
         {

             if (_state == CitizenState.IGNORANT)
             {
                 _state = CitizenState.NOT_INGORANT;

                 gossipTask = Context.System.Scheduler.ScheduleTellRepeatedlyCancelable(
                     TimeSpan.Zero,
                     TimeSpan.FromMilliseconds(500),
                     Self,
                     new MakePhoneCall(),
                     Self);
             }
         });

         Receive<GetState>(msg =>
         {
             Sender.Tell(new StateResponse(_name, _state));
         });

         Receive<MakePhoneCall>(msg =>
         {

             var k = _random.Next(_citizens.Count);
             var randomNeighbor = _citizens[k];
             randomNeighbor.Tell(new Rumor(_name));
         });
     }
 }

Rispetto all’attore progettato per il protocollo rumor mongering, in questo primo esempio non è necessario gestire il feedback al mittente (utile nel caso precedente per il passaggio dallo stato di “diffusore” a quello di “soffocante”). Testiamo il caso con 25 attori, comunicando a uno solo di essi la novità.

Per il testing ho riutilizzato i casi della lezione precedente, che vi invito a rileggere in caso di dubbi.

In una rete completamente connessa, l’esempio consente di raggiungere tutti i nodi in circa tre secondi. Tuttavia, anche dopo aver propagato il messaggio a tutti, il protocollo non termina: ciascun nodo continua a diffondere l’informazione, generando un notevole rumore nella rete.

Una alternativa alla variante PUSH è la sua controparte, la variante PULL. In questo caso ciascun nodo sceglie, a intervalli regolari, uno dei propri vicini e gli chiede se ha novità. In caso affermativo riceve il messaggio, aggiorna il proprio stato interno e si ferma, attendendo eventuali richieste dai vicini. Il vantaggio di questa modalità è che, pur avendo un avvio più lento (minor probabilità di raggiungere subito il nodo con l’informazione aggiornata), tende a una situazione di quiete: tutti i nodi si aggiornano e poi interrompono lo scambio di messaggi. Il rovescio della medaglia è la maggiore complessità della logica del ricevente, che deve determinare se dispone di una versione più recente dell’informazione (ad esempio tramite un numero di versione o un hash).

Modifichiamo ora l’attore precedente per gestire il caso PULL.

public enum CitizenState
{
    IGNORANT,
    NOT_INGORANT
}
public record SetCitizens(List<IActorRef> Citizens);
public record AnyNews();


public record News();
public record Rumor(String Sender = null);
public record MakePhoneCall();

public record Start();

public record GetState();
public record StateResponse(string Name, CitizenState State);

public class CitizenActor : ReceiveActor
{
    private List<IActorRef> _citizens;
    private String _name;

    private CitizenState _state;

    private readonly Random _random = new Random(); 
    private ICancelable gossipTask;

    public CitizenActor(string name)
    {
        _name = name;
        _state = CitizenState.IGNORANT;
        _citizens = new List<IActorRef>();



        Receive<SetCitizens>(msg =>
        {
            _citizens = new List<IActorRef>();
            foreach (var a in msg.Citizens)
            {
                if (a != Self)
                {
                    _citizens.Add(a);
                }
            }


        });

        Receive<Rumor>(msg =>
        {

            if (_state == CitizenState.IGNORANT)
            {
                if (gossipTask != null)
                {
                    gossipTask.Cancel();
                }

                _state = CitizenState.NOT_INGORANT;
            }
        });

        Receive<Start>(msg =>
        {
            if (_state == CitizenState.IGNORANT)
            {
                gossipTask = Context.System.Scheduler.ScheduleTellRepeatedlyCancelable(
                       TimeSpan.Zero,
                       TimeSpan.FromMilliseconds(500),
                       Self,
                       new MakePhoneCall(),
                       Self);
            }
        });

        Receive<GetState>(msg =>
        {
            Sender.Tell(new StateResponse(_name, _state));
        });

        Receive<AnyNews>(msg =>
        {
            if (_state == CitizenState.NOT_INGORANT)
            {
                Sender.Tell(new News());
            }
        });

        Receive<News>(msg =>
        {
            Self.Tell(new Rumor());
        });

        Receive<MakePhoneCall>(msg =>
        {

            var k = _random.Next(_citizens.Count);
            var randomNeighbor = _citizens[k];

            randomNeighbor.Tell(new AnyNews());
        });
    }
}

Per avviare il polling è necessario introdurre un ulteriore messaggio Start, inviato dopo la creazione della rete di nodi, per evitare che le prime iterazioni avvengano senza alcun interlocutore. In caso di ricezione di una nuova informazione, l’attore la integra nel proprio stato e interrompe il polling.

Nell’esempio non ho gestito il controllo di quale dei due attori disponga dell’informazione più recente, ma è possibile farlo facilmente rappresentando la versione dello stato con un numero intero: l’attore con il valore maggiore è quello aggiornato.

Rispetto al caso PUSH, la modalità PULL è più efficace nelle fasi finali del processo, garantendo una convergenza lenta ma con overhead ridotto: è quindi adatta alla sincronizzazione finale.

Possiamo combinare le due modalità operative? Sì: il protocollo PUSH-PULL consente un dialogo bidirezionale e una vera sincronizzazione fra i nodi. A ogni intervallo di tempo, un attore seleziona casualmente un vicino e gli invia il proprio stato. Il nodo ricevente risponde con il proprio. Ciascuno dei due utilizza quanto ricevuto per aggiornare il proprio stato. Eventuali conflitti possono essere risolti (in questa implementazione semplificata) scegliendo la versione con il numero maggiore. Anche al termine del processo gli attori continueranno a scambiarsi messaggi di sincronizzazione, generando overhead ma senza trasferire nuove informazioni. Per mitigare il problema, è possibile adottare strategie di adattamento che “allungano” il tempo tra due polling successivi in assenza di novità (adaptive gossip). Nella modalità PUSH-PULL la convergenza si raggiunge in O(log N) cicli di comunicazione; oltre tale soglia la probabilità di incontrare un nodo non informato è trascurabile (a meno di topologie particolari della rete, come visto nella lezione precedente).

Il nostro attore CitizenActor con protocollo PUSH-PULL può essere così descritto:

public record SetCitizens(List<IActorRef> Citizens);
public record PushPullRequest(int Version);
public record PushRequest(int Version);
public record MakePhoneCall();

public record Start();

public record GetVersion();
public record VersionResponse(string Name, int Version);

public class CitizenActor : ReceiveActor
{
    private List<IActorRef> _citizens;
    private String _name;
    private int _version;

    private readonly Random _random = new Random(); 
    private ICancelable gossipTask;

    public CitizenActor(string name)
    {
        _name = name;
        _version = 0;
        _citizens = new List<IActorRef>();



        Receive<SetCitizens>(msg =>
        {
            _citizens = new List<IActorRef>();
            foreach (var a in msg.Citizens)
            {
                if (a != Self)
                {
                    _citizens.Add(a);
                }
            }


        });

        Receive<PushRequest>(msg =>
        {

            if (msg.Version > _version)
            {
                _version = msg.Version;
            }
        });

        Receive<PushPullRequest>(msg =>
        {
            Sender.Tell(new PushRequest(_version));
            if (msg.Version > _version)
            {
                _version = msg.Version;
            }
        });

        Receive<Start>(msg =>
        {

                gossipTask = Context.System.Scheduler.ScheduleTellRepeatedlyCancelable(
                       TimeSpan.Zero,
                       TimeSpan.FromMilliseconds(500),
                       Self,
                       new MakePhoneCall(),
                       Self);

        });

        Receive<GetVersion>(msg =>
        {
            Sender.Tell(new VersionResponse(_name, _version));
        });
        Receive<MakePhoneCall>(msg =>
        {

            var k = _random.Next(_citizens.Count);
            var randomNeighbor = _citizens[k];

            var r = randomNeighbor.Ask<PushRequest>(new PushPullRequest(_version), timeout: TimeSpan.FromSeconds(10));

            r.PipeTo(Self);
        });
    }
}

Un aspetto interessante del nostro agente è l’utilizzo di un costrutto di Akka.NET che non avevamo ancora visto: PipeTo. PipeTo è un’estensione Akka.NET che “connette” un Task asincrono al sistema di messaggi. Quando il Task termina, con successo o con eccezione, PipeTo invia automaticamente il risultato come messaggio a un attore. In questo modo il nostro attore può, mentre attende la risposta a una chiamata Ask, continuare a smarcare i messaggi successivi presenti nella coda. Questa soluzione consente di mantenere attori reattivi e coerenti con il modello Akka, senza introdurre costrutti non idiomatici come await o Thread.Sleep.

Una osservazione interessante sulla modalità PUSH-PULL è che, anche introducendo più versioni dello stato dall’esterno, la versione più recente tende naturalmente a propagarsi, sostituendo le altre. In questo caso non abbiamo problemi di merge: la versione con il valore maggiore è sempre quella da considerare aggiornata.

Una variante interessante ed evoluta al protocollo PUSH-PULL è rappresentato dal protocollo Scuttlebutt. Il protocollo Scuttlebutt nasce come un’evoluzione del gossip classico che abbiamo visto, pensata per contesti in cui la connettività non è garantita, come reti con connettività intermittente (e.g.: NBIoT), ambienti distribuiti geograficamente o sistemi completamente decentralizzati. Ogni nodo in Scuttlebutt mantiene una replica locale dello stato globale, organizzata in una serie di sequenze di messaggi, una per ciascun nodo della rete che segue o con cui ha interagito in passato. Ciascun messaggio è immutabile e firmato digitalmente; la corretta sequenza dei messaggi(in un approccio che richiama per molti versi l’idea di blockchain) e la loro autenticità può essere verificata. Quando due nodi entrano in contatto, ciascuno comunica all’altro quale versione di ogni sequenza possiede (e.g.: io ho ricevuto 12 messaggi dall’attore X) e riceve solo i messaggi successivi mancanti(e.g.: ti mando i messaggi da 13 a 17 inviati da X). Questo approccio, simile a PUSH-PULL, consente di minimizzare la divergenza fra i nodi.

A differenza dei protocolli che abbiamo visto, che tendono a essere “stateless”, Scuttlebutt è deterministico e incrementale: due nodi che si incontrano più volte non duplicano lavoro, ma riprendono da dove avevano interrotto, conservando una “memoria” delle interazioni precedenti. Pensa ai vantaggi di un approccio di questo tipo quando ci troviamo a operare con dispositivi a ridotta connettività, o con connettività saltuaria (e.g.: un dispositivo NBIoT che si connette alla rete ad intervallo giornaliero). Un altro aspetto cruciale è la propagazione indiretta: se l’attore A non può parlare direttamente con C, ma entrambi comunicano con B, quest’ultimo può trasmettere gradualmente le informazioni dell’uno all’altro. Il modello quindi non richiede una connettività totale, ma solo la connettività transitiva nel tempo: è sufficiente che i nodi si incontrino periodicamente. Il grande vantaggio del modello è la robustezza: nessun singolo nodo è critico, e la perdita temporanea di connettività non impedisce la convergenza. Lo svantaggio principale è il costo di archiviazione e verifica: poiché ogni sequenza deve rispettare l’ordine e la autenticità, lo spazio cresce linearmente con il tempo e con il numero di nodi incontrati. Sono quindi necessari meccanismi di pruning, compressione o snapshot periodici.

Durante questo nostro approfondimento, più breve del solito, abbiamo potuto esplorare i protocolli che garantiscono la diffusione delle informazioni ad IoT City in modo distribuito, anche in caso di emergenza. Anche in assenza di novità, gli attori continuano a parlarsi, a confrontarsi, a verificare che la versione della realtà condivisa sia ancora valida. È una comunicazione costante e quasi impercettibile, che tiene viva la città: piccoli scambi di informazioni che, ripetuti milioni di volte, mantengono coeso un sistema distribuito. Il tutto senza una regia centralizzata: le regole di interazione fra gli attori determinano le modalità periodiche di aggiornamento, interrogazione e allineamento.

Nelle reti complesse — come nelle comunità umane e nella nostra città virtuale — la ridondanza non è un difetto, ma una forma di resilienza. L’anti-entropia è la capacità di un sistema di ritrovare l’ordine nonostante la casualità delle sue interazioni. Così, anche in presenza di errori, crash o disconnessioni, le informazioni corrette finiscono per prevalere e diffondersi ovunque.

La coerenza, come abbiamo imparato, in un mondo distribuito non è un punto d’arrivo, ma un processo continuo.

Qui ti ripoto un insieme di materiali sui quali ho basato queste riflessioni e che possono essere utili a complemento di quanto visto. Il secondo link è particolarmente interessante e introduce il concetto di directional gossip, ovvero gli algoritmi di gossip utilizzati in ambito WAN e LAN:

• https://highscalability.com/gossip-protocol-explained
• https://www.cs.utexas.edu/~lorenzo/corsi/cs395t/04S/notes/gossipnotes.pdf
• https://www.mitre.org/sites/default/files/pdf/11_5275.pdf
• https://arxiv.org/abs/1411.0948

Grazie mille per la tua attenzione. Nel prossimo incontro parleremo di Gossip Based Failure Detection e di utilizzo dei protocolli di gossip per gestire la salute di un insieme di nodi, con alcuni esempi relativi a casi reali dei nostri cittadini. A presto, grazie ancora, buona vita.