Orleans vs a Distributed Cache

I’m often asked what the difference is between Orleans and a distributed cache. It’s a good observation because on the face of it they have similar aims, to hold objects in memory for the right period of time to provide a performance benefit.

However, there are a number of differences.

1. The grains in Orleans are programmable, whereas caches normally just store data.
2. With a distributed cache you keep objects in memory, but there’s a good chance that the data doesn’t reside on the same machine that’s serving the request. This means that machine processing the request has to retrieve the data from another node. Orleans will send the request to the machine which holds the data, which in theory should be more efficient.
3. Orleans queues message to each grain and processes them one by one with a single threaded programming model. This stops concurrency problems and race conditions. With caches you have to build a locking system yourself, as described here for memcached.
4. A grain can register a timer so it can call itself without requiring an external request. Caches generally just respond to external requests.

People are certainly using Orleans as a cache, but I think it’s capable of much more.

Codename Orleans – Using Generics with Grains

In Orleans you can use generics with grain interfaces:

public interface IGenericGrain<T> : Orleans.IGrain
{
    Task SetValue(T value);

    Task GetValue();
}

When you talk to this grain from a client application, you can specify the value of ‘T’ on the factory class:

var grain1 = GenericGrainFactory<int>.GetGrain(1);
await grain1.SetValue(123);
await grain1.GetValue();

var grain2 = GenericGrainFactory<string>.GetGrain(2);
await grain2.SetValue("foo");
await grain2.GetValue();

You can even extend ‘T’ to the persistence layer:

public interface IMyGrainState<T> : IGrainState
{
    T Value { get; set; }
}

[StorageProvider(ProviderName = "AzureStore")]
public class GenericGrain<T> : Orleans.GrainBase<IMyGrainState<T>>, IDbGrain<T>
{
    public Task SetValue(T value)
    {
        this.State.Value = value;
        return this.State.WriteStateAsync();
    }

    public Task GetValue()
    {
        return Task.FromResult(this.State.Value);
    }
}

This seems to work well for value types such as string, int, double etc…

One word of caution; make sure that you always use the same value for ‘T’ when dealing with a particular grain ID, otherwise you’ll get cast exceptions.

Running Codename Orleans on Ubuntu

 

Yesterday I did a presentation on Orleans. Afterwards I was chatting to Richard Conway. He told me he was thinking of getting Orleans to run on Linux to solve a data analytics problem. “Good luck” I said.

Today I thought I’d give it a try. Here’s what I did:

1. Built an Orleans project using Visual Studio.

2. Took a blank Ubuntu 14 image.

3. Installed Mono:

  $ sudo apt-get install mono-complete

1. Copied the Orleans local silo binaries onto the Ubuntu machine. The files are located here:

  C:\Microsoft Codename Orleans SDK v0.9\SDK\LocalSilo\

1. Copied the grains DLLs into the same ‘LocalSilo’ directory.

(usually these are placed into an ‘Applications’ directory for you by a post-compilation step in the grain implementation project, but placing the binaries in this location didn’t seem to work for me).

1. Started the Silo:

  $ mono OrleansHost.exe

1. Started my client application in the same way:

  $ mono ConsoleApplication1.exe

It just worked :¬)

I don’t expect you’ll get any support running Orleans on Linux, but it’s interesting that you can!

Note: I didn’t have any persistence providers configured in the ‘OrleansConfiguration.xml’ file.

Reading the microphone’s audio stream in WinRT

I wanted to use the microphone on a Surface tablet to measure the audio level in the room. The technique I used was to implement my own stream, and then have the Microphone write to it.

Set up the MediaCapture class as usual:

 

var media = new MediaCapture();
var captureInitSettings = new MediaCaptureInitializationSettings();
captureInitSettings.StreamingCaptureMode = StreamingCaptureMode.Audio;
await media.InitializeAsync(captureInitSettings);
media.Failed += (_, ex) => new MessageDialog(ex.Message).ShowAsync();

var stream = new AudioAmplitudeStream(); // custom stream implementation
media.StartRecordToStreamAsync(MediaEncodingProfile.CreateWav(AudioEncodingQuality.Low), stream);
stream.AmplitudeReading += AmplitudeReading; // get an amplitude event

 

But you’ll notice I’m passing in an ‘AudioAmplitudeStream’. This is my custom stream implementation which just takes the average of the amplitude reported by the microphone. Here’s the code:

 

using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Runtime.InteropServices.WindowsRuntime;
using System.Threading.Tasks;
using Windows.Storage.Streams;

namespace IoT.WinRT
{
    class AudioAmplitudeStream : IRandomAccessStream
    {
        public bool CanRead
        {
            get { return false; }
        }

        public bool CanWrite
        {
            get { return true; }
        }

        public IRandomAccessStream CloneStream()
        {
            throw new NotImplementedException();
        }

        public IInputStream GetInputStreamAt(ulong position)
        {
            throw new NotImplementedException();
        }

        public IOutputStream GetOutputStreamAt(ulong position)
        {
            throw new NotImplementedException();
        }

        public ulong Position
        {
            get { return 0; }
        }

        public void Seek(ulong position)
        {

        }

        public ulong Size
        {
            get
            {
                return 0;
            }
            set
            {
                throw new NotImplementedException();
            }
        }

        public void Dispose()
        {

        }

        public Windows.Foundation.IAsyncOperationWithProgress<IBuffer, uint> ReadAsync(IBuffer buffer, uint count, InputStreamOptions options)
        {
            throw new NotImplementedException();
        }

        public Windows.Foundation.IAsyncOperation FlushAsync()
        {
            return AsyncInfo.Run(_ => Task.Run(() => true));
        }

        public Windows.Foundation.IAsyncOperationWithProgress<uint, uint> WriteAsync(IBuffer buffer)
        {

            return AsyncInfo.Run<uint, uint>((token, progress) =>
            {
                return Task.Run(() =>
                {
                    using (var memoryStream = new MemoryStream())
                    using (var outputStream = memoryStream.AsOutputStream())
                    {
                        outputStream.WriteAsync(buffer).AsTask().Wait();

                        var byteArray = memoryStream.ToArray();
                        var amplitude = Decode(byteArray).Select(Math.Abs).Average(x => x);

                        if (AmplitudeReading != null) this.AmplitudeReading(this, amplitude);

                        progress.Report((uint)memoryStream.Length);
                        return (uint)memoryStream.Length;
                    }
                });
            });
        }

        private IEnumerable Decode(byte[] byteArray)
        {
            for (var i = 0; i < byteArray.Length - 1; i += 2)
            {
                yield return (BitConverter.ToInt16(byteArray, i));
            }
        }

        public delegate void AmplitudeReadingEventHandler(object sender, double reading);

        public event AmplitudeReadingEventHandler AmplitudeReading;

    }
}

 

The stream is only partially implemented, not all the calls are required.

You can then subscribe to the event, and show the background noise level in your app!

 

void AmplitudeReading(object sender, double reading)
{
    Debug.WriteLine("Noise level: {0:0} dB", ToDb(reading));           
}

static double ToDb(double value)
{
    return 20 * Math.Log10(Math.Sqrt(value * 2));
}

Here is a sample application: https://github.com/richorama/WinRTdB

Have fun!

Aggregating Results in Orleans

Orleans is great for maintaining hundreds of thousands of gains, each with their own internal state. However, it’s sometimes difficult to get aggregate figures out the system, such as the total for a particular value across all grains. You can’t practically make a fan out request to all grains to retrieve a value, firstly such a request would time out, secondly you probably don’t know what grains you’ve got. The answer is to set up a single grain, which all other grains will report to. We’ll look at a simple scenario to see how this could be done.

Let’s suppose that we have grains that hold a score, and we’d like the total score for the entire system. Every time the score changes in the grains, we call a ‘total score grain’ which maintains the overall total. The grain looks like this:

 

    // Total Score Grain interface
    public interface ITotalScoreGrain : IGrain
    {
        Task AddScore(int value);

        Task GetTotalScore();
    }


    // Total Score Grain implementation
    public class TotalScoreGrain : GrainBase, ITotalScoreGrain
    {
        int total;

        public Task AddScore(int value)
        {
            this.total += value;
            return TaskDone.Done;
        }

        public Task GetTotalScore()
        {
            return Task.FromResult(total);
        }
    }

This looks simple enough, but it’s a really bad idea, for a couple of reasons:

• We now have a single grain acting as a bottleneck for the entire system. We’ve defeated the point in having a distributed system.
• If you’re running the system on several servers, the chances are that the ‘total score grain’ will not be in the same silo as the grain calling it. This makes the call to the ‘total score grain’ a remote call, which is more expensive (takes more time).

Let’s improve on this by introducing  a ‘sub total grain’ which will be responsible for collecting all scores from each silo, and reporting this up to the ‘total score grain’:

    // Sub Total Grain interface
    [StatelessWorker]
    public interface ISubTotalGrain : Orleans.IGrain
    {
        Task AddScore(int value);
    }

    // Sub Total Grain implementation
    public class SubTotalGrain : GrainBase, ISubTotalGrain
    {
        int score;

        public override Task ActivateAsync()
        {
            RegisterTimer(SendUpdate, null, TimeSpan.FromSeconds(5), TimeSpan.FromSeconds(5));
            return base.ActivateAsync();
        }

        async Task SendUpdate(object _)
        {
            if (score == 0) return;
            var totalScoreGrain = TotalScoreGrainFactory.GetGrain(0);
            await totalScoreGrain.AddScore(score);
            score = 0;
        }

        public Task AddScore(int value)
        {
            this.score += value;
            return TaskDone.Done;
        }
    }

This grain has two interesting properties. First of all it’s a Stateless Worker (see the attribute on the interface), secondly it has a timer, so it doesn’t forward every score update on, it just updates some internal state, and forwards the information if required every 5 seconds. This means that:

• The grain will auto scale as required, removing the bottleneck.
• The grain will always be created in the local silo. Which removes the need for network hop to report score changes. The cross-silo hop will only be triggered every 5 seconds.

This design introduces ‘eventual consistency’ to the total score, as the score could always be up to 5 seconds behind. However, the system should scale nicely with no bottlenecks.

Is it right to store state in a Stateless Worker? It’s not the intention of Stateless Worker grains to be used in this way, but they are activated and work in the same way as normal grains, there are just multiple copies of them, and they’re created locally. This is fine if you can reconcile the state distributed across all the activations, which in our case is easy, because we can just add it up.

Acknowledgements to Sergey Bykov for helping to foster the idea.