Saturday, July 23, 2005 1:20 PM bart

Adventures in Comega - part 6 (concurrency extensions)

Note: I wrote this post a couple of weeks ago on my way to TechEd Amsterdam somewhere between Brussels and Amsterdam on the Thalys train. Finally I found the time to finalize this first post on concurrency extensions in the Comega language. I hope you enjoy it. More on Comega will come later.

Introduction

In the previous posts I described how Comega helps to simplify the construction of data-driven applications thanks to the integration of SQL and XSD concepts into the programming language itself. Beside of these new data access constructs, Comega also extends the C# language with new asynchronous concurrency abstractions, that are partially derived from the "Polyphonic C#" project done by Microsoft Research.

Concepts

By default, methods in the C# language have a synchronous nature. When a method is called, the caller is blocked until the execution of method has come to an end and a result is returned (either a void or some value or object). Although the .NET Framework supports asynchronous method invocation too, through a bunch of class libraries, the language itself is not aware of the possible asynchronous nature of a method. Comega extends the notion of asynchronous methods by introducing a series of new language constructs that indicate that a method has an asynchronous nature. When calling an asynchronous method, the caller is not blocked until a result is returned but it can proceed immediately. Asynchronous methods are often called messages because of their one-way communication nature.

Now, how does Comega realize this mission statement? A first central concept when dealing with the concurrency extensions in Comega is the concept of a chord. In "classic languages" there is a one-on-one mapping between a method header and a method body: when method X is called that way (the "signature"), this code (the "body") has to be executed. Comega allows a method body to be associated with a set of methods, which we call a chord. This body can only be executed when all of the methods that are declared in the corresponding header have been called.

The async keyword

Okay, enough theory for now, let's take a look at some very basic samples.

Sample 1:

The first sample shows the use of the async keyword to indicate the asynchronousness of a method. When calling the method, the caller isn't blocked and can continue with its own work.

using System;

public class Test
{
   static async Do()
   {
      while(true);
   }

   static void Main()
   {
      Console.WriteLine("Before do");
      Do();
      Console.WriteLine("After do");
      Console.ReadLine();
   }
}

Take a look in the Windows Task Manager for the number of active threads for the process when executing this piece of code. When you comment out the Do() method invocation, the number of threads will be one less. So, when calling the Do-method another thread is started.

Sample 2:

In this second sample, I'll prove that parallellism introduced by the async method declaration using one of my favorite mathematical concepts, prime numbers :-).

using System;

public class Test
{
   static int n = 0;

   static
async Do(int max)
   {
      int id = ++n;
      Console.WriteLine("{0} - Started", id);

      for
(int i = 2; i <= max; i++)
      {
         bool prime = true;

         for
(int j = 2; j <= (int) Math.Sqrt(i); j++)
         {
            if (i % j == 0)
            {
               prime =
false;
               break;
            }
         }

         if
(prime)
            Console.WriteLine("{0} - {1}", id, i);
      }

     
Console.WriteLine("{0} - Finished", id);
   }

   static void Main()
   {
      Console.WriteLine("Before do");
      Do(1000000);
      Do(1000000);
      Console.WriteLine("After do");
      Console.ReadLine();
   }
}

When executing the piece of code, you should see clearly the parallellism of both method invocations as both threads will report the results of their prime search journey concurrently. Note the lack of a return type when declaring asynchronous methods.

Note: The sample above has a problem; people who're expierenced with concurrency programming should see it immediately. If you don't see it, try the next piece of code:

using System;

public class Test
{
  
static int k = 0;

  
static async Do2()
   {
      k++;
   }

   static void Main()
   {
      for (int i = 0; i < 1000; i++)
         Do2();

      Console.WriteLine(k);
   }
}

Before you do execute this piece of code, guess what the result should be.

Using chords

Time for the next step: chords. I'll kick off with the basic sample that's in the Comega documentation: the "buffer tutorial". Declare the following class:

using System;

public class Test
{
   static async Put(string s);

   static string Get() & Put(string s)
   {
      return s;
   }

   static void Main()
   {
      Put("Hello");
      string res = Get();
      Console.WriteLine(res);
      Console.ReadLine();
   }
}

The execution of this piece of code should write "Hello" on the screen. So, what happened? The call to Put takes the message as a string parameter that is. Nothing happens as the method body is missing in the class definition. Then, the Get method is called, which requires Get to be called first in order to "gain access" to the method body. As this is the case, the method can execute. In the body of the Get method, the parameter(s) from the Put method can be used in order to perform the job requested.

So, when you'd write this piece of code:

static void Main()
{
   string res = Get();
   Put("Hello");
   Console.WriteLine(res);
   Console.ReadLine();
}

nothing would happen, because the "entry condition" is not met. The following piece of code shows the queue nature of chords:

static void Main()
{
   Put("Hello");
   Put("World");
   Console.WriteLine(Get());
   Console.WriteLine(Get());
}

It should print two lines containing "Hello" and "World". The use of this kind of constructs becomes more useful when threads are used. One thread can wait on another one when a certain piece of code isn't ready executing the task requested. Let's go back to our primes sample, adapted with chords:

using System;

public class Test
{
   static async Do(int max);

   static int[] GetPrimes() & Do(int max)
   {
     
int count = 0;
      bool[] res = new bool[max + 1];

      for (int i = 2; i <= max; i++)
     
{
        
bool prime = true;

        
for (int j = 2; j <= (int) Math.Sqrt(i); j++)
         {
           
if (i % j == 0)
            {
               prime =
false;
              
break;
           
}
        
}

        
res[i] = prime;

        
if (prime)
           
count++;
     
}

     
int[] primes = new int[count];
     
int j = 0;

     
for (int i = 2; i <= max; i++)
         if (res[i])
            primes[j++] = i;

      return primes;
   }

   static void Main()
   {
     
Do(1000);

     
int[] primes = GetPrimes();
     
foreach (int p in primes)
        
Console.WriteLine(p);

     
Console.ReadLine();
   }
}

It should be clear that this code returns the right result. However, let's make some changes to show a somewhat more complicated construction:

using System;

public class Test
{
   static async Do(int max);

   static
int[] GetPrimes() & Do(int
max)
   {
      int count = 0;
      bool[] res = new bool[max + 1];

      for (int i = 2; i <= max; i++)
      {
         bool prime = true;

         for
(int j = 2; j <= (int
) Math.Sqrt(i); j++)
         {
            if (i % j == 0)
            {
               prime =
false;
               break;
            }
         }

         res[i] = prime;
         if (prime)
            count++;
      }

      int[] primes = new int[count];
      int j = 0;

      for (int i = 2; i <= max; i++)
         if (res[i])
            primes[j++] = i;

      return primes;
   }

   static async Worker()
   {
      Console.WriteLine("Worker started");
      System.Threading.Thread.Sleep(5000);
      Console.WriteLine("Okay, time for prime calculation now");
      Do(1000);
      Console.WriteLine("Command for prime calculation issued");
   }

   static void Main()
   {
      Worker();
      Console.WriteLine("I'm waiting...");

      int
[] primes = GetPrimes();
      foreach (int p in primes)
         Console.WriteLine(p);

      Console.ReadLine();
   }
}

So, what happens in here? The main thread is not sending a request for prime calculation this time. Instead, it decides to wait for a result. Normally, the sequence Wait-Start blocks forever, as we saw in the past. However, this time we launch an asynchronous Worker method that issues the command to calculate prime numbers somewhere in the future (in this case about 5 seconds after the call). Just as an example I'm using a Thread.Sleep call to wait for a fixed time, but in reality you might be waiting for an external stimulus. The next code snippet shows an extension of this "worker and waiter" concept:

static async Worker()
{
  
Console.WriteLine("Worker started");

   
Random r =
new Random();

  
while(true)
  
{
     
int wait = r.Next(1000, 10000);
     
System.Threading.Thread.Sleep(wait);
     
Console.WriteLine("Okay, time for prime calculation now");
      Do(r.Next(10000));
      Console.WriteLine("Command for prime calculation issued");
   }
}

static void Main()
{
   Worker();
   while(true
)
   {
      Console.WriteLine("I'm waiting...");
      int
[] primes = GetPrimes();
      foreach (int p in
primes)
         Console.WriteLine(p);
   }
   Console.ReadLine();
}

Note: This post was not fully completed because my train did arrive in Amsterdam before I could finish my writings. Additional stuff on concurrency extensions in Comega will be posted later on in a 6 bis (and maybe a 6 tris) post. Then I'll show some IL too :-).

Del.icio.us | Digg It | Technorati | Blinklist | Furl | reddit | DotNetKicks

Filed under:

Comments

No Comments