A thread-safe implementation of a generic container of type pair using shared_ptrs

Question

I have created a generic message queue for use in a multi-threaded application. Specifically, single producer, multi-consumer. Main code below.

1) I wanted to know if I should pass a shared_ptr allocated with new into the enqueue method by value, or is it better to have the queue wrapper allocate the memory itself and just pass in a genericMsg object by const reference?

2) Should I have my dequeue method return a shared_ptr, have a shared_ptr passed in as a parameter by reference (current strategy), or just have it directly return a genericMsg object?

3) Will I need signal/wait in enqueue/dequeue or will the read/write locks suffice?

4) Do I even need to use shared_ptrs? Or will this depend solely on the implementation I use? I like that the shared_ptrs will free memory once all references are no longer using the object. I can easily port this to regular pointers if that's recommended, though.

5) I'm storing a pair here because I'd like to discriminate what type of message I'm dealing with else w/o having to do an any_cast. Every message type has a unique ID that refers to a specific struct. Is there a better way of doing this?

Generic Message Type:

template<typename Message_T>
class genericMsg
{ 
  public:
    genericMsg()
    {
       id = 0;
       size = 0;
    }

    genericMsg (unsigned int &_id, unsigned int &_size, Message_T &_data)
    {
       id = _id;
       size = _size;
       data = _data;
    }

    ~genericMsg()
    {}

    unisgned int   id;
    unsigned int   size;
    Message_T      data; //All structs stored here contain only POD types
};

Enqueue Methods:

  // ----------------------------------------------------------------
  // -- Thread safe function that adds a new genericMsg object to the
  // -- back of the Queue.
  // -----------------------------------------------------------------
  template<class Message_T>
  inline void enqueue(boost::shared_ptr< genericMsg<Message_T> > data)
  {
     WriteLock w_lock(myLock);
     this->qData.push_back(std::make_pair(data->id, data));
  }

VS:

  // ----------------------------------------------------------------
  // -- Thread safe function that adds a new genericMsg object to the
  // -- back of the Queue.
  // -----------------------------------------------------------------
  template<class Message_T>
  inline void enqueue(const genericMsg<Message_T> &data_in)
  {
     WriteLock w_lock(myLock);
     boost::shared_ptr< genericMsg<Message_T> > data = 
          new genericMsg<Message_T>(data_in.id, data_in.size, data_in.data);
     this->qData.push_back(std::make_pair(data_in.id, data));
  }

Dequeue Method:

  // ----------------------------------------------------------------
  // -- Thread safe function that grabs a genericMsg object from the
  // -- front of the Queue.
  // -----------------------------------------------------------------
  template<class Message_T>
  void dequeue(boost::shared_ptr< genericMsg<Message_T> > &msg)
  {
     ReadLock r_lock(myLock);
     msg = boost::any_cast< boost::shared_ptr< genericMsg<Message_T> > >(qData.front().second);
     qData.pop_front();
  }

Get message ID:

  inline unsigned int getMessageID()
  {
     ReadLock r_lock(myLock);
     unsigned int tempID = qData.front().first;
     return tempID;
  }

Data Types:

 std::deque < std::pair< unsigned int, boost::any> > qData;

Edit:

I have improved upon my design. I now have a genericMessage base class that I directly subclass from in order to derive the unique messages.

Generic Message Base Class:

class genericMessage
{
   public:
      virtual ~genericMessage() {}
      unsigned int getID() {return id;}
      unsigned int getSize() {return size;}

   protected:
      unsigned int id;
      unsigned int size;   
};

Producer Snippet:

 boost::shared_ptr<genericMessage> tmp (new derived_msg1(MSG1_ID));
 theQueue.enqueue(tmp);

Consumer Snippet:

 boost::shared_ptr<genericMessage> tmp = theQueue.dequeue();         
 if(tmp->getID() == MSG1_ID)
 {
    boost::shared_ptr<derived_msg1> tObj = boost::dynamic_pointer_cast<derived_msg1>(tmp);
    tObj->printData();
 }

New Queue:

  std::deque< boost::shared_ptr<genericMessage> > qData;

New Enqueue:

void mq_class::enqueue(const boost::shared_ptr<genericMessage> &data_in)
{
   boost::unique_lock<boost::mutex> lock(mut);
   this->qData.push_back(data_in);
   cond.notify_one();
}

New Dequeue:

boost::shared_ptr<genericMessage> mq_class::dequeue()
{
   boost::shared_ptr<genericMessage> ptr;
   {
      boost::unique_lock<boost::mutex> lock(mut);
      while(qData.empty())
      {
         cond.wait(lock);
      }
      ptr = qData.front();
      qData.pop_front();
   }
   return ptr;
}

Now, my question is am I doing dequeue correctly? Is there another way of doing it? Should I pass in a shared_ptr as a reference in this case to achieve what I want?

Answer 1

Edit (I added answers for parts 1, 2, and 4).

1) You should have a factory method that creates new genericMsgs and returns a std::unique_ptr. There is absolutely no good reason to pass genericMsg in by const reference and then have the queue wrap it in a smart pointer: Once you've passed by reference you have lost track of ownership, so if you do that the queue is going to have to construct (by copy) the entire genericMsg to wrap.

2) I can't think of any circumstance under which it would be safe to take a reference to a shared_ptr or unique_ptr or auto_ptr. shared_ptrs and unique_ptrs are for tracking ownership and once you've taken a reference to them (or the address of them) you have no idea how many references or pointers are still out there expecting the shared_ptr/unique_ptr object to contain a valid naked pointer.

unique_ptr is always preferred to a naked pointer, and is preferred to a shared_ptr in cases where you only have a single piece of code (validly) pointing to an object at a time.

• https://softwareengineering.stackexchange.com/questions/133302/stdshared-ptr-as-a-last-resort
• http://herbsutter.com/gotw/_103/
• Bad practice to return unique_ptr for raw pointer like ownership semantics? (the answer explains why it is good practice not bad).

3) Yes, you need to use a std::condition_variable in your dequeue function. You need to test whether qData is empty or not before calling qData.front() or qData.pop_front(). If qData is empty you need to wait on a condition variable. When enqueue inserts an item it should signal the condition variable to wake up anyone who may have been waiting.

Your use of reader/writer locks is completely incorrect. Don't use reader/writer locks. Use std::mutex. A reader lock can only be used on a method that is completely const. You are modifying qData in dequeue, so a reader lock will lead to data races there. (Reader writer locks are only applicable when you have stupid code that is both const and holds locks for extended period of time. You are only keeping the lock for the period of time it takes to insert or remove from the queue, so even if you were const the added overhead of reader/writer locks would be a net lose.)

An example of implementing a (bounded) buffer using mutexes and condition_variables can be found at: Is this a correct way to implement a bounded buffer in C++.

4) unique_ptr is always preferred to naked pointers, and usually preferred to shared_ptr. (The main exception where shared_ptr might be better is for graph-like data structures.) In cases like yours where you are reading something in on side, creating a new object with a factory, moving the ownership to the queue and then moving ownership out of the queue to the consumer it sounds like you should be using unique_ptr.

5) You are reinventing tagged unions. Virtual functions were added to c++ specifically so you wouldn't need to do this. You should subclass your messages from a class that has a virtual function called do_it() (or better yet, operator()() or something like that). Then instead of tagging each struct, make each struct a subclass of your message class. When you dequeue each struct (or ptr to struct) just call do_it() on it. Strong static typing, no casts. See C++ std condition variable covering a lot of share variables for an example.

Also: if you are going to stick with the tagged unions: you can't have separate calls to get the id and the data item. Consider: If thread A calls to get the id, then thread B calls to get the id, then thread B retrieves the data item, now what happens when thread A calls to retrieve a data item? It gets a data item, but not with the type that it expected. You need to retrieve the id and the data item under the same critical section.

Answer 2

First of all, it's better to use 3rd-party concurrency containers than to implement them yourself, except it's for education purpose.

Your messages doesn't look to have costly constructors/destructor, so you can store them by value and forget about all your other questions. Use move semantics (if available) for optimizations.

If your profiler says "by value" is bad idea in your particular case:

I suppose your producer creates messages, puts them into your queue and loses any interest in them. In this case you don't need shared_ptr because you don't have shared ownership. You can use unique_ptr or even a raw pointer. It's implementation details and better to hide them inside the queue.

From performance point of view, it's better to implement lock-free queue. "locks vs. signals" depends completely on your application. For example, if you use thread pool and kind of a scheduler it's better to allow your clients to do something useful while queue is full/empty. In simpler cases reader/writer lock is just fine.

Answer 3

If I want to be thread safe, I usually use const objects and modify only on copy or create constructor. In this way you don't need to use any lock mechanism. In a threaded system, it is usually more effective than use mutexes on a single instance.

In your case only deque would need lock.