在Qt线程之间发送大量数据

Question

我有一个QThread定期生成相当大量的数据(每秒几兆字节),并且需要将其传输到父(GUI)线程.

我担心我在内部工作中并不确定,QThread所以我想要求最佳实践.

显然,传输数据的最直接方式是仅emit使用数组.但是,这有多高效？Qt是否知道它的使用位置并避免在发送和接收时对其进行深度复制？

如果没有,我很乐意只在主线程中分配内存并给出一个指向子线程的指针,它将写入数据(并且只有emit关于进度的短消息).这对我来说似乎不是最优雅的解决方案,这就是我要问的原因.

如果Qt避免在发送和接收时复制多个缓冲区中的数据,是否可以在所有系统中保证？我没有资源在各种操作系统下尝试对其进行基准测试.

Answer 1

QThread内部运作是无关紧要的:它们在事件循环的工作方式中没有任何作用.当您emit的信号位于QObject与插槽对象不同的线程中时,该信号将作为a发布QMetaCallEvent到接收线程的事件队列中.然后,在接收线程中运行的事件循环将对此事件起作用,并将调用执行到连接到发出信号的插槽中.

因此,无论发生什么,通过信号发送的任何数据最终都将作为QEvent派生类实例中的有效负载结束.

问题的关键是当QMetaCallEvent到达事件循环并且容器作为参数传递到槽中时.当然,复制构造函数可以在此过程中被多次调用.下面是一些简单的代码,演示了实际调用复制构造函数和默认构造函数的次数

• 关于隐式共享的写时复制容器(QVector)的数据成员的元素,

• 在代表容器的自定义类上.

你会惊喜:)

由于Qt容器是隐式共享的写时复制,因此它们的复制结构具有可忽略的成本:所有这些都是在构造时以原子方式递增的引用计数器.例如,没有复制任何数据成员.

唉,11之前的C++显示了它丑陋的一面:如果插槽代码以任何方式修改容器,就没有办法以一种让编译器知道不再需要原始容器的方式传递对插槽的引用.因此:如果插槽接收到容器的const引用,则保证不会生成副本.如果插槽接收到容器的可写副本并且您对其进行了修改,则会产生完全不必要的副本,因为不再需要在调用站点处存活的实例.在C++ - 11中,您将传递一个右值引用作为参数.在函数调用中传递右值引用会终止调用者中传递的对象的生命周期.

示例代码输出:

"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

在传递大缓冲区时,它对生产者线程中的new()缓冲区对象来说是"传统的",并且在加载时,队列/发出/任何*缓冲区到消费者线程,并立即new()另一个,(进入相同的*) buffer var),用于下一次加载数据.

问题:如果你的GUI线程无法跟上,除非你采取一些流量控制措施(例如预先分配一个*缓冲池和'循环'它们),否则你将失去内存失控.

我通常做的是在循环中预先分配一些缓冲区实例(在大型服务器中高达数千个),并将它们的实例推送到生产者 - 消费者"池队列".如果子线程想要将数据从某个网络连接加载到缓冲区,它必须从池中弹出一个并加载它.然后,它可以将任何缓冲区排队/发出/放入消费者线程,并为可能进入的任何其他数据弹出另一个缓冲区.使用者线程获取缓冲区,处理数据并将"已使用"缓冲区推回到池队列中再利用.这提供了流控制:如果子线程加载缓冲区的速度比消费者线程可以处理的速度快,它会发现池为空并阻塞它,直到使用者线程返回一些使用过的缓冲区,因此限制缓冲区/内存使用,(以及避免持续的新/处置,或GC支持它的那些语言).

我喜欢在1秒计时器上将池队列计数转储到GUI状态栏 - 这允许我观察缓冲区的使用,(并且如果有任何泄漏,则快速发现:).