7
\$\begingroup\$

My code is a variant on the Visitor Pattern and a "leaky bucket" variant. The goal is pretty straightforward: the "bucket" will collect a specified number of items (say, for example, 500) and then empty the queue, running some CPU-bound operation on ("visiting") each item. (Yes, I'm aware that there's probably existing code "out there" that does something similar to what I'm doing here - this code really isn't all that novel of an idea).

I've read documentation on how to implement the Visitor pattern before but this is truthfully the first time I've actually implemented one. Is what I have below a "valid" way to implement this? Also, have I properly implemented the concurrency/thread safety? I think it should be free from race conditions but it's also possible that I can implement this more simply.

/// <summary>
/// "Overflowing bucket" implementation
/// </summary>
/// <typeparam name="T">Type of the data structure we're operating on</typeparam>
/// <typeparam name="U">Type of the items in the data structure</typeparam>
/// <remarks>
/// The basic idea of this data structure is that it'll collect a certain number of items and then empty the queue.
/// 
/// The "overflowing bucket" metaphor isn't perfect because every time that the "bucket" is "filled to the brim" or starts
/// to "overflow" we just empty the whole thing.
/// </remarks>
public class OverflowingBucket<T, U>
{
    #region Fields
    // Not volatile - this will only be accessed by the background thread
    private readonly T itemToActOn;

    // Not volatile - this will only be accessed by the background thread
    private readonly Action<T, U> visitorOperation;

    // This action runs after we empty the queue
    private readonly Action<T> afterAction;

    // Must be concurrent because we could add to the queue while an operation is in place
    private readonly ConcurrentQueue<U> queue;

    // Will be accessed by multiple threads
    private volatile bool inProgress = false;

    // Obviously used for locking
    private readonly object lockObj = new object();
    #endregion Fields

    #region Constructor
    /// <summary>
    /// Create a new Overflowing Bucket
    /// </summary>
    /// <param name="itemToActOn">Item that is "visited." Required.</param>
    /// <param name="visitorOperation">Visitor operation. Required.</param>
    /// <param name="afterAction">Action that occurs after each time that the queue is emptied. Optional.</param>
    /// <param name="leakyNumber">Number of items that the queue may contain before the bucket starts "leaking."</param>
    public OverflowingBucket(T itemToActOn, Action<T, U> visitorOperation, Action<T> afterAction, int leakyNumber = 50)
    {
        this.itemToActOn = itemToActOn;
        this.visitorOperation = visitorOperation;
        this.afterAction = afterAction;
        queue = new ConcurrentQueue<U>();
        LeakyNumber = leakyNumber;
    }
    #endregion Constructor

    #region Properties
    /// <summary>
    /// Get or set the number at which we run an iteration
    /// </summary>
    public int LeakyNumber
    {
        get;
        set;
    }

    /// <summary>
    /// Get a value indicating whether a "visit" is already in progress (i.e. we are actively emptying the queue)
    /// </summary>
    public bool InProgress
    {
        get { return inProgress; }
    }
    #endregion Properties

    #region Public Methods
    /// <summary>
    /// Add an item, emptying the queue if necessary
    /// </summary>
    /// <param name="item">Item to add to the "bucket"</param>
    /// <returns>Handle to await the outcome</returns>
    public async Task Add(U item)
    {
        queue.Enqueue(item);

        if (queue.Count >= LeakyNumber && !inProgress)
        {
            await Visit();
        }
    }

    /// <summary>
    /// Visit the data structure. This will ignore successive calls (i.e. calls that occur while a previous"visit" operation is running).
    /// </summary>
    /// <returns>Handle to await the boolean result; result is <c>true</c> if we successfully ran the
    /// operation and <c>false</c> if the operation failed or never ran</returns>
    public async Task<bool> Visit()
    {
        // If there's already an operation in progress, ignore the call
        if (!inProgress)
        {
            bool result = true;
            await Task.Run(
                delegate ()
                {
                    lock (lockObj)
                    {
                        // "Double check" pattern to prevent race conditions
                        if (!inProgress)
                        {
                            inProgress = true;

                            // Completely empty the queue
                            while (queue.Any())
                            {
                                U item;

                                bool success = queue.TryDequeue(out item);

                                if (success) visitorOperation(itemToActOn, item);
                                else result = false;
                            }

                            // The "afterAction" operation is optional, so only
                            afterAction?.Invoke(itemToActOn);

                            inProgress = false;
                        } // End double-check condition
                    } // Release lock
                }); // End Task

            return result;
        } // End if inProgress

        // TODO: Should we throw an exception if we're already Visiting?
        return false;
    } // End Visit method
    #endregion Public Methods
} // End OverflowingBucket class

Here is an example of code that I wrote to call this (to give a better sense of what I'm trying to do).

It's probably not that great of an example because the operations in question aren't particularly CPU-bound (which is obviously the primary purpose of Task.Run operations), but it should still give an idea of what I'm trying to do.

private static void TestOverflowingBucket()
    {
        var doc = new XmlDocument();
        doc.LoadXml("<QATestLog></QATestLog>");

        Action<XmlDocument, XmlElement> operation =
            delegate (XmlDocument document, XmlElement element)
            {
                document.DocumentElement.AppendChild(element);
            };

        Action<XmlDocument> afterOperation = xmlDoc => xmlDoc.Save("TestDocument.xml");

        var bucket = new OverflowingBucket<XmlDocument, XmlElement>(doc, operation, afterOperation);

        // AsyncPump is to correct some "oddities" in the way that async/await works in a console application
        // Source: https://blogs.msdn.microsoft.com/pfxteam/2012/01/20/await-synchronizationcontext-and-console-apps/
        AsyncPump.Run(async () =>
        {
            List<Task> tasks = new List<Task>();

            Random random = new Random();

            for (int i = 0; i < 1003; i++)
            {
                XmlElement newElement = doc.CreateElement("step");

                XmlAttribute newAttribute = doc.CreateAttribute("number");
                newAttribute.Value = i.ToString();

                newElement.Attributes.Append(newAttribute);

                // Don't await, it's OK for us to continue to add to the queue while an execution is happening
                // This will generate a compiler warning since Add is async and we don't await
                Task task = bucket.Add(newElement);

                // For added realism, wait some random time between adds
                await Task.Delay(random.Next(1, 1000));

                if (!task.IsCompleted)
                    tasks.Add(task);
            }

            if (bucket.InProgress)
            {
                Task.WaitAll(tasks.ToArray());
            }
            else
            {
                // If there are any "left over" that weren't "picked up" by a previous Visit execution
                await bucket.Visit();
            }
        });
    }
\$\endgroup\$
6
\$\begingroup\$

First of all, IMHO the code is well written and has some meaningful comments.

Visitor Pattern

Actually, I am not sure if the OverflowingBucket has something to do with the visitor pattern. In my eyes, the visitor pattern can be used to add logic to a data structure by implementing visitors. A visitor visits each element of the data structure and decide what to do depending of the element's type.

The OverflowingBucket can be used to add logic to an object for processing other items in a special manner. Therfore I wouldn't use the term "visitor pattern" here.

OverflowingBucket

  • Because it is not a visitor in my eyes, I would rename the method Visit to Process or something like that.
  • Consider to make the method Process private and add another method ProcessPendingItems or something like that.
  • The only case where the method Visit returns false is, if queue.Any() is true and queue.TryDequeue(out item) is false. That should never happen in the current version of the class... therefore I would make the method void.

TestOverflowingBucket

  • The delay does not delay the Add-Task but the main thread. Adding Thread.Sleep() within the operation delegate would be more realistic.
  • There is no need to check if the task is completed. You can just add them all to the tasks list.
\$\endgroup\$
2
\$\begingroup\$

I prefer simplicity when it comes to async/await. await is useful and makes the code cleaner when you have code that executes after a Task completes. Using it just to pass tasks along methods (like in public async Task Add(U item)), creates noise.

In public async Task<bool> Visit() you only use await to wait for the task to finish so that you can return it's result. I would have rather made the delegate return a bool rather then capture result and modifying it and then I would have simply returned the task created by Task.Run or Task.FromResult in the case of your other branch.

\$\endgroup\$
0
\$\begingroup\$

I largely agree with JanDotNet.

I'm not sure if it's a good idea, but maybe OverflowingBucket should clean up after it self as a IDisposable?

\$\endgroup\$

Your Answer

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

Not the answer you're looking for? Browse other questions tagged or ask your own question.