быстрая передача данных через сокеты

кто нибудь знает способ передавать порядка тысячи сообщений в секунду?

сообщения размером 60000 байт
1-й поток принимает сообщения, ставит в очередь(очередь в оперативке)
2-й поток из очереди отправляет сообщения

на приеме-отправке сообщений смог добиться 400 сообщений/сек.
Надо больше!!!!!

использую АПИ функции сокетов

а теперь посчитаем...
60000 байт * 400 сообщений в секунду, что равно 24 000 000 байт в секунду
умножаем на 8 бит в байте получаем 192 000 000 бит в секунду.
Что для 100 мегабитки в полнодуплексном режиме это предел
и большего ты можешь добиться только увеличением скорости сети

=((
да, это я погорячился, конечно, 400 сообщений/сек было при очень небольшом размере

для сообщений 60000 байт: с помощью recv, send посылка их туда-сюда дает низкий результат, за минуту около 5200 сообщений, что составляет 85-90 в секунду.

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

Guest, И увеличивая скорость сети советую ставить свитч, а не хаб.
Головняка меньше в настройке и в плане работы, претензий не было.
Один раз хаб пришлось душным летом остужать вентилятором, что бв сеть не висла.


Добавлено 30.11.04, 09:11

Цитата

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

В трубу с пропускной способностью 100 ведер в секунду, не прогнать 1000 ведер в секунду.

Максимальный размер кадра 1500 байт
При отправке кадра драйвер сетевой карты переходит на прием, где проверяет, если сеть свободна, то пытается отправить новый кадр
Вот. В сумме загруженнность сети считается максимальным когда 80% от скорости сети.
Соответственно возрастает задержка при отправе кадра, в ожидании паузы.
Соотвественно сколько работает компьютеров в сети, как сильно интенсивно нагружена, то можно приблизительно посчитать, что максимальная скорость в сети для одного компа будет около 80% от скорости сети / количество компов в сети. Учитывая служебные пакеты, дополнительные обертки для данных, пока они доберутся до физического уровня скорость еще снижается.

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

Сдаётся мне, человек что-то типа "ping of death" пишет.
Помнится, Win98 валилась, если её пинговали из-под линукса нефрагментированным пакетом в 64К (или что-то в этом роде, я в этом не очень хорошо разбираюсь)

Цитата Guest, 30.11.04, 11:45, 527953

использую АПИ функции сокетов

Нарисуй код отправляющего / принимающего потока

Я бы тут делал на на отправляющей стороне по 100 (или больше) потоков в каждый
из которых бы передавал строку для отправки. То есть поток конектиться,
отправляет и убиваеться.

На серваке сделал бы прием accept в цикле и при каждем конекте передавал бы в новый
поток этот указатель на новый сокет, поток с него получает инфу и убиваеться.

Изврат конешно, ну а что тут еще можно придумать?

Для увеличения скорости передачи/приема надо пользоваться асинхронными функциями с перекрытием операций в/в
Сокет открывается тоже в асинхронном режиме:
Модель с перекрытием (ее также называют моделью совмещенных операций) – Overlapped I/O. Эта разновидность обмена реализуется только на сокетах, которые при формировании были объявлены для исполнения на них совмещенных операций c помощью функции WSASocket():

SOCKET WSASocket(
    int af,
    int type,
    int protocol,
    LPWSAPROTOCOL_INFO lpProtocolInfo,
    GROUP g,
    DWORD dwFlags);

Три первых параметра полностью соответствуют функции открытия сокета в стиле Беркли. Четвертым параметром является указатель на структуру WSAPROTOCOL_INFO. Если lpProtocolInfo равен NULL, тогда решение, какой сервис-провайдер будет избран, решает ws2_32.dll на основании параметров af, type и protocol. Если поле протокола равно 0, то выбирается тот провайдер, у которого эта характеристика прописана в его информационной структуре.
Если указатель lpProtocolInfo ссылается на конкретную запись провайдера, а первые три параметра сокетной функции записаны в виде предопределенной константы FROM_PROTOCOL_INFO ("брать из информации о протоколе"), тогда сокет создается в соответствии с характеристиками провайдера. Пятый параметр относится к "группам сокетов", и хотя и определен в спецификации WinSock 2.0, практически еще в Windows не реализован.
Последний параметр – это необязательный флаг, который может быть установлен при вызове. В настоящее время лишь один флаг играет большую роль – и этим флагом как раз является флаг WSA_FLAG_OVERLAPPED.
Модель I/O с перекрытием является самой производительной из всех до сих пор рассмотренных моделей. Она впервые появилась в спецификации WinSock 1.1, но была вначале реализована лишь на Windows NT.
Суть этой модели в том, что она позволяет приложению выдать "одновременно" несколько асинхронных запросов на исполнение операций ввода/вывода, используя специальную структуру данных с перекрытием. Затем, по прошествию некоторого времени, приложение может проверить, выполнены ли его запросы и обработать результаты завершившихся операций. В настоящее время эта модель доступна на всех современных версиях Windows, за исключением Windows CE.

=)
спасибо, попробую вариант с помощью WSASocket

проверил использование 2 сокетов:
клиент соединяется с сервером - 1-й сокет (для отправки сообщений)
принимает соединение от сервера - 2-й сокет (для приема сообщений)

создаются 2 потока:
один передает данные по первому сокету
второй принимает данные по второму сокету

передаю сообщения 60000 байт
при этом за минуту приходит 9300 сообщений (150 в сек)

уже лучше, но все равно мало, если прибавить к этому постановку сообщения в очередь и удаление из очереди после отправки

лаги....

bool InitClient()
{
///////////////////////////////// WSAStartup ///////////////////////////////////////////
    WORD wVersionRequested;
    WSADATA wsaData;
    int err;
 
    wVersionRequested = MAKEWORD( 2, 2 );
    err = WSAStartup( wVersionRequested, &wsaData );
 
    if ( err != 0 )
    {      
        child->AddToErrorLog("could not find a usable WinSock DLL");        
        return false;
    }
 
    // Confirm that the WinSock DLL supports 2.2
    if ( LOBYTE( wsaData.wVersion ) != 2 ||HIBYTE( wsaData.wVersion ) != 2 )
    {
        child->AddToErrorLog("could not find a usable WinSock DLL");                                
        WSACleanup();      
        return false;
    }
 
/////////////////////////////////// connect to server //////////////////////////////////////////        
    struct sockaddr_in peer;
    SOCKET skt=0;
    int res;
 
    memset(&peer, 0, sizeof(peer));
    peer.sin_family = AF_INET;              
    peer.sin_port = htons(4000);
    peer.sin_addr.s_addr = inet_addr(m_ActiveServer_ip);    
    
    if( (skt = socket(AF_INET, SOCK_STREAM, 0)) < 0)
    {
        child->ShowErrorMessage(WSAGetLastError());
        return false;
    }
 
    TRACE("\nconnecting to server. port=4000 ip=%s\n", m_ActiveServer_ip);  
 
    if( (res = connect(skt, (struct sockaddr*)&peer, sizeof(peer)) ) != 0 )
    {
        child->ShowErrorMessage(WSAGetLastError());
        closesocket(skt);                  
        return false;
    }
 
 
///////////////////////////// get server connection //////////////////////////////////
    SOCKET listen_skt, read_skt;
    struct sockaddr_in local;
    local.sin_family = PF_INET;        
    local.sin_port = htons(4001);
    local.sin_addr.s_addr = 0;
 
    // create socket
    if( (listen_skt = socket( PF_INET, SOCK_STREAM, 0) ) < 0)
    {
            child->ShowErrorMessage(WSAGetLastError());
    }
 
    // bind address and port to listening socket
    if( (res = bind(listen_skt, (struct sockaddr*)&local, sizeof(local)) ) < 0)
    {
        child->ShowErrorMessage(WSAGetLastError());
    }
    
    // mark socket as listening
    if( (res = listen(listen_skt, 5)) != 0)
    {  
        child->ShowErrorMessage(WSAGetLastError());
    }
        
    if( (read_skt = accept(listen_skt, NULL, NULL)) < 0)
    {      
        child->ShowErrorMessage(WSAGetLastError());
    }
 
    m_sendsocket = skt;
    m_readsocket = read_skt;
 
      AfxBeginThread(SendFunc, NULL);
    AfxBeginThread(GetFunc, NULL);
 
      return true;
}
 
//bool InitServer() - аналогично InitClient()
 
 
UINT SendFunc(LPVOID pParam)
{
    int skt = m_sendsocket;
    long counter=0;
    long size = 0;              
    tag_message1 message;
    
    for(long i=0; i<50000; i++)
        message.param[i] = 'W';
    message.param[i] = '\0';
 
    while(1)
    {
        size = sizeof(tag_message1);
                
        // send size
        if( send(skt, (char*)&size, sizeof(long), 0) < 0)
        {
            DWORD s = WSAGetLastError();
            break;
        }
            
        char* p = reinterpret_cast<char*>(&message);    
        
        // send structure
        if( send(skt, p, size, 0) < 0)
        {
            DWORD s = WSAGetLastError();
            break;      
        }
    }
 
    return 0;
}
 
 
 
 
UINT GetFunc(LPVOID pParam)
{
    int skt = m_readsocket;
    long counter=0;
    long ret;
    long size = 0;      
    tag_message1 message;
 
    for(long i=0; i<50000; i++)
        message.param[i] = 'W';
 
    message.param[i] = '\0';
 
    while(1)
    {
        // get parameter length
        size = sizeof(tag_message1);//strlen(message.param);        
 
        if( recv(skt, (char*)&size, sizeof(long), 0) <=0)
        {
            DWORD s = WSAGetLastError();
            break;
        }
        
        char* buff = new char [size];
        //tag_message1 pmsg = new tag_message1;
            
        long nBytes = recv( skt, buff, size, 0);
        tag_message1 *pmsg = reinterpret_cast<tag_message1*>(buff);
            
        TRACE("\ngot\t%d", counter++);
        delete [] buff;    
    }
 
    return 0;
}

Цитата Guest, 1.12.04, 11:57, 529060

if( recv(skt, (char*)&size, sizeof(long), 0) <=0)

Здесь бага, т.к. совсем необязательно ты примешь 4 байта.

И потом, для чистоты эксперимента TRACE необходимо вынести за цикл.

да, точно

исправил recv на ф-цию след вида

long readn(SOCKET skt, char* buffer, long length)
{
    long cnt, rc;
 
    cnt = length;
 
    while(cnt > 0)
    {
        rc = recv(skt, buffer, cnt, 0);
 
        if(rc < 0)
            return -1;  
 
        if(rc == 0)
            return length - cnt;
        
        buffer += rc;
        cnt -= rc;
    }
 
    return length;
}