Does the SocketAsyncEventArgs "pooled byte[]" style help reduce memory requirements?

Question

I can't see how the pooled SocketAsyncEventArgs style helps me reduce memory consumption for a server that serves many concurrent connections.

Yes, it provides an alternative to MS' Begin/End style that aforementioned MSDN page describes as requiring a System.IAsyncResult object be allocated for each asynchronous socket operation.

And inital research lead me to believe for some reason it would allow me to allocate only a handful of byte arrays at most and share them among my thousands of concurrently connected clients.

But it seems that if I want to wait for data on thousands of client connections, I have to call ReceiveAsync thousands of times, providing a different byte array (wrapped in a SocketAsyncEventArgs) every time, and those thousands of arrays will then just sit there until the time a client decides to send, which might well be 10 seconds.

So unless I call ReceiveAsync just around the time a client sends data (or after that, relying on some network stack buffers?) - which is at the client's discretion and unpredictable to the server, I'm out of luck and the byte arrays will sit there, idly waiting for a client to move his bottom.

I was hoping to listen on those thousands of connections with a single byte array (or maybe a single array per listening threads, if parallelizing makes sense), and once any of those connections sends something (which does have to get into some network stack buffer first anyway), it will be copied over into that array, my listener gets called, and once the listener is done the array can be reused.

Is this indeed not possible with Socket.*Async() methods?

Is something like this possible at all with .net's socket library?

Answer 1

It is not possible to share the same memory for multiple socket operations (or if you do you receive undefined results).

You can circumvent this problem by reading only 1 byte at first. When that read completes it is likely that more data will be coming. So for the next read you use a more efficient size such as 4KB (or you interrogate the DataAvailable property - this is about the only valid use case of that property).

Answer 2

The MSDN article explains how pooling works. Essentially:

a) If there is a pool instance available then use that, otherwise create a new instance.

b) After you are done with it return the instance to the pool so it can be reused.

Eventually the pool size will grow to accommodate all requests, or you might for example configure your pool to have a maximum instance count and block when there are requests for an instance, the max pool size has been reached, and the pool is currently empty. This strategy prevents the pool from growing in an uncontrolled way.

Answer 3

Here's the sketch of an implementation that incorporates usr's great byte[1] workaround suggestion, and shows how the somewhat cumbersome Socket.xxxAsync methods can be completely hidden away in a SimpleAsyncSocket, without sacrificing performance.

A simple asynchronous echo server using SimpleAsyncSocket could look like this.

readonly static Encoding Enc = new UTF8Encoding(false);
SimpleAsyncSocket _simpleSocket;

void StartEchoServer(Socket socket)
{
    _simpleSocket = new SimpleAsyncSocket(socket, OnSendCallback,
        _receiveBufferPool, OnReceiveCallback);
}

bool OnReceiveCallback(SimpleAsyncSocket socket,
    ArraySegment<byte> bytes)
{
    var str = Enc.GetString(bytes.Array, bytes.Offset, bytes.Count);
    _simpleSocket.SendAsync(new ArraySegment<byte>(Enc.GetBytes(str)));
    return false;
}

void OnSendCallback(SimpleAsyncSocket asyncSocket,
    ICollection<ArraySegment<byte>> collection, SocketError arg3)
{
    var bytes = collection.First();
    var str = Enc.GetString(bytes.Array, bytes.Offset, bytes.Count);
}

Here's a sketch of the implementation:

class SimpleAsyncSocket
{
    private readonly Socket _socket;
    private readonly Pool<byte[]> _receiveBufferPool;
    private readonly SocketAsyncEventArgs _recvAsyncEventArgs;
    private readonly SocketAsyncEventArgs _sendAsyncEventArgs;
    private readonly byte[] _waitForReceiveEventBuffer = new byte[1];
    private readonly Queue<ArraySegment<byte>> _sendBuffers = new Queue<ArraySegment<byte>>();

    public SimpleAsyncSocket(Socket socket, Action<SimpleAsyncSocket, ICollection<ArraySegment<byte>>, SocketError> sendCallback,
        Pool<byte[]> receiveBufferPool, Func<SimpleAsyncSocket, ArraySegment<byte>, bool> receiveCallback)
    {
        if (socket == null) throw new ArgumentNullException("socket");
        if (sendCallback == null) throw new ArgumentNullException("sendCallback");
        if (receiveBufferPool == null) throw new ArgumentNullException("receiveBufferPool");
        if (receiveCallback == null) throw new ArgumentNullException("receiveCallback");

        _socket = socket;

        _sendAsyncEventArgs = new SocketAsyncEventArgs();
        _sendAsyncEventArgs.UserToken = sendCallback;
        _sendAsyncEventArgs.Completed += SendCompleted;

        _receiveBufferPool = receiveBufferPool;
        _recvAsyncEventArgs = new SocketAsyncEventArgs();
        _recvAsyncEventArgs.UserToken = receiveCallback;
        _recvAsyncEventArgs.Completed += ReceiveCompleted;
        _recvAsyncEventArgs.SetBuffer(_waitForReceiveEventBuffer, 0, 1);
        ReceiveAsyncWithoutTheHassle(_recvAsyncEventArgs);
    }

    public void SendAsync(ArraySegment<byte> buffer)
    {
        lock (_sendBuffers)
            _sendBuffers.Enqueue(buffer);
        StartOrContinueSending();
    }
    private void StartOrContinueSending(bool calledFromCompleted = false)
    {
        lock (_waitForReceiveEventBuffer) // reuse unrelated object for locking
        {
            if (!calledFromCompleted && _sendAsyncEventArgs.BufferList != null)
                return; // still sending
            List<ArraySegment<byte>> buffers = null;
            lock (_sendBuffers)
            {
                if (_sendBuffers.Count > 0)
                {
                    buffers = new List<ArraySegment<byte>>(_sendBuffers);
                    _sendBuffers.Clear();
                }
            }
            _sendAsyncEventArgs.BufferList = buffers; // nothing left to send
            if (buffers == null)
                return;
        }

        if (!_socket.SendAsync(_sendAsyncEventArgs))
            // Someone on stackoverflow claimed that invoking the Completed
            // handler synchronously might end up blowing the stack, which
            // does sound possible. To avoid that guy finding my code and
            // downvoting me for it (and maybe just because it's the right
            // thing to do), let's leave the call stack via the ThreadPool
            ThreadPool.QueueUserWorkItem(state => SendCompleted(this, _sendAsyncEventArgs));
    }
    private void SendCompleted(object sender, SocketAsyncEventArgs args)
    {
        switch (args.LastOperation)
        {
            case SocketAsyncOperation.Send:
                {
                    try
                    {
                        var bytesTransferred = args.BytesTransferred;
                        var sendCallback = (Action<SimpleAsyncSocket, ICollection<ArraySegment<byte>>, SocketError>)args.UserToken;
                        // for the moment, I believe the following commented-out lock is not
                        // necessary, but still have to think it through properly
                        // lock (_waitForReceiveEventBuffer) // reuse unrelated object for locking
                        {
                            sendCallback(this, args.BufferList, args.SocketError);
                        }
                        StartOrContinueSending(true);
                    }
                    catch (Exception e)
                    {
                        args.BufferList = null;
                        // todo: log and disconnect
                    }


                    break;
                }
            case SocketAsyncOperation.None:
                break;
            default:
                throw new Exception("Unsupported operation: " + args.LastOperation);
        }
    }
    private void ReceiveCompleted(object sender, SocketAsyncEventArgs args)
    {
        switch (args.LastOperation)
        {
            case SocketAsyncOperation.Receive:
                {
                    var bytesTransferred = args.BytesTransferred;
                    var buffer = args.Buffer;
                    if (args.BytesTransferred == 0) // remote end closed connection
                    {
                        args.SetBuffer(null, 0, 0);
                        if (buffer != _waitForReceiveEventBuffer)
                            _receiveBufferPool.Return(buffer);

                        // todo: disconnect event
                        return;
                    }
                    if (buffer == _waitForReceiveEventBuffer)
                    {
                        if (args.BytesTransferred == 1)
                        {
                            // we received one byte, there's probably more!
                            var biggerBuffer = _receiveBufferPool.Take();
                            biggerBuffer[0] = _waitForReceiveEventBuffer[0];
                            args.SetBuffer(biggerBuffer, 1, biggerBuffer.Length - 1);
                            ReceiveAsyncWithoutTheHassle(args);
                        }
                        else
                            throw new Exception("What the heck");
                    }
                    else
                    {
                        var callback = (Func<SimpleAsyncSocket, ArraySegment<byte>, bool>)args.UserToken;
                        bool calleeExpectsMoreDataImmediately = false;
                        bool continueReceiving = false;
                        try
                        {
                            var count = args.Offset == 1
                                            // we set the first byte manually from _waitForReceiveEventBuffer
                                            ? bytesTransferred + 1
                                            : bytesTransferred;
                            calleeExpectsMoreDataImmediately = callback(this, new ArraySegment<byte>(buffer, 0, count));
                            continueReceiving = true;
                        }
                        catch (Exception e)
                        {
                            // todo: log and disconnect
                        }
                        finally
                        {
                            if (!calleeExpectsMoreDataImmediately)
                            {
                                args.SetBuffer(_waitForReceiveEventBuffer, 0, 1);
                                _receiveBufferPool.Return(buffer);
                            }
                        }
                        if (continueReceiving)
                            ReceiveAsyncWithoutTheHassle(args);
                    }
                    break;
                }
            case SocketAsyncOperation.None:
                break;
            default:
                throw new Exception("Unsupported operation: " + args.LastOperation);
        }
    }

    private void ReceiveAsyncWithoutTheHassle(SocketAsyncEventArgs args)
    {
        if (!_socket.ReceiveAsync(args))
            // Someone on stackoverflow claimed that invoking the Completed
            // handler synchronously might end up blowing the stack, which
            // does sound possible. To avoid that guy finding my code and
            // downvoting me for it (and maybe just because it's the right
            // thing to do), let's leave the call stack via the ThreadPool
            ThreadPool.QueueUserWorkItem(state => ReceiveCompleted(this, args));
    }
}