Miroscic CS1

Coding Session n.1

Paolo Bosetti

Università di Trento, Dipartimento di Ingegneria Industriale

Architettura Miroscic

A che punto ci siamo lasciati

ZeroMQ

Connessioni TCP

Le connessioni TCP sono sempre asimmetriche:
- 1 server (bind())
- N client (connect())

flowchart TD
  C1([Client]) --> S[Server]
  C2([Client]) --> S
  C3([Client]) --> S
  
  S -.-> C1
  S -.-> C2
  S -.-> C3

ZeroMQ PUB-SUB

In ZeroMQ l’architettura PUB-SUB è simile:
- 1 publisher (bind() oppure connect())
- N subscriber (connect() oppure bind())

flowchart TD
  P[Publisher]
  P --> C1([Subscriber]) 
  P --> C2([Subscriber]) 
  P --> C3([Subscriber])

ZeroMQ XPUB-XSUB

Fortunatamente, in ZeroMQ è indifferente chi fa il bind()

Ciò consente di risolvere il problema degli indirizzi dei publisher introducendo un broker

Solo il broker fa il bind(): è l’unico indirizzo che serve conoscere
I subscriber sono tutti client (connect())
I publisher sono tutti client (connect())
Ovviamente un singolo processo può essere sia publisher che subscriber (cioè invia e riceve)

flowchart LR
  P1([Publisher]) --> B[Broker]
  P2([Publisher]) --> B
  P3([Publisher]) --> B
  P4([Publisher]) --> B
  
  B --> S1([Subscriber])
  B --> S2([Subscriber])
  B --> S3([Subscriber])
  B --> S4([Subscriber])

Struttura della rete

La rete Miroscic:

flowchart LR
  A([Source agent]) --> Br[Broker]
  B([Source agent]) --> Br
  C([Source agent]) --> Br
  
  Br --> D([Filter agent])
  D --> Br
  Br --> E([Filter agent])
  E --> Br
  Br --> F([Filter agent])
  F --> Br
  
  Br --> S([Sink agent])
  S ==BSON==> Db[(Database)]
  style S fill:#f9f

Tre tipi di agenti:

Source: un agente che produce dati, raccogliendoli dal campo (GPIO, camere, sensori, …)
Filter: un agente che riceve dati dalla rete, li elabora e li restituisce alla rete
Sink: un agente che consuma dati dalla rete e li indirizza altrove (disco, database, GUI, ecc.)

In linea di principio, ogni agente può essere sviluppato indipendentemente, purché usi il protocollo ZeroMQ e le stesse convenzioni di serializzazione dei dati. Tuttavia, per robustezza, è meglio che gli agenti condividano quantomeno lo stesso codice che si occupa della comunicazione

Time sequence

sequenceDiagram
  actor Source
  participant Broker
  actor Filter
  actor Sink
  par Initialization
    Source ->> Broker: Settings please?
    Broker -->> Source: Settings!
  and
    Filter -->> Broker: Settings, please?
    Broker -->> Filter: Settings!
  end
  loop Anynchronous
    Source -->> Broker: Data
    Broker -->> Filter: Data
    activate Filter
    Filter -->> Broker: Data
    deactivate Filter
  end
  Broker -->> Sink: Data

Perché

Qualcuno ha detto:

Ma perché non facciamo un’unica applicazione grafica che si occupa di fare tutto, in modo da riciclare il più possibile quanto fatto in passato?

Le motivazioni

Rispetto ad un’applicazione monolitica multithread la architettura ad agenti multiprocesso ha questi vantaggi:

È scalabile: è immediato aggiungere un agente senza modificare il resto
È flessibile: diversi gruppi possono sviluppare indipendentemente diversi agenti (purché condividano la classe base)
È leggera: è possibile realizzare agenti compatti adatti a funzionare su edge-device
È efficiente: se serve i diversi processi possono essere messi su macchine diverse e non lottano per le stesse risorse (CPU, disco, rete). È più semplice evitare race condition
È automatizzabile: ogni agente può essere un demone che parte al boot; basta un’unica dashboard per comandarli tutti

GUI

Dashboard di controllo (on-line): è un agente
Applicazione di replay (off-line): si interfaccia solo con il database (non è un agente)

Agenti

Classe `Miroscic::Agent`

si connette al broker
ottiene i propri parametri
definisce i topic
riceve informazioni dalla rete
fornisce informazioni alla rete

Nota:

La comunicazione avviene in JSON compresso con Snappy
I parametri sono codificati in file con standard TOML
Agenti specifici devono ereditare la classe base

Sink

Al momento c’è un unico agente di tipo sink che si occupa della registrazione di tutto il traffico che passa per il broker su database MongoDB

MongoDB è un database non-relazionale (NoSQL) che organizza i dati gerarchicamente

flowchart RL
  D1[JSON Document] --> C1[Collection]
  D2[JSON Document] --> C1
  D3[JSON Document] --> C1
  D4[JSON Document] --> C2[Collection]
  D5[JSON Document] --> C2[Collection]
  C1 --> DB[Database]
  C2 --> DB[Database]
  DB --> S[Server]

Ogni documento, o record, è un oggetto JSON in struttura libera
Uno stesso server può avere uno o più database; ogni database contiene una o più collection
In Miroscic, ogni collection contiene tutti i messaggi con lo stesso topic

Nuovi sviluppi

Novità sul codice dall’ultima riunione

Obiettivi

semplificare la gestione degli eseguibili (runtime), in particolare il riavvio in modo da ricaricare le impostazioni quando queste cambino sulla macchina broker
consentire l’organizzazione degli esperimenti dimostrativi, soprattutto marcando istanti particolari o l’inizio e la fine di un test
semplificare lo sviluppo di nuovi agenti: nel caso in cui le dipendenze aumentino in numero può risultare difficile seguire lo stesso approccio finora utilizzato
proporre una soluzione per la sincronizzazione dei documenti, che sono generati in maniera asincrona da ogni agente

Gestione degli eseguibili

Tutti gli agenti ascoltano sempre su un topic dedicato, control, sul quale circolano comandi di controllo remoto

Questo topic consente, al momento, di comandare l’arresto o il riavvio di tutti gli agenti attivi

Al momento ci sono due modi per inviare questi comandi:

mediante lo script scripts/command, che internamente fa una chiamata all’agente bridge
mediante la GUI dedicata alla gestione dei metadati, sviluppata in QT6 e che si trova in GUI/MiroscicMetadata

Gestione degli eseguibili

Per supportare questa funzione la classe Agent dispone di due nuovi metodi:

Agent::enable_remote_control(): va chiamato subito prima di Agent::connect(); se l’agente è un puro publisher, è sufficiente questo
Agent::remote_control(): va chiamato all’interno del loop principale — solo per gli agenti che sono anche subscriber

Organizzazione dei test: GUI Dashboard

La GUI in QT6 ha al momento queste funzioni:

consente il controllo remoto degli altri agenti
consente l’inserimento di metadati
consente la marcatura:
- mark in: inizia un test
- bare mark: inserisce un marker nudo, cioè senza dati associati
- mark out: segnala la fine del test

Sviluppo di nuovi agenti: plugin

Per semplificare lo sviluppo di nuovi agenti ho introdotto un’architettura a plugin

Un plugin è una libreria dinamica che viene caricata runtime dall’eseguibile principale

Può essere sviluppato e compilato in maniera completamente separata, ma deve fornire un’interfaccia ABI standard, cioè deve esporre dei simboli (C o C++) prefissati, a cui l’eseguibile principale si collega quando carica il plugin

Generalmente la complessità di garantire questa interfaccia scoraggia questo approccio, ma la libreria pugg lo rende in realità molto semplice

Penalità: solo un trascurabile ritardo al primo caricamento della libreria (al lancio)

Potenzialità: i plugin potrebbero essere distribuiti direttamente dal broker assieme alle impostazioni, consentendone l’aggiornamento remoto!

Pugg

Caricare runtime una libreria richiede due funzioni C: dlopen() e dlsym()

dlopen() apre il file corrispondente e prepara le operazioni successive
dlsym() importa ciascun simbolo (funzione) che si desidera utilizzare; va usato una volta per ogni funzione

A causa del name mangling C++ è possibile caricare solo funzioni extern(C). Queste ultime però possono ritornare puntatori void * di cui è possibile fare il cast a classi C++ note all’eseguibile

La libreria pugg sfrutta questo meccanismo: definisce alcune classi base (destinazioni del cast) che devono essere note all’eseguibile. I plugin possono implementare classi derivate che però hanno sempre la stessa interfaccia (metodi base)

Pugg — classi base

È possibile definire tre tipi di classi base:

Source: produce informazioni ottenendole da periferiche di campo
Filter: opera su informaizoni ricevute in input per produrre un output
Sink: consuma informazioni e le smaltisce internamente (file, DB, etc.)

In maniera simmetrica, nel namespace Miroscic ci sono altrettanti agenti generici di tipo source, filter e sink

Ogni agente generico carica solo plugin di tipo corrispondente. In che formato si scambiano i dati di I/O tra eseguibile e plugin?

Si usano istanze della classe nlohman/json, la stessa usata per l’I/O a livello di network Miroscic

Pugg — esempio classe base `Filter`

template <typename Tin = vector<double>, typename Tout = vector<double>>
class Filter {
public:
  static const int version = 1;
  static const std::string server_name() { return "FilterServer"; }
  Filter() : _error("No error"), dummy(false) {}
  virtual ~Filter() {}
  virtual std::string kind() = 0;
  virtual return_type load_data(Tin &data) = 0; // Tin può essere nlohmann::json
  virtual return_type process(Tout *out) = 0;   // Tout può essere nlohmann::json
  virtual void set_params(void *params){};      // params può essere nlohmann::json (cast) 
  
  std::string error() { return _error; }
  bool dummy;

private:
  std::string _error;
};

Pugg — classi derivate

Lo sviluppatore di plugin deve:

ereditare da una delle tre classi base in un progetto separato
fare l’override dei metodi dichiarati come pure virtual
definire quanti altri metodi siano utili per il caso
compilare e distribuire il plugin

Nota: l’eseguibile principale non conosce le clasi derivate, quindi può interagire con il plugin SOLO mediante i metodi dichiarati nelle classi base.

Queste operazioni sono predisposte nel progetto github.com/miroscic/plugin_cpp

Il CMakeLists.txt fornito con il progetto consente di compilare, per ogni plugin, anche una versione direttamente eseguibile, utile per il debugging e lo sviluppo

Plugin — riassumendo

Per sviluppare un plugin si parte con un fork del progetto github.com/miroscic/plugin_cpp

Lo sviluppo può essere condotto in maniera assolutamente isolata dal resto di Miroscic, utilizzando la versione eseguibile del plugin per il testing.

Il nuovo progetto può utilizzare qualsiasi libreria necessaria. Se le librerie sono statiche, il plugin risultante sarà autosufficiente; se sono dinamiche, esse dovranno essere installate anche sulla macchina di destinazione

In Miroscic, gli agenti source e filter richiedono il file plugin come argomento e usano il nome del plugin come chiave per identificare topic e sezione del file INI.

Le variabili statiche Filter.version e Source.version vanno incrementate ogni volta che si cambia l’interfaccia. In questo modo plugin non compatibili segnalano l’errore al caricamento

Plugin — sviluppo futuro

Se necessario, sarà possibile modificare il broker in modo che quando un agente all’avvio richiede il file di impostazioni riceva in risposta, se è il caso, anche il file plugin.

Il broker dovrà mantenere un folder con tutti i plugin necessari, per le varie architetture

In questo modo, l’aggiornamento degli agenti può essere fatto direttamente e solo sul broker, comandando poi un riavvio di tutti gli agenti mediante GUI

Sincronizzazione dei documenti

Ogni agente genera in maniera asincrona dei documenti JSON che vengono raccolti dall’agente logger e salvati sul database MongoDB
Ogni documento contiene un timestamp in formato ISO8601 con risoluzione del millisecondo, ad es. 2024-02-14T09:52:31.453+00:00
Il database è organizzato in collezioni, una per ogni topic
In ogni collezione possiamo avere sequenze di documenti con timestamp regolari o sporadici, ma in ogni caso non sincronizzati con le altre collezioni
In ogni collezione possono alternarsi documenti con lo stesso timestamp ma contenuto diverso (complementare)
Fortunatamente la dinamica dei fenomeni è lenta (10 Hz)
Per la ricostruzione della scena e l’analisi ex-post abbiamo bisogno di una frequenza di campionamento sufficiente a una riproduzione a 25 fps

Sincronizzazione dei documenti

Proposta:

Operiamo un binning dei documenti sulla base dei tempi, raccogliendo tutti i documenti generati con timestamp in intervalli di 1/25 s (40 ms)

Per semplificare, ogni documento viene marcato, oltre che con il timestamp, con un campo timecode

Il timecode è il numero di secondi trascorsi dalla mezzanotte con una risoluzione di 40 ms. Tutti i documenti con lo stesso timecode appartengono allo stesso frame e possono essere quindi accorpati

Il meccanismo di aggregation di MongoDB consente di creare una nuova collezione dinamica read-only (in realtime!) che contiene i documenti aggregati utilizzando timecode come chiave di aggregazione

Nella repository di Miroscic, la cartella analysis contiene un notebook Python db_access.ipynb che mostra come aggregare i documenti della piattaforma di forza con quelli dell’RFID

Programma

Il programma della sessione è il seguente:

Giorno 1: setup del sistema di sviluppo; familiarizzazione con la struttura del progetto; compilazione e opzioni CMake; esecuzione del sistema multi-agente e descrizione dei meccanismo di logging su MongoDB; rudimenti di MongoDB
Giorno 2: sviluppo di un plugin
Giorno 3: abbozziamo uno degli agenti (dummy)

Per lo sviluppo del plugin partiremo da un esempio di human pose estimation che utilizza il toolkit OpenVINO per fare inferenza mediante un modello OpenPOSE

HPEQuick

HPEQuick è un esempio semplificato estratto da OpenVINO model zoo
Usa a un modello di inferenza OpenPOSE
Sarà adattato a funzionare come plugin per un agente Miroscic di tipo source
Prestazioni:
- Su M1 elabora un frame ogni 25 ms (CPU)
- Su NVIDIA TX2 in GPU dovrebbe avere prestazioni simili o superiori

Miroscic CS1

Architettura Miroscic

ZeroMQ

Connessioni TCP

ZeroMQ PUB-SUB

ZeroMQ XPUB-XSUB

Struttura della rete

La rete Miroscic:

Tre tipi di agenti:

Time sequence

Perché

Le motivazioni

GUI

Agenti

Classe Miroscic::Agent

Nota:

Sink

Nuovi sviluppi

Obiettivi

Gestione degli eseguibili

Gestione degli eseguibili

Organizzazione dei test: GUI Dashboard

Sviluppo di nuovi agenti: plugin

Pugg

Pugg — classi base

Pugg — esempio classe base Filter

Pugg — classi derivate

Plugin — riassumendo

Plugin — sviluppo futuro

Sincronizzazione dei documenti

Sincronizzazione dei documenti

Proposta:

Programma

HPEQuick

Classe `Miroscic::Agent`

Pugg — esempio classe base `Filter`