Robust C++: Queue Templates

• Download source - 10.5 KB

Introduction

If you're using the STL and need a queue, you'll probably opt for std::deque or std::list. For some systems, however, using these templates can be problematic:

1. Their memory overhead can be significant in a system with many short queues.
2. They allocate and manage memory, which can conflict with the need to provide predictable response times in a hard real-time system.
3. They hide the queue data structure, which makes it difficult for a robust system to recover from a queue corruption caused by an unguarded critical region or trampled memory.

This article introduces templates that can be used when these issues make it undesirable to use an STL template.

Background

The templates in this article come from the Robust Services Core (RSC), an open-source framework for developing robust C++ applications. The templates are

• Q1Way, for one-way queues, and
• Q2Way, for two-way queues.

Because their implementations are similar, this article only describes Q1Way, which is used more frequently. Q2Way is only used for performance reasons, when elements must often be removed from the middle of a relatively long queue. Q2Way also supports backwards iteration, but applications seldom need that capability.

Using the Code

The code has been edited to remove things that are not central to the templates. These things appear in the full versions of the code in the attached .zip file, so you can remove them if you're not using RSC.

The crux of the design is that a queueable object defines a Q1Link member which provides the link to the next object in the queue. Because of this, the template does not need to allocate memory for elements on the queue.

Here is the code for Q1Link, except for its destructor.

//  Link for an item on a one-way queue.  An object that resides on a one-way
//  queue includes this as a member and implements a LinkDiff function that
//  returns the distance between the top of the object and its Q1Link member.
//
class Q1Link
{
   template<class T> friend class Q1Way;
public:
   //  Public because an instance of this class is included in objects that
   //  can be queued.
   //
   Q1Link() : next(nullptr) { }

   //  Public because an instance of this class is included in objects that
   //  can be queued.
   //
   ~Q1Link();

   //  Deleted to prohibit copying.
   //
   Q1Link(const Q1Link& that) = delete;
   Q1Link& operator=(const Q1Link& that) = delete;

   //  Returns true if the item is on a queue.
   //
   bool IsQueued() const { return next != nullptr; }

   //  Returns a string for displaying the link.
   //
   std::string to_str() const
   {
      std::ostringstream stream;
      stream << next;
      return stream.str();
   }
private:
   //  The next item in the queue.  Because Q1Way uses circular queues, a
   //  value of nullptr means that the item is not on a queue.
   //
   Q1Link* next;
};

An object should be removed from a queue before it is deleted. But if it's still queued, the Q1Link destructor removes it to prevent a broken link from corrupting the queue and orphaning the objects that follow it:

Q1Link::~Q1Link()
{
   //  If the item is still queued, exqueue it.  This is a serious problem if
   //  it is the tail item, because it will leave the queue head pointing to
   //  a deleted item.
   //
   if(next == nullptr) return;

   auto prev = this;
   auto curr = next;
   auto i = INT16_MAX;

   while((i >= 0) && (curr != nullptr))
   {
      if(curr == this)
      {
         auto before = INT16_MAX - i;

         if(before == 0)
         {
            Debug::SwLog("tail item was still queued", before);
            prev->next = nullptr;
         }
         else
         {
            Debug::SwLog("item was still queued", before);
            prev->next = curr->next;
         }
         return;
      }

      --i;
      prev = curr;
      curr = curr->next;
   }

   Debug::SwLog("item is still queued", INT16_MAX - i);
}

Q1Link prohibited copying, which can corrupt a queue by duplicating a link. Because a queueable object defines a Q1Link member, this also makes the object uncopyable. If it must support copying, Q1Link's copy constructor and copy operator could be implemented by invoking its destructor (to make sure the item isn't queued) and then initializing next to nullptr so that the item's new incarnation remains unqueued.

Now we can start to look at the Q1Way template itself. Note that it points to the queue's tail element, which makes both enqueue and dequeue operations efficient. The queued objects actually form a circular queue.

//  One-way queue.  Recommended unless items are often exqueued, which
//  can be expensive.
//  o no items: tail_.next = nullptr
//  o one item: tail_.next = item, item.next = item (points to itself)
//  o two or more items: tail_.next = last item, last item.next = first
//    item, second last item.next = last item (circular queue)
//
template<class T> class Q1Way
{
public:
   //  Initializes the queue header to default values.  Before the queue
   //  can be used, Init must be invoked.
   //
   Q1Way() : tail_(nullptr), diff_(NilDiff) { }

   //  Cleans up the queue.
   //
   ~Q1Way()
   {
      if(tail_.next == nullptr) return;
      Purge();
   }

   //  Deleted to prohibit copying.
   //
   Q1Way(const Q1Way& that) = delete;
   Q1Way& operator=(const Q1Way& that) = delete;

   //  Initializes the queue so that it can be used.
   //
   void Init(ptrdiff_t diff)
   {
      tail_.next = nullptr;  // queue is empty
      diff_ = diff;          // distance from each item's vptr to its Q1Link
   }
private:
   //  For initializing diff_.
   //
   static const ptrdiff_t NilDiff = -1;

   //  The queue head, which actually points to the tail item.
   //  If the queue is empty, tail_.next is nullptr.
   //
   Q1Link tail_;

   //  The distance from an item's vptr to its Q1Link.
   //
   ptrdiff_t diff_;
};

In order to manipulate the queue links, Q1Way needs to know where each object's Q1Link is located. This is the purpose of diff_, which is the distance (in bytes) from the beginning of the object to its Q1Link member. All the objects in a queue must use the same offset, so a common base class is needed if different classes will have objects in the same queue.

The value of diff_ is returned by a grotty function that RSC classes always name LinkDiff. For example, RSC's Thread class defines a msgq_ member where messages for a thread are queued. These messages derive from MsgBuffer, which therefore needs a Q1Link member and a LinkDiff function that allows Thread to initialize msgq_. However, MsgBuffer derives from Pooled, which already provides a Q1Link member and LinkDiff function for all pooled objects. The outline (omitting other members) therefore looks like this:

class Pooled : public Object
{
public:
   static ptrdiff_t LinkDiff() const;
private:
   Q1Link link_;
};

ptrdiff_t Pooled ::LinkDiff()
{
   uintptr_t local;
   auto fake = reinterpret_cast<const Pooled*>(&local);
   return ptrdiff(&fake->link_, fake);
}

class Thread : public Permanent
{
private:
   Q1Way<MsgBuffer> msgq_;
};

msgq_.Init(Pooled::LinkDiff());  // in Thread's constructor

Once Init has been invoked, the Q1Way function Item can find the location of each element's Q1Link. This function only needs to be used internally and is therefore private:

Q1Link* Item(const T& elem) const
{
   if(diff_ == NilDiff) return nullptr;
   if(&elem == nullptr) return nullptr;
   return (Q1Link*) getptr2(&elem, diff_);  // adds diff_ bytes to &elem
}

An item is usually added to the back of the queue:

bool Enq(T& elem)
{
   auto item = Item(elem);
   if(item == nullptr) return false;        // error
   if(item->next != nullptr) return false;  // if item is queued, do nothing
   if(tail_.next != nullptr)                // if queue isn't empty
   {                                        // then
      item->next = tail_.next->next;        // item points to first element
      tail_.next->next = item;              // last element points to item
   }
   else
   {
      item->next = item;                    // else item points to itself
   }
   tail_.next = item;                       // tail points to item
   return true;
}

An item can also be added to the front of the queue:

bool Henq(T& elem)
{
   auto item = Item(elem);
   if(item == nullptr) return false;        // error
   if(item->next != nullptr) return false;  // if item is queued, do nothing
   if(tail_.next != nullptr)                // if queue isn't empty
   {
      item->next = tail_.next->next;        // item points to first element
      tail_.next->next = item;              // last element points to item
   }                                        // tail isn't changed, so item is
   else                                     // after last (and is thus first)
   {
      item->next = item;                    // item points to itself
      tail_.next = item;                    // tail points to item
   }
   return true;
}

An item can even be added to the middle of the queue:

bool Insert(T* prev, T& elem)
{
   if(prev == nullptr)                        // if nothing is previous
      return Henq(elem);                      // then put item at head
   auto item = Item(elem);
   if(item == nullptr) return false;          // error
   auto ante = (Q1Link*)
      getptr2(prev, diff_);                   // put item after previous
   if(item->next != nullptr) return false;    // item must not be queued
   if(ante->next == nullptr) return false;    // prev must be queued
   item->next = ante->next;
   ante->next = item;
   if(tail_.next == ante) tail_.next = item;  // update tail if item is last
   return true;
}

An item is usually removed from the front of the queue:

T* Deq()
{
   if(tail_.next == nullptr)          // if tail points to nothing
      return nullptr;                 // then queue is empty
   Q1Link* item = tail_.next->next;   // set item to first element
   if(tail_.next != item)             // if item wasn't alone
      tail_.next->next = item->next;  // then last now points to second
   else
      tail_.next = nullptr;           // else queue is now empty
   item->next = nullptr;              // item has been removed
   return (T*) getptr1(item, diff_);  // location of item's vptr (subtracts
                                      //   diff_ bytes from item)
}

If necessary, an item can be removed from anywhere on the queue:

bool Exq(T& elem)
{
   auto item = Item(elem);
   if(item == nullptr) return false;  // error
   if(item->next == nullptr)          // if the item isn't queued
      return true;                    // then return immediately
   if(tail_.next == nullptr)          // if the queue is empty
      return false;                   // then the item can't be there
   if(item->next == item)             // if the item points to itself
   {
      if(tail_.next == item)          // and if the item is also last
      {
         item->next = nullptr;        // then remove the item
         tail_.next = nullptr;        // and the queue is now empty
         return true;
      }
      return false;                   // the item is on another queue
   }
   auto curr = tail_.next;            // starting at the last element,
   while(curr->next != item)          // advance until the item is
   {
      curr = curr->next;              // the next element--but
      if(curr == tail_.next)          // stop after searching the
         return false;                // entire queue
   }
   curr->next = item->next;           // curr's next becomes item's next
   if(tail_.next == item)             // if the item was the tail
      tail_.next = curr;              // then the tail has to back up
   item->next = nullptr;              // the item has been removed
   return true;
}

The functions First and Next support queue traversal. They simply use a pointer to the current item rather than a distinct iterator. Traversal usually takes the form

for(auto item = queue_.First(); item != nullptr; queue_.Next(item))…

T* First() const
{
   if(diff_ == NilDiff) return nullptr;  // queue is not initialized
   Q1Link* item = tail_.next;            // set item to first element
   if(item == nullptr) return nullptr;   // queue is empty
   item = item->next;                    // advance to head
   return (T*) getptr1(item, diff_);     // location of item's vptr
}

//  Updates ELEM to the next item in the queue.  If ELEM is nullptr,
//  provides the first item.  Returns true if there was a next item.
//
bool Next(T*& elem) const
{
   if(diff_ == NilDiff) return false;    // queue is not initialized
   Q1Link* item;                         // item will hold result
   if(elem == nullptr)                   // nullptr means
   {
      item = tail_.next;                 // start at last element
      if(item == nullptr) return false;  // return if the queue is empty
   }
   else
   {
      item = (Q1Link*)
         getptr2(elem, diff_);           // start at the current item
      if(tail_.next == item)             // if that is the last element
      {
         elem = nullptr;                 // then there are no more
         return false;
      }
   }
   item = item->next;                    // advance to next element
   if(item == nullptr)                   // make sure the item was queued
   {
      elem = nullptr;
      return false;
   }
   elem = (T*) getptr1(item, diff_);     // location of item's vptr
   return true;
}

//  Returns the item that follows ELEM.
//
T* Next(const T& elem) const
{
   auto item = Item(elem);
   if(item == nullptr) return nullptr;     // error
   if(tail_.next == item) return nullptr;  // return if item was last
   item = item->next;                      // advance to the next element
   if(item == nullptr) return nullptr;     // return if item wasn't queued
   return (T*) getptr1(item, diff_);       // location of next item's vptr
}

There are also the usual functions for finding the queue's length:

bool Empty() const { return (tail_.next == nullptr); }

size_t Size() const
{
   if(diff_ == NilDiff) return 0;  // queue is not initialized
   Q1Link* item = tail_.next;      // start at the last item
   if(item == nullptr) return 0;   // check for an empty queue
   size_t count = 1;               // there is at least one element
   item = item->next;              // advance to the first element
   while(item != tail_.next)       // stop when we reach the tail
   {
      item = item->next;
      ++count;
   }
   return count;                   // report the result
}

And finally, there is the equivalent of clear:

void Purge()
{
   while(tail_.next != nullptr)
   {
      auto item = Deq();
      delete item;
   }
}

If you want to modify the templates, NtIncrement.cpp contains Q1WayCommands and Q2WayCommands, which allow Q1Way and Q2Way functions to be tested via RSC's CLI.

Points of Interest

The following function is provided to test software that tries to recover after a queue corruption:

//  Corrupts ELEM's next pointer for testing purposes.  If ELEM is
//  nullptr, the queue's tail pointer is corrupted instead.
//
void Corrupt(T* elem)
{
   if(elem == nullptr)                     // if no item provided
   {
      tail_.next = (Q1Link*) BAD_POINTER;  // corrupt queue header
      return;
   }
   auto item = (Q1Link*)                   // start at the current item
      getptr2(elem, diff_);
   item->next = (Q1Link*) BAD_POINTER;     // corrupt ELEM's next pointer
}

If the queue of available blocks in an object pool gets corrupted, how the object pool audit repairs it is described here.

History

• 13^th June, 2020: Initial version