Не пойму как заставить работать QTcpSocket в QThread

[CoderInside]

Подскажите как работать с QTcpSocket в отдельном потоке? Есть сервер (QTcpServer), когда появляется новый коннект - нужно вывести обработку этого клиента в отдельный поток. Для большего понимания я повставлял в код отладочные сообщения посылаемые на консоль через qDebug() (в начале каждой функции выводиться название функции и еще вывод перед отправкой в сокет и после). Первый раз, когда к серверу подключается клиент все проходит вроде нормально, клиент получает приветственное сообщение. Вот консоль:

Код:

coder@puh:~/projects/net/server$ ./server 5555
void XTcpServer::incomingConnection(int socketDescriptor)
void XClientThread::CONSTRUKTOR()
void XClientThread::run()
void XClientThread::sendMessage(QString msg) "Hello client!"
bvoid XClientThread::sendMessage(QString msg) -> Before write(block)
avoid XClientThread::sendMessage(QString msg) -> After write(block)

А если теперь клиентом послать серверу сообщение (текст "сообщение") - то на клиент ничего не приходит а в консоли выводиться вот это:

Код:

void XClientThread::reciveMessage()
void XClientThread::analizeRequest(QString newMessage) "сообщение"
void XClientThread::sendMessage(QString msg) "[[[сообщение]]]"
bvoid XClientThread::sendMessage(QString msg) -> Before write(block)
QSocketNotifier: socket notifiers cannot be enabled from another thread
avoid XClientThread::sendMessage(QString msg) -> After write(block)

Что это: "QSocketNotifier: socket notifiers cannot be enabled from another thread"?
Перепробовал уже все. Ничего не могу понять... А если запустить этот код на Windows - то винда иногда виснет , а иногда в Visual студии видно что поток с сокетом завершился сам по себе...

Вот код сервера:

Код:

#ifndef XCLIENT_THREAD_H
#define XCLIENT_THREAD_H

#include "common.h"

class XClientThread : public QThread
{
    Q_OBJECT

public:
    XClientThread(int socketDescriptor, QObject *parent=0);
    void run();

signals:
    void error(QTcpSocket::SocketError socketError);

private:
    QTcpSocket * clientSocket;
    int socketDescriptor;
    quint16 blockSize;

private slots:
    void analizeRequest(QString message);
    void sendMessage(QString msg);
    void deleteConnection();
    void reciveMessage();
};

#endif // XCLIENT_THREAD_H

Код:

#include "common.h"
#include "xclient_thread.h"
//---------------------------------------------------------------------------
XClientThread::XClientThread(int socketDescriptor, QObject *parent)
:QThread(parent),socketDescriptor(socketDescriptor)
{
    qDebug() << "void XClientThread::CONSTRUKTOR()";
}
//---------------------------------------------------------------------------
void XClientThread::run()
{
    // Создаем сокет и присваиваем ему полученный дескриптор
    qDebug() << "void XClientThread::run()";
    blockSize=0;
    clientSocket = new QTcpSocket();
    if (!clientSocket->setSocketDescriptor(socketDescriptor))
    {
        emit error(clientSocket->error());
        return;
    }

    // Подписываемся на нужные сигналы
    connect(clientSocket, SIGNAL(disconnected()), this, SLOT(deleteConnection()));
    connect(clientSocket, SIGNAL(readyRead()), this, SLOT(reciveMessage()));

    // Отсылаем клиенту приглашение
    sendMessage("Hello client!");

    exec();
}
//---------------------------------------------------------------------------
void XClientThread::sendMessage(QString msg)
{
    qDebug() << "void XClientThread::sendMessage(QString msg)" << msg;

    QByteArray block;
    QDataStream out(&block, QIODevice::WriteOnly);
    out.setVersion(QDataStream::Qt_4_0);

    out << (quint16)0;
    out << (QString)msg;
    out.device()->seek(0);
    out << (quint16)(block.size() - sizeof(quint16));

    qDebug() << "bvoid XClientThread::sendMessage(QString msg) -> Before write(block)";
    clientSocket->write(block);
    qDebug() << "avoid XClientThread::sendMessage(QString msg) -> After write(block)";
}
//---------------------------------------------------------------------------
void XClientThread::reciveMessage()
{
    qDebug() << "void XClientThread::reciveMessage()";

    QDataStream in(clientSocket);
    in.setVersion(QDataStream::Qt_4_0);
    if (blockSize == 0)
    {
        if (clientSocket->bytesAvailable() < (int)sizeof(quint16))
        return;
        in >> blockSize;
    }

    if (clientSocket->bytesAvailable() < blockSize)
    return;

    QString newMessage;
    in >> newMessage;
    blockSize=0;

    analizeRequest(newMessage);
}
//---------------------------------------------------------------------------
void XClientThread::analizeRequest(QString newMessage)
{
    qDebug() << "void XClientThread::analizeRequest(QString newMessage)" << newMessage;
    QString msg;
    msg=QString("[[["+newMessage+"]]]");
    // Отсылаем...
    sendMessage(msg);
}
//---------------------------------------------------------------------------
void XClientThread::deleteConnection()
{
    qDebug() << "void XClientThread::deleteConnection()";
    clientSocket->deleteLater();
    exit();
}
//---------------------------------------------------------------------------

[Dendy]

Дружище, проблема ясна и тривиальна. Не доконца понята система потоков Qt и передачи данньІх между ними. Знач так...

КаждьІй обьект существует в потоке, которьІй можно получить через QObject::thread(). Етот поток (что возвратится) означает екземпляр QThread'а, которьІй будет обслуживать собьІтия данного обьекта, всё просто. СобьІтия обьекта обслуживает один и только один поток, военного ничего нет, все калбеки (events, signals) будут обрабатьІваться физически в теле run() етого потока.

Итак, вот они грабли в твоём коде:

Код:

    // Подписываемся на нужные сигналы 
    connect(clientSocket, SIGNAL(disconnected()), this, SLOT(deleteConnection())); 
    connect(clientSocket, SIGNAL(readyRead()), this, SLOT(reciveMessage()));

Испускатель и получатель сигнала находятся в разньІх(!) потоках. ПервьІй - в XClientThread (так как создан в теле метода run()), второй - в главном потоке (так как создан в главном потоке). Соответственно, между етими обьектами будет создан Qt::QueuedConnection. То-есть тело принимателей собьІтия будет вьІполнено не в том же потоке, что и XClientThread, что влечёт за собой одновременньІй вьІзов методов QTcpSocket из разньІх потоков, а так как последний не потокобезопастньІй - получить кернел паник.

Ещё замечание: при завершении соединения тьІ останавливаешь обработку собьІтий потока и удаляешь сокет. Сам не знаю, но осмелюсь предположить, что deleteLater() не удалит обьект, так как некому будет передать собьІтие (цикл то остановлен).

Солюшн:

1. (Потенциально правильно) Создать класс, что будет принимать собьІтия от QTcpSocket и обрабатьІвать их. То-есть не XClientThread чтоб занимался етим, а самостоятельньІй обьект. Фишка в том, что етот обьект тьІ создашь в теле метода run() рядом с сокетом и между ними будет существовать Qt::DirectConnection.

2. (Вполне работоспособньІй) При коннекте указьІвать явно:

Код:

    // Подписываемся на нужные сигналы 
    connect(clientSocket, SIGNAL(disconnected()), this, SLOT(deleteConnection())б Qt::DirectConnection); 
    connect(clientSocket, SIGNAL(readyRead()), this, SLOT(reciveMessage()), Qt::DirectConnection);

Ето значит, что часть методов XClientThread будут вьІзьІваться из одного потока, часть из другого. Но при етом не нужно порождать лишние классьІ и есть доступ к данньІх одного догического подключения из разньІх потоков.

PS. Мелкие замечания по коду:

reciveMessage() пишется как receiveMessage().

В теле приёма сообщения должен существовать цикл, что разгребёт ВСЕ сообщения, которьІе пришли. Сейчас, если их пришло 5 (одним пакетом), получишь тьІ только одно, первое.

ПараметрьІ следует писать не (QString message), а (const QString & message). Ето секономит вьІзовьІ конструкторов/деструкторов и скроет реализацию метода.

Писал бьІ уже:

Код:

qDebug() << "void XClientThread::KONSTRUCTOR()";

В KDE стиле

PPS. Поздравьте меня! Ента маё 500-е сообщение  Я терь ГИПЕР АКТИВНЬІЙ ЖИТЕЛЬ

[CoderInside]

Спасибо. Буду разбираться.

PPS. Поздравьте меня! Ента маё 500-е сообщение Very Happy Very Happy Very Happy Я терь ГИПЕР АКТИВНЬІЙ ЖИТЕЛЬ

Поздравляю! Так держать!

[Myav]

Dendy, спасибо за подробное объяснение.

А не подскажет кто-нибудь, как организовать аналогичное связывание сигнала со слотом в случае, когда clientSocket не указатель на QTcpSocket, а собственно QTcpSocket и есть?

Если связывать так:

Код:

connect(&clientSocket, SIGNAL(disconnected()), this, SLOT(deleteConnection()), Qt::DirectConnection);

...то в момент связывания в консоль отладчика будет отправлено сообщение:

QObject: Cannot create children for a parent that is in a different thread.

...а в момент проишествия disconnected() при выполнении получим сообщение об ошибке.

P.S. Особенного смысла в размещении clientSocket'а в стеке, а не в куче, наверное, нет, просто у меня бзик - использовать указатели как можно реже :shock: А может быть конкретно в этом случае без указателей как раз нельзя обойтись?

[kotofay]

Вот нарыл в архивах, может тебе поможет.
Делать отдельный поток на сервис клиента, как это обычно принято, кмк, не стОит.

Код:

#include <QThread>
#include <QTcpServer>
#include <QTcpSocket>
#include <ещё чего-нибудь>

class MyService : public QObject
{
   Q_OBJECT
public:
   MyService(QTcpSocket *_tcpSocket, QObject *parent = 0);
public slots:
   void readyRead();
   void disconnected();
private:
   QTcpSocket *tcpSocket;
};
// --------------------------------------------------------------------------------------------------------------------------------
class MyServer : public QObject {
   Q_OBJECT
public:
   MyServer(QObject *parent);
public slots:
   void incomingConnection();   
protected:
     
     QTcpServer tcpServer;
};
// --------------------------------------------------------------------------------------------------------------------------------
class MyClient : public QObject{
   Q_OBJECT
public:
   MyClient(QObject *parent = 0);
   ~MyClient();
public slots:
   void readyRead();
   void connected();
   void bytesWritten(qint64);
protected:
   QTcpSocket tcpSocket;
};
// --------------------------------------------------------------------------------------------------------------------------------
class MyThread : public QThread {
   Q_OBJECT
public:
   MyThread(QObject *parent = 0);
   ~MyThread();
   void run();
};
// --------------------------------------------------------------------------------------------------------------------------------

Код:

// - Server -----------------------------------------------------------------------------------------------------------------------
MyService::MyService(QTcpSocket *_tcpSocket, QObject *parent)
: QObject(parent), tcpSocket(_tcpSocket){
   if (tcpSocket)
   {
      connect(tcpSocket, SIGNAL(readyRead()), this, SLOT(readyRead()));
      connect(tcpSocket, SIGNAL(disconnected()), this, SLOT(disconnected()));
   }
}
// --------------------------------------------------------------------------------------------------------------------------------
void MyService::disconnected(){
   std::cout << "\ndisconnected\n";
}
// --------------------------------------------------------------------------------------------------------------------------------
void MyService::readyRead(){
   QDataStream ds(tcpSocket);
   
   ds.setVersion(QDataStream::Qt_4_1);
   QString rs;
   QDateTime dt;// = QDateTime::currentDateTime();
   ds >> rs >> dt;
   std::cout << "\nServer read: " << rs.toStdString() << ", " << std::string(dt.toString("ddd MMMM d yyyy hh:mm:ss.zzz").toLocal8Bit());
   // compress stream, write to socket
   QByteArray ba; // buffer
   QDataStream _ds(&ba, QIODevice::WriteOnly); // stream
   _ds << rs.toUpper() << dt; // write data to stream
   // compressed, write to socket
   ds << QByteArray(qCompress(ba, 9));
}
// --------------------------------------------------------------------------------------------------------------------------------
MyServer::MyServer(QObject *parent) : QObject(parent){
   connect(&tcpServer, SIGNAL(newConnection()), this, SLOT(incomingConnection()));
   tcpServer.listen(QHostAddress::Any, (quint16)10000);
};
// --------------------------------------------------------------------------------------------------------------------------------
void MyServer::incomingConnection()
{
   new MyService(tcpServer.nextPendingConnection(), this);
}
// - Client -----------------------------------------------------------------------------------------------------------------------
MyClient::MyClient(QObject *parent):QObject(parent){
   connect(&tcpSocket, SIGNAL(connected()), this, SLOT(connected()));
   connect(&tcpSocket, SIGNAL(readyRead()), this, SLOT(readyRead()));
   connect(&tcpSocket, SIGNAL(bytesWritten(qint64)), this, SLOT(bytesWritten(qint64)));
   tcpSocket.connectToHost(QString("localhost"), (quint16)10000);
   tcpSocket.waitForConnected();
}
// --------------------------------------------------------------------------------------------------------------------------------
MyClient::~MyClient(){
   tcpSocket.disconnectFromHost();
   if(tcpSocket.state() == QAbstractSocket::ConnectedState)
      tcpSocket.waitForDisconnected();
};
// --------------------------------------------------------------------------------------------------------------------------------
void MyClient::readyRead(){
   //std::cout << "readyRead\n";
   qint64 sz = tcpSocket.bytesAvailable();
   if (sz)
   {
      QDataStream ds(&tcpSocket);
      ds.setVersion(QDataStream::Qt_4_1);

      QString rs;
      QDateTime dt;

      QByteArray zba, ba;
      ds >> zba;
      ba = qUncompress(zba);
      QDataStream _ds(ba);
      _ds >> rs >> dt;
      std::cout << "\nClient read: " << rs.toStdString() << ", " << std::string(dt.toString("ddd MMMM d yyyy hh:mm:ss.zzz").toLocal8Bit()) << ", " << _int64(sz);
   }
}
// --------------------------------------------------------------------------------------------------------------------------------
void MyClient::connected(){
   //std::cout << "\nconnected\n";
   //tcpSocket.write(QString(tr("hello!\n\r")).toAscii());

   QDataStream ds(&tcpSocket);
   ds.setVersion(QDataStream::Qt_4_1);
   ds << QString("hello ");
   QDateTime dt = QDateTime::currentDateTime();
   ds << dt;
}
// --------------------------------------------------------------------------------------------------------------------------------
void MyClient::bytesWritten(qint64 bw){
   //std::cout << "\nbytesWritten: " << bw << std::endl;
   //tcpSocket.write(QString(tr("Hello!\n\r")).toAscii());
}
// - Thread -----------------------------------------------------------------------------------------------------------------------
MyThread::MyThread(QObject *_parent) : QThread(_parent){
   start();
};
// --------------------------------------------------------------------------------------------------------------------------------
MyThread::~MyThread(){
   exit();
   wait();
}
// --------------------------------------------------------------------------------------------------------------------------------
void MyThread::run(){
   MyClient client;
   QTimer timer;
   timer.setInterval(1000);
   connect(&timer, SIGNAL(timeout()), &client, SLOT(connected()));
   timer.start();
   exec();
}
// - Thread -----------------------------------------------------------------------------------------------------------------------

вызовы:

Код:

   MyServer srv(this);
   MyThread thrd(this);
   thrd.run();

[Myav]

Отличный пример; большое спасибо.