Сокеты: как отправить данные клиенту, не ожидая их, когда они получают / анализируют

Question

У меня есть сервер сокетов, написанный на C ++ с использованием boost :: asio, и я отправляю данные клиенту.

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

Какой дизайн мне следует использовать на сервере, чтобы сервер просто записывал данные так быстро, как только мог, и никогда не ждал, пока клиент его проанализирует? Я полагаю, мне нужно сделать что-то асинхронное на сервере.

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

Я записываю данные в сокет так:

size_t bytesWritten = m_Socket.Write( boost::asio::buffer(buffer, bufferSize));

Обновление:

Я собираюсь попробовать использовать механизм Boost для асинхронной записи в сокет. Смотри http://www.boost.org/doc/libs/1_36_0/doc/html/boost_asio/tutorial/tutdaytime3/src.html

, например

 boost::asio::async_write(socket_, boost::asio::buffer(message_),
        boost::bind(&tcp_connection::handle_write, shared_from_this(),
          boost::asio::placeholders::error,
          boost::asio::placeholders::bytes_transferred));

• Alex

Answer 1

Продолжая комментарии к сообщению Стефана:

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

Теперь я недавно внедрил простую 'NetworkPipe', которая должна была функционировать как соединение между одним клиентом / сервером, сервером / клиентом, где внешний пользователь не знает / не заботится, является ли Pipe клиентом или сервером. Я реализовал ситуацию с буферизацией, похожую на ту, о которой вы спрашиваете, как? Ну, класс был многопоточным, это был единственный способ, которым я мог придумать для чистой буферизации данных. Вот основной процесс, которому я следовал, и обратите внимание, что я установил максимальный размер для Pipes:

1. Процесс 1 запускает канал, по умолчанию сервер. Теперь внутренний поток ожидает клиента.
2. Процесс 2 запускает канал, уже сервер, по умолчанию - Клиент.
3. Теперь мы подключены, первое, что нужно сделать, это обменяться максимальными размерами буфера.
4. Процесс 1 записывает данные (он отмечает, что на другом конце есть пустой буфер [см. # 3])
5. Внутренний поток процесса 2 (теперь ожидает выбора () для сокета) видит, что данные отправляются и читает их, буферизует их. Процесс 2 теперь отправляет обратно новый буферный размер P1.

Так что это действительно упрощенная версия, но в основном благодаря многопоточности я всегда могу ждать блокирующего вызова select, как только поступают данные, я могу читать и буферизировать, я отправляю обратно новый буферизованный размер. Вы можете сделать что-то подобное и слепо буферизировать данные, на самом деле это немного проще, потому что вам не нужно обмениваться размерами буфера, но, вероятно, это плохая идея. Таким образом, приведенный выше пример позволил внешним пользователям читать / записывать данные, не блокируя их поток (если буфер на другом конце не заполнен).

Answer 2

Когда вы передаете данные в сокет, он не ждет, пока получатель их обработает. Он даже не ждет передачи данных. Данные помещаются в исходящую очередь, которая обрабатывается ОС в фоновом режиме. Функция записи возвращает, сколько байтов было поставлено в очередь для передачи, а не сколько байтов было фактически передано.

Answer 3

Если вы установите свой сокет в неблокирующее состояние, тогда запись должна завершиться сбоем, если в противном случае они заблокируются. Затем вы можете поставить в очередь данные по своему усмотрению и договориться о следующей попытке сделать это позже. Я не знаю, как установить параметры сокетов в API-интерфейсе буст-сокетов, но это то, что вы ищете.

Но это, вероятно, больше проблем, чем оно того стоит. Вам нужно выбрать сокет, готовый к записи, предположительно из нескольких открытых одновременно, вставить в него больше данных, пока он не заполнится, и повторить. Я не знаю, имеет ли API бустерных сокетов эквивалент select, так что вы можете подождать несколько сокетов одновременно, пока один из них не будет готов к записи.

Причина, по которой серверы обычно запускают поток (или порождают процесс) для каждого клиентского соединения именно для того, чтобы они могли обслуживать других клиентов, ожидая ввода-вывода, избегая при этом реализации собственных очередей. Самый простой способ «организовать другую попытку позже» - просто заблокировать ввод-вывод в выделенном потоке.

То, что вы не можете сделать, если boost не сделал что-то необычное в своем сокете API, - это потребуйте, чтобы ОС или библиотека сокетов поставили в очередь произвольные объемы данных для вас без блокировки. Может существовать асинхронный API, который перезвонит вам при записи данных.

Answer 4

Вы можете обеспечить асинхронную связь, передавая данные не по TCP, а по UDP. Однако если вам нужно использовать TCP, позвольте клиенту быстро сохранить данные и обработать их в другом потоке или асинхронно с заданием cron.

Answer 5

Я реализовал решение, используя метод boost :: asio :: async_write.

В основном:

• У меня есть один поток на клиента (мои потоки выполняют работу, связанную с процессором)
• Поскольку каждый поток накапливает некоторое количество данных, он записывает их в сокет, используя async_write, не заботясь о том, что предыдущие записи завершили
• Код тщательно управляет временем жизни сокета и записываемых буферов данных, поскольку обработка ЦП завершается до того, как все данные будут записаны

Это хорошо работает для меня. Это позволяет серверному потоку завершать работу, как только он выполняет свою работу ЦП.

В целом время, необходимое клиенту для получения и анализа всех своих данных, сократилось. Точно так же время (часы на стене), которое сервер тратит на каждого клиента, уменьшается.

Фрагмент кода:

void SocketStream::Write(const char* data, unsigned int dataLength)
{
    // Make a copy of the data
    // we'll delete it when we get called back via HandleWrite
    char* dataCopy = new char[dataLength];
    memcpy( dataCopy,  data, dataLength );

    boost::asio::async_write
        (
        *m_pSocket,
        boost::asio::buffer(dataCopy, dataLength),
        boost::bind
            (
            &SocketStream::HandleWrite,                     // the address of the method to callback when the write is done
            shared_from_this(),                             // a pointer to this, using shared_from_this to keep us alive
            dataCopy,                                       // first parameter to the HandleWrite method
            boost::asio::placeholders::error,               // placeholder so that async_write can pass us values
            boost::asio::placeholders::bytes_transferred
            )
        );
}

void SocketStream::HandleWrite(const char* data, const boost::system::error_code& error, size_t bytes_transferred)
{
    // Deallocate the buffer now that its been written out
    delete data;

    if ( !error )
    {
        m_BytesWritten += bytes_transferred;
    }
    else
    {
        cout << "SocketStream::HandleWrite received error: " << error.message().c_str() << endl;
    }
}