Invio di grandi quantità di dati tra thread Qt

Question

Ho uno QThread che genera una quantità abbastanza grande di dati regolarmente (paio di megabyte al secondo) e deve trasmetterlo al thread padre (GUI).

Ho paura di non essere così sicuro nel funzionamento interno di QThread, quindi mi piacerebbe chiedere una buona pratica.

Ovviamente, il modo più diretto per trasmettere i dati è semplicemente un array emit. Tuttavia, quanto è efficiente questo? Qt sa dove viene utilizzato ed evita di copiarlo in profondità durante l'invio e la ricezione?

In caso contrario, posso semplicemente allocare la memoria nel thread principale e dare un puntatore al thread figlio in cui scriverà i dati (e solo emit messaggi brevi sullo stato di avanzamento). Questa non sembra la soluzione più elegante per me, è per questo che sto chiedendo.

Se Qt evita di copiare i dati in più buffer durante l'emissione e la ricezione, è garantito in tutti i sistemi? Non ho le risorse per provare il benchmarking sotto vari sistemi operativi.

Answer 1

QThread Le operazioni interne sono irrilevanti: non svolgono alcun ruolo nel modo in cui i loop degli eventi funzionano. Quando si emit un segnale in un QObject che vive in un thread diverso dall'oggetto dello slot, il segnale verrà registrato come QMetaCallEvent nella coda eventi del thread ricevente. Il ciclo degli eventi in esecuzione nel thread ricevente agirà quindi su questo evento ed eseguirà la chiamata nello slot collegato al segnale emesso.

Quindi, qualunque cosa accada, qualunque dato inviato attraverso il segnale finirà per essere un carico utile in un'istanza di classe derivata da QEvent.

La carne del problema è quando lo QMetaCallEvent raggiunge il ciclo degli eventi e il contenitore viene passato nello slot come argomento. Naturalmente i costruttori di copie potrebbero essere chiamati un sacco di volte lungo la strada.Di seguito alcune semplice codice che dimostra quante volte il costruttore di copia e costruttore di default sono infatti chiamati

• sugli elementi dei dati membro di un contenitore implicitamente condiviso copy-on-write (QVector),

• in una classe personalizzata che sostituisce un contenitore.

Sarete piacevolmente sorpresi :)

Dal contenitori Qt sono implicitamente copy-on-write condivisi, la loro copia costruzione ha costi trascurabili: tutto ciò che è fatto è un contatore di riferimento viene incrementato in modo atomico sulla costruzione . Ad esempio, nessuno dei membri dei dati viene copiato.

Alas, pre-11 C++ mostra il suo lato brutto: se il codice dello slot modifica il contenitore in qualche modo, non c'è modo di passare i riferimenti allo slot in modo tale da far sapere al compilatore che il contenitore originale è non serve più. Quindi: se lo slot riceve un riferimento const al container, hai la certezza che non verranno effettuate copie. Se lo slot riceve una copia scrivibile del contenitore e lo si modifica, verrà eseguita una copia completamente non necessaria poiché l'istanza in vita nel sito di chiamata non è più necessaria. In C++ - 11 si passa un riferimento rvalue come parametro. Il passaggio di un riferimento di rvalue in una chiamata di funzione termina la durata dell'oggetto passato nel chiamante.

Esempio uscita Codice:

"Started" copies: 0 assignments: 0 default instances: 0 
"Created Foo" copies: 0 assignments: 0 default instances: 100 
"Created Bar" copies: 0 assignments: 0 default instances: 100 
"Received signal w/const container" copies: 0 assignments: 0 default instances: 100 
"Received signal w/copy of the container" copies: 0 assignments: 0 default instances: 100 
"Made a copy" copies: 100 assignments: 1 default instances: 101 
"Reset" copies: 0 assignments: 0 default instances: 0 
"Received signal w/const class" copies: 2 assignments: 0 default instances: 1 
"Received signal w/copy of the class" copies: 3 assignments: 0 default instances: 1 

//main.cpp 
#include <QtCore> 

class Class { 
    static QAtomicInt m_copies; 
    static QAtomicInt m_assignments; 
    static QAtomicInt m_instances; 
public: 
    Class() { m_instances.fetchAndAddOrdered(1); } 
    Class(const Class &) { m_copies.fetchAndAddOrdered(1); } 
    Class & operator=(const Class &) { m_assignments.fetchAndAddOrdered(1); return *this; } 
    static void dump(const QString & s = QString()) { 
     qDebug() << s << "copies:" << m_copies << "assignments:" << m_assignments << "default instances:" << m_instances; 
    } 
    static void reset() { 
     m_copies = 0; 
     m_assignments = 0; 
     m_instances = 0; 
    } 
}; 

QAtomicInt Class::m_instances; 
QAtomicInt Class::m_copies; 
QAtomicInt Class::m_assignments; 

typedef QVector<Class> Vector; 

Q_DECLARE_METATYPE(Vector) 

class Foo : public QObject 
{ 
    Q_OBJECT 
    Vector v; 
public: 
    Foo() : v(100) {} 
signals: 
    void containerSignal(const Vector &); 
    void classSignal(const Class &); 
public slots: 
    void sendContainer() { emit containerSignal(v); } 
    void sendClass() { emit classSignal(Class()); } 
}; 

class Bar : public QObject 
{ 
    Q_OBJECT 
public: 
    Bar() {} 
signals: 
    void containerDone(); 
    void classDone(); 
public slots: 
    void containerSlotConst(const Vector &) { 
     Class::dump("Received signal w/const container"); 
    } 
    void containerSlot(Vector v) { 
     Class::dump("Received signal w/copy of the container"); 
     v[99] = Class(); 
     Class::dump("Made a copy"); 
     Class::reset(); 
     Class::dump("Reset"); 
     emit containerDone(); 
    } 
    void classSlotConst(const Class &) { 
     Class::dump("Received signal w/const class"); 
    } 
    void classSlot(Class) { 
     Class::dump("Received signal w/copy of the class"); 
     emit classDone(); 
     //QThread::currentThread()->quit(); 
    } 
}; 

int main(int argc, char ** argv) 
{ 
    QCoreApplication a(argc, argv); 
    qRegisterMetaType<Vector>("Vector"); 
    qRegisterMetaType<Class>("Class"); 

    Class::dump("Started"); 
    QThread thread; 
    Foo foo; 
    Bar bar; 
    Class::dump("Created Foo"); 
    bar.moveToThread(&thread); 
    Class::dump("Created Bar"); 
    QObject::connect(&thread, SIGNAL(started()), &foo, SLOT(sendContainer())); 
    QObject::connect(&foo, SIGNAL(containerSignal(Vector)), &bar, SLOT(containerSlotConst(Vector))); 
    QObject::connect(&foo, SIGNAL(containerSignal(Vector)), &bar, SLOT(containerSlot(Vector))); 
    QObject::connect(&bar, SIGNAL(containerDone()), &foo, SLOT(sendClass())); 
    QObject::connect(&foo, SIGNAL(classSignal(Class)), &bar, SLOT(classSlotConst(Class))); 
    QObject::connect(&foo, SIGNAL(classSignal(Class)), &bar, SLOT(classSlot(Class))); 
    QObject::connect(&bar, SIGNAL(classDone()), &thread, SLOT(quit())); 
    QObject::connect(&thread, SIGNAL(finished()), &a, SLOT(quit())); 
    thread.start(); 
    a.exec(); 
    thread.wait(); 
} 

#include "main.moc" 

Answer 2

Quando si comunicano buffer di grandi dimensioni, è "tradizionale" a new() oggetti buffer nel thread di produzione e, quando caricato, accoda/emette/qualunque sia il buffer * sul thread di consumo e immediatamente new() un altro, (nello stesso * buffer var), per il prossimo carico di dati.

Problema: se il thread della GUI non riesce a tenere il passo, si otterrà la fuga di memoria a meno che non si prenda qualche misura di controllo del flusso (ad esempio, pre-allocare un pool di * buffer e "farli circolare").

Quello che faccio di solito è pre-allocare alcune istanze di buffer in un ciclo, (fino a migliaia in un grande server), e spingere le loro istanze su una "coda di biliardo" produttore-consumatore. Se un thread figlio desidera caricare dati da una connessione di rete in un buffer, deve estrarne uno dal pool e caricarlo. Può quindi accodare/emettere/qualunque sia il buffer di un thread di consumo e inserire un altro buffer per altri dati che potrebbero entrare. Il thread di consumo ottiene il bufer, elabora i dati e spinge il buffer 'used' nella coda del pool per riutilizzo. Ciò fornisce il controllo del flusso: se il thread secondario carica i buffer più velocemente di quanto il thread consumer può elaborarli, troverà il pool vuoto e lo bloccherà fino a quando il thread consumer restituirà alcuni buffer usati, quindi limitando l'utilizzo di buffer/memoria (e anche evitando il continuo nuovo/smaltimento, o GC in quelle lingue che lo supportano).

Mi piace scaricare il conteggio della coda del pool su una barra di stato della GUI su un timer di 1 secondo: questo mi consente di osservare l'utilizzo del buffer (e individuare rapidamente eventuali perdite :).