Версия для печати темы

Автор: 512es 10.11.2008, 16:00

всем привет!

пишу многопоточный высокопроизводительный сервер на qt.

сервер прослушивает коннекты на определённый порт и на каждое входящее соединение создаёт отдельный поток.
при каждом пришедшем пакете поток отсылает главному потоку сообщение через customEvent.

вроде работает, только не знаю как правильнее заставить все дочерние потоки (или один выборочно) отослать определённый пакет клиентам?


закинул все потоки массив QVector и передаю в customEvent главного потока ид того кто отправил. правильный путь ли я выбрал?

Автор: Litkevich Yuriy 10.11.2008, 16:27

Цитата(512es @ 10.11.2008, 19:00)

Цитата(512es @ 10.11.2008, 19:00)

Автор: ViGOur 10.11.2008, 16:48

Если я тебя правильно понял, то тебе нужен менеджер соединений. А раз так, то обычным QVector тебе не обойтись, нужно создать класс для управления и просмотра соединений (сокетов), а потоки мониторить тебе не нужно, так как достаточно повесить событие на закрытие соединения для завершение потока.

Автор: 512es 10.11.2008, 19:05

Litkevich Yuriy, customEvent потому что хочу всё что относится к каждому сокету вывести в поток этого сокета. дабы не вешать главный поток при ддос атаках. а как я понял, система сигнал\слот выполняет код в главном потоке...


ViGOur, да, именно менеджер соединений. потоки максимум самостоятельны будут, но главный поток должен принимать и рассылать данные по дочерним потокам.

вообщем то я думаю даже не нужно заталкивать класс потоков в QVector...
надо просто как то присваивать каждому сокету=потоку свой ИД и по этому ИДу иметь возможность послать данные в поток, где открыт сокет. притом так, чтобы они обработались именно в дочернем потоке (где сокет открыт).

вот в том как это правильнее сделать и есть мой вопрос =)

Автор: Litkevich Yuriy 10.11.2008, 19:16

Цитата(512es @ 10.11.2008, 22:05)

Автор: ViGOur 11.11.2008, 0:19

Цитата(Litkevich Yuriy @ 10.11.2008, 19:16)

Цитата(512es @ 10.11.2008, 19:05)

Автор: 512es 11.11.2008, 11:53

connect(tcpSocket, SIGNAL(readyRead()), this, SLOT(onReadyRead()));

connect(tcpSocket, SIGNAL(readyRead()), this, SLOT(onReadyRead()),Qt::DirectConnection);

Таким образом максимальное количество потоков, которые можно создать при всех параметрах заданных по умолчанию, равняется примерно 2035
Все сказанное ниже справедливо для линейки Windows NT/2000/XP

Автор: ViGOur 11.11.2008, 13:23

Цитата(512es @ 11.11.2008, 11:53)

Автор: Litkevich Yuriy 11.11.2008, 15:05

Мысль в слух:
Нельзя ли сделать один поток для авторизации, и группу потоков для авторизованых пользователей, т.е. когда несколько соединений, например 10, крутятся в одном потоке, если авторизованый пользователь начнет гадить, то навредит только десятерым, а остальные нормально работают.

Вобще в юниксах обычно делают одно соединение - один процесс, админы говорят, что так удобнее следить, и когда явно надо кого-то сбросить, то вот его Id, вот его и будем Килл.

(в Unix-системах нет разделения понятий поток/процесс, там всё - потоки. Процессом обычно называют главный поток программы.)

Автор: 512es 12.11.2008, 0:43

Автор: ViGOur 12.11.2008, 8:29

Цитата(512es @ 12.11.2008, 0:43)

Автор: 512es 12.11.2008, 13:06

вообще, это мой первый проект на Qt, потому прошу сильно ногами не бить =)

вот как я пытаюсь делать:

class NetServer : public QTcpServer
{
    Q_OBJECT
public:
    NetServer(QObject *parent = 0);
    QHash<int, PThread*> Users;
private:
    int CurrentThreadNum;
protected:
    void incomingConnection(int socketDescriptor);
    void customEvent(QEvent *event);
};

void NetServer::incomingConnection(int socketDescriptor)
{
    Users.insert(++CurrentThreadNum,new PThread(socketDescriptor, this));
//    connect(this, SIGNAL(toSendMessage(QString msgString)), Users.value(CurrentThreadNum), SLOT(onSendMessage(QString msgString)));
    Users.value(CurrentThreadNum)->ThreadNumber = CurrentThreadNum;
    Users.value(CurrentThreadNum)->start();
}

void NetServer::customEvent(QEvent *event)
{
    QHashIterator<int, PThread*> i(Users);
    switch ((int)event->type()) {
    case 1234:
        MyNewMsgEvent *MyMsgEvent = (MyNewMsgEvent *)event;
        qDebug() << "Received custom Event text from user" << MyMsgEvent->SenderUser << MyMsgEvent->message;       
        while (i.hasNext()) {
            i.next();
//            i.value()->onSendMessage(QString::number(MyMsgEvent->SenderUser,10) + " " + MyMsgEvent->message);
//            qDebug() << i.value() << "ddd" << Users.value(1) << i.value()->currentThread();
            MyNewMsgEvent *ev = new MyNewMsgEvent();
            ev->message = MyMsgEvent->message;
            ev->SenderUser = MyMsgEvent->SenderUser;
            QCoreApplication::postEvent(i.value(), ev);
        }
        break;
    case 1109: // User disconnected
        EventDisconnected *EvDisconnected = (EventDisconnected *)event;
        qDebug() << "User" << EvDisconnected->SenderUser << "SignOff";
        Users.value(EvDisconnected->SenderUser)->deleteLater();
        Users.remove(EvDisconnected->SenderUser);
        qDebug() << "Stats" << Users.size();
        break;
    }
}

void PThread::onDisconnected()
{
    exit();
    EventDisconnected *ev = new EventDisconnected();
    ev->SenderUser = ThreadNumber;
    QCoreApplication::postEvent(this->parent(), ev);
}

void PThread::run()
{
    tcpSocket = new QTcpSocket;
    ...
    connect(tcpSocket, SIGNAL(readyRead()), this, SLOT(onReadyRead()),Qt::DirectConnection);
    connect(tcpSocket, SIGNAL(disconnected()), this, SLOT(onDisconnected()),Qt::DirectConnection);
    exec();
    ...
    delete tcpSocket;
}

void PThread::customEvent(QEvent *event)
{
    switch ((int)event->type()) {
    case 1234:
        forever;
        break;
    }
}

Автор: 512es 12.11.2008, 15:02

попробовал через сигнал\слот..

connect(this, SIGNAL(toSendMessage(QString)), Users.value(CurrentThreadNum), SLOT(onSendMessage(QString)),Qt::DirectConnection);

Автор: 512es 13.11.2008, 12:27

вообщем, всю голову сломал, не знаю как сделать(

http://trolltech.com/developer/faqs/faq.2007-10-02.9810372345
написано что есть 3 способа:
1) передача эвентов
2) QCoreApplication::invokeLater()
3) и сигнал\слот с QueuedConnection

но к сожалению всё сказаное там для явы..

пробовал эвенты и через сигнал\слот с QueuedConnection.

пробовал даже вызывать метод в котором эмитится сигнал по которому вызывается нужный слот, как тут:
http://forum.sources.ru/index.php?showtopic=245838&hl=

результат один и тот же(


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


на ум приходит ещё один оч кривой способ..
отказаться от родителя в дочерних потоках и добавить в нужные методы moveToThread(this)
иии.. вот только что родилась идея ещё хуже.. пересылать данные по TCP на локалхост соединению между главным и дочерними потоками =)) бред..

Автор: ViGOur 13.11.2008, 13:17

Цитата(512es @ 13.11.2008, 12:27)

Автор: ViGOur 13.11.2008, 14:30

Вот я пример набросал, правда с рисованием, но думаю идея будет понятна...

#include <QtCore/QCoreApplication>

#include "xThread.h"

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

    CxThread thread[2];
    for( int n=0; n < sizeof( thread)/sizeof( *thread); ++n)
    {
        thread[n].start();
    }

    while( true)
    {
        for( int n=0; n < sizeof( thread)/sizeof( *thread); ++n)
        {
            if( thread[n].isRunning())
                emit thread[n].drawText();
        }
        thread[0].sleep( 1);
        qDebug( "**********************");
    }

    return a.exec();
}

#ifndef XTHREAD_H
#define XTHREAD_H

#include <QThread>
#include <QMutex>

class CxPrint;

class CxThread : public QThread
{
    Q_OBJECT

private:
    static QMutex m_mutex;
    int        m_n;
    CxPrint *m_pPrint;

public:
    CxThread(QObject *parent=0);
    ~CxThread();

    void drawText();
    static void sleep ( unsigned long uls);
    int GetN()const { return m_n; }

protected:
    virtual void run();
    
};

class CxPrint: QObject
{
    Q_OBJECT

public:
    void drawText();

signals:
    void signalDrawText();

protected slots:
    void slotDraw();

public:
    CxPrint(void);
    virtual ~CxPrint(void);
};


#endif // XTHREAD_H

#include "xThread.h"
#include "xPrint.h"

QMutex CxThread::m_mutex;

CxThread::CxThread(QObject *parent)    : QThread(parent)
{
    m_n=0;
    m_pPrint = 0;
}

CxThread::~CxThread()
{
    delete m_pPrint;
    m_pPrint = 0;
}

void CxThread::drawText()
{
    if( m_pPrint)
        emit m_pPrint->drawText();
}

void CxThread::sleep ( unsigned long uls)
{
    QThread::sleep( uls);
}

void CxThread::run()
{
    {
        QMutexLocker locker(&m_mutex);
        static int n = 1;
        m_pPrint = new CxPrint;
        qDebug( "Runing thread with id: 0x%x, thread N: %d", QThread::currentThreadId(), n);
        m_n = n;
        n++;
    }

    CxThread::exec();
}

CxPrint::CxPrint(void)
{
    connect( this, SIGNAL( signalDrawText()), this, SLOT( slotDraw()));
}

CxPrint::~CxPrint(void)
{
}

void CxPrint::slotDraw()
{
    qDebug( "Draw in thread with id: 0x%x, thread N: %d", QThread::currentThreadId(), ((CxThread*)QThread::currentThread())->GetN());
}

void CxPrint::drawText()
{
    emit signalDrawText();
}

Автор: 512es 13.11.2008, 16:42

ViGOur, спасибо!) то что надо!))
у меня рождалась мысль что объект надо создавать в секции run() но уже не верилось что это заработает..
вообще жудко запутаная конструкция получилось, жаль что нельзя упростить.. хотяя...

Автор: ViGOur 13.11.2008, 18:01

Цитата(512es @ 13.11.2008, 16:42)

Цитата(512es @ 13.11.2008, 16:42)

Автор: 512es 15.11.2008, 1:19

разве что от CxThread::drawText() удалось избавиться)

emit thread[n].drawText().m_pPrint.drawText()

Автор: Litkevich Yuriy 15.11.2008, 1:29

Цитата(512es @ 15.11.2008, 4:19)

Автор: ViGOur 15.11.2008, 10:28

Цитата(Litkevich Yuriy @ 15.11.2008, 1:29)

Автор: 512es 20.11.2008, 13:52

ViGOur, провёл эксперимент:
подключился двумя клиентами. один из них отправлял файл на 200 мб а второй быстро посылал маленькие пакеты серверу и получал ответ. пока файл отправлялся подключился ещё двумя клиентами и тоже быстро отсылал небольшие пакеты. задержек вообще никаких не наблюдалось!) всё работало так, какбудто 200 меговый файл и не принимался.
разве что, когда закончилась передача файла и он поступил в главный поток для обработки (в данном случае для записи на диск), отсылка ответов клиентам приостановилась ненадолго. но все пакеты скапливались в очередь и сразу обработались как только запись файла завершилась =)
а если такие большие пакеты посылать в отдельный специальный поток, с низким приоритетом то вообще хоть терабайты пересылать можно, никто и не заметит =)

Автор: ViGOur 20.11.2008, 15:21

Цитата(512es @ 20.11.2008, 13:52)

Автор: 512es 20.11.2008, 16:59

тут просто разные нужды. мне важно чтобы все юзвери ансинхронно получали и отправляли маленькие пакеты. без задержек и очередей на приём\отправку. и оно отлично работает когда на каждого отдельный поток.

хотя я и планирую иногда передавать от сервера к юзверем файлы не больше 1 мб, это не файлообменник, где можно экономить на времени отклика, выстраивая отправку\приём в очередь.
вот наверное поэтому я всё никак с вами не соглашусь =))

32 соединения в поток, 320 соединений, это не 320 потоков, а 10, разница есть?

Автор: ViGOur 20.11.2008, 20:21

Цитата(512es @ 20.11.2008, 16:59)