OTS/OTSCPP/Tools/boost/statechart/fifo_worker.hpp

#ifndef BOOST_STATECHART_FIFO_WORKER_HPP_INCLUDED
#define BOOST_STATECHART_FIFO_WORKER_HPP_INCLUDED
//////////////////////////////////////////////////////////////////////////////
// Copyright 2002-2008 Andreas Huber Doenni
// Distributed under the Boost Software License, Version 1.0. (See accompany-
// ing file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt)
//////////////////////////////////////////////////////////////////////////////



#include <boost/assert.hpp>
#include <boost/noncopyable.hpp>
#include <boost/function/function0.hpp>
#include <boost/bind.hpp>
// BOOST_HAS_THREADS, BOOST_MSVC
#include <boost/config.hpp>

#include <boost/detail/allocator_utilities.hpp>

#ifdef BOOST_HAS_THREADS
#  ifdef BOOST_MSVC
#    pragma warning( push )
     // "conditional expression is constant" in basic_timed_mutex.hpp
#    pragma warning( disable: 4127 )
     // "conversion from 'int' to 'unsigned short'" in microsec_time_clock.hpp
#    pragma warning( disable: 4244 )
     // "... needs to have dll-interface to be used by clients of class ..."
#    pragma warning( disable: 4251 )
     // "... assignment operator could not be generated"
#    pragma warning( disable: 4512 )
     // "Function call with parameters that may be unsafe" in
     // condition_variable.hpp
#    pragma warning( disable: 4996 )
#  endif

#  include <boost/thread/mutex.hpp>
#  include <boost/thread/condition.hpp>

#  ifdef BOOST_MSVC
#    pragma warning( pop )
#  endif
#endif

#include <list>
#include <memory>   // std::allocator


namespace boost
{
namespace statechart
{



template< class Allocator = std::allocator< none > >
class fifo_worker : noncopyable
{
  public:
    //////////////////////////////////////////////////////////////////////////
    #ifdef BOOST_HAS_THREADS
    fifo_worker( bool waitOnEmptyQueue = false ) :
      waitOnEmptyQueue_( waitOnEmptyQueue ),
    #else
    fifo_worker() :
    #endif
      terminated_( false )
    {
    }

    typedef function0< void > work_item;

    // We take a non-const reference so that we can move (i.e. swap) the item
    // into the queue, what avoids copying the (possibly heap-allocated)
    // implementation object inside work_item.
    void queue_work_item( work_item & item )
    {
      if ( item.empty() )
      {
        return;
      }

      #ifdef BOOST_HAS_THREADS
      mutex::scoped_lock lock( mutex_ );
      #endif

      workQueue_.push_back( work_item() );
      workQueue_.back().swap( item );

      #ifdef BOOST_HAS_THREADS
      queueNotEmpty_.notify_one();
      #endif
    }

    // Convenience overload so that temporary objects can be passed directly
    // instead of having to create a work_item object first. Under most
    // circumstances, this will lead to one unnecessary copy of the
    // function implementation object.
    void queue_work_item( const work_item & item )
    {
      work_item copy = item;
      queue_work_item( copy );
    }

    void terminate()
    {
      work_item item = boost::bind( &fifo_worker::terminate_impl, this );
      queue_work_item( item );
    }

    // Is not mutex-protected! Must only be called from the thread that also
    // calls operator().
    bool terminated() const
    {
      return terminated_;
    }

    unsigned long operator()( unsigned long maxItemCount = 0 )
    {
      unsigned long itemCount = 0;

      while ( !terminated() &&
        ( ( maxItemCount == 0 ) || ( itemCount < maxItemCount ) ) )
      {
        work_item item = dequeue_item();

        if ( item.empty() )
        {
          // item can only be empty when the queue is empty, which only
          // happens in ST builds or when users pass false to the fifo_worker
          // constructor
          return itemCount;
        }

        item();
        ++itemCount;
      }

      return itemCount;
    }

  private:
    //////////////////////////////////////////////////////////////////////////
    work_item dequeue_item()
    {
      #ifdef BOOST_HAS_THREADS
      mutex::scoped_lock lock( mutex_ );

      if ( !waitOnEmptyQueue_ && workQueue_.empty() )
      {
        return work_item();
      }

      while ( workQueue_.empty() )
      {
        queueNotEmpty_.wait( lock );
      }
      #else
      // If the queue happens to run empty in a single-threaded system,
      // waiting for new work items (which means to loop indefinitely!) is
      // pointless as there is no way that new work items could find their way
      // into the queue. The only sensible thing is to exit the loop and
      // return to the caller in this case.
      // Users can then queue new work items before calling operator() again.
      if ( workQueue_.empty() )
      {
        return work_item();
      }
      #endif

      // Optimization: Swap rather than assign to avoid the copy of the
      // implementation object inside function
      work_item result;
      result.swap( workQueue_.front() );
      workQueue_.pop_front();
      return result;
    }

    void terminate_impl()
    {
      terminated_ = true;
    }


    typedef std::list<
      work_item,
      typename boost::detail::allocator::rebind_to<
        Allocator, work_item >::type
    > work_queue_type;

    work_queue_type workQueue_;

    #ifdef BOOST_HAS_THREADS
    mutex mutex_;
    condition queueNotEmpty_;
    const bool waitOnEmptyQueue_;
    #endif

    bool terminated_;
};



} // namespace statechart
} // namespace boost



#endif