Observing Changes to an Underlying Array

• Download source - 4.12 KB

Introduction

Recently, I came across the problem of visualizing an array using a WPF ItemsControl. The problem was that the array was constantly being modified by factors outside the control of my own code, so I could not raise the appropriate collection notifications because there was no fixed flow of control where these notifications should be raised. Furthermore, the array consisted of a few hundred thousands of elements, so simply resetting the ItemsControl's ItemsSource at fixed time intervals was out of the question.

So I came up with a solution thanks to the helpful feedback of John Simmons, a fellow CodeProject user. The idea behind this solution is to subclass the ObservableCollection<T> class so as to provide an INotifyCollectionChanged interface to the consumer, while in the background running a worker thread that keeps watch of the underlying array and moves the changed items back into the ObservableCollection.

Background

The main problem that has to be tackled in some way is that simply wrapping the underlying array around an ObservableCollection will not suffice, because the ObservableCollection constructor copies the array elements and emits CollectionChanged events when somebody modifies the copy through the ObservableCollection's ICollection<T> (or ICollection, or IList) interface. But what happens when for some reason the generating array is modified in some way? What if our purpose is to observe the generating array itself, not some copy?

So our solution inherits the ObservableCollection class and keeps a reference to, not a copy of, the underlying array. At fixed time intervals (checkpoints), it checks the contents of the underlying array and updates its instance so that it becomes in sync again with the array.

In addition, our solution tackles the possibility that the array may contain many items. To overcome this, it makes use of the excellent parallel processing classes found in .NET v4. Assuming that the array may contain millions of elements, but not many of them may change from one checkpoint to the next, we can easily monitor this array with minimal overhead.

Finally, as a "bonus", our new observable collection class is capable of projecting its elements to another type, given a projection function in its constructor. For example, we may have an array of a thousand integers, wrap it around our collection class and give it a projection function like i => i * i so that an observer of our collection sees the squares of those integers.

Using the Code

We have provided extension methods so that we can wrap an array around a monitor like so:

// Create an array
int[] theArray = new int[1000000];
// Populate the array
for (var i = 0; i < theArray.Length; i++)
    theArray[i] = i;

// Create a monitor wrapper
var mon = theArray.AsMonitored();
// Create a monitor wrapper which presents
// to its observer the squares of the array's elements.
var monProj = theArray.AsMonitoredProjected(i => i * i);

The extension methods are as follows:

• AsMonitored<T>() - Monitors the underlying array of T every 100 milliseconds
• AsMonitored<T>(int period) - Monitors the underlying array of T every period milliseconds
• AsMonitoredProjected<T, P>(Func<T, P> project) - Monitors the underlying array of T every 500 milliseconds, while at the same time projecting it to type P; and
• AsMonitoredProjected<T, P>(Func<T, P> project, int period) - Monitors the underlying array of T every period milliseconds, while at the same time projecting it to type P.

Implementation Details

The main class which implements our collection is called MonitoredProjectedArray.

public class MonitoredProjectedArray<T, P>
    : ObservableCollection<P>, IDisposable
{
    // Fields
    protected T[] _monitoredArray;
    protected Func<T, P> _project;

    // Construction
    public MonitoredProjectedArray(T[] a, int period, Func<T, P> project)
        : base(a.AsParallel().AsOrdered().Select(project))
    {
        _monitoredArray = a;
        _project = project;

        /* Code that sets up changed items */

        /* Code that sets up a timer which runs every "period"
           milliseconds and calls method QueueChangedItems() */
    }

    public MonitoredProjectedArray(T[] a, Func<T, P> project)
        : this(a, 500, project)
    {
    }

    /* More helper methods */
}

It is important to note that class MonitoredProjectedArray keeps a reference to the array that is monitored. Secondly, note that the underlying array is an array of items of type T, however it inherits ObservableColletion<P>. This happens because every item in the array is projected to type P using the function _project. This happens during the instance creation (note the base call).

The core of the class is method QueueChangedItems(), which is called periodically by a timer. This method first compares the elements of the underlying array with the elements of the monitored array instance. If it finds differences, it creates ChangedItem instances and stores them in a queue. Then, for each item in the queue, it dequeues it and updates the monitored array at the corresponding indices.

A ChangedItem is defined as follows:

struct ChangedItem
{
    public int Index;     // The index in the array of the changed item
    public T NewValue;    // The new value at the specified index
}

To take advantage of the Framework's new parallel processing capabilities, we make use of the Parallel class in our QueueChangedItems() method:

ConcurrentQueue<ChangedItem> _changedItems;

protected void QueueChangedItems()
{
    // Collect the changed items
    Parallel.For(0, _monitoredArray.Count(), i =>
    {
        if (!_project(_monitoredArray[i]).Equals(this[i]))
        {
            var ci = new ChangedItem() { Index = i, NewValue = _monitoredArray[i] };
            if (!_changedItems.Contains(ci))
                _changedItems.Enqueue(ci);
        }
    });

    // The following action updates this instance with the changed items.
    // Use _project to project the new values to type P.
    Action updateAction = () =>
    {
        ChangedItem item;
        while (_changedItems.TryDequeue(out item))
            this[item.Index] = _project(item.NewValue);
    };

    // Start four concurrent updateActions to consume the _changedItems queue
    Parallel.Invoke(updateAction, updateAction, updateAction, updateAction);
}

Note that instead of using a simple Queue, we elect to use a ConcurrentQueue object to represent our queue of changed items. ConcurrentQueue resides in the System.Collections.Concurrent namespace and is thread-safe, which is a requirement in our case since we consume the queue from four different threads.

As a final point of interest, note the fact that in our class we have overridden the OnNotifyCollectionChanged() method in such a way as to make use of the Dispatcher. This was done simply because we modify the collection from a different thread (actually from four different threads, the ones started with the Parallel.Invoke() method). The Dispatcher is the only thread-safe way to change an ObservableCollection from a different thread:

// Override OnCollectionChanged so that we make use of the Dispatcher
protected override void OnCollectionChanged
	(System.Collections.Specialized.NotifyCollectionChangedEventArgs e)
{
    using (BlockReentrancy())
    {
        // Get the CollectionChanged event handler
        System.Collections.Specialized.NotifyCollectionChangedEventHandler 
				eventHandler = CollectionChanged;
        if (eventHandler != null)
        {
            foreach (var handler in eventHandler.GetInvocationList())
            {
                DispatcherObject dispatcherObject = handler.Target as DispatcherObject;
                if (dispatcherObject != null && dispatcherObject.CheckAccess() == false)
                    dispatcherObject.Dispatcher.Invoke
			(DispatcherPriority.DataBind, handler, this, e);
                else
                    (handler as System.Collections.Specialized.
		NotifyCollectionChangedEventHandler)(this, e);
            }
        }
    }
}

We acquire all the delegates attached to the CollectionChanged event, and if any of them is on a different thread, we use the Dispatcher to invoke it. Otherwise, we proceed as normal.

The Final Word

The MonitoredProjectedArray class, along with its extension methods, is definitely not production-quality code. Some of its shortcomings are:

• It only monitors arrays, not general collections (i.e. implementors of ICollection<T>, IList, etc);
• Users cannot configure the monitoring strategy: it's hardcoded into the QueueChangedItems() method;
• It makes the assumption that not many array elements change between two consecutive checkpoints, which is a sensible assumption for most real-world applications, however in the general case, we should not take such things for granted.

Despite its shortcomings however, the class is quite usable, so I hope it will present itself as a solution in similar cases in your programs. Have fun!

History

• February 28, 2011: First version published
• March 1, 2011: A few minor changes in the presentation