Somewhat fancier BlockingCollection<T> used to distribute work to multiple threads

EventQueueDispatcher.cs

#region File Header
// /*
//    EventQueueDispatcher.cs
//    Copyright 2013 Gibraltar Software, Inc.
//    
//    Licensed under the Apache License, Version 2.0 (the "License");
//    you may not use this file except in compliance with the License.
//    You may obtain a copy of the License at
// 
//        http://www.apache.org/licenses/LICENSE-2.0
// 
//    Unless required by applicable law or agreed to in writing, software
//    distributed under the License is distributed on an "AS IS" BASIS,
//    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//    See the License for the specific language governing permissions and
//    limitations under the License.
// */

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Data.SqlClient;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Gibraltar.Analyst.Data;
using Gibraltar.Monitor;
using Gibraltar.Server.Data;
using Environment = System.Environment;
using LogMessageSeverity = Gibraltar.Analyst.Data.LogMessageSeverity;
using MetricDefinition = Gibraltar.Monitor.MetricDefinition;

#endregion File Header

namespace Gibraltar.Server.Notifications
{
    /// <summary>
    /// Handles polling for events for the server.
    /// </summary>
    public class EventQueueDispatcher : IDisposable, IQueuedEventPublisher
    {
        internal const string LogCategory = "Loupe.Server.Events.Dispatcher";

        private volatile bool m_IsShuttingDown; //indicates if the event engine is attempting to shut down processing
        private volatile bool m_IsRunning; //indicates if the event engine is currently in a running state
        private readonly object m_Lock = new object();
        private readonly ServerContext m_ServerContext;
        private readonly EventQueue m_EventQueue;
        private readonly List<IQueuedEventSubscriber> m_EventSubscribers = new List<IQueuedEventSubscriber>(); 
        private readonly Dictionary<string, Dictionary<string, List<IQueuedEventSubscriber>>> m_EventSubscriptions = new Dictionary<string, Dictionary<string, List<IQueuedEventSubscriber>>>(StringComparer.OrdinalIgnoreCase); 
        private readonly BlockingCollection<QueuedEvent> m_PendingEvents = new BlockingCollection<QueuedEvent>(1); //we only have one on deck at any moment
        private readonly CancellationTokenSource m_CancellationTokenSource = new CancellationTokenSource();
        private readonly Task m_EventProducerTask; //reads repositories to find work
        private readonly Task[] m_EventConsumerTasks; //processes work
        private readonly object m_PollSignalLock = new object();
        private bool m_ForcePoll; //used with poll signal lock to poke the dequeue thread
        
        private static CustomSampledMetric s_QueueDepthMetric;
        private static EventMetric s_QueueEventMetric;

        /// <summary>
        /// Create a new event dispatcher engine.
        /// </summary>
        public EventQueueDispatcher(ServerContext context)
        {
            if (context == null)
                throw new ArgumentNullException("context");

            m_ServerContext = context;

            m_EventQueue = new EventQueue(m_ServerContext);
            m_EventQueue.QueueChanged += (sender, args) => WakeUp();

            int numberOfThreads = Environment.ProcessorCount;
            m_EventConsumerTasks = new Task[numberOfThreads];
            for (int i = 0; i < m_EventConsumerTasks.Length; i++)
            {
                m_EventConsumerTasks[i] = new Task(AsyncProcessEvent, m_CancellationTokenSource.Token, TaskCreationOptions.LongRunning);
            }

            m_EventProducerTask = new Task(AsyncPollQueue, m_CancellationTokenSource.Token, TaskCreationOptions.LongRunning);
        }

        #region Public Properties and Methods

        /// <summary>
        /// Indicates if the dispatcher is currently running 
        /// </summary>
        public bool IsRunning { get { return m_IsRunning; } }

        /// <summary>
        /// The server context
        /// </summary>
        public ServerContext Context { get { return m_ServerContext; } }

        void IQueuedEventPublisher.Register(IQueuedEventSubscriber subscriber, string sourceName, string eventName)
        {
            IList<IQueuedEventSubscriber> subscribers = GetSubscribers(sourceName, eventName);

            if (subscribers.Contains(subscriber) == false)
            {
                subscribers.Add(subscriber);
            }
        }


        /// <summary>
        /// Start processing the events in the system
        /// </summary>
        public void Start()
        {
            Log.Write(LogMessageSeverity.Verbose, LogCategory, "Event Queue Dispatcher Start Requested", null);

            lock (m_Lock)
            {
                if (m_IsRunning)
                {
                    Log.Write(LogMessageSeverity.Verbose, LogCategory, "Ignoring Event Queue Start request because queue is already running", null);
                }
                else
                {
                    m_IsShuttingDown = false;

                    //release anything we may have left running...
                    m_EventQueue.Start();

                    RegisterSubscribers();

                    //kick off all of the tasks
                    foreach (Task task in m_EventConsumerTasks.Concat(new[] {m_EventProducerTask}))
                    {
                        task.Start();
                    }

                    m_IsRunning = true;

                    Log.Write(LogMessageSeverity.Verbose, LogCategory, "Event Queue Dispatcher Start Completed", null);
                }

                System.Threading.Monitor.PulseAll(m_Lock);
            }
        }

        /// <summary>
        /// Stop all event processing as soon as feasible
        /// </summary>
        public void Stop()
        {
            Log.Write(LogMessageSeverity.Verbose, LogCategory, "Event Queue Dispatcher Shutdown Requested", "Current shutdown status is {0}, it will be set to true.", m_IsShuttingDown);

            lock(m_Lock)
            {
                if (m_IsRunning)
                {
                    m_IsShuttingDown = true;

                    //signal that we're done adding so enumerations will exit (and we'll stop queueing)
                    m_PendingEvents.CompleteAdding();

                    // cancel all pending work
                    m_CancellationTokenSource.Cancel();

                    // wait for the tasks to complete
                    Task.WaitAll(m_EventConsumerTasks, 1000);

                    //release anything we may have left running...
                    m_EventQueue.Stop();

                    m_IsRunning = false;
                    Log.Write(LogMessageSeverity.Verbose, LogCategory, "Event Queue Dispatcher Shutdown Completed", null);
                }
            }
        }

        /// <summary>
        /// Called to ask the dispatcher to check immediately for new work if not busy
        /// </summary>
        public void WakeUp()
        {
            lock(m_PollSignalLock)
            {
                m_ForcePoll = true;

                System.Threading.Monitor.PulseAll(m_PollSignalLock);
            }
        }

        /// <summary>
        /// Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources.
        /// </summary>
        /// <filterpriority>2</filterpriority>
        public void Dispose()
        {
            Stop();

            for (int i = 0; i < m_EventConsumerTasks.Length; i++)
            {
                var task = m_EventConsumerTasks[i];
                if (task.IsCompleted) task.Dispose(); // you'll get an exception if you dispose something that's not completed.
            }

            m_CancellationTokenSource.Dispose();
            m_PendingEvents.Dispose();
            GC.SuppressFinalize(this);
        }

        #endregion

        #region Private Properties and Methods

        /// <summary>
        /// The worker thread main loop for processing events
        /// </summary>
        /// <param name="state">Unused</param>
        private void AsyncProcessEvent(object state)
        {
            try
            {
                // GetConsumingEnumerable will block if there's nothing to process
                foreach (QueuedEvent queuedEvent in m_PendingEvents.GetConsumingEnumerable(m_CancellationTokenSource.Token))
                {
#if DEBUG
                    Console.WriteLine("Events:  Starting Processing of Event {0:N0}", queuedEvent.Identity);
#endif
                    queuedEvent.StartProcessing();

                    //resolve who should handle this event.
                    var subscribers = GetSubscribers(queuedEvent.Source, queuedEvent.EventName);

                    if (subscribers.Count == 0)
                    {
#if DEBUG
                        Console.WriteLine("Events:  No one cares about {0} {1} events, so we will skip it.", queuedEvent.Source, queuedEvent.EventName);
#endif
                    }

                    bool anyFailed = false;
                    foreach (var queuedEventSubscriber in subscribers)
                    {
                        try
                        {
                            queuedEventSubscriber.Process(queuedEvent);
                        }
                        catch (Exception ex)
                        {
                            anyFailed = true;
                            Log.Write(LogMessageSeverity.Error, LogWriteMode.Queued, ex, LogCategory, string.Format("{0} {1} Event processing failed due to {3} exception in {2}", queuedEvent.Source, queuedEvent.EventName, queuedEventSubscriber.GetType(), ex.GetType()),
                                      "While asking the subscriber to process the event it threw an exception.  This will be considered a permanent failure for the event and will not be retried.\r\nEvent: {0}\r\n{1}: {2}", queuedEvent, ex.GetType(), ex.Message);
                        }
                    }

                    queuedEvent.CompleteProcessing();
                    RecordEventMetric(queuedEvent, subscribers.Count, !anyFailed);
                    queuedEvent.Remove();                  
                }
            }
            catch (OperationCanceledException ex)
            {
                // we can safely ignore this - our cancellation token was cancelled
#if DEBUG
                Console.WriteLine("Events:  Event Task exiting due to shutdown");
#endif
                Log.Write(LogMessageSeverity.Information, LogWriteMode.Queued, ex, LogCategory, "Event Task exiting due to shutdown", "Received operation canceled exception, task thread will exit.");
            }
            catch (Exception ex)
            {
                // make sure to catch any exceptions that occur... letting this bubble could crash the process.
#if DEBUG
                Console.WriteLine("Events:  Failed Processing of Event:\r\n  {0}", ex);
#endif
                Log.Write(LogMessageSeverity.Error, LogWriteMode.Queued, ex, LogCategory, "Error While Performing Indexing", null);
            }
        }

        /// <summary>
        /// Poll the event queue, lookin for work.
        /// </summary>
        /// <param name="state"></param>
        private void AsyncPollQueue(object state)
        {
            try
            {
                while (m_CancellationTokenSource.IsCancellationRequested == false)
                {
                    //Keep feeding our processing queue
                    do
                    {
                        m_ForcePoll = false;
                        var nextEvent = m_EventQueue.DequeueEvent();

                        if (nextEvent == null)
                        {
                            //the persistent queue is empty.  Don't pound the database, go quiet.
                            DateTimeOffset nextBackupPoll = DateTimeOffset.UtcNow.AddMinutes(5);
                            lock(m_PollSignalLock)
                            {
                                while ((m_ForcePoll == false)
                                       && (m_CancellationTokenSource.IsCancellationRequested == false)
                                       && (DateTimeOffset.UtcNow < nextBackupPoll))
                                {
                                    System.Threading.Monitor.Wait(m_PollSignalLock, 1000); //we want to check for shutdown every second or so
                                }
                            }
                        }
                        else
                        {
                            m_PendingEvents.Add(nextEvent, m_CancellationTokenSource.Token);
                        }
                    } while (m_CancellationTokenSource.IsCancellationRequested == false);
                }
            }
            catch (OperationCanceledException ex)
            {
                // we can safely ignore this - our cancellation token was cancelled
#if DEBUG
                Console.WriteLine("Events:  Displatcher Task exiting due to shutdown");
#endif
                Log.Write(LogMessageSeverity.Information, LogWriteMode.Queued, ex, LogCategory, "Event Displatcher Task exiting due to shutdown", "Received operation canceled exception, task dispatcher thread will exit.");
            }
            catch (Exception ex)
            {
                // make sure to catch any exceptions that occur... letting this bubble could crash the process.
#if DEBUG
                Console.WriteLine("Events:  Dispatcher Failed:\r\n  {0}", ex);
#endif
                Log.Write(LogMessageSeverity.Error, LogWriteMode.Queued, ex, LogCategory, "Event Dispatcher Poll Failed",
                          "We aren't able to continue dispatching events for the repository because of an exception ({0}): {1}",
                          ex.GetType().FullName, ex.Message);
            }
        }

        /// <summary>
        /// Find the list of subscribers 
        /// </summary>
        /// <param name="sourceName"></param>
        /// <param name="eventName"></param>
        /// <returns></returns>
        private IList<IQueuedEventSubscriber> GetSubscribers(string sourceName, string eventName)
        {
            Dictionary<string, List<IQueuedEventSubscriber>> events;
            if (m_EventSubscriptions.TryGetValue(sourceName, out events) == false)
            {
                events = new Dictionary<string, List<IQueuedEventSubscriber>>(StringComparer.OrdinalIgnoreCase);
                m_EventSubscriptions.Add(sourceName, events);
            }

            List<IQueuedEventSubscriber> subscribers;
            if (events.TryGetValue(eventName, out subscribers) == false)
            {
                subscribers = new List<IQueuedEventSubscriber>();
                events.Add(eventName, subscribers);
            }

            return subscribers;
        }

        /// <summary>
        /// Dispose each of our existing subscribers
        /// </summary>
        private void DisposeSubscribers()
        {
            var subscribers = m_EventSubscribers.ToArray();
            m_EventSubscribers.Clear();
            foreach (var queuedEventSubscriber in subscribers)
            {
                IDisposable disposableSubscriber = queuedEventSubscriber as IDisposable;
                if (disposableSubscriber != null) disposableSubscriber.Dispose();
            }
        }

        /// <summary>
        /// Register and initialize all of the subscribers.
        /// </summary>
        private void RegisterSubscribers()
        {
            //Get rid if any we have right now...
            DisposeSubscribers();

            //and load them all up again.  Here is where a greater man than I would have some fancy-pants IoC DI Arcitecture FUNLAND going on.
            m_EventSubscribers.Add(new NotificationEventProcessor());
            m_EventSubscribers.Add(new IndexingEventProcessor());
            m_EventSubscribers.Add(new CalculatedValuesEventProcessor());
            m_EventSubscribers.Add(new SessionAnalysisProcessor());
            m_EventSubscribers.Add(new ProfilesEventProcessor());

            foreach (var queuedEventSubscriber in m_EventSubscribers)
            {
                try
                {
                    queuedEventSubscriber.Initialize(this);
                }
                catch (Exception ex)
                {
                    Log.Write(LogMessageSeverity.Error, LogWriteMode.Queued, ex, LogCategory, string.Format("Event subscriber registration for {0} due to a {1}", queuedEventSubscriber.GetType(), ex.GetType()),
                              "While asking the subscriber to register for its events it threw an exception.  It will still be used for any events it managed to register.\r\n{0}: {1}", ex.GetType(), ex.Message);
                }
            }
        }

        private static void RecordQueueLength(int queueLength)
        {
            //init our metrics
            lock (Log.Metrics.Lock)
            {
                if (s_QueueDepthMetric == null)
                {
                    string sampledMetricKey = MetricDefinition.GetKey(Log.ThisLogSystem, LogCategory, "Queue Length");

                    //there's a shorter path that uses CustomSampledMetric directly but it doesn't support all display strings.
                    IMetricDefinition queueDepthMetricDefinitionRaw;
                    if (!Log.Metrics.TryGetValue(sampledMetricKey, out queueDepthMetricDefinitionRaw))
                    {
                        //the constructor does an auto-add
                        queueDepthMetricDefinitionRaw = new CustomSampledMetricDefinition(Log.ThisLogSystem, LogCategory, "Queue Length", MetricSampleType.RawCount, "Sessions", "The number of sessions waiting to be analyzed");
                    }
                    CustomSampledMetricDefinition queueDepthMetricDefinition = (CustomSampledMetricDefinition)queueDepthMetricDefinitionRaw;

                    CustomSampledMetric queueDepthMetric;
                    if (!queueDepthMetricDefinition.Metrics.TryGetValue(null, out queueDepthMetric))
                    {
                        queueDepthMetric = queueDepthMetricDefinition.Metrics.Add(null);
                    }
                    s_QueueDepthMetric = queueDepthMetric;
                }
            }

            Log.Write(s_QueueDepthMetric.CreateSample(queueLength));
        }

        private static void RecordEventMetric(QueuedEvent request, int subscribers, bool success)
        {
            //init our metrics
            lock (Log.Metrics.Lock)
            {
                if (s_QueueEventMetric == null)
                {
                    string eventMetricKey = MetricDefinition.GetKey(Log.ThisLogSystem, LogCategory, "Queue");

                    IMetricDefinition eventMetricDefinitionRaw;
                    EventMetricDefinition eventMetricDefinition;
                    if (!Log.Metrics.TryGetValue(eventMetricKey, out eventMetricDefinitionRaw))
                    {
                        eventMetricDefinitionRaw = new EventMetricDefinition(Log.ThisLogSystem, LogCategory, "Queue");
                        eventMetricDefinition = (EventMetricDefinition)eventMetricDefinitionRaw;
                        var values = (EventMetricValueDefinitionCollection)eventMetricDefinition.Values;
                        values.Add("QueuedTimestamp", request.EntryDateTime.GetType(), "Queued Timestamp", null);
                        values.Add("Latency", request.Latency.GetType(), "Latency", "How long the request waited in the queue");
                        eventMetricDefinition.DefaultValue = values.Add("Processing", request.ProcessingDuration.GetType(), "Processing Time", "How long the request took to process");
                        values.Add("Source", typeof(string), "Source", "The type of souce of the event (e.g. issue, application event, session, etc.)");
                        values.Add("Event", typeof(string), "Event", "The event that was raised");
                        values.Add("ItemId", typeof(Guid), "Item Id", "The number of tasks that subscribed to this event");
                        values.Add("Subscribers", typeof(int), "Subscribers", "The number of subscribers to the event");
                        values.Add("Success", typeof(bool), "Success", "True if the event was processed successfully");
                        eventMetricDefinitionRaw = eventMetricDefinition.Register();
                    }
                    eventMetricDefinition = (EventMetricDefinition)eventMetricDefinitionRaw;

                    s_QueueEventMetric = EventMetric.AddOrGet(eventMetricDefinition, null);
                }
            }

            var eventSample = s_QueueEventMetric.CreateSample();
            eventSample.SetValue("QueuedTimestamp", request.EntryDateTime);
            eventSample.SetValue("Latency", request.Latency);
            eventSample.SetValue("Processing", request.ProcessingDuration);
            eventSample.SetValue("Source", request.Source);
            eventSample.SetValue("Event", request.EventName);
            eventSample.SetValue("Subscribers", subscribers);
            eventSample.SetValue("ItemId", request.ItemId);
            eventSample.SetValue("Success", success);
            Log.Write(eventSample);
        }

        #endregion

    }
}