소켓 : 수신/구문 분석시 '대기'하지 않고 클라이언트에 데이터를 보내는 방법

Question

나는 boost :: asio를 사용하여 C++로 작성된 소켓 서버를 가지고 있으며 클라이언트에 데이터를 보내고 있습니다.

서버는 데이터를 청크로 보내고 클라이언트는 수신 한대로 각 청크를 구문 분석합니다. 둘 다 꽤 많은 싱글 스레드입니다.

서버가 최대한 빨리 데이터를 쓰고 클라이언트에서 구문 분석을 기다리지 않도록 서버에서 어떤 디자인을 사용해야합니까? 서버에서 비동기식 작업을해야한다고 생각합니다.

클라이언트에서도 변경 작업을 수행 할 수 있다고 생각하지만 이상적으로 서버는 클라이언트 작성 방법에 관계없이 클라이언트를 기다리지 않아야합니다.

이 같은 소켓에 데이터를 쓰고 있어요 :

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

업데이트 : 소켓에 비동기 적으로 작성하는 부스트의 메커니즘을 사용하려고 하겠어

. 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)); 

• 알렉스

Answer 1

소켓을 비 블로킹으로 설정하면, 그렇지 않으면 블로킹이 실패하면 쓰기가 실패합니다. 그런 다음 원하는 데이터를 큐에 넣고 나중에 작성하려고 다시 시도 할 수 있습니다. 부스트 소켓 API에 소켓 옵션을 설정하는 방법을 모르겠지만, 그게 당신이 찾고있는 것입니다.

하지만 이보다 더 큰 문제는 아닙니다. 쓰기가 준비된 소켓을 선택해야합니다. 아마도 여러 개의 소켓을 동시에 열어서 더 많은 데이터를 가득 채울 때까지 밀어 넣고 반복해야합니다. 부스트 소켓 API에 select과 동등한 항목이 있는지 모르겠다. 그래서 쓸 수있을 때까지 한 번에 여러 소켓을 기다릴 수있다.

서버가 일반적으로 클라이언트 연결 당 스레드를 시작 (또는 프로세스를 생성)하는 이유는 I/O를 기다리는 동안 다른 클라이언트가 자신의 큐를 구현하지 않고도 계속 처리 할 수 ​​있기 때문입니다. "나중에 다른 시도를 준비하는"가장 간단한 방법은 전용 스레드에서 I/O를 차단하는 것입니다.

부스트가 소켓 API에서 비정상적인 작업을 수행하지 않으면 수행 할 수없는 작업은 OS 또는 소켓 라이브러리가 차단하지 않고 임의의 양의 데이터를 대기열에 저장해야합니다. 데이터가 기록 될 때 다시 전화 할 수있는 비동기 API가있을 수 있습니다.

Answer 2

당신은하지 TCP를 통한하지만 UDP를 통해 데이터를 전송하여 비동기 통신을 보장 할 수 있습니다. 그러나 TCP를 사용해야하는 경우 클라이언트가 데이터를 빠르게 저장하고 다른 스레드에서 또는 cron 작업과 비동기 적으로 처리하도록하십시오.

Answer 3

데이터를 소켓에 전달할 때 수신기에서 처리하기를 기다리지 않습니다. 데이터가 전송되기를 기다리지도 않습니다. 데이터는 백그라운드에서 OS에 의해 처리되는 아웃 바운드 대기열에 저장됩니다. 쓰기 함수는 실제로 전송 된 바이트 수가 아니라 전송 대기중인 바이트 수를 반환합니다.

Answer 4

스테판의 게시물에 코멘트에서 계속 :

클라이언트 또는 서버 양쪽에 버퍼 확실히 가능하다. 그러나 Neil이 쓴 것을 고려해야합니다. 맹목적으로 데이터를 버퍼링하기 시작한 경우 처리가 전송을 따라 잡을 수없는 경우 우리 버퍼는 원하지 않는 방식으로 커집니다.

저는 최근에 단순한 'NetworkPipe'를 구현했습니다.이 간단한 'NetworkPipe'는 단일 클라이언트/서버, 서버/클라이언트 사이의 연결로 작동하도록 의도되었습니다. 외부 사용자는 파이프가 클라이언트인지 서버인지를 모릅니다. . 당신이 묻는 것과 비슷한 버퍼링 상황을 구현했습니다, 어떻게? 글쎄, 클래스가 스레드 되었다면, 이것은 데이터를 깨끗하게 버퍼링 할 수있는 유일한 방법이었습니다. 다음은 내가 수행 한 기본 프로세스이며 파이프에 최대 크기를 설정했습니다.

1. 프로세스 1은 파이프를 시작하고 기본값은 서버입니다. 이제 내부 스레드가 클라이언트를 기다립니다.
2. 프로세스 2 시작 파이프 (이미 서버)는 기본적으로 클라이언트입니다.
3. 이제 연결이 완료되면 가장 먼저해야 할 일은 최대 버퍼 크기를 교환하는 것입니다.
4. 프로세스 1은 데이터를 씁니다 (다른 쪽 끝에는 빈 버퍼가 있음을 나타냅니다. [참조 # 3])
5. 프로세스 2의 내부 스레드 (이제 소켓의 select()를 기다림) 그것, 그것을 완충시킨다. 프로세스 2는 이제 새로운 버퍼링 된 크기를 P1로 되돌려 보냅니다.

정말 간단한 버전이긴하지만 기본적으로 스레딩을 통해 차단 선택 호출을 기다릴 수 있습니다. 데이터가 도착하면 즉시 읽을 수 있고 버퍼링 할 수 있습니다. 새 버퍼링 된 크기를 다시 전송합니다. 비슷한 것을 할 수 있고 데이터를 맹목적으로 버퍼링 할 수 있습니다. 실제로 버퍼 크기를 교환 할 필요가 없기 때문에 조금 더 간단합니다.하지만 나쁜 생각 일 수 있습니다. 따라서 위 예제는 외부 사용자가 스레드를 차단하지 않고 데이터를 읽거나 쓸 수있게했습니다 (다른 쪽 버퍼가 가득 찼다면 제외).

Answer 5

boost :: asio :: async_write 메서드를 사용하여 솔루션을 구현했습니다.

기본적으로 각 스레드가 데이터의 일부 금액을 누적으로

• 나는 그것을 쓰지 않는, async_write를 사용하여 소켓에 기록, 클라이언트 당 하나 개의 스레드
• (내 스레드가 CPU 바인딩 작업을하고있다)가 이전 쓰기가
• 를 완료 한 경우 코드는 모든 데이터가

기입되기 전에 CPU 처리가 완료 때문에 기입되는 소켓의 수명과 데이터 버퍼를 관리하는주의입니다

이것은 잘 작동합니다. 이렇게하면 서버 스레드가 CPU 작업을 마자 마자 마칠 수 있습니다.

전체적으로 클라이언트가 모든 데이터를 수신하고 구문 분석하는 시간이 단축되었습니다. 마찬가지로 서버가 각 클라이언트에서 소비하는 시간 (벽 시간의 시계)이 감소합니다.

코드 스 니펫 :

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; 
    } 
}