Re: Socket switch delay

From: Alexander Nickolov (agnickolov_at_mvps.org)
Date: 08/24/04 

Date: Tue, 24 Aug 2004 11:32:05 -0700

I went over all your code so far. My first impression is it is
very inefficient, both at the client and at the server (and why
would you set the send buffer size to zero on a non-overlapped
socket escapes me...). One glaring error is your client does
not check it sent all data in one go (synchronous send can
still accept less data than you offered it), though I have no
idea if this is your problem. So when you use a single socket,
what exactly happens with this code and on which side?

As far as design, do not attempt to force packet structure
on the socket I/O level. Use it as a stream and compose/
decompose your packets in your own buffers. Overlapped
send and receive (and I noticed you don't do overlapped
receive at the server!) means you queue buffers, you don't
wait for the socket with select. For sending, your buffers
will be used directly by the TCP driver bypassing the socket
layer buffer, so several queued overlapped send operations
at any time are essential to performance. If you want efficient
overlapped receive, you need to queue many receive buffers.
On the client, provide a large receive buffer. Use at least
8KB send buffer (this is the default), or performance may
degrade (though if your requests are small and/or relatively
rare this does not impact performance much).

Finally, I've never mixed overlapped with non-overlapped I/O
on the same socket. Could that be your problem? 

-- 
=====================================
Alexander Nickolov
Microsoft MVP [VC], MCSD
email: agnickolov@mvps.org
MVP VC FAQ: http://www.mvps.org/vcfaq
=====================================
"mil" <mil@discussions.microsoft.com> wrote in message 
news:B411F92B-6991-494F-A110-5802E0604426@microsoft.com...
> "Alexander Nickolov" wrote:
>
>>BTW, it may be my own ignorance, but overlapped I/O is
>>_designed_ to work with non-blocking sockets, no? I'll
>>strongly second Alun here that you should rethink your
>>code and work with non-blocking sockets (and possibly
>>a single thread for read/write).
>
>
> I am using non-blocking sockets. But for the Overlapped IO (in the 
> server),
> although I've added the code to make them non-blocking, it doesn't really
> matter, because it was working fine without it. Unless you mean that 
> because
> I was using blocking sockets in the server before (even with the 
> Overlapped
> IO) it was causing the problem of not being able to send from client and
> server at the same time.
>
>
>>No, you are wrong. Show some code, you probably have a
>>subtle bug...
>
> I hope I do have a bug and you find it :).but I did run lots of tests and
> always, when the client was sending data while the server was also sending
> data there was a problem.
> (Although when I was running the tests the server side was using blocking
> sockets).
>
> Keep in mind also that when the client is sending data and the server is
> sending replies, the client is receiving the replies through another 
> thread.
>
>
> Here it is.
>
>
> THE SERVER CREATES THE LISTEN PORT:
>
>
>
>
> //...Create the listening socket
> m_scktListen = WSASocket(AF_INET,
> SOCK_STREAM,
> IPPROTO_TCP,
> NULL,
> 0,
> WSA_FLAG_OVERLAPPED);
> if(m_scktListen!=INVALID_SOCKET)
> {
> si_addrlocal.sin_family = AF_INET;
> si_addrlocal.sin_port = htons(m_ushPort);
> si_addrlocal.sin_addr.s_addr = m_lIPAddressListen;
>
> nRet = bind(m_scktListen, (struct sockaddr *)&si_addrlocal,
> sizeof(si_addrlocal));
> if(nRet!=SOCKET_ERROR)
> {
> if(m_pRtmContextRef->dwMaxConnListenBacklog>SOMAXCONN)
> {
> m_pRtmContextRef->dwMaxConnListenBacklog=SOMAXCONN;
> }
> else if(m_pRtmContextRef->dwMaxConnListenBacklog<1)
> {
> m_pRtmContextRef->dwMaxConnListenBacklog=1;
> }
>
> nRet=listen(m_scktListen, 
> (int)(m_pRtmContextRef->dwMaxConnListenBacklog));
> if(nRet!=SOCKET_ERROR)
> {
> int nScktBufferSize;
>
> if(!m_bDisableZeroSendBuffering)
> {
> nScktBufferSize=0;
> //...Disable send buffering on the socket.  Setting SO_SNDBUF to 0
> causes winsock to stop
> //...bufferring sends and perform sends directly from our buffers,
> thereby reducing CPU usage.
> nRet=setsockopt(m_scktListen, SOL_SOCKET, SO_SNDBUF, (char
> *)&nScktBufferSize, sizeof(nScktBufferSize));
> if(nRet==SOCKET_ERROR)
> {
> hr=OM8ERR_WSA_SO_SNDBUF;
> m_errorLogHlpr.ReportErrorSTRWSA(hr,szActionCtx);
> }
> }
>
> if(SUCCEEDED(hr))
> {
> //...Set the receive buffer size
> nScktBufferSize=OM8_MAXIMUM_PACKET_SIZE+OM8_PACKETHEADER_SIZE;
> nRet=setsockopt(m_scktListen, SOL_SOCKET, SO_RCVBUF, (char
> *)&nScktBufferSize, sizeof(nScktBufferSize));
> if(nRet!=SOCKET_ERROR)
> {
> LINGER lingerStruct;
>
> lingerStruct.l_onoff  = 1;
> lingerStruct.l_linger = 0;
>
> nRet = setsockopt(m_scktListen, SOL_SOCKET, SO_LINGER,(char
> *)&lingerStruct, sizeof(lingerStruct) );
> if(nRet!=SOCKET_ERROR)
> {
>
>
>
> THE SERVER SENDS A NUMBER OF PACKETS:
>
> dwSendBytes=dwFlags=0;
> ZeroMemory(&(pIOCtxPacketOUT->Overlapped),sizeof(WSAOVERLAPPED));
>
> //...Need to call this function before we execute the OverlappedIO 
> function,
> //...because it may free the IO thread before we have registered the 
> seesion
> //...with the pendingIO sessions list and then we will be out of sync
> IncrementIO();
> IncrementSENDIOStats();
>
> //...Always use the secondary connection for sending
> //...feedback messages
> iRet=WSASend(m_scktConnectionSecondary,
> pIOCtxPacketOUT->wsabuf,
> pIOCtxPacketOUT->dwNumberOfBuffersUsed,
> &dwSendBytes,
> dwFlags,
> &(pIOCtxPacketOUT->Overlapped),
> NULL);
>
> if(iRet==SOCKET_ERROR)
> {
> int iLastWSError=WSAGetLastError();
>
> if(iLastWSError==ERROR_IO_PENDING)
> {
> //...Do not allow the caller to re-use this packet
> pIOCtxPacketOUT=NULL;
> }
> else
> {
> hr=OM8ERR_SEND;
> ATLASSERT(SUCCEEDED(hr));
> DecrementIO();
> DecrementSENDIOStats();
> }
> }
> else
> {
> //...Do not allow the caller to re-use this packet
> pIOCtxPacketOUT=NULL;
> }
>
>
>
> THE SERVER RECEIVES DATA
>
>
>
> if(pIOCtxPacket->operationState==OM8IO_OPST_SOCKET_RECV_PACKET_HDR)
> {
> pIOCtxPacket->wsabuf[0].buf=(((char*)&(pIOCtxPacket->packetHeader))+pIOCtxPacket->dwBytesSoFar);
> pIOCtxPacket->wsabuf[0].len=pIOCtxPacket->dwBytesTotal-pIOCtxPacket->dwBytesSoFar;
> pIOCtxPacket->dwNumberOfBuffersUsed=1;
>
> //...Need to call this function before we execute the OverlappedIO 
> function,
> //...because it may free the IO thread before we have registered the 
> seesion
> //...with the pendingIO sessions list and then we will be out of sync
> IncrementIO();
> IncrementRECVIOStats();
>
> ZeroMemory(&(pIOCtxPacket->Overlapped),sizeof(WSAOVERLAPPED));
> iRet=WSARecv(m_scktConnectionPrimary,
> pIOCtxPacket->wsabuf,
> pIOCtxPacket->dwNumberOfBuffersUsed,
> &dwRecvBytes,
> &dwFlags,
> &(pIOCtxPacket->Overlapped),
> NULL);
>
> if(iRet==SOCKET_ERROR)
> {
> int iLastWSError=WSAGetLastError();
> if(iLastWSError!=ERROR_IO_PENDING)
> {
> hr=OM8ERR_RECEIVE;
> ATLASSERT(SUCCEEDED(hr));
> DecrementIO();
> DecrementRECVIOStats();
> }
> }
> }//...It is a bit more complex because we may receive more than one blocks
> at once
> else if(pIOCtxPacket->operationState==OM8IO_OPST_SOCKET_RECV_PACKET_BODY)
> {
>
>
>
> THE CLIENT SETS UP THE CONNECTION
>
> //...Create the socket
> m_socketPrimary=socket(AF_INET,SOCK_STREAM,IPPROTO_TCP);
> if(m_socketPrimary==INVALID_SOCKET)
> {
> hr=OM8ERR_CREATESOCKET;
> m_errorLogHlpr.ReportErrorSTRWSA(hr,lpActionCtx);
> }
> else
> {
> if(dwConnectionFlags & OM8CLIENTSESSION_CONNECT_FLAGS_SINGLECONNECTION)
> {
> m_socketSecondary=m_socketPrimary;
> m_bSingleConnection=TRUE;
> }
> else
> {
> m_socketSecondary=socket(AF_INET,SOCK_STREAM,IPPROTO_TCP);
> if(m_socketSecondary==INVALID_SOCKET)
> {
> hr=OM8ERR_CREATESOCKET;
> m_errorLogHlpr.ReportErrorSTRWSA(hr,lpActionCtx);
> }
> }
> }
>
> //...Set the sockets to non-blocking mode
> if(SUCCEEDED(hr))
> {
> if(ioctlsocket(m_socketPrimary,FIONBIO,&ul)==SOCKET_ERROR)
> {
> hr=OM8ERR_FIONBIO;
> m_errorLogHlpr.ReportErrorSTRWSA(hr,lpActionCtx);
> }
> else if(!m_bSingleConnection)
> {
> ul=1;
> if(ioctlsocket(m_socketSecondary,FIONBIO,&ul)==SOCKET_ERROR)
> {
> hr=OM8ERR_FIONBIO;
> m_errorLogHlpr.ReportErrorSTRWSA(hr,lpActionCtx);
> }
> }
> }
>
> //...Finally connect
> if(SUCCEEDED(hr))
> {
> int iScktBufferSize;
> OM8_OSVERSION osVersion;
> OM8_OSTYPE osType;
> OM8_OSSUITE osSuite;
> DWORD dwMaxBufferSize=OM8_MAXIMUM_PACKET_SIZE+OM8_PACKETHEADER_SIZE;
>
> //...Need to know which OS are we running on
> OM8GetOSVersion(osVersion,osType,osSuite);
>
> //...We need a different setting for different OS versions
> if(osVersion<OM8OSVERSION_WINDOWS_XP)
> {
> //...It seems that Windows 2000 doesn't like 0 send buffer in the client
> part
> iScktBufferSize=dwMaxBufferSize;
> }
> else
> {
> //...Disable send buffering on the socket.  Setting SO_SNDBUF to 0 causes
> winsock to stop
> //...bufferring sends and perform sends directly from our buffers, thereby
> reducing CPU usage.
> iScktBufferSize=0;
> }
>
> iRet=setsockopt(m_socketPrimary, SOL_SOCKET, SO_SNDBUF, (char
> *)&iScktBufferSize, sizeof(iScktBufferSize));
> if((iRet!=SOCKET_ERROR) && !m_bSingleConnection)
> {
> iRet=setsockopt(m_socketSecondary, SOL_SOCKET, SO_SNDBUF, (char
> *)&iScktBufferSize, sizeof(iScktBufferSize));
> if(iRet==SOCKET_ERROR)
> {
> hr=OM8ERR_WSA_SO_SNDBUF;
> m_errorLogHlpr.ReportErrorSTRWSA(hr,lpActionCtx);
> }
> }
> else
> {
> hr=OM8ERR_WSA_SO_SNDBUF;
> m_errorLogHlpr.ReportErrorSTRWSA(hr,lpActionCtx);
> }
>
> if(iRet!=SOCKET_ERROR)
> {
> //...Set the receive buffer size to 4 times the packet+header size, that
> way the server
> //...can send data faster without having to rely on our receiver thread to
> process data
> //...fast enough
> iScktBufferSize=(OM8_MAXIMUM_PACKET_SIZE+OM8_PACKETHEADER_SIZE)<<2;
>
> iRet=setsockopt(m_socketPrimary, SOL_SOCKET, SO_RCVBUF, (char
> *)&iScktBufferSize, sizeof(iScktBufferSize));
> if((iRet!=SOCKET_ERROR) && !m_bSingleConnection)
> {
> iRet=setsockopt(m_socketSecondary, SOL_SOCKET, SO_RCVBUF, (char
> *)&iScktBufferSize, sizeof(iScktBufferSize));
> if(iRet==SOCKET_ERROR)
> {
> hr=OM8ERR_WSA_SO_RCVBUF;
> m_errorLogHlpr.ReportErrorSTRWSA(hr,lpActionCtx);
> }
> }
> else
> {
> hr=OM8ERR_WSA_SO_RCVBUF;
> m_errorLogHlpr.ReportErrorSTRWSA(hr,lpActionCtx);
> }
> }
>
>
>
> THE CLIENT CONNECTS
>
>
> HRESULT hr=OM8ERR_CONNECT;
>
> if(connect(sck,(struct sockaddr*)pServer,sizeof(struct
> sockaddr_in))!=SOCKET_ERROR)
> {
> hr=S_OK;
> }
> else
> {
> switch(WSAGetLastError())
> {
> case WSAEINPROGRESS:
> //...Pass through
> case WSAEWOULDBLOCK:
> {
> struct  timeval tmVal;
> fd_set fdWrite;
>
> //...Wait for 'lSecsWait' seconds to establish a connection
> tmVal.tv_sec  = lSecsWait;
> tmVal.tv_usec = 0;
>
> FD_ZERO(&fdWrite);
> FD_SET(sck,&fdWrite);
>
> if(select(0,NULL,&fdWrite,NULL,&tmVal)==1)
> {
> hr=S_OK;
> }
> }
> break;
>
> default:
> break;
> }
> }
>
> return hr;
>
>
>
> THE CLIENT SENDS DATA
>
> It actually calls this routine with a number of data blocks, one for the
> header and many for the body. Then it blocks (or not) on an event, that is
> signaled by the receiver thread.
>
>
>
> while(IsConnectionAlive())
> {
> tmVal.tv_sec  = PCKIN_TIMETWAIT;
> tmVal.tv_usec = 0;
>
> FD_ZERO(&fdWrite);
> FD_SET(socket,&fdWrite);
>
> iRet=select(0,NULL,&fdWrite,NULL,&tmVal);
> if(iRet==SOCKET_ERROR)
> {
> if(WSAGetLastError()!=WSAEINPROGRESS)
> {
> m_bSendDataStatus=FALSE;
> return OM8ERR_SEND;
> }
> }
> else if(iRet==0)
> {
> dwTotalTimeOut+=PCKIN_TIMETWAIT;
> if(dwTotalTimeOut>PCKSEND_TIMEOUT)
> {
> m_bSendDataStatus=FALSE;
> return OM8ERR_SENDTIMEOUT;
> }
> }
> else
> {
> dwNumberOfBytesSent=0;
>
> #ifdef _DEBUG
> ATLTRACE(_T("CLOCK: 'sendMultipleBuffers - select' total time
> %u\n"),clock()-tStart);
> #endif
>
> iRet=WSASend( socket,
> lpBuffers,
> dwBufferCount,
> &dwNumberOfBytesSent,
> flags,
> NULL,
> NULL);
>
> if(iRet==SOCKET_ERROR)
> {
> int iLastError=WSAGetLastError();
>
> switch(iLastError)
> {
> case WSAEINPROGRESS:
> //...Pass through
> case WSAEWOULDBLOCK:
> continue;
>
> default:
> if(IsRunning())
> {
> ShutDown(false); //...close the connection since there is a problem
> m_bSendDataStatus=FALSE;
> return OM8ERR_SEND;
> }
> else
> {
> m_bSendDataStatus=FALSE;
> return OM8ERR_CONNECTIONCLOSED;
> }
> break;
> }
>
> m_bSendDataStatus=FALSE;
> return SOCKET_ERROR;
> }
> else
> {
> #ifdef _DEBUG
> ATLTRACE(_T("CLOCK: 'sendMultipleBuffers - WSASend' total time
> %u\n"),clock()-tStart);
> #endif
> m_bSendDataStatus=FALSE;
> return S_OK;
> }
> }
> }
>
>
>
> THE CLIENT RECEIVES DATA
>
> This function is called to receive the header block and any subsequent
> blocks repeatedly from the extra client thread (actually 2 threads when 2
> sockets are used).
>
>
>
> while(IsConnectionAlive())
> {
> tmVal.tv_sec  = PCKIN_TIMETWAIT;
> tmVal.tv_usec = 0;
>
> FD_ZERO(&fdRead);
> FD_SET(socket,&fdRead);
>
> //...Wait for th esocket to get some data
> iRet=select(0,&fdRead,NULL,NULL,&tmVal);
> if(iRet==SOCKET_ERROR)
> {
> if(WSAGetLastError()!=WSAEINPROGRESS)
> {
> m_bRecvDataStatus=FALSE;
> return OM8ERR_RECEIVE;
> }
> }
> else if(iRet==0)
> {
> //...just keep waiting
> continue;
> }
> else if(iBytesRecvSoFar<len)
> {
> //...Read the remaining bytes in the buffer
> iRet=recv(socket,buf+iBytesRecvSoFar,len-iBytesRecvSoFar,flags);
> if(iRet==SOCKET_ERROR)
> {
> int iLastError=WSAGetLastError();
>
> switch(iLastError)
> {
> case WSAEINPROGRESS:
> //...Pass through
> case WSAEWOULDBLOCK:
> continue;
>
> default:
> break;
> }
>
> m_bRecvDataStatus=FALSE;
> return OM8ERR_RECEIVE;
> }
> else
> {
> if(!iRet)
> {
> m_bRecvDataStatus=FALSE;
> return OM8ERR_RECEIVE;
> }
>
> iBytesRecvSoFar+=iRet;
> }
>
>
>
> mil
>
>

Relevant Pages

  • Re: bin/131567: Update for regression/sockets/unix_cmsg
    ... -Usually each test does following steps: create Server, fork Client, ... Sometimes Client sends several ... +If Client sends some ancillary data object, ... Receiving sockcred (listening socket has LOCAL_CREDS) ...
    (freebsd-net)
  • Re: bin/131567: Update for regression/sockets/unix_cmsg
    ... -Usually each test does following steps: create Server, fork Client, ... Sometimes Client sends several ... +If Client sends some ancillary data object, ... Receiving sockcred (listening socket has LOCAL_CREDS) ...
    (freebsd-net)
  • Re: bin/131567: Update for regression/sockets/unix_cmsg
    ... -Usually each test does following steps: create Server, fork Client, ... Sometimes Client sends several ... +in the msg_iov field from struct msghdr. ... Receiving sockcred (listening socket has LOCAL_CREDS) ...
    (freebsd-net)
  • Re: Locking on async calls
    ... you must synchronize the entire SendMessage routine as an atomic ... operation to prevent mixed messages from being transmitted to the server. ... You are correct that read and write on the socket do not interfere with each ... If you want to handle multiple client connections from one server object ...
    (microsoft.public.dotnet.general)
  • Re: Socket switch delay
    ... The server uses blocking sockets just because I am also using Overlapped IO ... structures to send the packets. ... The reason I am using a 0 bytes send buffer in my socket (i.e. ...
    (microsoft.public.win32.programmer.networks)