Schedulers in Rxcpp

Question

Sto cercando di capire il modello di schedulazione nel C++ version di Rx.

Conoscere la versione C# in cui esiste un'interfaccia semplice con un metodo Schedule; La versione C++ sembra piuttosto complessa, con cose come scheduler, worker e coordinazione.

Uno dei pezzi mancanti per me è un'implementazione di un programma di pianificazione del pool di thread, esiste con qualche altro nome? Come potrei implementarlo da solo? Devo scriverlo sopra PPL (Windows)? Se ho bisogno di un osservatore serializzato (Actor like) sopra di esso, cosa dovrei usare? Peeking here e here può mostrare che questo non è un compito banale.

Sarebbe davvero utile ottenere una sorta di panoramica sull'argomento, dal momento che la documentazione di official viene generata automaticamente e ancora molto sparsa.

Answer 1

Sì, i documenti generati sono nuovi e la pianificazione non è ancora documentata.

Lo scheduler in rxcpp v2 è basato sul scheduler e lavoratore costrutti che RxJava utilizza (Eric Meijer è stato coinvolto) la documentazione per RxJava avranno una spiegazione per scheduler e l'operaio. rxcpp aggiunge pianificazione, coordinazione e coordinatore.

scheduler possiede una sequenza temporale esposta dal metodo now(). scheduler è anche una fabbrica per worker s in quella sequenza temporale. poiché uno schedulatore possiede una timeline è possibile costruire scheduler che viaggiano nel tempo. il virtual-scheduler è una base per il test-scheduler che usa questo per effettuare test di più secondi completi in ms.

worker possiede una coda di attesa schedulable s per la cronologia e ha una durata. quando viene raggiunto il tempo di schedulable viene eseguito il numero schedulable. La coda mantiene l'ordine di inserimento in modo tale che quando N schedulable s abbia la stessa ora di destinazione vengono eseguiti nell'ordine in cui sono stati inseriti nella coda. Lo worker garantisce che ogni schedulable venga completato prima dell'avvio del successivo schedulable. quando la vita di worker viene annullata, tutti i messaggi in sospeso schedulable vengono scartati.

schedulable possiede una funzione e ha un lavoratore e una durata. quando la vita di schedulable è annullata, la funzione schedulable non verrà chiamata. il schedulable viene passato alla funzione e consente alla funzione di riprogrammare se stessa o pianificare qualcos'altro sullo stesso operatore.

I nuovi concetti sono coordinazione e coordinatore. Ho aggiunto questi per semplificare le implementazioni degli operatori e introdurre il pay-for-use nelle implementazioni degli operatori. Nello specifico, in Rx.NET e RxJava, gli operatori utilizzano operazioni atomiche e primitive di sincronizzazione per coordinare i messaggi da più flussi anche quando tutti gli stream si trovano sullo stesso thread (come gli eventi dell'interfaccia utente). I coordinamenti identity_. . . in rxcpp vengono utilizzati per impostazione predefinita e non hanno un sovraccarico. I coordinamenti syncronize_. . . e observe_on_. . . utilizzano rispettivamente mutex e queue-on-a-worker per interlacciare più flussi in modo sicuro.

coordination è una fabbrica per coordinator s e ha un scheduler.

coordinator ha un worker, ed è una fabbrica di coordinate observable s, subscriber s e schedulable funzioni.

Tutti gli operatori che eseguono più flussi o gestiscono in tempo (anche subscribe_on e observ_on) utilizzano un parametro di coordinamento, non uno scheduler.

Ecco alcune funzioni fornite che produrranno un coordinamento utilizzando un particolare programmatore.

• identity_immediate()
• identity_current_thread()
• identity_same_worker (w lavoratore)
• serialize_event_loop()
• serialize_new_thread()
• serialize_same_worker (lavoratore w)
• observe_on_event_loop()
• observ_on_new_thread()

Non esiste ancora un programmatore di pool di thread. Un programma di pianificazione del pool di thread richiede l'assunzione di una dipendenza da un'implementazione del pool di thread poiché non desidero scrivere un pool di thread. Il mio piano è di creare uno scheduler per il pool di thread di Windows e il pool di thread Apple e il pool di boost asio executor. Una domanda a cui rispondere è se questi costrutti specifici della piattaforma debbano vivere nel repository rxcpp o avere repository specifici della piattaforma.

Contributi, opinioni e idee sono benvenuti!