8
\$\begingroup\$

I been working on getting the SocketAsyncEventArgs to work the way I want it to work. Now I was wondering: is this going to work how it should work?

/// <summary>
/// The settings to use with this ServerSocket.
/// </summary>
ServerSocketSettings Settings;

/// <summary>
/// The buffer manager for allocation a buffer block to a SocketAsyncEventArgs.
/// </summary>
BufferManager BufferManager;

/// <summary>
/// The semaphore used for controlling the max connections to the server.
/// </summary>
SemaphoreSlim MaxConnectionsEnforcer;

/// <summary>
/// The socket used for listening for incoming connections.
/// </summary>
Socket ListenSocket;

/// <summary>
/// The pool of re-usable SocketAsyncEventArgs for accept operations.
/// </summary>
SocketAsyncEventArgsPool PoolOfAcceptEventArgs;

/// <summary>
/// The pool of re-usable SocketAsyncEventArgs for receiving data.
/// </summary>
SocketAsyncEventArgsPool PoolOfRecEventArgs;

/// <summary>
/// The pool of re-usable SocketAsyncEventArgs for sending data.
/// </summary>
SocketAsyncEventArgsPool PoolOfSendEventArgs;

/// <summary>
/// Initializes a new instance of the Non-blocking I/O ServerSocket.
/// </summary>
/// <param name="settings">The settings to use with this ServerSocket.</param>
public ServerSocket(ServerSocketSettings settings)
{
    this.Settings = settings;

    this.BufferManager = new BufferManager((this.Settings.BufferSize * this.Settings.NumOfSaeaForRec) + (this.Settings.BufferSize * this.Settings.NumOfSaeaForSend) * this.Settings.OpsToPreAllocate,
        this.Settings.BufferSize * this.Settings.OpsToPreAllocate);
    this.PoolOfAcceptEventArgs = new SocketAsyncEventArgsPool(this.Settings.MaxSimultaneousAcceptOps);
    this.PoolOfRecEventArgs = new SocketAsyncEventArgsPool(this.Settings.NumOfSaeaForRec);
    this.PoolOfSendEventArgs = new SocketAsyncEventArgsPool(this.Settings.NumOfSaeaForSend);

    this.MaxConnectionsEnforcer = new SemaphoreSlim(this.Settings.MaxConnections, this.Settings.MaxConnections);
}

internal void Init()
{
    this.BufferManager.InitBuffer();

    for (int i = 0; i < this.Settings.MaxSimultaneousAcceptOps; i++)
    {
        SocketAsyncEventArgs acceptEventArg = new SocketAsyncEventArgs();
        acceptEventArg.Completed +=
            new EventHandler<SocketAsyncEventArgs>(AcceptEventArg_Completed);

        this.PoolOfAcceptEventArgs.Push(acceptEventArg);
    }

    // receive objs
    for (int i = 0; i < this.Settings.NumOfSaeaForRec; i++)
    {
        SocketAsyncEventArgs eventArgObjectForPool = new SocketAsyncEventArgs();
        this.BufferManager.SetBuffer(eventArgObjectForPool);

        eventArgObjectForPool.Completed +=
            new EventHandler<SocketAsyncEventArgs>(IO_ReceiveCompleted);
        eventArgObjectForPool.UserToken = new Connection(null, this);
        this.PoolOfRecEventArgs.Push(eventArgObjectForPool);
    }

    // send objs
    for (int i = 0; i < this.Settings.NumOfSaeaForSend; i++)
    {
        SocketAsyncEventArgs eventArgObjectForPool = new SocketAsyncEventArgs();
        this.BufferManager.SetBuffer(eventArgObjectForPool);

        eventArgObjectForPool.Completed +=
            new EventHandler<SocketAsyncEventArgs>(IO_SendCompleted);
        eventArgObjectForPool.UserToken = new SendDataToken();
        this.PoolOfSendEventArgs.Push(eventArgObjectForPool);
    }
}

public void StartListen()
{
    this.ListenSocket = new Socket(this.Settings.Endpoint.AddressFamily, SocketType.Stream, ProtocolType.Tcp);
    this.ListenSocket.Bind(this.Settings.Endpoint);
    this.ListenSocket.Listen(this.Settings.Backlog);
}

internal void StartAccept()
{
    SocketAsyncEventArgs acceptEventArgs;

    if (this.PoolOfAcceptEventArgs.TryPop(out acceptEventArgs))
    {
        this.MaxConnectionsEnforcer.Wait();
        bool willRaiseEvent = this.ListenSocket.AcceptAsync(acceptEventArgs);

        if (!willRaiseEvent)
        {
            ProcessAccept(acceptEventArgs);
        }
    }
}

private void AcceptEventArg_Completed(object sender, SocketAsyncEventArgs e)
{
    ProcessAccept(e);
}

private void ProcessAccept(SocketAsyncEventArgs acceptEventArgs)
{
    if (acceptEventArgs.SocketError != SocketError.Success)
    {
        HandleBadAccept(acceptEventArgs);
        return;
    }

    StartAccept();

    SocketAsyncEventArgs recEventArgs;

    if (this.PoolOfRecEventArgs.TryPop(out recEventArgs))
    {
        ((Connection)recEventArgs.UserToken).Socket = acceptEventArgs.AcceptSocket;

        acceptEventArgs.AcceptSocket = null;
        this.PoolOfAcceptEventArgs.Push(acceptEventArgs);

        StartReceive(recEventArgs);
    }
    else
    {
        HandleBadAccept(acceptEventArgs);
        throw new InvalidOperationException("We starved the receive pool for objects, make sure it matches the max connections.");
    }
}

private void IO_SendCompleted(object sender, SocketAsyncEventArgs e)
{
    ProcessSend(e);
}

private void IO_ReceiveCompleted(object sender, SocketAsyncEventArgs e)
{
    ProcessReceive(e);
}

private void StartReceive(SocketAsyncEventArgs receiveEventArgs)
{
    Connection token = (Connection)receiveEventArgs.UserToken;

    bool willRaiseEvent = token.Socket.ReceiveAsync(receiveEventArgs);

    if (!willRaiseEvent)
    {
        ProcessReceive(receiveEventArgs);
    }
}

private void ProcessReceive(SocketAsyncEventArgs receiveEventArgs)
{
    Connection con = (Connection)receiveEventArgs.UserToken;

    if (receiveEventArgs.BytesTransferred > 0 && receiveEventArgs.SocketError == SocketError.Success)
    {

        // NEED TO ADD RECEIVE DATA HERE ETC

        StartReceive(receiveEventArgs);
    }
    else
    {
        CloseClientSocket(receiveEventArgs);
        ReturnReceiveSaea(receiveEventArgs);
    }
}

public void SendData(Socket socket, byte[] data)
{
    SocketAsyncEventArgs sendEventArgs;
    this.PoolOfSendEventArgs.TryPop(out sendEventArgs);

    SendDataToken token = (SendDataToken)sendEventArgs.UserToken;
    token.DataToSend = data;

    sendEventArgs.AcceptSocket = socket;
    StartSend(sendEventArgs);
}

private void StartSend(SocketAsyncEventArgs sendEventArgs)
{
    SendDataToken token = (SendDataToken)sendEventArgs.UserToken;

    if (token.SendBytesRemainingCount <= this.Settings.BufferSize)
    {
        sendEventArgs.SetBuffer(sendEventArgs.Offset, token.SendBytesRemainingCount);
        Buffer.BlockCopy(token.DataToSend, token.BytesSentAlreadyCount, sendEventArgs.Buffer, sendEventArgs.Offset, token.SendBytesRemainingCount);
    }
    else
    {
        sendEventArgs.SetBuffer(sendEventArgs.Offset, this.Settings.BufferSize);
        Buffer.BlockCopy(token.DataToSend, token.BytesSentAlreadyCount, sendEventArgs.Buffer, sendEventArgs.Offset, this.Settings.BufferSize);
    }

    bool willRaiseEvent = sendEventArgs.AcceptSocket.SendAsync(sendEventArgs);

    if (!willRaiseEvent)
    {
        ProcessSend(sendEventArgs);
    }
}

private void ProcessSend(SocketAsyncEventArgs sendEventArgs)
{
    SendDataToken token = (SendDataToken)sendEventArgs.UserToken;

    if (sendEventArgs.SocketError == SocketError.Success)
    {
        token.SendBytesRemainingCount = token.SendBytesRemainingCount - sendEventArgs.BytesTransferred;

        if (token.SendBytesRemainingCount == 0)
        {
            token.Reset();
            this.PoolOfSendEventArgs.Push(sendEventArgs);
        }
        else
        {
            token.BytesSentAlreadyCount += sendEventArgs.BytesTransferred;
            StartSend(sendEventArgs);
        }
    }
    else
    {
        token.Reset();
        CloseClientSocket(sendEventArgs);
        ReturnSendSaea(sendEventArgs);
    }
}

private void CloseClientSocket(SocketAsyncEventArgs args)
{
    Connection con = (Connection)args.UserToken;

    try
    {
        con.Socket.Shutdown(SocketShutdown.Both);
    }
    catch (Exception) { }

    con.Socket.Close();
    con.OnConnectionClose();
}

private void ReturnReceiveSaea(SocketAsyncEventArgs args)
{
    this.PoolOfRecEventArgs.Push(args);
    this.MaxConnectionsEnforcer.Release();
}

private void ReturnSendSaea(SocketAsyncEventArgs args)
{
    this.PoolOfSendEventArgs.Push(args);
}

private void HandleBadAccept(SocketAsyncEventArgs acceptEventArgs)
{
    acceptEventArgs.AcceptSocket.Close();
    this.PoolOfAcceptEventArgs.Push(acceptEventArgs);
}

internal void Shutdown()
{
    this.ListenSocket.Shutdown(SocketShutdown.Receive);
    this.ListenSocket.Close();

    DisposeAllSaeaObjects();
}

private void DisposeAllSaeaObjects()
{
    this.PoolOfAcceptEventArgs.Dispose();
    this.PoolOfSendEventArgs.Dispose();
    this.PoolOfRecEventArgs.Dispose();
}

Is it OK to have 1 receive operation looping and receiving data, but having a pool of multiple send operations? The reason behind this is because I don't want to send data straight after a receive. I want to send data when I wish to, but for receive operations its different, I want it to loop constantly accepting data for a client. Is this going to be FAST and effective?

The only thing I need to handle now is receiving data properly with this code and decoding the data.

| improve this question | | | | |
\$\endgroup\$
  • 2
    \$\begingroup\$ I have found that reuse of the SAEA at extreme rates can actually cause the CLR to fail because of the underlying buffer. I would recommend using a queue of SAEA and process them one after the other. It could have been my approach, or system, but thought i'd flag it for you anyway! If you want to see how I implemented shout back and I will drop a link to you :) \$\endgroup\$ – Stuart Blackler Feb 28 '14 at 13:09
5
\$\begingroup\$

Just a couple observations, unfortunately not directly related to your primary concerns but still [hopefully] valuable input from a code review perspective:

  • The class is tightly coupled with several other classes. This may or may not be problematic, just thought I'd point it out - I consider instantiating objects as a concern on its own, I don't like having new instructions all over my code, so I'd try to move these instructions outside the class if possible.
  • Init() has too many responsibilities, I would extract 3 methods from that code (one for each loop).
  • SocketAsyncEventArgs.UserToken is an object, for convenience. It seems you're sometimes putting a Connection in there, and other times a SendDataToken. This is likely to get confusing and result in InvalidCastException occurrences as the code gets maintained and evolves.

You could create a class that cleanly exposes what you need, and only cast UserToken to that type:

public class CustomUserToken
{
    public class CustomUserToken(Connection connection, SendDataToken token)
    {
        _connection = connection;
        _token = token;
    }

    private readonly Connection _connection;
    public Connection Connection { get { return _connection; } }

    private readonly SendDataToken _token;
    public SendDataToken Token { get { return _token; } }
}

And now you can consistently cast SockedAsyncEventArgs.UserToken to CustomUserToken, and get the Connection and Token from there.

| improve this answer | | | | |
\$\endgroup\$
0
\$\begingroup\$

I've noticed that the socket you use for SendAsync is the SAEA's AcceptSocket. However for ReceiveAsync, you use your UserToken's Socket. Could you use the same instance for both purposes?

| improve this answer | | | | |
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.