[Boost-users] 2. TRY boost::thread - problem with worker as an inner class of a queue

From: Lars Kunert (lkunert_at_[hidden])
Date: 2007-01-27 04:06:03

Dear Moderator!

Yesterday I posted a mail to boost-users. I got a reply, that my mail
awaits moderation. That mail also contained a link. The link is dead since
yesterday evening. Which does usually mean that the mail was posted,
but it did not show up. I have enclosed the mail again below,

Thanks, Lars Kunert

The following attached (downsized) program implements a queue which
contains a single worker process. Int's that are pushed into the queue
are popped by
the worker and processed... The problem is, the program ends up with an
segmentation fault. Valgrind tells me that the stack is overwritten.

==4927== Conditional jump or move depends on uninitialised value(s)
==4927==    at 0x8073536: strstr (in
/local/home/lkunert/projects/pg-build-4.0.3/src/cdag/threads/queue_error)
==4927==    by 0x80609F5: pthread_initialize (smp.h:47)
==4927==    by 0x80B2DE1: (within
/local/home/lkunert/projects/pg-build-4.0.3/src/cdag/threads/queue_error)
==4927==    by 0x80480E8: (within
/local/home/lkunert/projects/pg-build-4.0.3/src/cdag/threads/queue_error)
==4927==    by 0x80641A0: __libc_csu_init (in
/local/home/lkunert/projects/pg-build-4.0.3/src/cdag/threads/queue_error)
==4927==    by 0x8063FC2: __libc_start_main (in
/local/home/lkunert/projects/pg-build-4.0.3/src/cdag/threads/queue_error)
==4927==    by 0x8048120: ??? (start.S:102)
==4927==
==4927== Syscall param write(buf) points to uninitialised byte(s)
==4927==    at 0x8063044: write (in
/local/home/lkunert/projects/pg-build-4.0.3/src/cdag/threads/queue_error)
==4927==    by 0x804C241: boost::thread::thread(boost::function0<void,
std::allocator<boost::function_base> > const&) (in
/local/home/lkunert/projects/pg-build-4.0.3/src/cdag/threads/queue_error)
==4927==    by 0x804AD26: QueueError::QueueError(unsigned)
(QueueError.cxx:74)
==4927==    by 0x8048287: main (QueueError.cxx:150)
==4927==  Address 0x52BFE04C is on thread 1's stack

Line 74 corresponds to the initailisation of the worker. The
thread-manual writes:

*explicit* thread(*const* boost::function0<*void*>& threadfunc);

*Effects*: Starts a new thread of execution and constructs a thread
<http://www.boost.org/doc/html/thread.html> object representing it.
Copies |threadfunc| (which in turn copies the function object wrapped by
|threadfunc|) to an internal location which persists for the lifetime of
the new thread of execution. Calls |operator()| on the copy of the
|threadfunc| function object in the new thread of execution.

Which sound good to me. Were is the mistake.

class QueueError
{

// *** Typedefs
*************************************************************

  private:
    typedef std::queue<int> Queue;
    typedef boost::thread Thread;

  public:
    typedef Queue::size_type ST;

    
// *** Inner Classes
********************************************************

  private:
    
    class Worker
    {
      private:
        QueueError* outer_p_;

      public:
        Worker( QueueError* outer_p )
                : outer_p_( outer_p )
        {};
        
        
      public:
        void
        operator()()
        {
            outer_p_->worker_run_();
        };
    };
    

// *** Attributes
************************************************************

  public:
  private:
    ST max_queue_size_;
    Queue  queue_;

    Thread thread_;
    
    boost::mutex monitor_;
    boost::condition space_available_, job_available_;
    
    
// *** Constructors / Destructors
********************************************

  public:
    explicit
    QueueError( ST max_queue_size =1000 )
            : max_queue_size_( max_queue_size )
            , thread_( Worker( this ))
    {};

    
// *** Methods
***************************************************************

  public:
    void
    push( int i )
    {
        boost::mutex::scoped_lock lock( monitor_ );
        
        while( queue_.size() >= max_queue_size_ )
        {
            space_available_.wait( lock );
        };
        queue_.push( i );
        
        job_available_.notify_one();
    };
    

  protected:
    void
    worker_run_()
    {
        int i =0;

        while( true )
        {
            i = worker_pop_();

            worker_work_( i );
        };
    };

    
    int
    worker_pop_()
    {
        boost::mutex::scoped_lock lock( monitor_ );

        while( queue_.empty())
        {
            job_available_.wait( lock );
        };
        
        int i = queue_.front();
        queue_.pop();
        
        space_available_.notify_one();
        
        return i;
    };

    void
    worker_work_( int i )
    {
        for( int ii= 0; ii < i; ++ii )
        {
            // do some work
            double divident = 10000;
            double sum;
            for( uint divisor =1; divisor < divident; ++divisor )
            {
                sum = divident / divisor;
            };
        };
    };
    
};  // class QueueError

int main()
{
    QueueError q;

    q.push( 10 );
    q.push( 9 );
    q.push( 8 );
    q.push( 7 );
    q.push( 6 );

    return 0;
};