2
\$\begingroup\$

I'm working on an ASP.Net MVC 5/WebApi 2 project. I'm using EF 6 code first for my database access.

I have a WebApi action method in which I've to update some data in my database based on the request. To correctly update the data it is very important that:

  1. The method doesn't change the same records concurrently
  2. The method does not read records while they're getting updated

I've tried to implement a queue like class, which has an internal queue for each combination of a key and a value: e.g. Key: "userId" value: 100.

The action method enqueues an item for each parameters (which is relevant for DB access) and waits until the queue has dequeued all items. After dequeuing an item each queue waits until the action method has done its work, before the queue dequeues the next item.

namespace Movie.Web
{
    public class MyController : ApiController
    {
        public void Test(String userId, String itemNo)
        {
            //Create a queue item for each parameters
            var items = new List<SyncQItem>
            {
                new SyncQItem
                {
                    Key = "userId",
                    Value = userId
                },
                new SyncQItem
                {
                    Key = "itemNo",
                    Value = itemNo
                }
            };

            //Add the items to a queue
            SyncQManager.EnqueueItems(items);
            //Wait until all items are dequeued
            WaitHandle.WaitAll(items.Select(x => x.QResetEvent.WaitHandle).ToArray());

            //do some work with data from EF....

            //Unblock the queues, so that the can dequeue the next item.
            items.ForEach(x =>
            {
                x.WorkResetEvent.Set();
                x.WorkResetEvent.Dispose();
            });
        }
    }

    /// <summary>
    /// Class managing all the queues.
    /// </summary>
    public static class SyncQManager
    {
        #region Fields
        private static readonly ConcurrentDictionary<String, ConcurrentDictionary<String, SyncQ>> QRegistration
            = new ConcurrentDictionary<String, ConcurrentDictionary<String, SyncQ>>(); 
        private static readonly Object SyncRoot = new Object();
        #endregion

        /// <summary>
        /// Adds the given items to a queue.
        /// </summary>
        /// <param name="items">The items to add.</param>
        public static void EnqueueItems( List<SyncQItem> items )
        {
            try
            {
                lock ( SyncRoot )
                {
                    items.ForEach( item =>
                    {
                        var attributeQ = QRegistration.GetOrAdd( item.Key,
                                                                  x =>
                                                                  new ConcurrentDictionary<String, SyncQ>() );
                        SyncQ q;
                        attributeQ.TryGetValue( item.Value, out q );
                        if ( q != null )
                        {
                            q.Enqueue( item );
                        }
                        else
                        {
                            q = new SyncQ( item.Key, item.Value );
                            attributeQ.TryAdd( item.Value, q );
                            q.Empty += Queue_Empty;
                            q.Enqueue( item );
                        }
                    } );
                }

            }
            catch ( Exception ex )
            {
                var e = ex;
            }
        }

        /// <summary>
        /// Removes a empty queue.
        /// </summary>
        /// <param name="itemKey">The key.</param>
        /// <param name="itemValue">The value.</param>
        private static void Queue_Empty(String itemKey, String itemValue)
        {
            try
            {
                ConcurrentDictionary<String, SyncQ> dictionary;
                QRegistration.TryGetValue( itemKey, out dictionary );
                if ( dictionary == null )
                    return;

                SyncQ q;
                dictionary.TryRemove( itemValue, out q );
            }
            catch ( Exception ex )
            {
                var e = ex;
            }
        }
    }

    /// <summary>
    /// Queue class.
    /// </summary>
    public class SyncQ
    {
        #region Fields
        private readonly ConcurrentQueue<SyncQItem> _queue = new ConcurrentQueue<SyncQItem>();
        private volatile Boolean _canDequeueItem = true;
        private readonly Object _syncRoot = new Object();
        private readonly String _key;
        private readonly String _value;
        #endregion

        #region Properties
        /// <summary>
        /// Gets or sets whether the queue can dequeue an item or not.
        /// </summary>
        private Boolean CanDequeueItem
        {
            get
            {
                lock ( _syncRoot )
                {
                    return _canDequeueItem;
                }
            }
            set
            {
                lock ( _syncRoot )
                {
                    _canDequeueItem = value;
                }
            }
        }
        #endregion

        /// <summary>
        /// Event fired when the queue is empty.
        /// </summary>
        public event Action<String, String> Empty;

        #region Ctor

        public SyncQ(String key, String value)
        {
            _key = key;
            _value = value;
        }

        #endregion

        /// <summary>
        /// Adds the given item to the queue.
        /// </summary>
        /// <param name="item">The item to add.</param>
        public void Enqueue(SyncQItem item)
        {
            _queue.Enqueue( item );

            if (CanDequeueItem)
                Dequeue();
        }

        /// <summary>
        /// Dequeues an item form the internal queue.
        /// </summary>
        private void Dequeue()
        {
            //Check if the queue can dequeue an item
            if (!CanDequeueItem)
                return;

            //Dequeue an item on a new thread
            Task.Run( () =>
            {
                if ( !CanDequeueItem )
                    return;
                CanDequeueItem = false;

                SyncQItem outItem;
                if ( ! _queue.TryDequeue( out outItem ) )
                {
                    return;
                }
                //Unblock the action method
                outItem.QResetEvent.Set();
                outItem.QResetEvent.Dispose();

                //Wait until the action method has done its work
                outItem.WorkResetEvent.Wait();

                //DEqu next item or fire empty event
                CanDequeueItem = true;               
                if ( !_queue.IsEmpty )
                    Dequeue();
                else if ( Empty != null )
                    Empty( _key, _value );

            } );
        }
    }

    /// <summary>
    /// Class representing an queue item.
    /// </summary>
    public class SyncQItem
    {
        /// <summary>
        /// The key of the item.
        /// </summary>
        public String Key { get; set; }
        /// <summary>
        /// The value of the item.
        /// </summary>
        public String Value { get; set; }
        /// <summary>
        /// Reset event to block the action method until the item is dequeued.
        /// </summary>
        public ManualResetEventSlim QResetEvent { get; set; }
        /// <summary>
        /// Reset event to block the queue until the action method has done it's work.
        /// </summary>
        public ManualResetEventSlim WorkResetEvent { get; set; }

        public SyncQItem()
        {
            QResetEvent = new ManualResetEventSlim();
            WorkResetEvent = new ManualResetEventSlim();
        }
    }
}

My questions:

  1. Is it ok to do something like this with a queue, or is it a totally wrong way to do it?

  2. I've tested my code and I think it's quite slow. Do you have any idea how I can improve the performance?

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.