idea for wxMailbox Topic is solved

[carlosbecker]

Hi everybody, I've been using wxWidgets for a while now. Some time ago I needeed something like a mailbox (message queue) to send data between threads, so I made this little class (template actually). I thought it could be useful for other users here, since there are many posts on the forum about thread communication.

who knows, maybe someday this template could be part of the wxWidgets release.. with some modifications of course.

here it goes:

wxMailbox.hpp

Code: Select all

#ifndef _wxMailbox_h_
#define _wxMailbox_h_

#include <wx/wx.h>

#include <vector>

/****************************************************

wxMailbox is useful for working with threads.
Elements are passed by copy.

*****************************************************/

template <class T>
class	wxMailbox
{
	public:
		//constructor: max cantidad pendientes, y si debe hacer delete de los que restan al destruirse
		wxMailbox( unsigned int maxCount = 10 );
		wxMailbox::~wxMailbox();
	
		//Inserts an element at the end of the list
		bool	Put( T &element );
		
		//Tries to get an element out of the Mailbox
		bool	TryGet( T &element );

	
	private:
		wxMutex		*m_mutex;	//mutex for Put & Get
		
		std::vector< T >	m_vector;	//maybe not the most efficient allocator
		
		unsigned int	m_maxElements;
};

#include "../../common/wxMailbox.tpp"	//implementations

#endif

wxMailbox.tpp

Code: Select all

#include "wxMailbox.hpp"

template <class T>
wxMailbox<T>::wxMailbox( unsigned int maxCount )
{
	m_maxElements = maxCount;
	
	m_mutex = new wxMutex();
}

template <class T>
bool	wxMailbox<T>::Put( T& element )
{
	//lock mutex
	wxMutexLocker	lock( *m_mutex );

	//max size? => out of here
	if ( m_vector.size() > m_maxElements-1 )
		return false;
	
	//push!
	m_vector.push_back( element );

	return true;
}

template <class T>
bool	wxMailbox<T>::TryGet( T& element )
{
	wxMutexLocker	lock( *m_mutex );
	
	//empty?
	if ( m_vector.empty() )
		return false;
	
	//copy
	element = m_vector[0];
	
	//erase
	m_vector.erase( m_vector.begin() );	

	return true;
}

template <class T>
wxMailbox<T>::~wxMailbox()
{
	//finally..
	delete m_mutex;
}

The .tpp extension is just because that file doesn't have to be compiled, just included, but it's easier to keep apart implementation from definition.

Any ideas welcomed,
thanks again to everyone who helps building wxWidgets!

Carlos

[eranif]

Hi,
Several comments:

1. I am missing here a Get(int timeout) method which blocks and accepts timeout parameter.

2. The current implementation forces my queue objects to have copy constructor & assignment operator, not to mention that for performance, I would keep pointers allocated on the heap, so adding items to the queue will be something like this:

Code: Select all

m_queue.Put(new Item());

The threads who 'Get' the message, is also responsible for deleting it.

3. API to clear all elements in the queue.

Eran

[carlosbecker]

Eran,
thanks for the comments.
Here there is a new version of the code

wxMailbox.hpp

Code: Select all

#ifndef _wxMailbox_h_
#define _wxMailbox_h_

#include <wx/wx.h>

#include <vector>

/****************************************************

wxMailbox is useful for working with threads.
Elements are passed by copy.

*****************************************************/

template <class T>
class	wxMailbox
{
	public:
		//constructor: max queue length
		wxMailbox( unsigned int maxCount = 10 );
		wxMailbox::~wxMailbox();
	
		//Inserts an element at the end of the list
		bool	Put( T &element );
		bool	Put( T *element );	//another way
		
		//Tries to get an element out of the Mailbox
		bool	TryGet( T &element );
		
		//get w/timeout
		bool	Get( T& element, unsigned long millisec );
		
		//empty the queue
		void	Empty();
	
	private:
		wxMutex		*m_mutex;	//mutex for Put & Get
		
		wxMutex		*m_mutexEmpty;	//mutex to signal if it's empty or not
		wxCondition	*m_condEmpty;
		
		std::vector< T >	m_vector;	//maybe not the most efficient allocator
		
		unsigned int	m_maxElements;
};

#include "../../common/wxMailbox.tpp"	//implementations

#endif

wxMailbox.tpp

Code: Select all

#include "wxMailbox.hpp"

template <class T>
wxMailbox<T>::wxMailbox( unsigned int maxCount )
{
	m_maxElements = maxCount;
	
	m_mutex = new wxMutex();
	
	m_mutexEmpty = new wxMutex();
	m_condEmpty = new wxCondition( *m_mutexEmpty );
}

template <class T>
bool	wxMailbox<T>::Put( T& element )
{
	wxMutexLocker	lock( *m_mutex );

	//max size? => out of here
	if ( m_vector.size() > m_maxElements-1 )
		return false;

	//push!
	m_vector.push_back( element );
	
	wxMutexLocker	lockEmpty( *m_mutexEmpty );
	m_condEmpty->Broadcast();	//signal the condition

	return true;
}

template <class T>
bool	wxMailbox<T>::Put( T *element )
{
	return this->Put( *element );
}

template <class T>
bool	wxMailbox<T>::TryGet( T& element )
{
	wxMutexLocker	lock( *m_mutex );
	
	//empty?
	if ( m_vector.empty() )
		return false;
	
	//copy
	element = m_vector[0];
	
	//erase
	m_vector.erase( m_vector.begin() );	

	return true;
}


template <class T>
bool	wxMailbox<T>::Get( T& element, unsigned long millisec )
{
	m_mutex->Lock();
	
	//empty?
	if ( m_vector.empty() )
	{
		m_mutex->Unlock();	//to let the other thread put elements.
		
		wxMutexLocker	lockEmpty( *m_mutexEmpty );
		if ( m_condEmpty->WaitTimeout( millisec ) != wxCOND_NO_ERROR )
		{
			return false;
		}
	}
	else
		m_mutex->Unlock();

	//lock
	wxMutexLocker	lock( *m_mutex );
	
	//copy
	element = m_vector[0];
	
	//erase
	m_vector.erase( m_vector.begin() );	

	return true;
}

template <class T>
void	wxMailbox<T>::Empty()
{
	wxMutexLocker	lock( *m_mutex );
	
	m_vector.clear();
	
	return;
}

template <class T>
wxMailbox<T>::~wxMailbox()
{
	//finally..
	delete m_condEmpty;
	delete m_mutexEmpty;
	
	delete m_mutex;
}

To be able to put elements using heap, just like you pointed out:

Code: Select all

m_queue.Put(new Item());

this could be used:

Code: Select all

wxMailbox< MyClass * > m_queue;

and then

Code: Select all

m_queue.Put( *new MyClass() );

or just

Code: Select all

m_queue.Put( new MyClass() );

I haven't tried this fully yet, but I think it should work alright.

oh, btw, I clicked "Accept" accidentally, or at least I think so, so the topic was marked as solved. Anyway, not much to solve here, but thanks to Eran.

Carlos

Hi,
Several comments:

1. I am missing here a Get(int timeout) method which blocks and accepts timeout parameter.

2. The current implementation forces my queue objects to have copy constructor & assignment operator, not to mention that for performance, I would keep pointers allocated on the heap, so adding items to the queue will be something like this:

Code: Select all

m_queue.Put(new Item());

The threads who 'Get' the message, is also responsible for deleting it.

3. API to clear all elements in the queue.

Eran

[pete_b]

I'd suggest adding a copy constructor and a copy assignment operator to wxMailbox (even if private to prevent copy/assign).. At the moment, its dangerous.

I always use std::list or std::deque rather than std::vector for this kind of thing.

deque is cheaper to erase items at the beginning than a vector (which is why it has a pop_front member function).

list, however has the splice functions which are really quick - ideal for when a mutex is locked..

Resizing the list also does not require any existing elements to be copied.

Over time, using a free list and a used list should stabilize so that you no longer get allocations. It will also stabilize memory-wise with a vector, but when there is a large number of items, erasing the front of the vector means that every item in the vector must be copied which is a bit of a waste.

Code: Select all

typedef std::list<T> list_type;
list_type free_;
list_type used_;

...

void X::push(T const& t)
{
    wxMutexLocker lock( *mutex_ );
    if (free_.empty()) {
        free_.push_front(t);
    }
    else {
        free_.front() = t;
    }
    // move first element of free_ to the back of used_
    used_.splice(used_.end(), free_, free_.begin());
}

void X::pop(T& t)
{
    wxMutexLocker lock( *mutex_ );
    t = used_.front();
    // move first element of used_ to the front of free_
    free_.splice(free_.begin(), used_, used_.begin());
}

Another slant on this is to add items to a local list on the stack before locking the mutex. Then you lock the mutex and splice the item into your mutexed list.. This means you have no memory allocations while the mutex is locked, which can be critical for some things.
There are also ways to do away with the mutex altogether but it gets fairly complicated.




You can also re-use the same mutex for the condition variable that you are using to protect the list. One of the really important features of a condition variable is that waiting on it will unlock the associated mutex!

i.e.

Code: Select all

// in the ctor get rid of m_mutexEmpty
        m_condEmpty = new wxCondition( *m_mutex );

...


template <class T>
bool    wxMailbox<T>::Get( T& element, unsigned long millisec ) {
        wxMutexLocker   lock( *m_mutex );

        if ( m_vector.empty() )
        {       
                // waiting on the condition variable will 
                // unlock the mutex and re-lock it when woken.        
                if ( m_condEmpty->WaitTimeout( millisec ) != wxCOND_NO_ERROR )
                {
                        return false;
                }
        }
        // mutex is locked
       
        //copy
        element = m_vector[0];
       
        //erase
        m_vector.erase( m_vector.begin() );     

        return true;

[phlox81]

2. The current implementation forces my queue objects to have copy constructor & assignment operator, not to mention that for performance, I would keep pointers allocated on the heap, so adding items to the queue will be something like this:

Code: Select all

m_queue.Put(new Item());

The threads who 'Get' the message, is also responsible for deleting it.

3. API to clear all elements in the queue.

You already can do this with the first implementation:
wxMailBox<T*> is possible.
And when implementing such a feature, I wouldn't prefere raw pointers.
Theres always a problem, when more then one Thread is reading the messages from the box.
Therefore a shared ptr would be the better implementation, f.e. boost::shared_ptr.
The first implementation leaves all this to the user, and its probably best leave this to the user.
With a deletion / delegation policy as a template parameter you probably could create a very clean interface.

phlox