Passing data between threads.

jenya7

I have a class

class MyUDP : public QObject
{
    Q_OBJECT

    public:
    explicit MyUDP(QObject *parent = nullptr);
    void Send(QString data, QString remote_ip, quint16 remote_port);
    void Send(QByteArray data, QString remote_ip, quint16 remote_port);
    uint8_t Start(QString ip, quint16 port);

    QByteArray udp_buffer;

    signals:
    void GetUdpMessage(const SENS_NET_PARAM& param, const QByteArray data);

    public slots:
    void ReadyRead();
    void SocketError(QAbstractSocket::SocketError socketError);
    void StateChanged(QAbstractSocket::SocketState socketState);
    private:
};

Here I get a message

void MyUDP::ReadyRead()
{
    // when data comes in
    int size = static_cast<int>(socket->pendingDatagramSize());
    udp_buffer.resize(size);

    socket->readDatagram(udp_buffer.data(), udp_buffer.size(),
                         &sender, &senderPort);
    
        //udp_buffer add to queue 
}

And I want to put the message udp_buffer.data() into a queue (list?). And other thread have to pull the messages from the queue. For instance

//in main.cpp
#include "udp.h"

MyUDP udp;

int main(int argc, char *argv[])
{
    QCoreApplication a(argc, argv);

    //load parameters
    m_sys.ParamInit(nullptr);
    udp.Start(sys_params.local_ip, sys_params.local_port);

    QFuture<void> future = QtConcurrent::run(&Listener);

    return a.exec();
}

void Listener()
{
    while(1)
    {
         if (!queue.isEmpty())
                queue.dequeue() ;
    }
}

How can I do it? Can it be just a global queue with enqueue/dequeue or I'll get a cross thread exception?

jsulm

@jenya7 said in Passing data between threads.:

How can I do it?

I would rather emit a signal from the thread which gets the data. This signal would have a parameter containing the message data. In the other thread I would have a slot connected to that signal and this other thread would put the message data into a queue. This way you avoid accessing same variable from two threads which requires synchronisation (mutex).

KroMignon

@jenya7 said in Passing data between threads.:

How can I do it? Can it be just a global queue with enqueue/dequeue or I'll get a cross thread exception?

First questions I have to you:
a. why do you think you need to use an additional thread?
b. why not simply use signals/slots mechanism to handle message queue?

jenya7

@KroMignon said in Passing data between threads.:

@jenya7 said in Passing data between threads.:

How can I do it? Can it be just a global queue with enqueue/dequeue or I'll get a cross thread exception?

First questions I have to you:
a. why do you think you need to use an additional thread?
b. why not simply use signals/slots mechanism to handle message queue?

It'll be 3 sockets getting messages at very high rate (streaming video) and putting it into a queue. So I thought it would be better to have a separate thread to pull and parse messages. I have 4 cores so I hope for some parallelism.

jsulm

@jenya7 I would not invent threads without knowing whether I really need them. Start without threads and see how it works. Only if needed add threads.

jenya7

@jsulm said in Passing data between threads.:

@jenya7 I would not invent threads without knowing whether I really need them. Start without threads and see how it works. Only if needed add threads.

Ok. I'll try.

KroMignon

@jenya7 said in Passing data between threads.:

It'll be 3 sockets getting messages at very high rate (streaming video) and putting it into a queue. So I thought it would be better to have a separate thread to pull and parse messages. I have 4 cores so I hope for some parallelism.

As @jsulm already wrote, I would also suggest you first to try it out without using additional thread.
And second, do not use QConcurrent::run() to create a thread, because this thread will not contain an event loop and then you cannot use signals/slots!

To use additional thread, prefer something like this:

auto reader = new MyUDP();
auto thread = new QThread();
// on thread start, start UDP socket ==> create all QObject there to ensure they are working in the right thread!!
connect(thread, &QThread::started, reader, &MyUDP::start);
// on reader end, stop working thread
connect(reader, &MyUDP::aboutToQuit, thread, &QThread::quit);
// on working thread end, delete all instances (cleanup)
connect(thread, &QThread::finished, thread, &QObject::deleteLater);
connect(thread, &QThread::finished, reader, &QObject::deleteLater);

// start working thread
thread->start();

jenya7

@KroMignon
Thank you. And how do I pass a queue between a reader and a consumer (parser) ?

jsulm

@jenya7 said in Passing data between threads.:

And how do I pass a queue between a reader and a consumer (parser) ?

I suggested to not to pass/share the queue. Instead pass the actual data and manage the queue only where you need it.

KroMignon

@jenya7 said in Passing data between threads.:

Thank you. And how do I pass a queue between a reader and a consumer (parser) ?

I don't understand why you think you need to use an additional queue?
Messages are already queued by the UDP socket, why not processing them on reception?

jenya7

I see. Just can't get rid of embedded programming concepts.

jenya7

Just to clarify. If I instantiate 3 sockets.

static MyUDP udp_1;
static MyUDP udp_2;
static MyUDP udp_3;

int main(int argc, char *argv[])
{
    QCoreApplication a(argc, argv);

    udp_1.Start("192.176.0.1", 8001);
    udp_2.Start("192.176.0.2", 8002);
    udp_3.Start("192.176.0.3", 8003);

    return a.exec();
}

How it gets messages simultaneously in one thread?

jsulm

@jenya7 said in Passing data between threads.:

How it gets messages simultaneously in one thread?

Please read https://doc.qt.io/qt-5/qtnetwork-index.html and https://doc.qt.io/qt-5/signalsandslots.html
Qt is an asynchronous framework.

jenya7

@jsulm said in Passing data between threads.:

@jenya7 said in Passing data between threads.:

How it gets messages simultaneously in one thread?

Please read https://doc.qt.io/qt-5/qtnetwork-index.html and https://doc.qt.io/qt-5/signalsandslots.html
Qt is an asynchronous framework.

I see. Looks like Qt takes care 90% of stuff I have to worry in embedded programming. :)

KroMignon

@jenya7 said in Passing data between threads.:

Just to clarify. If I instantiate 3 sockets.

Please, be kind and never do something like this!!!

Never create global static instances of QObject base classes!!!
This is not supported be Qt.
QApplication, QCodeApplication or QGuiApplication must be create before any other QObject base class instance or your application will crash somehow!

Do something like this:

int main(int argc, char *argv[])
{
    QCoreApplication a(argc, argv);

    MyUDP udp_1;
    MyUDP udp_2;
    MyUDP udp_3;

    udp_1.Start("192.176.0.1", 8001);
    udp_2.Start("192.176.0.2", 8002);
    udp_3.Start("192.176.0.3", 8003);

    return a.exec();
}

jenya7

I added a signal

class UDP : public QObject
{
    Q_OBJECT

    public:
    UDP(QObject *parent = nullptr);

     //////////////////////////////////////////////////////

    signals:
    void ReadyForReader(const NET_PARAM& param, const QByteArray data);

    //////////////////////////////////////////////////////////
};

and in a reader I add a slot

class READER
{
    public:
    READER();

    public slots:
    void ReadyForReader(const NET_PARAM& param, const QByteArray data);
};

Now in main

static UDP udp_1;
static UDP udp_2;

static READER reader;

int main(int argc, char *argv[])
{
    QCoreApplication a(argc, argv);

    QObject::connect(&udp_1, &UDP::ReadyForReader, &reader, &READER::ReadyForReader);
    QObject::connect(&udp_2, &UDP::ReadyForReader, &reader, &READER::ReadyForReader);

    udp_1.Start("192.176.0.1", 8001);
    udp_1.Start("192.176.0.2", 8002);

    return a.exec();
}

I get

C:\Qt\5.12.0\mingw73_64\include\QtCore\qobject.h:250: error: no matching function for call to 'QObject::connectImpl(const Object*&, void**, const Object*&, void**, QtPrivate::QSlotObject<void (READER::)(const NET_PARAM&, QByteArray), QtPrivate::List<const NET_PARAM&, QByteArray>, void>, Qt::ConnectionType&, const int*&, const QMetaObject*)'
return connectImpl(sender, reinterpret_cast<void **>(&signal),
~^~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
receiver, reinterpret_cast<void **>(&slot),
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
new QtPrivate::QSlotObject<Func2, typename QtPrivate::List_Left<typename SignalType::Arguments, SlotType::ArgumentCount>::Value,
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
typename SignalType::ReturnType>(slot),
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
type, types, &SignalType::Object::staticMetaObject);
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

What I did wrong?

Christian Ehrlicher

@jenya7 said in Passing data between threads.:

static UDP udp_1;
static UDP udp_2;

static READER reader;

Still static...

For signals/slots your class must be derived from QObject. Please read the documentation.

jenya7

@Christian-Ehrlicher
Thank you.
Without static I get warning - warning: no previous extern declaration for non-static variable 'udp_1'.
I can do like this

int main(int argc, char *argv[])
{
    UDP udp_1;
    UDP udp_2;

    READER reader;

    QCoreApplication a(argc, argv);

    QObject::connect(&udp_1, &UDP::ReadyForReader, &reader, &READER::ReadyForReader);
    QObject::connect(&udp_2, &UDP::ReadyForReader, &reader, &READER::ReadyForReader);

    udp_1.Start("192.176.0.1", 8001);
    udp_2.Start("192.176.0.2", 8002);

    return a.exec();
}

But this way the objects allocated on stack.

Christian Ehrlicher

@jenya7 said in Passing data between threads.:

But this way the objects allocated on stack.

And what's the problem with it? And if it's really a problem (for whatever reason) you can allocate them with new and delete them later on or use a shared_ptr or similar.

jsulm

@jenya7 You should more carefully read what others write!
As @KroMignon wrote: all QObject based class instances have to be created AFTER QCoreApplication a(argc, argv)!
And as also was mentioned here: your classes have to be subclassed from QObject to use signals/slots. Please read the links I gave you!

"no previous extern declaration for non-static variable 'udp_1'." - did you include the header file?