work_stream.h

// ---------------------------------------------------------------------
//
// Copyright (C) 2008 - 2015 by the deal.II authors
//
// This file is part of the deal.II library.
//
// The deal.II library is free software; you can use it, redistribute
// it, and/or modify it under the terms of the GNU Lesser General
// Public License as published by the Free Software Foundation; either
// version 2.1 of the License, or (at your option) any later version.
// The full text of the license can be found in the file LICENSE at
// the top level of the deal.II distribution.
//
// ---------------------------------------------------------------------

#ifndef dealii__work_stream_h
#define dealii__work_stream_h


#include <deal.II/base/config.h>
#include <deal.II/base/graph_coloring.h>
#include <deal.II/base/multithread_info.h>
#include <deal.II/base/thread_management.h>
#include <deal.II/base/template_constraints.h>
#include <deal.II/base/std_cxx11/function.h>
#include <deal.II/base/std_cxx11/bind.h>
#include <deal.II/base/thread_local_storage.h>
#include <deal.II/base/parallel.h>

#ifdef DEAL_II_WITH_THREADS
#  include <deal.II/base/thread_management.h>
#  include <tbb/pipeline.h>
#endif

#include <vector>
#include <utility>
#include <memory>


DEAL_II_NAMESPACE_OPEN



namespace WorkStream
{

#ifdef DEAL_II_WITH_THREADS

  namespace internal
  {

//TODO: The following classes all use std_cxx11::shared_ptr, but the
//  correct pointer class would actually be std::unique_ptr. make this
//  replacement whenever we have a class that provides these semantics
//  and that is available also as a fall-back whenever via boost or similar

    namespace Implementation2
    {
      template <typename Iterator,
                typename ScratchData,
                typename CopyData>
      class IteratorRangeToItemStream : public tbb::filter
      {
      public:
        struct ItemType
        {
          struct ScratchDataObject
          {
            std_cxx11::shared_ptr<ScratchData> scratch_data;
            bool                               currently_in_use;

            ScratchDataObject ()
              :
              currently_in_use (false)
            {}

            ScratchDataObject (ScratchData *p,
                               const bool in_use)
              :
              scratch_data (p),
              currently_in_use (in_use)
            {}

//TODO: when we push back an object to the list of scratch objects, in
//  Worker::operator(), we first create an object and then copy
//  it to the end of this list. this involves having two objects
//      of the current type having pointers to it, each with their own
//      currently_in_use flag. there is probably little harm in this because
//      the original one goes out of scope right away again, but it's
//      certainly awkward. one way to avoid this would be to use unique_ptr
//      but we'd need to figure out a way to use it in non-C++11 mode
            ScratchDataObject (const ScratchDataObject &o)
              :
              scratch_data (o.scratch_data),
              currently_in_use (o.currently_in_use)
            {}
          };


          typedef std::list<ScratchDataObject> ScratchDataList;

          std::vector<Iterator> work_items;

          std::vector<CopyData> copy_datas;

          unsigned int          n_items;

          Threads::ThreadLocalStorage<ScratchDataList> *scratch_data;

          const ScratchData *sample_scratch_data;

          bool currently_in_use;


          ItemType ()
            :
            n_items (0),
            scratch_data (0),
            sample_scratch_data (0),
            currently_in_use (false)
          {}
        };


        IteratorRangeToItemStream (const Iterator       &begin,
                                   const Iterator       &end,
                                   const unsigned int    buffer_size,
                                   const unsigned int    chunk_size,
                                   const ScratchData    &sample_scratch_data,
                                   const CopyData       &sample_copy_data)
          :
          tbb::filter (/*is_serial=*/true),
          remaining_iterator_range (begin, end),
          item_buffer (buffer_size),
          sample_scratch_data (sample_scratch_data),
          chunk_size (chunk_size)
        {
          // initialize the elements of the ring buffer
          for (unsigned int element=0; element<item_buffer.size(); ++element)
            {
              Assert (item_buffer[element].n_items == 0,
                      ExcInternalError());

              item_buffer[element].work_items.resize (chunk_size,
                                                      remaining_iterator_range.second);
              item_buffer[element].scratch_data = &thread_local_scratch;
              item_buffer[element].sample_scratch_data = &sample_scratch_data;
              item_buffer[element].copy_datas.resize (chunk_size,
                                                      sample_copy_data);
              item_buffer[element].currently_in_use = false;
            }
        }


        virtual void *operator () (void *)
        {
          // find first unused item. we know that there must be one
          // because we have set the maximal number of tokens in flight
          // and have set the ring buffer to have exactly this size. so
          // if this function is called, we know that less than the
          // maximal number of items in currently in flight
          //
          // note that we need not lock access to this array since
          // the current stage is run sequentially and we can therefore
          // enter the following block only once at any given time.
          // thus, there can be no race condition between checking that
          // a flag is false and setting it to true. (there may be
          // another thread where we release items and set 'false'
          // flags to 'true', but that too does not produce any
          // problems)
          ItemType *current_item = 0;
          for (unsigned int i=0; i<item_buffer.size(); ++i)
            if (item_buffer[i].currently_in_use == false)
              {
                item_buffer[i].currently_in_use = true;
                current_item = &item_buffer[i];
                break;
              }
          Assert (current_item != 0, ExcMessage ("This can't be. There must be a free item!"));

          // initialize the next item. it may
          // consist of at most chunk_size
          // elements
          current_item->n_items = 0;
          while ((remaining_iterator_range.first !=
                  remaining_iterator_range.second)
                 &&
                 (current_item->n_items < chunk_size))
            {
              current_item->work_items[current_item->n_items]
                = remaining_iterator_range.first;

              ++remaining_iterator_range.first;
              ++current_item->n_items;
            }

          if (current_item->n_items == 0)
            // there were no items
            // left. terminate the pipeline
            return 0;
          else
            return current_item;
        }

      private:
        std::pair<Iterator,Iterator> remaining_iterator_range;

        std::vector<ItemType>        item_buffer;

        Threads::ThreadLocalStorage<typename ItemType::ScratchDataList> thread_local_scratch;

        const ScratchData &sample_scratch_data;

        const unsigned int           chunk_size;

        void init_buffer_elements (const unsigned int element,
                                   const CopyData    &sample_copy_data)
        {
          Assert (item_buffer[element].n_items == 0,
                  ExcInternalError());

          item_buffer[element].work_items
          .resize (chunk_size, remaining_iterator_range.second);
          item_buffer[element].scratch_data
            = &thread_local_scratch;
          item_buffer[element].sample_scratch_data
            = &sample_scratch_data;
          item_buffer[element].copy_datas
          .resize (chunk_size, sample_copy_data);
        }
      };



      template <typename Iterator,
                typename ScratchData,
                typename CopyData>
      class Worker : public tbb::filter
      {
      public:
        Worker (const std_cxx11::function<void (const Iterator &,
                                                ScratchData &,
                                                CopyData &)> &worker,
                bool copier_exist=true)
          :
          tbb::filter (/* is_serial= */ false),
          worker (worker),
          copier_exist (copier_exist)
        {}


        void *operator () (void *item)
        {
          // first unpack the current item
          typedef
          typename IteratorRangeToItemStream<Iterator,ScratchData,CopyData>::ItemType
          ItemType;

          ItemType *current_item = static_cast<ItemType *> (item);

          // we need to find an unused scratch data object in the list that
          // corresponds to the current thread and then mark it as used. if
          // we can't find one, create one
          //
          // as discussed in the discussion of the documentation of the
          // IteratorRangeToItemStream::scratch_data variable, there is no
          // need to synchronize access to this variable using a mutex
          // as long as we have no yield-point in between. this means that
          // we can't take an iterator into the list now and expect it to
          // still be valid after calling the worker, but we at least do
          // not have to lock the following section
          ScratchData *scratch_data = 0;
          {
            typename ItemType::ScratchDataList &
            scratch_data_list = current_item->scratch_data->get();

            // see if there is an unused object. if so, grab it and mark
            // it as used
            for (typename ItemType::ScratchDataList::iterator
                 p = scratch_data_list.begin();
                 p != scratch_data_list.end(); ++p)
              if (p->currently_in_use == false)
                {
                  scratch_data = p->scratch_data.get();
                  p->currently_in_use = true;
                  break;
                }

            // if no object was found, create one and mark it as used
            if (scratch_data == 0)
              {
                scratch_data = new ScratchData(*current_item->sample_scratch_data);

                typename ItemType::ScratchDataList::value_type
                new_scratch_object (scratch_data, true);
                scratch_data_list.push_back (new_scratch_object);
              }
          }

          // then call the worker function on each element of the chunk we were
          // given. since these worker functions are called on separate threads,
          // nothing good can happen if they throw an exception and we are best
          // off catching it and showing an error message
          for (unsigned int i=0; i<current_item->n_items; ++i)
            {
              try
                {
                  if (worker)
                    worker (current_item->work_items[i],
                            *scratch_data,
                            current_item->copy_datas[i]);
                }
              catch (const std::exception &exc)
                {
                  Threads::internal::handle_std_exception (exc);
                }
              catch (...)
                {
                  Threads::internal::handle_unknown_exception ();
                }
            }

          // finally mark the scratch object as unused again. as above, there
          // is no need to lock anything here since the object we work on
          // is thread-local
          {
            typename ItemType::ScratchDataList &
            scratch_data_list = current_item->scratch_data->get();

            for (typename ItemType::ScratchDataList::iterator p =
                   scratch_data_list.begin(); p != scratch_data_list.end();
                 ++p)
              if (p->scratch_data.get() == scratch_data)
                {
                  Assert(p->currently_in_use == true, ExcInternalError());
                  p->currently_in_use = false;
                }
          }

          // if there is no copier, mark current item as usable again
          if (copier_exist==false)
            current_item->currently_in_use = false;


          // then return the original pointer
          // to the now modified object
          return item;
        }


      private:
        const std_cxx11::function<void (const Iterator &,
                                        ScratchData &,
                                        CopyData &)> worker;

        bool copier_exist;
      };



      template <typename Iterator,
                typename ScratchData,
                typename CopyData>
      class Copier : public tbb::filter
      {
      public:
        Copier (const std_cxx11::function<void (const CopyData &)> &copier)
          :
          tbb::filter (/*is_serial=*/true),
          copier (copier)
        {}


        void *operator () (void *item)
        {
          // first unpack the current item
          typedef
          typename IteratorRangeToItemStream<Iterator,ScratchData,CopyData>::ItemType
          ItemType;

          ItemType *current_item = static_cast<ItemType *> (item);

          // initiate copying data. for the same reasons as in the worker class
          // above, catch exceptions rather than letting it propagate into
          // unknown territories
          for (unsigned int i=0; i<current_item->n_items; ++i)
            {
              try
                {
                  if (copier)
                    copier (current_item->copy_datas[i]);
                }
              catch (const std::exception &exc)
                {
                  Threads::internal::handle_std_exception (exc);
                }
              catch (...)
                {
                  Threads::internal::handle_unknown_exception ();
                }
            }

          // mark current item as usable again
          current_item->currently_in_use = false;


          // return an invalid item since we are at the end of the
          // pipeline
          return 0;
        }


      private:
        const std_cxx11::function<void (const CopyData &)> copier;
      };

    }


    namespace Implementation3
    {
      template <typename Iterator,
                typename ScratchData,
                typename CopyData>
      struct ScratchAndCopyDataObjects
      {
        std_cxx11::shared_ptr<ScratchData> scratch_data;
        std_cxx11::shared_ptr<CopyData>    copy_data;
        bool                               currently_in_use;

        ScratchAndCopyDataObjects ()
          :
          currently_in_use (false)
        {}

        ScratchAndCopyDataObjects (ScratchData *p,
                                   CopyData *q,
                                   const bool in_use)
          :
          scratch_data (p),
          copy_data (q),
          currently_in_use (in_use)
        {}

//TODO: when we push back an object to the list of scratch objects, in
//      Worker::operator(), we first create an object and then copy
//      it to the end of this list. this involves having two objects
//      of the current type having pointers to it, each with their own
//      currently_in_use flag. there is probably little harm in this because
//      the original one goes out of scope right away again, but it's
//      certainly awkward. one way to avoid this would be to use unique_ptr
//      but we'd need to figure out a way to use it in non-C++11 mode
        ScratchAndCopyDataObjects (const ScratchAndCopyDataObjects &o)
          :
          scratch_data (o.scratch_data),
          copy_data (o.copy_data),
          currently_in_use (o.currently_in_use)
        {}
      };





      template <typename Iterator,
                typename ScratchData,
                typename CopyData>
      class WorkerAndCopier
      {
      public:
        WorkerAndCopier (const std_cxx11::function<void (const Iterator &,
                                                         ScratchData &,
                                                         CopyData &)> &worker,
                         const std_cxx11::function<void (const CopyData &)> &copier,
                         const ScratchData    &sample_scratch_data,
                         const CopyData       &sample_copy_data)
          :
          worker (worker),
          copier (copier),
          sample_scratch_data (sample_scratch_data),
          sample_copy_data (sample_copy_data)
        {}


        void operator() (const tbb::blocked_range<typename std::vector<Iterator>::const_iterator> &range)
        {
          // we need to find an unused scratch and corresponding copy
          // data object in the list that corresponds to the current
          // thread and then mark it as used. If we can't find one,
          // create one as discussed in the discussion of the documentation
          // of the IteratorRangeToItemStream::scratch_data variable,
          // there is no need to synchronize access to this variable
          // using a mutex as long as we have no yield-point in between.
          // This means that we can't take an iterator into the list
          // now and expect it to still be valid after calling the worker,
          // but we at least do not have to lock the following section.
          ScratchData *scratch_data = 0;
          CopyData    *copy_data    = 0;
          {
            ScratchAndCopyDataList &scratch_and_copy_data_list = data.get();

            // see if there is an unused object. if so, grab it and mark
            // it as used
            for (typename ScratchAndCopyDataList::iterator
                 p = scratch_and_copy_data_list.begin();
                 p != scratch_and_copy_data_list.end(); ++p)
              if (p->currently_in_use == false)
                {
                  scratch_data = p->scratch_data.get();
                  copy_data    = p->copy_data.get();
                  p->currently_in_use = true;
                  break;
                }

            // if no element in the list was found, create one and mark it as used
            if (scratch_data == 0)
              {
                Assert (copy_data==0, ExcInternalError());
                scratch_data = new ScratchData(sample_scratch_data);
                copy_data    = new CopyData(sample_copy_data);

                typename ScratchAndCopyDataList::value_type
                new_scratch_object (scratch_data, copy_data, true);
                scratch_and_copy_data_list.push_back (new_scratch_object);
              }
          }

          // then call the worker and copier functions on each
          // element of the chunk we were given.
          for (typename std::vector<Iterator>::const_iterator p=range.begin();
               p != range.end(); ++p)
            {
              try
                {
                  if (worker)
                    worker (*p,
                            *scratch_data,
                            *copy_data);
                  if (copier)
                    copier (*copy_data);
                }
              catch (const std::exception &exc)
                {
                  Threads::internal::handle_std_exception (exc);
                }
              catch (...)
                {
                  Threads::internal::handle_unknown_exception ();
                }
            }

          // finally mark the scratch object as unused again. as above, there
          // is no need to lock anything here since the object we work on
          // is thread-local
          {
            ScratchAndCopyDataList &scratch_and_copy_data_list = data.get();

            for (typename ScratchAndCopyDataList::iterator p =
                   scratch_and_copy_data_list.begin(); p != scratch_and_copy_data_list.end();
                 ++p)
              if (p->scratch_data.get() == scratch_data)
                {
                  Assert(p->currently_in_use == true, ExcInternalError());
                  p->currently_in_use = false;
                }
          }

        }

      private:
        typedef
        typename Implementation3::ScratchAndCopyDataObjects<Iterator,ScratchData,CopyData>
        ScratchAndCopyDataObjects;

        typedef std::list<ScratchAndCopyDataObjects> ScratchAndCopyDataList;

        Threads::ThreadLocalStorage<ScratchAndCopyDataList> data;

        const std_cxx11::function<void (const Iterator &,
                                        ScratchData &,
                                        CopyData &)> worker;

        const std_cxx11::function<void (const CopyData &)> copier;

        const ScratchData    &sample_scratch_data;
        const CopyData       &sample_copy_data;
      };
    }

  }


#endif // DEAL_II_WITH_THREADS


  template <typename Worker,
            typename Copier,
            typename Iterator,
            typename ScratchData,
            typename CopyData>
  void
  run (const std::vector<std::vector<Iterator> > &colored_iterators,
       Worker                                     worker,
       Copier                                     copier,
       const ScratchData                         &sample_scratch_data,
       const CopyData                            &sample_copy_data,
       const unsigned int queue_length = 2*MultithreadInfo::n_threads(),
       const unsigned int                         chunk_size = 8);


  template <typename Worker,
            typename Copier,
            typename Iterator,
            typename ScratchData,
            typename CopyData>
  void
  run (const Iterator                          &begin,
       const typename identity<Iterator>::type &end,
       Worker                                   worker,
       Copier                                   copier,
       const ScratchData                       &sample_scratch_data,
       const CopyData                          &sample_copy_data,
       const unsigned int queue_length = 2*MultithreadInfo::n_threads(),
       const unsigned int                       chunk_size = 8)
  {
    Assert (queue_length > 0,
            ExcMessage ("The queue length must be at least one, and preferably "
                        "larger than the number of processors on this system."));
    (void)queue_length; // removes -Wunused-parameter warning in optimized mode
    Assert (chunk_size > 0,
            ExcMessage ("The chunk_size must be at least one."));
    (void)chunk_size; // removes -Wunused-parameter warning in optimized mode

    // if no work then skip. (only use operator!= for iterators since we may
    // not have an equality comparison operator)
    if (!(begin != end))
      return;

    // we want to use TBB if we have support and if it is not disabled at
    // runtime:
#ifdef DEAL_II_WITH_THREADS
    if (MultithreadInfo::n_threads()==1)
#endif
      {
        // need to copy the sample since it is marked const
        ScratchData scratch_data = sample_scratch_data;
        CopyData    copy_data    = sample_copy_data;

        for (Iterator i=begin; i!=end; ++i)
          {
            // need to check if the function is not the zero function. To
            // check zero-ness, create a C++ function out of it and check that
            if (static_cast<const std_cxx11::function<void (const Iterator &,
                                                            ScratchData &,
                                                            CopyData &)>& >(worker))
              worker (i, scratch_data, copy_data);
            if (static_cast<const std_cxx11::function<void (const CopyData &)>& >
                (copier))
              copier (copy_data);
          }
      }
#ifdef DEAL_II_WITH_THREADS
    else // have TBB and use more than one thread
      {
        // Check that the copier exist
        if (static_cast<const std_cxx11::function<void (const CopyData &)>& >(copier))
          {
            // create the three stages of the pipeline
            internal::Implementation2::IteratorRangeToItemStream<Iterator,ScratchData,CopyData>
            iterator_range_to_item_stream (begin, end,
                                           queue_length,
                                           chunk_size,
                                           sample_scratch_data,
                                           sample_copy_data);

            internal::Implementation2::Worker<Iterator, ScratchData, CopyData> worker_filter (worker);
            internal::Implementation2::Copier<Iterator, ScratchData, CopyData> copier_filter (copier);

            // now create a pipeline from these stages
            tbb::pipeline assembly_line;
            assembly_line.add_filter (iterator_range_to_item_stream);
            assembly_line.add_filter (worker_filter);
            assembly_line.add_filter (copier_filter);

            // and run it
            assembly_line.run (queue_length);

            assembly_line.clear ();
          }
        else
          {
            // there is no copier function. in this case, we have an
            // embarrassingly parallel problem where we can
            // essentially apply parallel_for. because parallel_for
            // requires subdividing the range for which operator- is
            // necessary between iterators, it is often inefficient to
            // apply it directory to cell ranges and similar iterator
            // types for which operator- is expensive or, in fact,
            // nonexistent. rather, in that case, we simply copy the
            // iterators into a large array and use operator- on
            // iterators to this array of iterators.
            //
            // instead of duplicating code, this is essentially the
            // same situation we have in Implementation3 below, so we
            // just defer to that place
            std::vector<std::vector<Iterator> > all_iterators (1);
            for (Iterator p=begin; p!=end; ++p)
              all_iterators[0].push_back (p);

            run (all_iterators,
                 worker, copier,
                 sample_scratch_data,
                 sample_copy_data,
                 queue_length,
                 chunk_size);
          }
      }
#endif
  }


  // Implementation 3:
  template <typename Worker,
            typename Copier,
            typename Iterator,
            typename ScratchData,
            typename CopyData>
  void
  run (const std::vector<std::vector<Iterator> > &colored_iterators,
       Worker                                     worker,
       Copier                                     copier,
       const ScratchData                         &sample_scratch_data,
       const CopyData                            &sample_copy_data,
       const unsigned int                         queue_length,
       const unsigned int                         chunk_size)
  {
    Assert (queue_length > 0,
            ExcMessage ("The queue length must be at least one, and preferably "
                        "larger than the number of processors on this system."));
    (void)queue_length; // removes -Wunused-parameter warning in optimized mode
    Assert (chunk_size > 0,
            ExcMessage ("The chunk_size must be at least one."));
    (void)chunk_size; // removes -Wunused-parameter warning in optimized mode

    // we want to use TBB if we have support and if it is not disabled at
    // runtime:
#ifdef DEAL_II_WITH_THREADS
    if (MultithreadInfo::n_threads()==1)
#endif
      {
        // need to copy the sample since it is marked const
        ScratchData scratch_data = sample_scratch_data;
        CopyData    copy_data    = sample_copy_data;

        for (unsigned int color=0; color<colored_iterators.size(); ++color)
          for (typename std::vector<Iterator>::const_iterator p = colored_iterators[color].begin();
               p != colored_iterators[color].end(); ++p)
            {
              // need to check if the function is not the zero function. To
              // check zero-ness, create a C++ function out of it and check that
              if (static_cast<const std_cxx11::function<void (const Iterator &,
                                                              ScratchData &,
                                                              CopyData &)>& >(worker))
                worker (*p, scratch_data, copy_data);
              if (static_cast<const std_cxx11::function<void (const CopyData &)>& >(copier))
                copier (copy_data);
            }
      }
#ifdef DEAL_II_WITH_THREADS
    else // have TBB and use more than one thread
      {
        // loop over the various colors of what we're given
        for (unsigned int color=0; color<colored_iterators.size(); ++color)
          if (colored_iterators[color].size() > 0)
            {
              typedef
              internal::Implementation3::WorkerAndCopier<Iterator,ScratchData,CopyData>
              WorkerAndCopier;

              typedef
              typename std::vector<Iterator>::const_iterator
              RangeType;

              WorkerAndCopier worker_and_copier (worker,
                                                 copier,
                                                 sample_scratch_data,
                                                 sample_copy_data);

              tbb::parallel_for (tbb::blocked_range<RangeType>
                                 (colored_iterators[color].begin(),
                                  colored_iterators[color].end(),
                                  /*grain_size=*/chunk_size),
                                 std_cxx11::bind (&WorkerAndCopier::operator(),
                                                  std_cxx11::ref(worker_and_copier),
                                                  std_cxx11::_1),
                                 tbb::auto_partitioner());
            }
      }
#endif
  }





  template <typename MainClass,
            typename Iterator,
            typename ScratchData,
            typename CopyData>
  void
  run (const Iterator                          &begin,
       const typename identity<Iterator>::type &end,
       MainClass                               &main_object,
       void (MainClass::*worker) (const Iterator &,
                                  ScratchData &,
                                  CopyData &),
       void (MainClass::*copier) (const CopyData &),
       const ScratchData                       &sample_scratch_data,
       const CopyData                          &sample_copy_data,
       const unsigned int queue_length =        2*MultithreadInfo::n_threads(),
       const unsigned int chunk_size =          8)
  {
    // forward to the other function
    run (begin, end,
         std_cxx11::bind (worker,
                          std_cxx11::ref (main_object),
                          std_cxx11::_1, std_cxx11::_2, std_cxx11::_3),
         std_cxx11::bind (copier,
                          std_cxx11::ref (main_object),
                          std_cxx11::_1),
         sample_scratch_data,
         sample_copy_data,
         queue_length,
         chunk_size);
  }

}




DEAL_II_NAMESPACE_CLOSE




//----------------------------   work_stream.h     ---------------------------
// end of #ifndef dealii__work_stream_h
#endif
//----------------------------   work_stream.h     ---------------------------