Socket OnData event

Question

I'm currently in a little bit of a pickle. I have some code that reads data from a socket whenever data is available but currently it is in a while loop on a separate thread and chews through 50% of the CPU when the function returns because of no data available. What I would really like is a blocking function for Sockets that blocks until data is available, or at least a OnData event that could be listened on. I originally converted this code from AS3 (Flash) but their sockets class has the OnData event I need... just in the wrong language.

I currently have this code in the code that handles a client connecting:

ServerThread = new Thread(() =>
{
    while (server.Connected && ServerContinue)
    {
        ReceiveFromServer(server, client);
    }

    Disconnect(server, client, false);
});

ServerThread.Start();

And this is the code in ReceiveFromServer:

bool isReady = false;
int messageLength = 0;
int dataAvailable = 0;
UInt16 packetSize = 0;
byte[] temp = new byte[2];
do
{
    dataAvailable = server.Available;
    if (isReady)
    {
        if (dataAvailable >= messageLength)
        {
            byte[] temp1 = new byte[2000];
            int bytesRead = server.Receive(temp1, 0, messageLength, SocketFlags.None);

            byte[] data = new byte[bytesRead + 2];
            Buffer.BlockCopy(temp1, 0, data, 2, messageLength);

            Helpers.ByteArray tempo = data;
            tempo.writeByte(temp[1]);
            tempo.writeByte(temp[0]);
            if (!VersionCheckPass)
            {
                Send(tempo, client);
                return;
            }

            ServerPacketHandler(tempo, client);
            messageLength = 0;
            isReady = false;
            temp = new byte[2];
        }
    }
    else if(dataAvailable > 2)
    {
        server.Receive(temp, 0, 2, SocketFlags.None);
        temp = temp.Reverse().ToArray();
        packetSize = BitConverter.ToUInt16(temp, 0);
        if (packetSize > 0)
        {
            messageLength = packetSize;
            isReady = true;
        }
    }
}
while (dataAvailable > 2 && dataAvailable >= messageLength && ServerContinue);

But the issue here is that when dataAvailable is 0 the function simply returns, and then RecevieFromServer is called again in the thread. This means that alot of the CPU is used by simply calling ReceiveFromServer and then returning again.

I currently have Thread.Sleep(10) after ReceiveFromServer in the ServerThread but this is inefficient. So my question is, Is there a way to block until data is available or is there an event that I can handle? Or does anyone else have any suggestions on how to do the same thing I am currently doing but it doesn't loop endlessly whilst there is no data available.

I've used the [Nito Async Sockets](http://nitoasync.codeplex.com/) library with success. It provides an event-based approach to socket programming. — Jonathon Reinhart, Oct 13 '12 at 05:00
Thanks, I'll take a look, but I would really hate to have to re-write it all. — jduncanator, Oct 13 '12 at 05:04

score 1 · Accepted Answer · answered Oct 18 '12 at 05:54

Found a really easy (and obvious) solution to block until data is available. Call Socket.Receive with a receive size of 0. The socket blocks until there is data to receive, then reads 0 bytes from the socket, and unblocks. Its really quite marvelous :) Heres how I implemented it:

ServerThread = new Thread(() =>
{
    byte[] zero = new byte[0];
    while (Server.Connected && ServerContinue)
    {
        server.Receive(zero, 0, SocketFlags.None);
        ReceiveFromServer(server, client);
    }

    Disconnect(server, client, false);
});

Thanks for all the help.

Josh

score 0 · Answer 2 · edited May 23 '17 at 12:33

There is not a lot of re-write needed. Your code looks like it's just receiving the message and then passing off to another routine to process it.

My reply to this thread pretty much covers what you're wanting to do:

C# Sockets and Multithreading

My socketReadCallBack function is:

    private void OnDataReceived(IAsyncResult asyn)
    {
        ReceiveState rState = (ReceiveState)asyn.AsyncState;
        Socket client = rState.Client;
        SocketError socketError = SocketError.TypeNotFound;

        if (!client.Connected)
        {
            // Not Connected anymore ?
            return;
        }

        _LastComms = DateTime.Now;
        _LastIn = _LastComms;

        int dataOffset = 0; 
        int restOfData = 0;
        int dataRead = 0;
        Boolean StreamClosed = false;
        long rStateDataLength = 0;
        long LastrStateDataLength = 0;

        try
        {

            dataRead = client.EndReceive(asyn, out socketError);
        }
        catch (Exception excpt)
        {
            // Handle error - use your own code..

        }

        if (socketError != SocketError.Success)
        {
            // Has Connection been lost ?
            OnConnectionDropped(client);
            return;
        }

        if (dataRead <= 0)
        {
            // Has connection been lost ?
            OnConnectionDropped(client);
            return;
        }

        while (dataRead > 0)
        {
            //check to determine what income data contain: size prefix or message
            if (!rState.DataSizeReceived)
            {
                //there is already some data in the buffer
                if (rState.Data.Length > 0)
                {
                    restOfData = PrefixSize - (int)rState.Data.Length;
                    rState.Data.Write(rState.Buffer, dataOffset, restOfData);
                    dataRead -= restOfData;
                    dataOffset += restOfData;
                }
                else if (dataRead >= PrefixSize)
                {  //store whole data size prefix
                    rState.Data.Write(rState.Buffer, dataOffset, PrefixSize);
                    dataRead -= PrefixSize;
                    dataOffset += PrefixSize;
                }
                else
                {  // store only part of the size prefix
                    rState.Data.Write(rState.Buffer, dataOffset, dataRead);
                    dataOffset += dataRead;
                    dataRead = 0;
                }

                if (rState.Data.Length == PrefixSize)
                {  //we received data size prefix 
                    rState.DataSize = BitConverter.ToInt32(rState.Data.GetBuffer(), 0);
                    rState.DataSizeReceived = true;
                    //reset internal data stream             
                    rState.Data.Position = 0;
                    rState.Data.SetLength(0);
                }
                else
                {  //we received just part of the prefix information 
                    //issue another read
                    client.BeginReceive(rState.Buffer, 0, rState.Buffer.Length,
                       SocketFlags.None, new AsyncCallback(socketReadCallBack),
                          rState);
                    return;
                }
            }



            //at this point we know the size of the pending data

            // Object disposed exception may raise here
            try
            {
                rStateDataLength = rState.Data.Length;
                LastrStateDataLength = rStateDataLength;
            }
            catch (ObjectDisposedException Ode)
            {
                StreamClosed = true;
            }
            if (!StreamClosed)
            {

                if ((rStateDataLength + dataRead) >= rState.DataSize)
                {   //we have all the data for this message

                    restOfData = rState.DataSize - (int)rState.Data.Length;

                    rState.Data.Write(rState.Buffer, dataOffset, restOfData);
                    //Console.WriteLine("Data message received. Size: {0}",
                    //   rState.DataSize);

                    // Is this a heartbeat message ?
                    if (rState.Data.Length == 2)
                    {
                        // Yes
                        HeartBeatReceived();
                    }
                    else
                    {
                        //charArray = new char[uniEncoding.GetCharCount(
                        //byteArray, 0, count)];
                        //uniEncoding.GetDecoder().GetChars(
                        //    byteArray, 0, count, charArray, 0);
                        //Console.WriteLine(charArray);

                        //rState.Data.Position = 0;

                        DecodeMessageReceived(GetStringFromStream(rState.Data));
                    }

                    dataOffset += restOfData;
                    dataRead -= restOfData;

                    //message received - cleanup internal memory stream
                    rState.Data = new MemoryStream();
                    rState.Data.Position = 0;
                    rState.DataSizeReceived = false;
                    rState.DataSize = 0;

                    if (dataRead == 0)
                    {  
                        //no more data remaining to process - issue another receive
                        if (_IsConnected)
                        {
                            client.BeginReceive(rState.Buffer, 0, rState.Buffer.Length,
                               SocketFlags.None, new AsyncCallback(socketReadCallBack),
                                  rState);
                            return;
                        }
                    }
                    else
                        continue; //there's still some data to process in the buffers
                }
                else
                {  //there is still data pending, store what we've 
                    //received and issue another BeginReceive
                    if (_IsConnected)
                    {
                        rState.Data.Write(rState.Buffer, dataOffset, dataRead);

                        client.BeginReceive(rState.Buffer, 0, rState.Buffer.Length,
                           SocketFlags.None, new AsyncCallback(socketReadCallBack), rState);

                        dataRead = 0;
                    }
                }
            }
            else
            {
                // Stream closed, but have we read everything ?
                if (LastrStateDataLength + dataRead == rState.DataSize)
                {
                    // We're equal, get ready for more
                    //no more data remaining to process - issue another receive
                    if (_IsConnected)
                    {
                        client.BeginReceive(rState.Buffer, 0, rState.Buffer.Length,
                           SocketFlags.None, new AsyncCallback(socketReadCallBack),
                              rState);
                    }
                    return;
                }
                else
                {
                    // We should have more..
                    // Report Error
                }
            }

            // If we've been disconnected, provide a graceful exit
            if (!_IsConnected)
                dataRead = -1;

        }
     }

I've got a few more things in here than you need like provision for a heartbeat message and raising events on connection dropped etc.

Socket OnData event

2 Answers2