синхронизация сокетов в разных потоках Опции

[borune]

Ребят, вопросик.

Есть сеть. На каждой машине сети запускается клиентское ПО, которое производит тестирование сети путем передачи пакетов между машиной, на которой оно запущенно, и всеми остальными машинами по типу "точка-точка". Т.е. если в сети n компьютеров, то каждый клиент будет общаться с n-1 компьютером.

Как это организовано. При инициализации клиента ему передается список адресов машин, с которыми ему нужно обмениваться сообщениями, а также список портов (обмен с каждой машиной осуществляется по уникальному для каждой пары порту). В клиенте происходит следующее:

1. Для каждого компа из переданного списка создается поток, в котором открывается QTcpServer на прослушивание заданного порта, при приходе сигнала newConnection() создается сокет для чтения.

Раскрывающийся текст

void read_thread::begin() const
{
   // qDebug(tr("listen to port %1").arg(port).toLatin1());
    srv->listen(QHostAddress::Any,port);
    connect(srv,SIGNAL(newConnection()),this,SLOT(processIncomingConnection()));
}

Раскрывающийся текст

void read_thread::processIncomingConnection()
{
    if(isConnected) return;
    read_soc = srv->nextPendingConnection();    
    connect(read_soc,SIGNAL(readyRead()),this,SLOT(read()));
    isConnected = true;
}

2. Для каждого компа из переданного списка создается поток, содержащий сокет, который пытается подключиться к компу

Раскрывающийся текст

bool send_thread::begin()
{      
    write_soc = new QTcpSocket(this);
    write_soc->connectToHost(QHostAddress(addr),port,QIODevice::WriteOnly);
    isConnected = write_soc->waitForConnected();

    if(write_soc->state() == QAbstractSocket::ConnectedState)
    {
        qDebug(tr("write_socket has been created").toLatin1());                
        return true;
    }
    return false;
}

Потоки запускаю так:

Раскрывающийся текст

for(int iter=0; iter<addresses.size(); ++iter)  // создаем слушающие потоки
    {
        QThread* listen_thread = new QThread();
        read_thread* lstn_thread = new read_thread(iter,ports[iter],(quint32)period*time);

        connect(listen_thread,SIGNAL(started()),lstn_thread,SLOT(begin()));
        connect(lstn_thread,SIGNAL(thread_finished()),listen_thread,SLOT(quit()));
        connect(lstn_thread,SIGNAL(thread_finished()),lstn_thread,SLOT(deleteLater()));
        connect(lstn_thread,SIGNAL(thread_finished()),this,SLOT(calc_threads()));
        connect(this,SIGNAL(letsfinish()),lstn_thread,SLOT(end()));

        lstn_thread->moveToThread(listen_thread);

        listen_thread->start();      
    }

    for(int iter=0; iter<addresses.size(); ++iter)  // создаем пишущие потоки
    {
        QThread* write_thread = new QThread();

        send_thread* snd_thread = new send_thread(iter,addresses[iter],ports[iter],(quint32)period*time,period);

        connect(write_thread,SIGNAL(started()),snd_thread,SLOT(begin()));
        connect(snd_thread,SIGNAL(thread_finished()),write_thread,SLOT(quit()));
        connect(snd_thread,SIGNAL(thread_finished()),snd_thread,SLOT(deleteLater()));
        connect(snd_thread,SIGNAL(thread_finished()),this,SLOT(calc_threads()));
        connect(this,SIGNAL(letsfinish()),snd_thread,SLOT(end()));

        connect(this,SIGNAL(go()),snd_thread,SLOT(startSending()));

        snd_thread->moveToThread(write_thread);

        write_thread->start();

        while(!snd_thread->connected()){}     
    }
}

Такая конструкция работает нестабильно - иногда все нормально, а иногда некоторые сокеты не создаются. Я точно не знаю, в чем дело, но вот если в обработчике read_thread::processIncomingConnection() убрать условие if(isConnected) return;, то для одного и того же хоста может создастся несколько сокетов, то есть сигнал newConnection() может приходить неоднократно. Непонятно, из-за чего такое может происходить.

Как грамотно организовать создание потоков в моем случае?

Сообщение отредактировал borune - 23.10.2013, 18:14

[Алексей1153]

не пытайся всё отладить сразу. Пусть сначала будет всего 2 машины в сети. Тестируй. Потом добавишь третью. Тестируй. И всё увидишь

Сообщение отредактировал Алексей1153 - 25.10.2013, 9:41

[Yuriy]

... При инициализации клиента ему передается список адресов машин, с которыми ему нужно обмениваться сообщениями, а также список портов (обмен с каждой машиной осуществляется по уникальному для каждой пары порту). В клиенте происходит следующее:

1. Для каждого компа из переданного списка создается поток, в котором открывается QTcpServer на прослушивание заданного порта, при приходе сигнала newConnection() создается сокет для чтения.
...
2. Для каждого компа из переданного списка создается поток, содержащий сокет, который пытается подключиться к компу
...
Такая конструкция работает нестабильно - ...

Как грамотно организовать создание потоков в моем случае?

Во-первых - слишком много потоков. Сервер нужен один, который слушает все порты. В случае, если происходит новое соединение из списка, то тогда можно создать сокет и поместить его в отдельный поток. Зачем запускать кучу серверов в отдельных потоках - не понятно. Одного сокета на соединение достаточно, зачем нужно их два (на чтение и запись) - мне не оч. понятно.
Ну и прежде всего - начните с проекта. Есть ли общий сервер в сети, откуда берется список адресов-портов, кто инициатор соединения, что происходит при включении-отключении и т.д.

[borune]

Во-первых - слишком много потоков. Сервер нужен один, который слушает все порты. В случае, если происходит новое соединение из списка, то тогда можно создать сокет и поместить его в отдельный поток. Зачем запускать кучу серверов в отдельных потоках - не понятно.

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

Одного сокета на соединение достаточно, зачем нужно их два (на чтение и запись) - мне не оч. понятно.

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

upd а как кстати заставить сервер слушать все порты?

Сообщение отредактировал borune - 29.10.2013, 10:37