3

I have the the following two scenarios that I am testing and one works but the other does not. I have socket server and socket client application running on two different machines

both the scenarios are using the socketasynceventargs

Scenario 1 (Works) create 40k socket clients in a loop, wait for all connections to be established and then all clients send messages to the server at the same time and receive response from the server 10 times(i.e. send/receive happens 10 times).

Scenario 2 (Does not work. I get a lot of connection refusal errors) create 40k socket clients in a loop and send/receive the same 10 messages to the server as soon as each client is connected instead of waiting for the 40k connections to be established.

I cant figure out why my second scenario would fail. i understand that in scenario 1 the server is not doing much until all the 40k connections are made. but it is able to communicate with all the clients at the same time. any ideas??

Thank you for you patience.

here is the socket server code

public class SocketServer
    {

   private static System.Timers.Timer MonitorTimer = new System.Timers.Timer();
        public static SocketServerMonitor socket_monitor = new SocketServerMonitor();
        private int m_numConnections; 
        private int m_receiveBufferSize;
        public static BufferManager m_bufferManager;  
        Socket listenSocket;           

        public static SocketAsyncEventArgsPool m_readWritePool;
        public static int m_numConnectedSockets;    
        private int cnt = 0;

        public static int Closecalled=0;


        public SocketServer(int numConnections, int receiveBufferSize)
        {
            m_numConnectedSockets = 0;
            m_numConnections = numConnections;
            m_receiveBufferSize = receiveBufferSize;


            m_bufferManager = new BufferManager(receiveBufferSize * numConnections ,
               receiveBufferSize);

            m_readWritePool = new SocketAsyncEventArgsPool(numConnections);

        }


        public void Init()
        {
            MonitorTimer.Interval = 30000;
            MonitorTimer.Start();
            MonitorTimer.Elapsed += new System.Timers.ElapsedEventHandler(socket_monitor.Log);


            m_bufferManager.InitBuffer();


            SocketAsyncEventArgs readWriteEventArg;

            for (int i = 0; i < m_numConnections; i++)
            {

                readWriteEventArg = new SocketAsyncEventArgs();

                m_readWritePool.Push(readWriteEventArg);
            }

        }


        public void Start(IPEndPoint localEndPoint)
        {

            listenSocket = new Socket(localEndPoint.AddressFamily, SocketType.Stream, ProtocolType.Tcp);
            listenSocket.Bind(localEndPoint);

            listenSocket.Listen(1000);


            StartAccept(null);
        }

        public void Stop()
        {
            if (listenSocket == null)
                return;
            listenSocket.Close();
            listenSocket = null;


            Thread.Sleep(15000);
        }

        private void StartAccept(SocketAsyncEventArgs acceptEventArg)
        {
            if (acceptEventArg == null)
            {
                acceptEventArg = new SocketAsyncEventArgs();
                acceptEventArg.Completed += new EventHandler<SocketAsyncEventArgs>(AcceptEventArg_Completed);
            }
            else
            {
                // socket must be cleared since the context object is being reused
                acceptEventArg.AcceptSocket = null;
            }

            try
            {
                bool willRaiseEvent = listenSocket.AcceptAsync(acceptEventArg);
                if (!willRaiseEvent)
                {
                    ProcessAccept(acceptEventArg);
                }
            }
            catch (Exception e)
            {

            }
        }


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

        private void ProcessAccept(SocketAsyncEventArgs e)
        {
            Interlocked.Increment(ref m_numConnectedSockets);
            socket_monitor.IncSocketsConnected();


            SocketAsyncEventArgs readEventArgs = m_readWritePool.Pop();
            m_bufferManager.SetBuffer(readEventArgs);

            readEventArgs.UserToken = new AsyncUserToken { id = cnt++, StarTime = DateTime.Now };
            readEventArgs.AcceptSocket = e.AcceptSocket;
            SocketHandler handler=new SocketHandler(readEventArgs);

            StartAccept(e);
        }

    }   






class SocketHandler
    {
        private SocketAsyncEventArgs _socketEventArgs;

        public SocketHandler(SocketAsyncEventArgs socketAsyncEventArgs)
        {
            _socketEventArgs = socketAsyncEventArgs;
            _socketEventArgs.Completed += new EventHandler<SocketAsyncEventArgs>(IO_Completed);
            StartReceive(_socketEventArgs);
        }


        private void StartReceive(SocketAsyncEventArgs receiveSendEventArgs)
        {

            bool willRaiseEvent = receiveSendEventArgs.AcceptSocket.ReceiveAsync(receiveSendEventArgs);
            if (!willRaiseEvent)
            {
                ProcessReceive(receiveSendEventArgs);
            }
        }


        private void ProcessReceive(SocketAsyncEventArgs e)
        {
            // check if the remote host closed the connection
            AsyncUserToken token = (AsyncUserToken)e.UserToken;
            //token.StarTime = DateTime.Now;
            if (e.BytesTransferred > 0 && e.SocketError == SocketError.Success)
            {
                // process the data here

                //reply to client
                byte[] AckData1 = BitConverter.GetBytes(1);
                SendData(AckData1, 0, AckData1.Length, e);


                StartReceive(e);
            }
            else
            {
                CloseClientSocket(e);
            }
        }


        private void IO_Completed(object sender, SocketAsyncEventArgs e)
        {
            // determine which type of operation just completed and call the associated handler 
            switch (e.LastOperation)
            {
                case SocketAsyncOperation.Receive:
                    ProcessReceive(e);
                    break;
                case SocketAsyncOperation.Send:
                    ProcessSend(e);
                    break;
                default:
                    throw new ArgumentException("The last operation completed on the socket was not a receive or send");
            }

        }





        private void CloseClientSocket(SocketAsyncEventArgs e)
        {

            AsyncUserToken token = e.UserToken as AsyncUserToken;

            // close the socket associated with the client 
            try
            {

                e.AcceptSocket.Shutdown(SocketShutdown.Send);
            }

            catch (Exception ex)
            {

            }

            e.AcceptSocket.Close();

            Interlocked.Decrement(ref SocketServer.m_numConnectedSockets);
            SocketServer.socket_monitor.DecSocketsConnected();

            SocketServer.m_bufferManager.FreeBuffer(e);

            e.Completed -= new EventHandler<SocketAsyncEventArgs>(IO_Completed);

            SocketServer.m_readWritePool.Push(e);

        }


        public void SendData(Byte[] data, Int32 offset, Int32 count, SocketAsyncEventArgs args)
        {

            try
            {

                Socket socket = args.AcceptSocket;
                if (socket.Connected)
                {
                    var i = socket.Send(data, offset, count, SocketFlags.None);
                }
            }
            catch (Exception Ex)
            {

            }
        }
    }

here is the client code that throws the error in the connectcallback method

// State object for receiving data from remote device.
public class StateObject
{
    // Client socket.
    public Socket workSocket = null;
    // Size of receive buffer.
    public const int BufferSize = 256;
    // Receive buffer.
    public byte[] buffer = new byte[BufferSize];
    // Received data string.
    public StringBuilder sb = new StringBuilder();

    public int count = 0;
}

public class AsynchronousClient
{
    // The port number for the remote device.
    private const int port = 11000;

    private static int closecalled = 0;
    private static bool wait = true;
    // ManualResetEvent instances signal completion.
    private static ManualResetEvent connectDone =
        new ManualResetEvent(false);
    private static ManualResetEvent sendDone =
        new ManualResetEvent(false);
    private static ManualResetEvent receiveDone =
        new ManualResetEvent(false);

    // The response from the remote device.
    private static String response = String.Empty;

    private static void StartClient(Socket client, IPEndPoint remoteEP)
    {
        // Connect to a remote device.
        try
        {
            // Connect to the remote endpoint.
            client.BeginConnect(remoteEP,
                new AsyncCallback(ConnectCallback), new StateObject { workSocket = client });

        }
        catch (Exception e)
        {
            Console.WriteLine(e.ToString());
        }
    }

    private static void ConnectCallback(IAsyncResult ar)
    {
        try
        {
            // Retrieve the socket from the state object.
            StateObject state = (StateObject)ar.AsyncState;

            var client = state.workSocket;
            // Complete the connection.
            client.EndConnect(ar);
            var data = "AA5500C08308353816050322462F01020102191552E7D3FA52E7D3FB1FF85BF1FE9F201000004AB80000000500060800001EFFB72F0D00002973620000800000FFFFFFFF00009D6D00003278002EE16D0000018500000000000000000000003A0000000100000000828C80661FF8B436FE9EA9FC000000120000000700000000000000000000000400000000000000000000000000000000000000000000281E0000327800000000000000000000000000AF967D00000AEA000000000000000000000000";

                     Send(state, data);
        }
        catch (Exception e)
        {
            Console.WriteLine(e.ToString());
        }
    }

    private static void Receive(StateObject state)
    {
        try
        {
          Socket client = state.workSocket;
            // Begin receiving the data from the remote device.
            client.BeginReceive(state.buffer, 0, StateObject.BufferSize, 0,
                new AsyncCallback(ReceiveCallback), state);
        }
        catch (Exception e)
        {
            Console.WriteLine(e.ToString());
        }
    }

    private static void ReceiveCallback(IAsyncResult ar)
    {
        try
        {
            StateObject state = (StateObject)ar.AsyncState;
            Socket client = state.workSocket;

            // Read data from the remote device.
            int bytesRead = client.EndReceive(ar);

            //if (wait)
            //{
            //    connectDone.WaitOne();
            //}

            if (bytesRead > 0)
            {
                state.count = state.count + 1;
                byte[] b = new byte[bytesRead];
                Array.Copy(state.buffer, b, 1);
                if (b[0] == 1)
                {

                    if (state.count < 10)
                    {
                        var data = "AA5500C08308353816050322462F01020102191552E7D3FA52E7D3FB1FF85BF1FE9F201000004AB80000000500060800001EFFB72F0D00002973620000800000FFFFFFFF00009D6D00003278002EE16D0000018500000000000000000000003A0000000100000000828C80661FF8B436FE9EA9FC000000120000000700000000000000000000000400000000000000000000000000000000000000000000281E0000327800000000000000000000000000AF967D00000AEA000000000000000000000000";
                        Send(state, data);
                    }
                    else
                    {
                        Interlocked.Increment(ref closecalled);
                        Console.WriteLine("closecalled:-" + closecalled + " at " + DateTime.Now);
                        client.Close();
                    }

                }
                else
                {
                    // Get the rest of the data.
                    client.BeginReceive(state.buffer, 0, StateObject.BufferSize, 0,
                        new AsyncCallback(ReceiveCallback), state);
                }
            }
            else
            {
                client.Close();
            }
        }
        catch (Exception e)
        {
            Console.WriteLine(e.ToString());
        }
    }

    private static void Send(StateObject state, String data)
    {
        try
        {
            Socket client = state.workSocket;
            var hexlen = data.Length;
            byte[] byteData = new byte[hexlen / 2];
            int[] hexarray = new int[hexlen / 2];
            int i = 0;
            int k = 0;
            //create the byte array
            while (i < data.Length / 2)
            {
                string first = data[i].ToString();
                i++;
                string second = data[i].ToString();
                string x = first + second;
                byteData[k] = (byte)Convert.ToInt32(x, 16);
                i++;
                k++;
            }
            // Begin sending the data to the remote device.
            client.BeginSend(byteData, 0, byteData.Length, 0,
                new AsyncCallback(SendCallback), state);
        }
        catch (Exception e)
        {
            Console.WriteLine(e.ToString());
        }
    }

    private static void SendCallback(IAsyncResult ar)
    {
        try
        {
            // Retrieve the socket from the state object.
            StateObject state = (StateObject)ar.AsyncState;
            Socket client = state.workSocket;
            // Complete sending the data to the remote device.
            int bytesSent = client.EndSend(ar);
            Receive(state);
        }
        catch (Exception e)
        {
            Console.WriteLine(e.ToString());
        }
    }

    public static int Main(String[] args)
    {
        Start();
        Console.ReadLine();
        return 0;
    }

    private static void Start()
    {
        IPAddress ipaddress = IPAddress.Parse("10.20.2.152");
        IPEndPoint remoteEP = new IPEndPoint(ipaddress, port);

        for (int i = 0; i < 40000; i++)
        {
            Thread.Sleep(1);
            // Create a TCP/IP socket.
            try
            {
                Socket client = new Socket(AddressFamily.InterNetwork,
                    SocketType.Stream, ProtocolType.Tcp);
                StartClient(client, remoteEP);
            }
            catch (Exception e)
            {
                Console.WriteLine(e.ToString());
            }
            if (i == 39999)
            {
                Console.WriteLine("made all conns at " + DateTime.Now);
            }
        }
    }
}
rakesh
  • 301
  • 1
  • 3
  • 12
  • Is the server code identical in the two scenarios? If so, there is a possible bug. Could you show us some of its code in this case? – Luca Cremonesi Jul 17 '14 at 17:11
  • 1
    Sounds like you exhausted the OS backlog for incoming connections. Post some code and the full exception. – usr Jul 17 '14 at 17:25
  • I have added part of the code above. the error i get is "No connection could be made because the target machine actively refused it" – rakesh Jul 17 '14 at 20:15
  • Where is the code that initiates the connections? Specify a higher backlog (like 40k). – usr Jul 17 '14 at 22:51
  • @usr i have added the client code . i tried 40k as backlog but no luck. can you review the client code please? – rakesh Jul 18 '14 at 08:28
  • Looks pretty good. This is the highest quality socket code that I have seen in a long time on Stack Overflow. In fact in all this code I did not find a single API usage error. Gold medal for you. – usr Jul 18 '14 at 09:57
  • I still cant figure out why scenario 2 will not work. it works but some of the connections are refused. i can always reconnect to the client. could the OS backlog be exhausted like you said? – rakesh Jul 18 '14 at 10:56
  • @rakesh, is there any reason you stick with the low-level socket and synchronization APIs? Related: http://stackoverflow.com/q/21013751/1768303 – noseratio Jul 19 '14 at 03:48
  • @Noseratio only because it was suggested as the best approach for handling high volume of connections. does the async/await approach have any added benefits? I will try out the code in your link as well – rakesh Jul 21 '14 at 08:25
  • @rakesh, `Task`-based TAP Sockets APIs are just very thin wrappers around `BeginXXX/EndXXX` APM APIs. With TAP APIs, you can use `async/await`, `TcpClient.GetStream` w/ `ReadAsync/WriteAsync`, and create your own TAP wappers like [this](http://stackoverflow.com/a/22237307/1768303). – noseratio Jul 21 '14 at 08:49
  • What OS are you running? Server or Desktop windows? – Aron Jul 22 '14 at 15:01
  • scenario 2 seems to work if I comment out the code in the processreceive method which just pushes data to a rabbitMq – rakesh Jul 22 '14 at 15:29

1 Answers1

0

I would use a linear queue to accept incoming connections. Something like this:

public async Task Accept40KClients()
{
    for (int i = 0; i < 40000; i++)
    {
        // Await this here   -------v
        bool willRaiseEvent = await listenSocket.AcceptAsync(acceptEventArg);
        if (!willRaiseEvent)
        {
            ProcessAccept(acceptEventArg);
        }
    }
}

If that's not fast enough, maybe you can do 10 waits at a time, but I think this is good enough... I might be wrong on this though.

Philippe Paré
  • 4,279
  • 5
  • 36
  • 56