Throttling

[Syrinx]

I have been coding a httpd using asynchronous functions with non-blocking sockets.It's almost done but I fail to find a method to limit the file transfer speed.Any good idea?

Thanks

[darkone]

Throttling isn really a trivial task, but implementation is dependant on wheter you use asynchronous, synchronous or overlapped io.

With overlapped io, you have to assume that every read/write you post - is going to return immeaditely. Before performing the read/write, you will have to check if more reads/writes are allowed.

If operation is allowed, you will truncate the request to fixed size of ie. 1024bytes, and reduce the amount from available bandwidth.

If no more reads/writes are allowed, request is to be queued. Requests from queue are released, once more bandwidth becomes available. By default io allows more requests every 200ms. If limit of allowed read/writes requests wasn't reached during last cycle, remainder can be added to current cycle. However, to prevent number of allowed requests from growing to infinite, it limits number of allowed requests to 1.5x of requests allowed per cycle.

... so: If daemon used 100kb/sec out of 500kb/sec that was allowed during 200ms cycle, next cycle has 'min(500kb/sec * 1.5, 500kb/sec + (500kb/sec - 100kb/sec))' = '750kb/sec' to spare.


With async sockets, you can return the allocated bandwidth to pool, if send/read returns (WSAEWOULDBLOCK) .

Right.. I've been coding ssl for last 13hours, so my head is feeling dizzy... I can't really paste any code examples atm, because what I'm using would require pasting code for the whole core of current io. (which could potentially reveal exploits)

[Syrinx]

First of all,great thanks for the post even I still don't really understand it.
I know all about overlapped IO stuff,I just can't get the calculation right and accurate.
Thanks again for the post

[darkone]

Ay, it's sometimes hard to understand what I'm saying. Here's old send routine:

BOOL SendOverlapped(LPIOSOCKET lpSocket, LPSOCKETOVERLAPPED lpOverlapped)
{
	LPIODEVICE	lpDevice;
	LPBANDWIDTH	lpBandwidth;
	DWORD		dwBytesSent, dwToSend, n;

	if (dwSchedulerUpdateSpeed &&
		(lpDevice = lpSocket->lpDevice) &&	
		(lpDevice->Outbound.bGlobalBandwidthLimit || lpSocket->Options.dwSendLimit))
	{
		lpBandwidth	= &lpDevice->Outbound;
		dwToSend	= 1024;
		//	Calculate maximum send amount
		for (n = 0;n < lpOverlapped->dwBuffers;n++)
		{
			if (lpOverlapped->Buffer[n].len > dwToSend)
			{
				lpOverlapped->Buffer[n].len	= dwToSend;
				lpOverlapped->dwBuffers	= n + 1;
				break;
			}
			else dwToSend	-= lpOverlapped->Buffer[n].len;
		}

		while (InterlockedExchange(&lpBandwidth->lLock, TRUE)) SwitchToThread();
		//	Check available bandwidth on device
		if (! lpBandwidth->dwGlobalBandwidthLeft)
		{
			//	Push item to queue
			if (! lpBandwidth->lpIOQueue[0][HEAD])
			{
				lpBandwidth->lpIOQueue[0][HEAD]	= lpOverlapped;
			}
			else lpBandwidth->lpIOQueue[0][TAIL]->lpNext	= lpOverlapped;
			lpBandwidth->lpIOQueue[0][TAIL]	= lpOverlapped;
			lpBandwidth->dwIOQueue[0]++;
			InterlockedExchange(&lpBandwidth->lLock, FALSE);
			return -1;
		}
#ifdef _REGISTERED
		//	Check available bandwidth for user
		if (lpSocket->Options.dwSendLimit &&
			! lpSocket->dwBandwidthLimit[0]--)
		{
			if (! lpBandwidth->lpIOQueue[1][HEAD])
			{
				lpBandwidth->lpIOQueue[1][HEAD]	= lpOverlapped;
			}
			else lpBandwidth->lpIOQueue[1][TAIL]->lpNext	= lpOverlapped;
			lpBandwidth->lpIOQueue[1][TAIL]	= lpOverlapped;
			lpBandwidth->dwIOQueue[1]++;
			lpSocket->dwBandwidthLimit[0]	= lpSocket->Options.dwSendLimit - 1;
			InterlockedExchange(&lpBandwidth->lLock, FALSE);
			return -1;
		}
#endif
		lpBandwidth->dwGlobalBandwidthLeft--;
		InterlockedExchange(&lpBandwidth->lLock, FALSE);
	}
	//	Send data
	if (WSASend(lpSocket->Socket, lpOverlapped->Buffer, lpOverlapped->dwBuffers,
		&dwBytesSent, 0, (LPWSAOVERLAPPED)lpOverlapped, NULL) == SOCKET_ERROR &&
		WSAGetLastError() != WSA_IO_PENDING) return -2;
	return -1;
}

As you can notice it throws request to queue, if bandwidth counter hits zero. Queued items are released by dedicated thread every few hundred milliseconds.

[Syrinx]

Thanks for the code,I would take a very close look.By the way,
if WSASend() return WSAEWOULDBLOCK error code,which means
There are too many outstanding overlapped I/O requests,and will the system deliver the data later?

[darkone]

If you use overlapped, it will never return WSAEWOULDBLOCK. Only non-error return values for overlapped are, TRUE or FALSE with GetLastError() returning WSA_IO_PENDING.

That error only shows up with WSAEventSelect(), WSAAsyncSelect(), select(). And it means that no data was moved to/copied from socket buffers (either socket buffer was empty or full).

[Syrinx]

Thanks for explaining.

By the way
I think of an interesting issue today about the pasv mode data connection in FTPD.
If the FTPD uses asynchronous functions with non-blocking sockets or IO Completion Ports model,passive mode data connections can be accepted both before and after a file transfer command (RETR, STOR, LIST, etc.) is issued,which is pretty weird,
isn't it?

[darkone]

Connection should be accepted after the transfer command has been issued.


... and btw, if you do use iocp with multiple simultanous reads/writes per handle/socket, that method doesn't work for throttling.

[Syrinx]

I agree that "Connection should be accepted after the transfer command has been issued",but some FTP clients also use asynchronous functions with non-blocking sockets such as CUTE FTP and Flashfxp.The clients may send the command before connect to the data port,but sometimes due to the asynchronous
nature,data connection is accepted before the command has been received.

[darkone]

Well, it's up to you to accept it or not at that point. I personally prefer to do the accept() after the connection has been established.