Il pattern architetturale Pipes and Filters

Pipes and filters è un pattern architetturale del catalogo POSA (Pattern Oriented Software Architecture) nella categoria “Data stream processing”. Un esempio di un’applicazione del genere viene dagli e-commerce, in cui dobbiamo trasformare un flusso di ordini in un flusso di spedizioni (si parla in questo caso di dataflow-driven application) o dalle applicazioni di analisi dei dati.

Indice dei contenuti:

1. Definizione del pattern
2. Applicabilità
3. Scenari possibili
   • Scenario 1: push
   • Scenario 2: pull
   • Scenario 3: push-pull con un solo elemento attivo
   • Scenario 4: push-pull con più elementi attivi
4. Impatti
5. Esempio

DEFINIZIONE DEL PATTERN

Nella definizione di questo pattern abbiamo visto l’introduzione di due elementi architetturali, le pipe e i filters. La parola “filter” tuttavia può essere fuorviante, dal momento che suggerisce un qualcosa che seleziona i dati. Qui il termine più corretto sarebbe processor.

Una cosa importante da tenere a mente qui è la coesione dei filtri: ogni filter si deve occupare di un solo compito, e non ci deve essere sovrapposizione tra i compiti dei vari filtri.

I filtri che fanno parte di una pipeline possono essere eseguiti su macchine diverse, e questo, ragionando in un contesto cloud, permette di avvantaggiarsi dell’elasticità di questi ambienti, dal momento che ogni filtro potrebbe scalare indipendentemente dagli altri (specie in contesti come lo scenario 4, che verrà illustrato in seguito). Un filtro che richiede elevata potenza di calcolo può essere eseguito su macchine più potenti, mentre filtri meno “esigenti” possono essere eseguiti su macchine meno costose.

APPLICABILITÀ

Questo pattern può essere impiegato nelle seguenti situazioni:

• L’elaborazione richiesta può essere facilmente suddivisa in step indipendenti
• I diversi step dell’elaborazione hanno requisiti prestazionali (in termini di scalabilità e potenza di calcolo richiesta) differenti
• È richiesta flessibilità in termini di possibilità di cambiare l’ordine dei vari step o di aggiungerne/rimuoverne.
• Si può avere un beneficio nelle prestazioni distribuendo i vari processi su diversi server
• Si vuole una soluzione affidabile che minimizzi le conseguenze del fallimento di uno step

Potrebbe essere controproducente, se non addirittura dannoso, quando:

• Alcuni step dell’elaborazione non sono indipendenti, o devono essere eseguiti insieme come parte della stessa transazione
• Le informazioni relative al contesto o allo stato richieste da uno step rende questo approccio inefficiente, a causa dell’alto numero di informazioni trasferite in seguito al cambio di contesto

I 4 SCENARI POSSIBILI

Ci sono quattro scenari possibili per Pipes and filters: push, pull, push-pull con un solo elemento attivo, push pull con più elementi attivi. I quattro scenari differiscono per la gestione del controllo e degli elementi attivi. Nei primi tre che vedremo c’è un solo elemento attivo. Nel quarto, più complesso (ma anche il più comune) abbiamo più flussi attivi contemporaneamente.

Scenario 1: push

Scenario 2: pull

Scenario 3: push-pull con un solo elemento attivo

Scenario 4: push-pull con più elementi attivi

Dei quattro scenari presentati questo è quello che permette prestazioni migliori: dal momento che ciascun filtro vive in un processo o thread separato ognuno di essi può svolgere il proprio compito indipendentemente dagli altri, con l’ulteriore vantaggio che del buffering dei dati, se ne occupano le pipe (che ci potrebbero essere messe a disposizione da strumenti di Messaging come RabbitMQ.

IMPATTI

Come ogni pattern architetturale anche pipes and filters ha impatti sulle diverse qualità del sistema, e se si decide di implementarlo bisogna tenere a mente le sue conseguenze sul sistema, senza farsi prendere dalla sindrome del martello (“Quando hai un martello in mano vedi chiodi dappertutto”).

ESEMPIO

Per comprendere meglio i concetti illustrati mostreremo ora i punti salienti dell’implementazione di Pipes & filters proposta da Microsoft. Si tratta di un’implementazione specifica per Azure che sfrutta alcuni suoi strumenti (come il Service Bus), ma il nucleo concettuale resta lo stesso.
Può essere interessante andare ad esaminare la classe ServiceBusPipeFilter, che non è altro che un connettore tra due filtri. In particolare, il metodo OnPipeFilterMessageAsync si occupa di ricevere messaggi dalla coda e fare un push al prossimo filtro per avere un messaggio elaborato da inviare alla prossima coda.

public void OnPipeFilterMessageAsync(Func<ServiceBusReceivedMessage, Task<ServiceBusMessage» asyncFilterTask)
        { 
            this.processor.ProcessMessageAsync +=
                async args =>
            {
                ServiceBusReceivedMessage message = args.Message;
                pauseProcessingEvent.WaitOne();
                if (message.DeliveryCount > Constants.MaxServiceBusDeliveryCount) 
                { 
                     await args.DeadLetterMessageAsync(message);
                     Trace.TraceWarning("Maximum Message Count Exceeded: {0) for MessageID: {1) ", Constants.
MaxServiceBusDeliveryCount, message.Messageld);
                     return;
                }
                var outMessage = await asyncFilterTask(message);
                if (sender != null) 
                { 
                    await this.sender.SendMessageAsync(outMessage); 
                }
           });
           this.processor.ProcessErrorAsync += this.OptionsOnExceptionReceived;
           this.processor.StartProcessingAsync();
        }

Si può notare che, come abbiamo detto, si tratta di implementazione un po’ diversa da quella “canonica”, dal momento che è la pipe che si occupa di inviare dati alla coda successiva invece che il filtro, che nella versione proposta nel catalogo POSA è colui che fa push verso la pipe e quindi verso il filtro successivo.
Il secondo elemento da analizzare è la classe PipeFilterARoleEntry.

public override bool OnStart() 
         { 
            ServicePointManager.DefaultConnectionlimit = 12;
            this.pipeFilterA = new ServiceBusPipeFilter(
            Settings.ServiceBusConnectionString,
            Constants.QueueAPath,
            Constants.QueueBPath);
         this.pipeFilterA.Start();
         return base.OnStart(); 
} 

public override void Run()
        { 
            this.pipeFilterA.OnPipeFilterMessageAsync(async (msg) =>
            { 
                var newMsg = new ServiceBusMessage(msg);
                await Task.Delay(500);
                Trace.TraceInformation("Fitler A processed message:{0} at {1}", msg.Messageld, DateTime.UtcNow);
                newMsg.ApplicationProperties.Add(Constants.FilterAMessagekey, "Complete");
                return newMsg; 
            });
            this.stopRunningEvent.WaitOne(); 
} 

Non è molto importante la logica del filtro (non fa altro che fare un clone del messaggio ricevuto), quanto la scelta degli “strumenti”. Qui i filtri vengono implementati come WorkerRole, e questo significa che ogni filtro può scalare indipendentemente dagli altri, qualora gli sia richiesta potenza di calcolo maggiore. Inoltre, più istanze di un singolo WorkerRole possono essere eseguite in parallelo, a vantaggio del throughput.
L’esempio riporta anche due ruoli addizionali chiamati InitialSenderRoleEntry e FinalReceiverRoleEntry. InitialSenderRoleEntry produce il primo messaggio nella pipeline, tramite i metodi OnStart (per connettersi alla coda) e Run (per pubblicare messaggi su di essa). La classe FinalReceiverRoleEntry è la destinazione finale dei messaggi:

public override bool OnStart()
        { 
            ServicePointManager.DefaultConnectionLimit = 12;
            this.queueFinal = new ServiceBusPipeFilter(Settings.ServiceBusConnectionString,
Constants.QueueFinalPath);
            this.queueFinal.Start();
            return base.OnStart(); 
        }

public override void Run()
        { 
            this.queueFinal.OnPipeFilterMessageAsync(
                async (msg) =>
                { 
                    await Task.Delay(500);
                    Trace.TraceInformation(
                        "Pipeline Message Complete - FilterA:{0} FilterB:{1}",
                        msg.ApplicationProperties[Constants.FilterAMessageKey],
                        msg.ApplicationProperties[Constants.FilterBMessageKey]);
                    return null; 
                }); 

            this.stopRunningEvent.WaitOne(); 
        }

Anche le ultime due classi presentate sono implementare come WorkerRole, quindi beneficiano di tutti i vantaggi circa scalabilità e possibile parallelismo visti per PipeFilterARoleEntry e PipeFilterBRoleEntry.

Fonti e Bibliografia

Questo articolo è basato su “Luca Cabibbo. 2021. Architettura del software – strutture e qualità. Edizioni Efesto”.

Il pattern Pipes and Filters è stato trattato in “Frank Buschmann, Regine Meunier, Hans Rohnert, Peter Sommerlad, and Michael Stal. 2008. Pattern-oriented software architecture: A system of patterns. John Wiley & Sons.”, da cui sono stati tratti i diagrammi di sequenza.

L’implementazione di Pipes & filters proposta da Microsoft è reperibile all’indirizzo https://docs.microsoft.com/en-us/azure/architecture/patterns/pipes-and-filters.