random thoughts, formed in the twisted mind of a coder... RSS 2.0
# Friday, November 07, 2008

MultiCore CPU Chips

Parallel computing is hot. And that’s not without a good reason…
The chip factories are unable to produce faster cpu cores. There seems to be a magical limit around the 2 GHz. But no limit without a challenge. So they came up with multi-core cpu chips. These chips enable the software to do multiple tasks at the same time. Each core its own task.

This is not the same as multithreading. Multithreading can be done on a single core cpu. In such a case, two threads can never execute at the same time on the cpu. The operating system (which enables multi-threading for you) divides the time of the processor between all open threads (this is the case for  time-sliced OSes like Microsoft Windows). When you have too many executing threads, your system slows down, as there is not enough time to be sliced for all the threads to run at full speed.

By creating more cores, the chip producers enable the OSes to have more threads before the system slows down, as the OS is able to divide the threads over multiple cores. The OS can do this by itself for all current software. But although the OS divides the threads over the available cores for you, you can definitely get more out of your multi-core cpu by helping the OS a little.

You can do this by writing your software in a way that you leave the threading up to the OS (or any other underlying system). You don’t do the threading yourself, but hand the OS small tasks to perform. That way the OS is able execute the tasks on the core that is least busy and therefore finds out the best way to run your software on the particular cpu it runs on.


Concurrent vs. Parallel

I have attended the PDC 2008 in Los Angeles and also went to the pre-conference sessions. There it was mentioned what the difference is between concurrent programming (programming in threads) and parallel programming (programming in tasks). I could try to explain it all in words, but there they showed one simple diagram that nearly says it all.

It looked something like this…


 

To explain a little, concurrent applications tend to create a thread that handles a whole series of tasks. For example… When you write a server application, you create a thread for an incoming client connection and handle the whole protocol between client and server in that thread until the connection is closed.
Most of the time, concurrent applications create threads because they need an isolated process for a concurrent outerworld event.

Parallel applications divide the process into small tasks which are executed on separate threads. Because the tasks are small, the threads can be divided evenly over the cores, resulting in very efficient use of your multi-core cpu.
Parallel programming is also applicable for server applications which need to handle multiple concurrent client connections. I’ll show you in a bit.


Concurrent Server Application Example

Now, let’s take a look at a typical sever application using the concurrent model. This example application simply waits for a client to connect, and when so, it echoes all the text that the client sends back to the client.

using System;
using System.Net;
using System.Net.Sockets;
using System.Threading;
using System.IO;

namespace EchoServer
{
  class ThreadServer
  {
    //---------------------------------------------------------------
    // FIELDS
    //---------------------------------------------------------------
    private Thread  serverThread  = null;
    private Socket  serverSocket  = null;
    private bool    running       = false;

    //---------------------------------------------------------------
    // METHODS
    //---------------------------------------------------------------
    public void start()
    {
      if (!running)
      {
        //start a new backgroundthread for accepting new connections
        serverThread = new Thread(new ThreadStart(waitForConnections));
        serverThread.Start();

        //set the running flag
        running = (serverThread.ThreadState == ThreadState.Running);
      }
    }

    //---------------------------------------------------------------
    public void stop()
    {
      if (running)
      {
        //reset the running flag
        running = false;
        
        try
        {
          //close server socket
          serverSocket.Close();
        }
        catch (SocketException e) {;}

        try
        {
          //stop server thread (give it max 5 seconds)
          serverThread.Join(5000);
        }
        catch (ThreadStateException e) {;}
      }
    }

    //---------------------------------------------------------------
    private void waitForConnections()
    {
      try
      {
        //setup server socket
        serverSocket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
        serverSocket.Bind(new IPEndPoint(IPAddress.Parse("127.0.0.1"), 1234));
        serverSocket.Listen(1000);

        while (running)
        {
          //wait for a client to connect
          Socket client = serverSocket.Accept();

          //prep client
          client.Blocking = true;
          client.ReceiveTimeout = 30 * 1000; //30 seconds

          //run client in new thread
          Thread clientThread = new Thread(new ParameterizedThreadStart(handleClient));
          clientThread.Start(client);
        }
      }
      catch (SocketException e)
      { ; }
      finally
      {
        //stop running
        running = false;

        //try to close the socket
        try{ serverSocket.Close();  }catch (Exception e) {;}
      }
    }

    //---------------------------------------------------------------
    private void handleClient(Object _client)
    {
      //configure client networking objects
      Socket        client        = (Socket)_client;
      NetworkStream clientStream  = new NetworkStream(client, false);
      StreamReader  clientIn      = new StreamReader(clientStream);
      StreamWriter  clientOut     = new StreamWriter(clientStream);
      clientOut.AutoFlush = true;

      try
      {
        //receive lines of text from the client
        string text = null;
        while (running && (text = clientIn.ReadLine()) != null)
        {
          //echo text to client
          clientOut.WriteLine(text);
        }
      }
      catch (SocketException e)
      { ; }
      catch (IOException e)
      { ; }
      finally
      {
        //try to close the client socket
        try{ clientIn.Close();      }catch (Exception e) {;}
        try{ clientOut.Close();     }catch (Exception e) {;}
        try{ clientStream.Close();  }catch (Exception e) {;}
        try{ client.Close();        }catch (Exception e) {;}
      }
    }

    //---------------------------------------------------------------
  }
}

Ok, so what are we looking at…

First there are two methods for starting and stopping the server. Here the main server thread is created and destroyed.

Then we have the waitForConnections method. This method is ran by the main server thread. It creates the server socket and waits for any client to connect. When a client does, the method creates a new thread for that client connection, passes client control over to that thread and starts waiting for a client to connect again.

The handleClient method is ran by the client thread. It creates a bunch of networking helper objects and starts waiting for any input from the client (on a single line). When data is received, it’s simply written back to the client.


Concurrent Programming Upsides and Downsides

The advantages of this model for server applications are, that it’s quite easy and simple to write an application like this, because you have a one-to-one relationship between a socket and a thread. And you can make use of blocking sockets, and thus make use of several network helper classes, which make life a lot easier for you.

One of the major downsides of this concurrent application is, that your threads are not always busy but remain running on a core, leaving less time for other threads to run on that core.
For example… Your server might be waiting for a client to send some text  (this might even take more than a second), while doing so, it is not executing any of your code. But as the thread is still running, the time slicing OS still provides the thread with cpu time.

Another major downside to this concurrent server application, is that it creates a thread for each new client connection. This way you can never handle a lot of concurrent connections, because threads need time and memory. If you would get 1000 concurrent connections, you would have 1000 threads all wanting their slice of time. But also, because in Windows, every thread you create gets its own 1 MB of stack space. With 1000 threads on a 32 bit cpu (with 32 bit memory addressing) you’ll get stuck very quickly.
This means that you must limit the amount of concurrent connections to the maximum amount of threads your system can handle. On a Windows Server machine (form Win 2k3 R2 and later) you can handle up to 5000 concurrent client connections. So if you don’t limit the amount of concurrent connections of your server application, a hacker (or simply flawed client software) could easily create enough concurrent connections to crash your system.


Parallel Server Application Example

Now let’s take a look at the example code for the parallel server application. It has the same functionality as the concurrent (threading) server…

using System;
using System.Net;
using System.Net.Sockets;
using System.Threading;
using System.IO;
using System.Text;

namespace EchoServer
{
  class TaskServer
  {
    //---------------------------------------------------------------
    // FIELDS
    //---------------------------------------------------------------
    private Socket  serverSocket  = null;
    private bool    running       = false;

    //---------------------------------------------------------------
    // METHODS
    //---------------------------------------------------------------
    public void start()
    {
      if (!running)
      {
        try
        {
          //set the running flag
          running = true;

          //setup server socket
          serverSocket = new Socket(AddressFamily.InterNetwork, SocketType.Stream, ProtocolType.Tcp);
          serverSocket.Bind(new IPEndPoint(IPAddress.Parse("127.0.0.1"), 1234));
          serverSocket.Listen(1000);

          //accept client connections
          serverSocket.BeginAccept(new AsyncCallback(handleConnections), null);
        }
        catch (SocketException excpt)
        {
          //reset the running flag
          running = false;

          //try to close the socket
          try{ serverSocket.Close();  }catch (Exception e) {;}
        }
      }
    }

    //---------------------------------------------------------------
    public void stop()
    {
      if (running)
      {
        //reset the running flag
        running = false;

        try
        {
          //close server socket
          serverSocket.Close();
        }
        catch (SocketException e) {;}
      }
    }

    //---------------------------------------------------------------
    private void handleConnections(IAsyncResult _aResult)
    {
      Socket client = null;

      try
      {
        //get client socket
        client = serverSocket.EndAccept(_aResult);

        //accept new client connections
        serverSocket.BeginAccept(new AsyncCallback(handleConnections), null);

        //start receiving form client
        ClientState state = new ClientState(client);
        client.BeginReceive(state.buffer, 0, state.buffer.Length, SocketFlags.None, new AsyncCallback(handleClientReceive), state);
      }
      catch (SocketException excpt)
      {
        //try to close the client connection
        try{ client.Close();  }catch(Exception e){;}
      }
    }

    //---------------------------------------------------------------
    private void handleClientReceive(IAsyncResult _aResult)
    {
      //get the state
      ClientState state = (ClientState)_aResult.AsyncState;

      try
      {
        //read the buffered text
        int count = state.client.EndReceive(_aResult);
        if (count > 0)
        {
          //read the text from the buffer
          state.received += Encoding.ASCII.GetString(state.buffer, 0, count);

          //check for line end
          int idx = state.received.IndexOf("\r\n");
          if (idx > -1)
          {
            //get a single line from the received buffer
            string line = state.received.Substring(0, idx+2);
            state.received = state.received.Substring(idx+2);

            //echo to client
            byte[] buffer = Encoding.ASCII.GetBytes(line);
            state.client.BeginSend(buffer, 0, buffer.Length, SocketFlags.None, new AsyncCallback(handleClientSend), state);
          }
          else
          {
            //start receiving form client
            state.client.BeginReceive(state.buffer, 0, state.buffer.Length, SocketFlags.None, new AsyncCallback(handleClientReceive), state);
          }
        }
      }
      catch (SocketException excpt)
      {
        //try to close the client connection
        try{ state.client.Close();  }catch(Exception e){;}
      }
    }

    //---------------------------------------------------------------
    private void handleClientSend(IAsyncResult _aResult)
    {
      //get the state
      ClientState state = (ClientState)_aResult.AsyncState;

      try
      {
        //end sending process
        state.client.EndSend(_aResult);

        //start receiving form client
        state.client.BeginReceive(state.buffer, 0, state.buffer.Length, SocketFlags.None, new AsyncCallback(handleClientReceive), state);
      }
      catch (SocketException excpt)
      {
        //try to close the client connection
        try{ state.client.Close();  }catch(Exception e){;}
      }
    }

    //===============================================================
    class ClientState
    {
      //-------------------------------------------------------------
      public Socket client    = null;
      public byte[] buffer    = null;
      public string received  = "";

      //-------------------------------------------------------------
      public ClientState(Socket _client)
      {
        client = _client;
        buffer = new byte[1024];
      }

      //-------------------------------------------------------------
    }

    //===============================================================
  }
}

As you can see, this parallel example basically has the same format as the concurrent example, but works quite different.

The stat and stop methods are still there, but there is no longer a main thread created. Instead the start method already creates the server socket and starts listening for client connections. This is done with the BeginAccept method on the server socket. As a parameter the handleConnections method is passed, which is called when a new client connects.

But what is happening under the covers? Basically the BeginAccept method defers the action by using the static ThreadPool class. It pushes a task into the queue which checks if a new client connection has been made. When so, the task ends by accepting the client and calling the callback function (handleConnections).

After setting up the client connection, the handleConnections method starts accepting new clients by calling the BeginAccept method again, thus pushing a new task into the ThreadPool queue.
After that, the same is done with the BeginReceive method on the client socket. It creates a new task, “wait for client data to come in”, and puts that task into the ThreadPool queue.


Parallel Programming Downsides

When data comes in from the client, the callback function handleClientReceive is called. Here you can see some of the downsides of the current implementation in .Net of sockets and parallelism. You cannot use the ReadLine method of the StreamReader (+NetworkStream) anymore. So, basically you have to read the bytes. And because you cannot be sure that a whole line is passed at once (in one IP packet), you have to accumulate incoming data until you have a complete line.

And that is where the next downside comes in. Because you’re not running all the client handling in one thread, you have to transfer state objects between the tasks. The state object I’ve used (the ClientState class) contains the client socket and the received bytes.

When a full line is received, another task is started. The handleClientSend sends the received bytes back to the client.


Parallel Frameworks and Helper Classes

So in basic, it all comes down to splitting up your normal process in little tasks and running those tasks via the ThreadPool. The pool is then able to quickly balance your tasks over the running threads and the OS is able to balance the ThreadPool threads over the multiple cores on your cpu.
Please note that the ThreadPool is used underwater by many .Net functions and classes. When you’re using a FileStream, you’re using the ThreadPool.

Now when you’re writing parallel applications which use sockets you’re in a tough spot. I wasn’t able to find any helper classes or frameworks to help me, which doesn’t mean that there aren’t any. :-)
But as far as I can see, when writing a socket based application you have to program all the parallelism yourself.
It’s not too hard, but it’s also not as sweet as some parallel frameworks, like the Task Parallel Library (TPL) for .Net or like PLINQ. These are all additions to the .Net framework to help you create parallel applications in your .Net environment. Though this is a good thing as it helps you and makes it easy to get the most out of your cpu, it also clutters up imperative programming languages with functional statements.


Imperative Programming vs. Functional Programming

Functional programming languages are at heart the best languages for parallelism. This is because with functional languages you program the computer what to do instead of how to do it. This gives the compiler the power of finding out the best way to subdivide the program into tasks.

Microsoft has just launched a new functional programming language called F#. It has all the advantages of a functional language, but it’s also fully compatible with the .Net library. But the best thing is, that you are able to write your parallel code in the F# language, and use those F# classes (!) in any other .Net language like C# or VB.Net. That way you can combine the power of F# with the power of C#!

In my opinion such a combination is far more preferable than trying to fit in some functional frameworks into an imperative language. Both languages have their own power, let’s keep it that way. :-)

 

References:

Task Parallel Library blog:
http://blogs.msdn.com/pfxteam/archive/2007/11/29/6558543.aspx

All about Parallelism at Microsoft and beyond:
http://msdn.microsoft.com/en-us/concurrency/default.aspx

Slides from the PDC pre-conference session:
http://blogs.msdn.com/pfxteam/attachment/9019731.ashx

Wikipedia topics:
http://en.wikipedia.org/wiki/Functional_programming
http://en.wikipedia.org/wiki/Parallel_computing
http://en.wikipedia.org/wiki/Parallel_algorithm
http://en.wikipedia.org/wiki/Parallel_programming_model

F# (F-Sharp):
http://msdn.microsoft.com/en-us/fsharp/default.aspx

 

Friday, November 07, 2008 10:46:56 PM (W. Europe Standard Time, UTC+01:00)  #    Comments [4]
.Net | Parallelism
Thursday, February 05, 2009 3:51:51 PM (W. Europe Standard Time, UTC+01:00)
Again I enjoyed reading this article. Why did you choose to use sockets specifically though? It strikes me that you are making life quite difficult here - there are libraries & standards (such as MPI, BSP) which sit on top of this low level and allow to send messages in many different forms without worry about the nitty gritty details (such as synchronising the calls, error checking the data etc...)

I agree there has been much work done on parallel functional languages, but I am not convinced they are nescesarily the answer to parallel programming - the problem (as I see it) is that, due to their their abstract nature, most parallelization details are hidden from the programmer. The programmer has no control over the vital parallel issues (communication, sync etc..) and performance is often not great when the compiler infers (or guesses) the best solution. There has been considerable research on Skeleton functions, which aim to address this and might very well become a viable solution.
Monday, September 21, 2009 5:55:19 AM (W. Europe Daylight Time, UTC+02:00)
Dear Rednael,

How do we implement parallel with UDP socket? please give me sample about that

Thanks Rednael,
Canh
Monday, September 21, 2009 3:32:24 PM (W. Europe Daylight Time, UTC+02:00)
Hi Quang,

What would be different in your opinion? You can still use the BeginReceive, can't you?

Wednesday, September 23, 2009 4:41:09 AM (W. Europe Daylight Time, UTC+02:00)
hi Quang, Martiijn

Please review it & feedback, this is sample using parallel programming with UDP socket


class Program
{
static Socket serverSocket;
static byte[] byteData = new byte[1024];
static void Main(string[] args)
{
serverSocket = new Socket(AddressFamily.InterNetwork,
SocketType.Dgram, ProtocolType.Udp);
IPEndPoint ipEndPoint = new IPEndPoint(IPAddress.Any, 10001);

serverSocket.Bind(ipEndPoint);

IPEndPoint ipeSender = new IPEndPoint(IPAddress.Any, 0);
EndPoint epSender = (EndPoint)ipeSender;

//Start receiving data
ClientState client = new ClientState(serverSocket);
serverSocket.BeginReceiveFrom(client.buffer, 0, client.buffer.Length, SocketFlags.None, ref epSender, new AsyncCallback(HandleReceive), client);
Console.ReadLine();
}


private static void HandleReceive(IAsyncResult ar)
{
ClientState state = (ClientState)ar.AsyncState;
IPEndPoint ipeSender = new IPEndPoint(IPAddress.Any, 0);
EndPoint epSender = (EndPoint)ipeSender;
int count = state.client.EndReceiveFrom(ar, ref epSender);
state.client.BeginReceiveFrom(state.buffer, 0, state.buffer.Length, SocketFlags.None, ref epSender, new AsyncCallback(HandleReceive), state);
string msg = ASCIIEncoding.ASCII.GetString(state.buffer, 0, count);
Console.WriteLine(string.Format("Thread: {0} {1}", Thread.CurrentThread.GetHashCode(), msg));
Thread.Sleep(30);
state.client.BeginSendTo(state.buffer, 0, count, SocketFlags.None, epSender, new AsyncCallback(HandleSend), state);
}

private static void HandleSend(IAsyncResult ar)
{
ClientState state = (ClientState)ar.AsyncState;
state.client.EndSend(ar);
}


}

class ClientState
{
//-------------------------------------------------------------
public Socket client = null;
public byte[] buffer = null;

//-------------------------------------------------------------
public ClientState(Socket _client)
{
client = _client;
buffer = new byte[1024];
}

//-------------------------------------------------------------
}
All comments require the approval of the site owner before being displayed.
Name
E-mail
Home page

Comment (HTML not allowed)  

Enter the code shown (prevents robots):

Live Comment Preview
About the author/Disclaimer

Disclaimer
The opinions expressed herein are my own personal opinions and do not represent my employer's view in any way.

© Copyright 2014
Martijn Thie
Sign In
Statistics
Total Posts: 18
This Year: 0
This Month: 0
This Week: 0
Comments: 161
All Content © 2014, Martijn Thie
DasBlog theme adapted from 'Business' (originally by delarou)