Create in-process asynchronous services in C#

• Download demo - 13.1 Kb

Introduction

This article describes how to make an in-process asynchronous service in your C# applications and why you would want to. The classes for the same provide what is often referred to as background processing. However, they also provide a foundation for building stable and robust user interfaces centered around message passing (like Windows itself) and state machines. The InprocessAsynchronousService class provided in the sample project is a base class that you can inherit your own classes from to build custom services. The sample project shows how to create these custom services.

Background

There are many instances where synchronous processing has negative side effects. Over the years, I have moved more and more away from synchronous programming to asynchronous programming. At first, it is easier to write synchronous code solutions. However, if you force yourself to start using asynchronous multithreaded programming all the way through a project's life cycle, you will come to appreciate the benefits that come later on as your project becomes larger and larger. The growth of an application usually introduces bottlenecks, bugs, artificial barriers, and unnecessary dependencies caused by synchronous programming. Some of the situations where you might want to operate in a multithreaded way include: updating a progress bar in the middle of a long running process, saving data to a database over the Internet while allowing the user to continue working with the UI, using a state machine to process requests from a UI thread instead of putting code behind click events or selected item changed events, uploading a video file while another one is being recorded, or polling a message queue like MSMQ or MQSeries while allowing the user to continue working in the UI.

Using the code

The following code shows the InprocessAsynchronousService base class. The main features of this class include:

• A thread synchronized way to drop messages into its queue.
• Supports Start, Stop, and Pause functions.
• Has a throttle control mechanism to 'prioritize' the aggressive use of processor cycles by increasing or decreasing the Thread.Sleep time in the MainLoop.
• Uses a ManualResetEvent to pause/unpause a thread in an efficient manner
• Provides events to alert listeners to any change in throttle speed or run state (start, stop, pause).
• Uses the method attribute [MethodImpl(MethodImplOptions.Synchronized)] to force threads calling into this class to take turns accessing a particular method.

using System;
using System.Threading;
using System.Runtime.CompilerServices;
using System.Collections;

namespace WindowsApplication1
{
    public delegate void PulseHandler(object args);
    public delegate void SpeedChangeHandler(int milliseconds);
    public delegate void RunStateChangeHandler(RunState state);
    public enum RunSpeed{Fast=1,Faster=2, Fastest=3, 
                         Slow=4,Slower=5,Slowest=6};
    public enum RunState{Started=1,Stoped=2,Paused=3};

    /// <summary>
    /// Summary description for InprocessAsynchronousService.
    /// </summary>
    public class InprocessAsynchronousService
    {
        //Used to put the thread into a paused 
        //state and to unpause that thread when
        //the Start method is called 
        //The thread is paused when the Pause 
        //method is called or temporarily paused
        //as part of the throttle mechanism
        //Suggested by yfoulon 
        //http://www.codeproject.com/csharp/
        //   InprocessAsynServicesInCS.asp#xx1276684xx
        private ManualResetEvent PauseTrigger = 
                       new ManualResetEvent(false);

        private Queue q = null;
        private Queue synchQ = null;

        //Delegate used to signal that it is time to process
        protected PulseHandler Heartbeat = null;

        //Used to adjust run speed of this threaded service
        protected Throttle SpeedControl = null;

        //Event that is fired off whenever 
        //the throttle speed gets changed
        public event SpeedChangeHandler SpeedChanged;
        //Event that is fired off 
        //whenever the runstate gets changed
        public event RunStateChangeHandler RunStateChanged;

        

        //Allows us to remember what run state we are in
        private int runstate = (int)RunState.Stoped;

        
        public InprocessAsynchronousService()
        {
            //Create the queue to hold messages
            q = new Queue();
            //Get a synchronized version so we can 
            //do multithreaded access to the queue
            synchQ = Queue.Synchronized(q);
        }

        #region Messaging
        [MethodImpl(MethodImplOptions.Synchronized)]
        public void SendMessage(ServiceMessage Message)
        {
            synchQ.Enqueue(Message);            
        }
        #endregion

        #region Public throttle control
        [MethodImpl(MethodImplOptions.Synchronized)]
        public void SetSpeed(RunSpeed speed)
        {
            switch(speed)
            {
                case RunSpeed.Fast:
                    SpeedControl.Fast();
                    break;

                case RunSpeed.Faster:
                    SpeedControl.Faster();
                    break;

                case RunSpeed.Fastest:
                    SpeedControl.Fastest();
                    break;

                case RunSpeed.Slow:
                    SpeedControl.Slow();
                    break;

                case RunSpeed.Slower:
                    SpeedControl.Slower();
                    break;

                case RunSpeed.Slowest:
                    SpeedControl.Slowest();
                    break;
            }

            
            //Let listeners know we just changed the speed
            if(SpeedChanged != null)
            {
                SpeedChanged(SpeedControl.Speed);
            }
        }
        [MethodImpl(MethodImplOptions.Synchronized)]
        public void SetSpeed(int milliseconds)
        {
            SpeedControl.GoThisFast(milliseconds);

            

            //Let listeners know we just changed the speed
            if(SpeedChanged != null)
            {
                SpeedChanged(SpeedControl.Speed);
            }
        }
        #endregion

        #region Run Control
        [MethodImpl(MethodImplOptions.Synchronized)]
        public void Start()
        {
            if(runstate != (int)RunState.Started)
            {
                if(runstate == (int)RunState.Paused)
                {
                    //Set State
                    runstate = (int)RunState.Started;

                    //Unpause the thread
                    PauseTrigger.Set();

                    //Let listeners know the state changed
                    if(RunStateChanged != null)
                    {
                        RunStateChanged((RunState)runstate);
                    }
                }
                else if(runstate == (int)RunState.Stoped)
                {
                    //Set State
                    runstate = (int)RunState.Started;

                    //Let listeners know the state changed
                    if(RunStateChanged != null)
                    {
                        RunStateChanged((RunState)runstate);
                    }

                    //Delegate execution of this 
                    //thread to the MainLoop method
                    ThreadStart ts = new ThreadStart(MainLoop);
                    //Create the thread
                    Thread t = new Thread(ts);
                    //Begin
                    t.Start();
                }
            }
        }

        [MethodImpl(MethodImplOptions.Synchronized)]
        public void Stop()
        {
            //Kill the thread
            if(runstate != (int)RunState.Stoped)
            {
                //Set the throttle to stop
                SpeedControl.GoThisFast(0);

                //Put Stop message in the queue
                synchQ.Enqueue(new ServiceMessage(RunState.Stoped));

                //Set State
                runstate = (int)RunState.Stoped;

                //Let listeners know the state changed
                if(RunStateChanged != null)
                {
                    RunStateChanged((RunState)runstate);
                }
            }
        }

        [MethodImpl(MethodImplOptions.Synchronized)]
        public void Pause()
        {
            if(runstate != (int)RunState.Paused)
            {
                //Set state
                runstate = (int)RunState.Paused;

                //Let listeners know the state changed
                if(RunStateChanged != null)
                {
                    RunStateChanged((RunState)runstate);
                }
            }
        }
        #endregion

        #region Thread lives in a infinite loop 
           within this method until Stop is called
        private void MainLoop()
        {
            for(; ; )
            {
                if(runstate == (int)RunState.Paused)
                {
                    PauseTrigger.WaitOne(-1, false);
                    //Reset the trigger so 
                    //it can be triggered again
                    PauseTrigger.Reset();
                }

                //If there are no messages 
                //to pass on then just fire with a null
                if(q.Count == 0)
                {
                    //Fire!
                    Heartbeat.DynamicInvoke(new object[] { null });
                }
                else
                {
                    //Get the message from the queue
                    ServiceMessage msg = 
                       (ServiceMessage)synchQ.Dequeue();

                    //If we got a message then see 
                    //if it contains a run state change
                    if(msg.ChangeToRunState != 0)
                    {
                        //If so then see if this 
                        //is ONLY a run state change
                        if(msg.Args != null)
                        {
                            //If not then fire off the 
                            //heartbeat along with the args
                            Heartbeat.DynamicInvoke(new object[] { msg.Args });

                            //If the requested run state 
                            //change is to stop then return which kills
                            //this threads execution loop
                            if(((RunState)msg.ChangeToRunState) == 
                                               RunState.Stoped)
                            {
                                return;
                            }
                        }
                        else if(((RunState)msg.ChangeToRunState) 
                                             == RunState.Stoped)
                        {
                            //If the requested run state change 
                            //is to stop then return which kills
                            //this threads execution loop
                            return;
                        }
                    }
                    else
                    {
                        //If not then just pass on the args
                        Heartbeat.DynamicInvoke(new object[] { msg.Args });
                    }
                }


                //Pause the thread until the throttle period is over or until
                //the Start method is called
                PauseTrigger.WaitOne(SpeedControl.Speed, false);
            }
        }
        #endregion
    }
}

Since this is a generic base class, no real work should be done in the MainLoop method. Instead, your custom class should inherit from this class and set the Heartbeat delegate to a method in your custom class. Then, each time the MainLoop pulses the Heartbeat delegate, your custom method will be called where you can do all the real work you want your class to do on this service thread. For example...

Heartbeat = new PulseHandler(OnHeartbeat);

with the handler looking like...

private void OnHeartbeat(object args)
{
    if(args != null)
    {
        if(ChangeColor != null)
        {
            ChangeColor((System.Drawing.Color)args);
        }
    }

    if(Tick != null)
    {
        Tick(DateTime.Now);
    }
}

Here is the full MyService class code from the sample project. This class inherits from the InprocessAsynchronousService class and does the actual work to be done on each heartbeat.

using System;

namespace WindowsApplication1
{

    public delegate void LapseHandler(DateTime time);
    public delegate void ColorChangeHandler(System.Drawing.Color c);

    /// <summary>
    /// Summary description for MyService.
    /// </summary>
    public class MyService : InprocessAsynchronousService
    {
        

        public MyService()
        {
            //Attach the parents Heartbeat delegate 
            //to my local method called OnHeartbeat
            Heartbeat = new PulseHandler(OnHeartbeat);
            //Setup throttle
            SpeedControl = new Throttle(2000,1000, 
                            500,5000,10000,15000);
        }

        public MyService(int Fast, int Faster, int Fastest, 
                                   int Slow, int Slower, int Slowest)
        {
            //Attach the parents Heartbeat delegate 
            //to my local method called OnHeartbeat
            Heartbeat = new PulseHandler(OnHeartbeat);
            //Setup throttle
            SpeedControl = new Throttle(Fast,Faster, 
                               Fastest,Slow,Slower,Slowest);
        }

        private void OnHeartbeat(object args)
        {
            //If a ServiceMessage was added to the message 
            //queue and args were provided then...
            if(args != null)
            {
                //If someone has wired up to my ChangeColor event...
                if(ChangeColor != null)
                {
                    //Then fire the event
                    ChangeColor((System.Drawing.Color)args);
                }
            }

            //Every time the Heartbeat pulses even 
            //if there was no message in the queue
            //fire off the Tick event with the current time
            if(Tick != null)
            {
                Tick(DateTime.Now);
            }
        }

        public event LapseHandler Tick;
        public event ColorChangeHandler ChangeColor;
    }
}

The messages that are passed into the service class are not just simple object wrappers. Instead, they hold two bits of information. One is the args that you want passed to the Heartbeat handler in your custom service class. The other is a RunState change. Right now, the only RunState change that the InprocessAsynchronousService class supports is RunState.Stoped. If this RunState is set then when the MainLoop in the InprocessAsynchronousService class processes a message from the queue, the MainLoop will kill its thread by returning from the MainLoop method.

With this code, you can write your own in-process asynchronous services that perform all kinds of background processing, asynchronous UI control, and even implement more complex constructs such as state machines.

To see this code in action, check out the sample project provided. It demonstrates all the abilities of the InprocessAsynchronousService class. Try running the project, clicking the 'play' button then the 'pause' button, and then send multiple messages into the queue by clicking the Background Color button several times. Then press the 'play' button again. What this will do is it will stack up several color change messages in the MyService queue and when you 'press' play again, the messages will be executed one by one making the background color of the form change with each Heartbeat.

Points of Interest

• Multithreaded processing encapsulated in an easy to use base class.
• Uses queuing to allow requests to be cached while the thread works on dispatching them.
• Provides methods similar to the Windows Services in the Control Panel Admin Tools (Start, Stop, Pause).

History

• January 2006 - Added support for ManualResetEvent so that pausing the MainLoop thread doesn't take up 100% processor cycles.