QLocalSocket проблема с многопоточностью

[JAkutenshi]

Здравствуйте!
Проблема следующая: есть написанный и отлаженный клиент, а есть сервер, у которого проблемы с многопоточностью ибо все моих рук дело.
Что хочется: чтобы каждый клиент обрабатывался в QThread::run().
В чем проблема:
QObject: Cannot create children for a parent that is in a different thread.
(Parent is QLocalSocket(0x758fa8), parent's thread is QThread(0x8216b0), current
 thread is NewConnectionThread(0x7581f8)

Код run-a:

Код:

void NewConnectionThread::run()
{
    clientSocket =
            parentLocalServerRef->getServerRef()->nextPendingConnection();
    // connect(clientSocket, SIGNAL(readyRead()), this, SLOT(read()));

    while (clientSocket->bytesAvailable() < (int)sizeof(quint32)) {
            clientSocket->waitForReadyRead(1);
        }

    connect(clientSocket, SIGNAL(disconnected()),
            clientSocket, SLOT(deleteLater()));

    QByteArray newClientName = clientSocket->peek(30);
    qDebug() << "   New client name : " << newClientName << "\n";

    QString currentDate = parentLocalServerRef->getCurrentDate();
    qDebug() << "   Sending to " << newClientName
             << " current date : " << currentDate << "\n";
    QByteArray response = currentDate.toLatin1();

    clientSocket->write(response);
    clientSocket->flush();

    qDebug() << "   Finish sending. Close the Thread.";

    stopProcess = true;

    //while (!stopProcess) {}
}

Причину, кажется, понимаю: clientSocket->write(response) рискует просуществовать дольше своего потока. Но при этом не очень понимаю, как исправить без костылей.
Еще есть второй момент, в clientSocket->waitForReadyRead(1). Судя по докам, когда данные готовы к чтению, он не дожидаясь дефолтного счетчика (30000 мс) возвращает готовность. Но не тут-то было и в этой вариации оно ждет полностью счетчик. В абсолютно таком же run-e, разве-что в реализации с гуи (вешепредставленны ран из консоли, но мне кажется это не должно влиять на выполнение метода Оо), оно сразу выходит из ожидания. ЧЯДНТ?

Немного подумав, я вспомнил, что клиент читал малость по-другому и ожидал через сигнал readReady(). Ну, ладно, делаем, объявляю слот read() у потока и в итоге:

Код:

NewConnectionThread::NewConnectionThread(LocalServer* parentLocalServerRef,
                                         QObject *parent) :
    QThread(parent),
    parentLocalServerRef(parentLocalServerRef)
{
    stopProcess = false;
}

void NewConnectionThread::run()
{
    clientSocket =
            parentLocalServerRef->getServerRef()->nextPendingConnection();
    connect(clientSocket, SIGNAL(readyRead()), this, SLOT(read()));

    //while (!stopProcess) {}
}

void NewConnectionThread::read()
{
    connect(clientSocket, SIGNAL(disconnected()),
            clientSocket, SLOT(deleteLater()));

    QByteArray newClientName = clientSocket->peek(30);
    qDebug() << "   New client name : " << newClientName << "\n";

    QString currentDate = parentLocalServerRef->getCurrentDate();
    qDebug() << "   Sending to " << newClientName
             << " current date : " << currentDate << "\n";
    QByteArray response = currentDate.toLatin1();

    clientSocket->write(response);
    clientSocket->flush();

    qDebug() << "   Finish sending. Close the Thread.";

    stopProcess = true;
}

Казалось бы, все распрекрасно, и ошибок нет и читает быстро. Вот только многопоточности почему-то нету... Т.е., нету ее потому что run() заканчивается после коннекта? И потом read() выполняется в основном потоке, как я понял, судя по дебагу (там всегда 1 поток был на обработку клиента, менялся только айдишник).

Ниже, весь код сервера:

Код:

_________localserver.h

#ifndef LOCALSERVER_H
#define LOCALSERVER_H

#include <QObject>
#include <QString>
#include <QLocalServer>
#include <QMutex>
#include <QDateTime>

class LocalServer : public QObject
{
    Q_OBJECT

    QString pipeName;
    QLocalServer* localServer;
    QDateTime currentDate;
    QMutex mutex;


public:
    explicit LocalServer(QString pipeName = "pipeJAkutenshi",
                         QObject *parent = 0);
    QLocalServer *getServerRef();
    QString getCurrentDate();

signals:

public slots:
    void newLocalSocketConnection();

};

#endif // LOCALSERVER_H


_______localserver.cpp

#include "localserver.h"
#include "newconnectionthread.h"

LocalServer::LocalServer(QString pipeName, QObject *parent) :
    QObject(parent), pipeName(pipeName)
{
    localServer = new QLocalServer();
    localServer->listen(pipeName);
    qDebug() << "Server pipe" << pipeName
             << "is started and waiting for new clients...\n";

    connect(localServer, SIGNAL(newConnection()),
            this, SLOT(newLocalSocketConnection()));
}

QLocalServer *LocalServer::getServerRef() { return localServer; }

QString LocalServer::getCurrentDate()
{
    mutex.lock();
    currentDate = QDateTime::currentDateTime();
    QString stringCurrentDate =
            currentDate.toString("dd.MM.yyyy hh:mm:ss");
    mutex.unlock();

    return stringCurrentDate;
}

void LocalServer::newLocalSocketConnection()
{
    qDebug() << "New client!\n";    
    NewConnectionThread* newClientConnection =
            new NewConnectionThread(this);

    newClientConnection->start();
    qDebug() << "Server is created new Thread and waiting for new clients...\n";
}

_________newconnectionthread.h

#ifndef NEWCONNECTIONTHREAD_H
#define NEWCONNECTIONTHREAD_H

#include <QThread>
#include "localserver.h"

class NewConnectionThread : public QThread
{
    Q_OBJECT

    QLocalSocket* clientSocket;
    bool stopProcess;

public:
    explicit NewConnectionThread(LocalServer* parentLocalServerRef,
                                 QObject *parent = 0);
    void run();
    LocalServer* parentLocalServerRef;
signals:

public slots:
    void read();
};

#endif // NEWCONNECTIONTHREAD_H

_________newconnectionthread.cpp

NewConnectionThread::NewConnectionThread(LocalServer* parentLocalServerRef,
                                         QObject *parent) :
    QThread(parent),
    parentLocalServerRef(parentLocalServerRef)
{
    stopProcess = false;
}

void NewConnectionThread::run()
{
    clientSocket =
            parentLocalServerRef->getServerRef()->nextPendingConnection();
    connect(clientSocket, SIGNAL(readyRead()), this, SLOT(read()));

    //while (!stopProcess) {}
}

void NewConnectionThread::read()
{
    connect(clientSocket, SIGNAL(disconnected()),
            clientSocket, SLOT(deleteLater()));

    QByteArray newClientName = clientSocket->peek(30);
    qDebug() << "   New client name : " << newClientName << "\n";

    QString currentDate = parentLocalServerRef->getCurrentDate();
    qDebug() << "   Sending to " << newClientName
             << " current date : " << currentDate << "\n";
    QByteArray response = currentDate.toLatin1();

    clientSocket->write(response);
    clientSocket->flush();

    qDebug() << "   Finish sending. Close the Thread.";

    stopProcess = true;
}

_________main.cpp

#include <QCoreApplication>
#include "localserver.h"

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

    LocalServer* server =
            new LocalServer();

    return a.exec();
}

Слезливо прошу о помощи, ибо дедлайн буквально завтра, впереди корба и соап, а на пайпы я потратил по таким мелочам больше 2х ночей и моральных сил сделать нормально без костылей уже нету.
Спасибо!

[Пантер]

А нужна ли тут многопоточность? Работа с сокетами ассинхронна, главное, работать через сигналы/слоты.

nextPendingConnection нужно вызывать в том же потоке, в котором живет сервер.

[JAkutenshi]

Да, многопоточность с методами синхронизации нужны как условие лабы. Да и почему нет? Если каждый клиент будет ждать, пока будут обработаны предыдущие - как минимум не выполняется условие РВС о прозрачности  . Поэтому каждый клиент должен обрабатываться при подключении отдельным потоком.

[Пантер]

Не обязательно. Ассинхронка же.
В общем, тебе нельзя никакие блокирующие методы юзать типа waitForReadyRead. Все на сигналах/слотах (сигналы readyRead, bytesWritten).

[Пантер]

Лови мое тестовое задание на одну из вакансий. Там достаточно просто реализована многопоточная работа с клиентами. Думаю, разобраться не составит сложности.

[JAkutenshi]

Можете пожалуйста пояснить про синхронность/ассинхронность? Несколько путаю понятия. Ну, разве что имею в голове пример udp/tcp как пример ассинхронного/синхронного протоколов взаимодействия по сети, но эти понятия зависят от контекста, если не ошибаюсь... Что Вы подразумеваете под этим здесь?

Т.е. вариант с сигналом readReady() - таки правильное решение было?

Спасибо большое за пример! Разберусь.

[Пантер]

Ну, смотри, подключились к тебе 100 клиентов, но данные к тебе попадают ассинхронно - то есть, ты не должен сразу 100 клиентов читать и отвечать им. Как только пришла порция данных от клиента, обработчик этого клиента читает данные и что-то с ними делает.

[JAkutenshi]

Простите, можно еще?
В разных потоках я создаю соединения и помещаю их в ожидание данных. Данные получаются ассинхронно, да. Могут придти в любое время в любом порядке. Но что мешает их ассинхронно принимать и отвечать? По готовности соединения начать чтение и ответ...  И как?
И что с той ошибкой ны clientSocket->write(response)? Тем было в сети много с этим но как решить проблему не очень понял. Что то с moveToThread я так понимаю, но в голове с этим каша.

[Пантер]

Сокеты ты создаешь в главном потоке. делаешь свой класс, который будет работать с конкретным сокетом и помещаешь его в поток. В классе ты на readyRead читаешь данные и работаешь с ними (при создании не забудь вычитать то, что уже пришло от клиента). Посмотри внимательно пример, который я привел - там все просто.

[JAkutenshi]

Я посмотрел Ваш проект и возник вопрос:

ClientConnection *clientConnection = new ClientConnection (socket);
QThread *thread = Core::moveToThread (clientConnection);

Как мы тред помещаем в QObject? Можете пояснить это место?

[Пантер]

Мы QObject помещаем в тред.

Код

C++ (Qt)
QThread *moveToThread (QObject *object)
	{
		QThread *thread = new QThread;
		object->moveToThread (thread);
 
		QObject::connect (thread, SIGNAL (finished ()), object, SLOT (deleteLater ()));
		QObject::connect (thread, SIGNAL (finished ()), thread, SLOT (deleteLater ()));
		thread->start ();
 
		return thread;
	}
 

[JAkutenshi]

Читаю док по QObject и там только вариация : void moveToThread(QThread* targetThread)

ps: qt5.3 если что. И еще не вижу, что thread стартует Оо

[Пантер]

Читаю док по QObject и там только вариация : void moveToThread(QThread* targetThread)

Ну, object->moveToThread (thread);. Я просто вынес это в отдельную функцию.

[JAkutenshi]

Все получилось, спасибо большое!