A Generic Collection Class with a Failsafe and Bi-directional Enumerator

• Download source code - 24.46 KB

Introduction

The Pool class can be used as a stack, or a queue, or simply as a list of type T items.

But it is more than a simple collection. It's more like a gearbox or transmission for multithreading.

Pool class features:

• Threadsafe operation managed completely within and by the Pool class, requiring no external code cooperation.
• Failsafe and bi-directional enumerator (first to last or last to first).
• Enumerator RemoveCurrent and ReplaceCurrent operations on underlying pool.
• Pool search from either end (for replace/remove/contains).
• Recycled nodes to minimize garbage collection (GC) issues.
• Stepped closure, to prevent additions while shutting down the pool.
• Thread wait for first-item-added, or last-item-removed.
• Permits duplicate items, including default(T).

Background

Pool.cs was the first class I built in C# 1.0. It was to be, and still is, a key player in managing objects in their multithreaded lifetime, especially threads (ie: thread pools) such as service, display, TCP, and named pipe message consumers and producers. When C# 2.0 generics arrived, I upgraded it to Pool<T>. And, of course, I enhanced and beautified it for this article.

Pool nodes are doubly linked and recyclable. The application can, at any time, change the size of the node cache in an effort to minimize GC issues. Adding to or removing from the ends of the pool is an <nobr>O(1+) operation. The "+" part only occurs when a node is created or destroyed. Count is an <nobr>O(1) operation. Unlike the LinkedList<T>, Pool nodes are not exposed, and so the Pool is not a linked list.

The Modules Under Discussion (in the Download)

• Pool.cs - the Pool and Pool.Enumerator classes.
• App*.cs - the partial class modules to test and demonstrate Pool.cs
• Runner.cs - a rig to evaluate/develop/investigate code segments.

Points of Interest

The Pool.Enumerator is more like a roaming cursor. It has an IEnumerator interface, and indeed, you can use the forward enumerator in a foreach statement. But, unlike the usual enumerator, it does not throw exceptions when the underlying pool is changed. This allows administrator threads to move smoothly through the pool looking for items to administer. And it permits direct removal (or replacement) of the current item from the underlying pool.

The WaitForAnyOrClosing() and WaitForEmpty() methods (see below) allow threads to pause while waiting for items to be added or removed by other threads.

The App* modules illustrate most of the features of the Pool and Pool.Enumerator. And if you are into extreme multithreading ... on my machine (Vista32, VS2005, 2GIG, 2Core, 2GHZ), I could run up to 1,376 players (threads) before I ran out of memory. For those of you who have a 4 or 8 core machine, I'd be very interested in your results.

And, no, we don't recommend using any more application threads than are needed. But approaching the limits of the machine does provide an interesting study of how the OS handles massive thread scheduling.

Pool Class Members

new Pool<T>(int theRecycledLimit)
- returns a pool instance of type T.

new Pool<T>(IEnumerable<T> items, bool firstToLastSequence)
- returns a pool instance of type T filled with items in the sequence specified.

int RecycledCount - current count of recycled nodes
int RecycledLimit - current limit of recycled nodes
void SetRecycledLimit(int theRecycledLimit)
- Sets RecycledLimit (negative number sets int.MaxValue).
- Excess nodes are released immediately.
- Nodes are created only when they are needed.

bool AddedLast(T theItem) <nobr>or bool AddedFirst(T theItem)
- Returns true if theItem was added as the last or first item.
- Returns false if the pool is closed or unable to allocate a new node.

bool AddedEach(IEnumerable<T> theItems, bool eachToLast_elseFirst)
- Returns true if each and every item was added as specified.
- Returns false if the pool is closed, or unable to allocate a new node.
note: may throw exception if theItems are modified.

bool RemovedFirst(out T returnItem) <nobr>or bool RemovedLast(out T returnItem)
- Returns true if first or last item was removed to returnItem.
- Returns false and returnItem=default(T) if the pool is empty

bool Removed (T matchingItem, bool searchFirstToLast)
bool Replaced (T matchingItem, bool searchFirstToLast, T replaceWithItem)
- Returns true if matchingItem found and node was removed or item was replaced.
- Returns false if the item was not found.

int Count - Count of active nodes in the list.
void Clear() - Removes all items without closing.
void Clear(out T[] returnArray, bool firstToLast) - Clear to array
void Close() - Prevents items from being added.
void Dispose() - Close(), Clear() and dispose.

T[] ToArray(bool firstToLast)
- Returns T[] snapshot of pool items in the order specified.

void WaitForEmpty(int theMillisecondsToWait)
- If the pool is NOT EMPTY, blocks the calling thread
- until the pool is empty,
- or theMillisecondsToWait elapses.

void WaitForAnyOrClosing(int theMillisecondsToWait)
- If the pool IS EMPTY, blocks the calling thread
- until the first item is added,
- or the pool is closed,
- or theMillisecondsToWait elapses.

delegate void FirstItemAddedDo(Pool<T> thePool)
event FirstItemAddedDo OnFirstItemAdded
- Event triggered when first item is added to an empty pool.

Pool<T>.Enumerator GetEnumerator() - first to last sequence
Pool<T>.Enumerator GetEnumerator(bool firstToLastSequence)
- Gets enumerator for the pool, in sequence specified;

Pool.Enumerator Class Members

bool IsReset - True upon instantiation
void Reset() - Keeps same move sequence
void Reset(bool firstToLastSequence) - Sets move sequence
- Reset sets Current = default(T) and deactivates the enumerator.

NOTE: a reset enumerator (IsReset == true) has no active attachments.
There is NO significant penalty for retaining an enumerator for subsequent use.
Active attachment to the underlying pool occurs upon the first MoveNext().

bool IsFirstToLast - which way are we moving
bool MoveNext()
- Moves through the pool in the established direction.
- Returns true if a Current item has been found.
- If no more items, does a Reset(), and returns false.

T Current - the current item
T CurrentAndDispose() - return Current and Dispose().
T CurrentAndReset() - return Current and Reset().
void Dispose() - Reset() and dispose the enumerator.

bool RemovedCurrent()
bool ReplacedCurrent(T replaceWithItem)
- Returns true if the current node was removed or item was replaced.
- Returns false if the current node has already been removed.
- note: this is an O(1) operation (ie: immediate).

Using the Pool Class

Here are some methods that illustrate a pool's either-directional relationship to other lists. Note: any number of threads could safely use any of these methods on a given pool at any time.

static void copy_list_to_pool(List<Foo> theList, Pool<Foo> thePool
                , bool eachToLast_elseFirst)
{
    // as individual operations for each item
    foreach (Foo x in theList)
        if ( !  ( eachToLast_elseFirst
                ? thePool.AddedLast(x)
                : thePool.AddedFirst(x)
                )
            ) throw thePool.IsClosed
                ? (Exception) new ObjectDisposedException("Pool")
                : (Exception) new OutOfMemoryException();
    
    // OR as a single operation on thePool
    if (!thePool.AddedEach(theList, eachToLast_elseFirst))
        throw thePool.IsClosed
            ? (Exception)new ObjectDisposedException("Pool")
            : (Exception)new OutOfMemoryException();
}

static void copy_pool_to_list(Pool<Foo> thePool, List<Foo> theList
               , bool firstToLastSequence)
{
    // as individual operations for each item
    if (firstToLastSequence)
        theList.AddRange(thePool);
    else
    {
        Pool<Foo>.Enumerator x = thePool.GetEnumerator(false);
        while (x.MoveNext())
            theList.Add(x.Current);
    }

    // OR as a single operation on (and snapshot of) thePool
    theList.AddRange(thePool.ToArray(firstToLastSequence));
}


static void empty_pool_to_list(Pool<Foo> thePool, List<Foo> theList
                , bool firstToLastSequence)
{      Foo x;
    // as individual operations for each item
    if (firstToLastSequence) 
            while (thePool.RemovedFirst(out x)) 
                theList.Add(x);
    else    while (thePool.RemovedLast(out x)) 
                theList.Add(x);

    // OR as a single operation on thePool
    Foo[] xSnapshot;
    thePool.Clear(out xSnapshot, firstToLastSequence);
    theList.AddRange(xSnapshot);
}

static Foo get_Foo33_or_eeek(Pool<Foo> thePool, bool searchFirstToLast)
{
    Pool<Foo>.Enumerator x = thePool.GetEnumerator(searchFirstToLast);

    while (x.MoveNext())
        if (x.Current.Code == 3 && x.Current.Contents == "3")
            return x.CurrentAndDispose();

    x.Dispose();
    throw new ApplicationException("Foo33 not found ... eeek!");
}

Example: Pool Message Consumer

The following code will process messages (if any), do some other assigned task (idle time processing), and if the pool is empty, will pause for 5 seconds, or whenever the first item is added, or the pool is closed. It repeats until the pool is closed, and then does one more final message processing.

struct Msg { internal int Code; internal string Contents; };
static void consumeMsg(Msg theMsg) { /* whatever */ }
static void administrateSomething() { /* whatever */ }

static void consumeMsgs(Pool<Msg> thePool)
{   Msg vMsg;
    while (!thePool.IsClosed)
    {
        while (thePool.RemovedFirst(out vMsg))
            consumeMsg(vMsg);

        administrateSomething();

        thePool.WaitForAnyOrClosing(5000);
    }   
    while (thePool.RemovedFirst(out vMsg))
        consumeMsg(vMsg); // consume any messages left
}

Example: Pool Used To Manage Worker Threads

Let's assume you have running worker threads that were added to a Pool of theWorkers. Given the following worker methods ...

class worker : IDisposable
{
    Pool<worker> myPool; Thread myThread;
    bool pleaseShutdown = false;

    internal void PleaseShutdownASAP()
    {   pleaseShutdown = true;
    }
    private void myThread_StartLogic()
    {   while (!pleaseShutdown)
            {/* do my duty, again and again */}
        Dispose();
    }
    public void Dispose()
    {   myPool.Removed(this, true);
        // then dispose stuff, THEN:
        if (Thread.CurrentThread != myThread)
            myThread.Abort();
}   }

... the following manager thread code will attempt to shutdown those workers politely, and after 10 seconds, will shut them down in a rude way. Note: If the workers clear out in 1 second, the manager thread will start right up.

static void closing_time_for_workers(Pool<worker> theWorkers)
{
    theWorkers.Close(); // prevents new workers from showing up

    Pool<worker>.Enumerator 
        x = theWorkers.GetEnumerator();

    while (x.MoveNext()) // enumerator resets upon last MoveNext
        x.Current.PleaseShutdownASAP(); // polite request
    
    theWorkers.WaitForEmpty(10000); // 10 seconds to rudeness
         
    while (x.MoveNext())        // reusing the reset enumerator
        x.Current.Dispose();    // rude awakening, then silence
}

The rest of this article is an overview of the App*.cs modules.

BadgeGrabbers (App*.cs)

So you have opened BadgeGrabbers.exe.sln and you are looking in the Solution Explorer:

• App.~Main.cs - static Main() method, with some trivial App supporting members.
• App.+PlayTheGame.cs - App PlayTheGame() method, NOT-trivial, AND discussed below.
• App.+reporting.cs - App reporting methods, NOT trivial, but NOT discussed.
• App.Badge.cs - trivial class representing something to grab.
• App.Player.cs - NON-trivial class AND discussed below.
• App.Team.cs - trivial class, good hook if you want to add a Coach thread.
• Pool.cs - the actual code of what this article is about.
• PoolExamples.cs - examples, nothing else.
• Runner.cs - briefly mentioned below.

Runner.cs is a rig I use to evaluate, develop, and investigate code segments. Think of it as a console app in a text box. It runs the specified code segment in a separate thread, surrounded by a try statement so the code can fail without affecting runner. The only "application command" is Ctrl-S to toggle runner states READY, RUNNING, STOPPING, FINISHED. The primary methods used by the "code segment" are <nobr>Put() and <nobr>PutLine() ... which append a string to the text box.

The original Pool tester was fairly simplistic. It verified that pool methods were using the lock{} statement appropiately, and that the secret event would do the right thing in the enumerator. Otherwise, it was an unnecessary test of the C# lock{} statement;

For this article, I wanted some code to demonstrate most of the Pool class features, especially the Pool.Enumerator. After an hour of scaling up my old Pool tester, I decided I needed a metaphor to help me keep track of and express my thoughts ... hence, BadgeGrabbers ... an app whose purpose is to thread thrash pools. It is not a game to play ... it's a game to play with.

ELEMENTS:

• Game - Contains Pool<Team> TeamPool.
• Badge - the thing to be grabbed in order to win.
• Team - Contains a Pool<Player> PlayerPool, and Pool<Badge> BadgesClaimed.
• Player - An active thread, enumerating through other team's BadgesClaimed,
  using an assigned strategy to attempt to grab a badge.

App.PlayTheGame() (The Game Thread)

This method runs on it's own thread assigned by runner. It is the game thread, or the playing field thread, and does the following:

• Retrives the game parameters from the runner entry list.
• Loads the teams and their players onto the field.
• Hands out badges to each team.
• Fires the starting gun.
• Monitors the players and teams as they play.
• Manages catastrophic failure, and/or the end of the game.
• Verifies team claims and that all badges are accounted for.
• Reports results, and team standings.
• Clears the field for the next game.

The game thread runs at ThreadPriority.Highest. If it does not, it never gets a chance to do any reporting. It is the only thread that actively outputs to runner, in order to isolate runner from the game. The game and runner threads will affect the game, but mostly, the players fight with each other on their own thread bashing field.

App.Player class (The Game Action)

The void PlayingLoop() is where all of the game action occurs. Here is a simplified view of that loop:

void PlayingLoop()
{   WaitForStartingGun();
    try
    {   while (WeArePlayingTheGame)
        {   while (... got a "grabbable badge" ...)
            {   AttemptGrab(... of that badge ...);
    }   }   }   
    catch (Exception e){ ...will cancel the game ... }
}

A "grabbable badge" is one found while enumerating through another team's BadgesClaimed pool. It can only be "grabbable" if the other team's BadgesClaimed.Count is greater or equal to the player's BadgesClaimed.Count, adjusted by a "HandicapCount". The bigger the "HandicapCount", the harder it is for a leading team to find a "grabbable badge".

The void AttemptGrab(..) method attempts to remove the badge and add it to the player's team badges. It records success or failure. Upon a successful grab, a flag is set that will cause the player to sleep(1) just before trying to grab the next "grabbable badge" it finds.

The void DetectProcessor() method detects a players current processor and if it is different that the last processor detected for that player (thread), records it for the end of game totals. DetectProcessor() calls are peppered throughout the code.

Players run at ThreadPriority.BelowNormal to prevent them from swallowing the CPUS ... but for giant games, they will do it anyway.

Player methods (perhaps positions):

• When scanning for other teams, scan TeamPool first to last, else reverse.
• When scanning for a badge to grab, scan BadgesClaimed first to last, else reverse.
• When attempting a grab (remove), remove searching first to last, else reverse.
• For a successful grab, AddedLast() badge to team's BadgesClaimed else AddedFirst().

Methods are permutated onto the players as they are loaded into the team. If there are 16 players per team, all method permutations will be present.

Conclusion

The Pool class was designed to be one of those "go to the freezer and get the box" tools, not sophisticated, but solid in what it does. It does a lock{} upon every operation (including MoveNext), which may be an issue in a few applications. Otherwise, it is a very simple and effective tool for coordinating multiple threads.

Please notify me if you find any typos or misplaced logic. I will pass on the fixes.

History

• 2009-01-12 - first published