From: turcksin Date: Wed, 25 Sep 2013 16:12:27 +0000 (+0000) Subject: Use graph coloring whith work_stream. X-Git-Url: https://gitweb.dealii.org/cgi-bin/gitweb.cgi?a=commitdiff_plain;h=f528c7544e4be4147aedba2abb632da3313bd97e;p=dealii-svn.git Use graph coloring whith work_stream. git-svn-id: https://svn.dealii.org/trunk@30921 0785d39b-7218-0410-832d-ea1e28bc413d --- diff --git a/deal.II/include/deal.II/base/graph_coloring.h b/deal.II/include/deal.II/base/graph_coloring.h index 3bf6c6abf5..b6d49e7d2d 100644 --- a/deal.II/include/deal.II/base/graph_coloring.h +++ b/deal.II/include/deal.II/base/graph_coloring.h @@ -38,7 +38,7 @@ namespace graph_coloring { template std::vector > create_partitioning(Iterator const &begin, typename identity::type const &end, - std::vector (*get_conflict_indices) (Iterator &)) + std::vector (*get_conflict_indices) (Iterator const&)) { std::vector > partitioning(1,std::vector (1,begin)); // Number of iterators. @@ -111,7 +111,7 @@ namespace graph_coloring { */ template std::vector > make_dsatur_coloring(std::vector &partition, - std::vector (*get_conflict_indices)(Iterator &)) + std::vector (*get_conflict_indices)(Iterator const&)) { std::vector > partition_coloring; // Number of zones composing the partitioning. @@ -343,7 +343,7 @@ namespace graph_coloring { template std::vector > make_graph_coloring(Iterator const &begin,typename identity::type const &end, - std::vector (*get_conflict_indices)(Iterator &)) + std::vector (*get_conflict_indices)(Iterator const&)) { // Create the partitioning. std::vector > partitioning = create_partitioning(begin,end, diff --git a/deal.II/include/deal.II/base/work_stream.h b/deal.II/include/deal.II/base/work_stream.h index d06f06cbec..51b1655d4d 100644 --- a/deal.II/include/deal.II/base/work_stream.h +++ b/deal.II/include/deal.II/base/work_stream.h @@ -19,6 +19,7 @@ #include +#include #include #include #include @@ -261,9 +262,9 @@ namespace WorkStream */ ItemType () : - n_items (0), - scratch_data (0), - sample_scratch_data (0) + n_items (0), + scratch_data (0), + sample_scratch_data (0) {} }; @@ -277,32 +278,67 @@ namespace WorkStream * copier function invokation. */ 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) + 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), - ring_buffer (buffer_size), - sample_scratch_data (sample_scratch_data), - n_emitted_items (0), - chunk_size (chunk_size) + tbb::filter (/*is_serial=*/true), + remaining_iterator_range (new std::pair (begin, end)), + ring_buffer (buffer_size), + sample_scratch_data (sample_scratch_data), + color(false), + n_emitted_items (0), + chunk_size (chunk_size) { - // initialize the elements of the ring buffer + // initialize the elements of the ring buffer for (unsigned int element=0; elementsecond); + ring_buffer[element].scratch_data = &thread_local_scratch; + ring_buffer[element].sample_scratch_data = &sample_scratch_data; + ring_buffer[element].copy_datas.resize (chunk_size, + sample_copy_data); + } + } + + IteratorRangeToItemStream (const typename std::vector::iterator &begin, + const typename std::vector::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), + color_remaining_iterator_range (new std::pair::iterator, + typename std::vector::iterator> (begin,end)), + ring_buffer (buffer_size), + sample_scratch_data (sample_scratch_data), + color(true), + n_emitted_items (0), + chunk_size (chunk_size) + { + for (unsigned int element=0; elementfirts) because + // iterator may not have a default constructor and + // *(color_remaining_iterator_range->second) is invalid. + ring_buffer[element].work_items.resize (chunk_size, + *(color_remaining_iterator_range->first)); + ring_buffer[element].scratch_data = &thread_local_scratch; + ring_buffer[element].sample_scratch_data = &sample_scratch_data; + ring_buffer[element].copy_datas.resize (chunk_size, + sample_copy_data); + } } /** @@ -320,15 +356,28 @@ namespace WorkStream // 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)) + if (color==false) + 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; - ++remaining_iterator_range.first; + ++remaining_iterator_range->first; + ++current_item->n_items; + } + else + while ((color_remaining_iterator_range->first != + color_remaining_iterator_range->second) + && + (current_item->n_items < chunk_size)) + { + current_item->work_items[current_item->n_items] + = *(color_remaining_iterator_range->first); + + ++color_remaining_iterator_range->first; ++current_item->n_items; } @@ -337,10 +386,10 @@ namespace WorkStream // left. terminate the pipeline return 0; else - { - ++n_emitted_items; - return current_item; - } + { + ++n_emitted_items; + return current_item; + } } private: @@ -349,7 +398,15 @@ namespace WorkStream * be worked on. This range will shrink * over time. */ - std::pair remaining_iterator_range; + std_cxx1x::shared_ptr > remaining_iterator_range; + + /** + * When graph coloring is used the iterators to be worked on are given + * in a vector defined by a pair of iterators. + */ + std_cxx1x::shared_ptr::iterator, + typename std::vector::iterator> > + color_remaining_iterator_range; /** * A ring buffer that will store items. @@ -396,6 +453,11 @@ namespace WorkStream */ const ScratchData &sample_scratch_data; + /** + * This flag is used to know if graph coloring is used or not. + */ + bool color; + /** * Counter for the number of emitted * items. Each item may consist of up @@ -423,19 +485,19 @@ namespace WorkStream * the ring buffer. */ void init_buffer_elements (const unsigned int element, - const CopyData &sample_copy_data) + const CopyData &sample_copy_data) { Assert (ring_buffer[element].n_items == 0, - ExcInternalError()); + ExcInternalError()); ring_buffer[element].work_items - .resize (chunk_size, remaining_iterator_range.second); + .resize (chunk_size, remaining_iterator_range->second); ring_buffer[element].scratch_data = &thread_local_scratch; ring_buffer[element].sample_scratch_data = &sample_scratch_data; ring_buffer[element].copy_datas - .resize (chunk_size, sample_copy_data); + .resize (chunk_size, sample_copy_data); } }; @@ -451,135 +513,135 @@ namespace WorkStream template - class Worker : public tbb::filter - { - public: - /** - * Constructor. Takes a - * reference to the object on - * which we will operate as - * well as a pointer to the - * function that will do the - * assembly. - */ - Worker (const std_cxx1x::function &worker) - : - tbb::filter (/* is_serial= */ false), - worker (worker) - {} - - - /** - * Work on an item. - */ - void *operator () (void *item) - { - // first unpack the current item - typedef - typename IteratorRangeToItemStream::ItemType - ItemType; - - ItemType *current_item = static_cast (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; in_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; - } - } - - - - // then return the original pointer - // to the now modified object - return item; - } - - - private: - /** - * Pointer to the function - * that does the assembling - * on the sequence of cells. - */ - const std_cxx1x::function worker; - }; + class Worker : public tbb::filter + { + public: + /** + * Constructor. Takes a + * reference to the object on + * which we will operate as + * well as a pointer to the + * function that will do the + * assembly. + */ + Worker (const std_cxx1x::function &worker) + : + tbb::filter (/* is_serial= */ false), + worker (worker) + {} + + + /** + * Work on an item. + */ + void *operator () (void *item) + { + // first unpack the current item + typedef + typename IteratorRangeToItemStream::ItemType + ItemType; + + ItemType *current_item = static_cast (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; in_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; + } + } + + + + // then return the original pointer + // to the now modified object + return item; + } + + + private: + /** + * Pointer to the function + * that does the assembling + * on the sequence of cells. + */ + const std_cxx1x::function worker; + }; @@ -594,76 +656,76 @@ namespace WorkStream template - class Copier : public tbb::filter - { - public: - /** - * Constructor. Takes a - * reference to the object on - * which we will operate as - * well as a pointer to the - * function that will do the - * copying from the - * additional data object to - * the global matrix or - * similar. - */ - Copier (const std_cxx1x::function &copier) - : - tbb::filter (/* is_serial= */ true), - copier (copier) - {} - - - /** - * Work on a single item. - */ - void *operator () (void *item) - { - // first unpack the current item - typedef - typename IteratorRangeToItemStream::ItemType - ItemType; - - ItemType *current_item = static_cast (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; in_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 (); - } - } - - - // return an invalid - // item since we are at - // the end of the - // pipeline - return 0; - } - - - private: - /** - * Pointer to the function - * that does the copying of - * data. - */ - const std_cxx1x::function copier; - }; + class Copier : public tbb::filter + { + public: + /** + * Constructor. Takes a + * reference to the object on + * which we will operate as + * well as a pointer to the + * function that will do the + * copying from the + * additional data object to + * the global matrix or + * similar. + */ + Copier (const std_cxx1x::function &copier,bool is_serial) + : + tbb::filter (is_serial), + copier (copier) + {} + + + /** + * Work on a single item. + */ + void *operator () (void *item) + { + // first unpack the current item + typedef + typename IteratorRangeToItemStream::ItemType + ItemType; + + ItemType *current_item = static_cast (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; in_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 (); + } + } + + + // return an invalid + // item since we are at + // the end of the + // pipeline + return 0; + } + + + private: + /** + * Pointer to the function + * that does the copying of + * data. + */ + const std_cxx1x::function copier; + }; } @@ -731,15 +793,15 @@ namespace WorkStream typename Iterator, typename ScratchData, typename CopyData> - void - run (const Iterator &begin, + void + run (const Iterator &begin, const typename identity::type &end, Worker worker, Copier copier, const ScratchData &sample_scratch_data, const CopyData &sample_copy_data, const unsigned int queue_length = 2*multithread_info.n_default_threads, - const unsigned int chunk_size = 8) + const unsigned int chunk_size = 8) { Assert (queue_length > 0, ExcMessage ("The queue length must be at least one, and preferably " @@ -760,14 +822,15 @@ namespace WorkStream // create the three stages of the // pipeline internal::IteratorRangeToItemStream - iterator_range_to_item_stream (begin, end, - queue_length, - chunk_size, - sample_scratch_data, - sample_copy_data); + iterator_range_to_item_stream (begin, end, + queue_length, + chunk_size, + sample_scratch_data, + sample_copy_data); + internal::Worker worker_filter (worker); - internal::Copier copier_filter (copier); + internal::Copier copier_filter (copier,true); // now create a pipeline from // these stages @@ -803,6 +866,222 @@ namespace WorkStream + /** + * This is the main function of the + * WorkStream concept, doing work as + * described in the introduction to this + * namespace. + * + * This is the function that can be used + * for worker and copier objects that are + * either pointers to non-member + * functions or objects that allow to be + * called with an operator(), for example + * objects created by std::bind. + * + * The argument passed as @p end must be + * convertible to the same type as + * @p begin, but doesn't have to be of the + * same type itself. This allows to write + * code like + * WorkStream().run(dof_handler.begin_active(), + * dof_handler.end(), ... where + * the first is of type + * DoFHandler::active_cell_iterator + * whereas the second is of type + * DoFHandler::raw_cell_iterator. + * + * The two data types + * ScratchData and + * CopyData need to have a + * working copy + * constructor. ScratchData + * is only used in the + * worker function, while + * CopyData is the object + * passed from the worker + * to the copier. + * + * The @p get_conflict_indices argument, is a function + * that given an iterator computes the conflict indices + * necessary for the graph_coloring. + * + * The @p queue_length argument indicates + * the number of items that can be live + * at any given time. Each item consists + * of @p chunk_size elements of the input + * stream that will be worked on by the + * worker and copier functions one after + * the other on the same thread. + * + * @note If your data objects are large, + * or their constructors are expensive, + * it is helpful to keep in mind + * that queue_length + * copies of the ScratchData + * object and + * queue_length*chunk_size + * copies of the CopyData + * object are generated. + */ + template + void + run (const Iterator &begin, + const typename identity::type &end, + Worker worker, + Copier copier, + const ScratchData &sample_scratch_data, + const CopyData &sample_copy_data, + std::vector (*get_conflict_indices) (const Iterator &), + const unsigned int queue_length = 2*multithread_info.n_default_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; + +#ifdef DEAL_II_WITH_THREADS + // color the graph + std::vector > coloring = graph_coloring::make_graph_coloring( + begin,end,get_conflict_indices); + + for (unsigned int color=0; color + iterator_range_to_item_stream (coloring[color].begin(), coloring[color].end(), + queue_length, + chunk_size, + sample_scratch_data, + sample_copy_data); + + internal::Worker worker_filter (worker); + internal::Copier copier_filter (copier,false); + + // 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 + + // 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) + { + if (static_cast >(worker)) + worker (i, scratch_data, copy_data); + if (static_cast > + (copier)) + copier (copy_data); + } +#endif + } + + + + /** + * This is the main function of the + * WorkStream concept, doing work as + * described in the introduction to this + * namespace. + * + * This is the function that can be + * used for worker and copier functions + * that are member functions of a class. + * + * The argument passed as @p end must be + * convertible to the same type as + * @p begin, but doesn't have to be of the + * same type itself. This allows to write + * code like + * WorkStream().run(dof_handler.begin_active(), + * dof_handler.end(), ... where + * the first is of type + * DoFHandler::active_cell_iterator + * whereas the second is of type + * DoFHandler::raw_cell_iterator. + * + * The @p queue_length argument indicates + * the number of items that can be live + * at any given time. Each item consists + * of @p chunk_size elements of the input + * stream that will be worked on by the + * worker and copier functions one after + * the other on the same thread. + * + * @note If your data objects are large, + * or their constructors are expensive, + * it is helpful to keep in mind + * that queue_length + * copies of the ScratchData + * object and + * queue_length*chunk_size + * copies of the CopyData + * object are generated. + */ + template + void + run (const Iterator &begin, + const typename identity::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*multithread_info.n_default_threads, + const unsigned int chunk_size = 8) + { + // forward to the other function + run (begin, end, + std_cxx1x::bind (worker, + std_cxx1x::ref (main_object), + std_cxx1x::_1, std_cxx1x::_2, std_cxx1x::_3), + std_cxx1x::bind (copier, + std_cxx1x::ref (main_object), + std_cxx1x::_1), + sample_scratch_data, + sample_copy_data, + queue_length, + chunk_size); + } + + + + /** * This is the main function of the * WorkStream concept, doing work as @@ -825,6 +1104,10 @@ namespace WorkStream * whereas the second is of type * DoFHandler::raw_cell_iterator. * + * The @p get_conflict_indices argument, is a function + * that given an iterator computes the conflict indices + * necessary for the graph_coloring. + * * The @p queue_length argument indicates * the number of items that can be live * at any given time. Each item consists @@ -857,6 +1140,7 @@ namespace WorkStream void (MainClass::*copier) (const CopyData &), const ScratchData &sample_scratch_data, const CopyData &sample_copy_data, + std::vector (MainClass::*get_conflict_indices)(const Iterator &), const unsigned int queue_length = 2*multithread_info.n_default_threads, const unsigned int chunk_size = 8) { @@ -870,6 +1154,7 @@ namespace WorkStream std_cxx1x::_1), sample_scratch_data, sample_copy_data, + get_conflict_indices, queue_length, chunk_size); } diff --git a/tests/base/graph_coloring_01.cc b/tests/base/graph_coloring_01.cc index 0dfbbe35c1..63f35366e9 100644 --- a/tests/base/graph_coloring_01.cc +++ b/tests/base/graph_coloring_01.cc @@ -35,7 +35,7 @@ template std::vector get_conflict_indices_cfem( - typename DoFHandler::active_cell_iterator &it) + typename DoFHandler::active_cell_iterator const &it) { std::vector local_dof_indices(it->get_fe().dofs_per_cell); it->get_dof_indices(local_dof_indices); diff --git a/tests/base/graph_coloring_02.cc b/tests/base/graph_coloring_02.cc index 845e251d0c..7079cc8ae6 100644 --- a/tests/base/graph_coloring_02.cc +++ b/tests/base/graph_coloring_02.cc @@ -35,7 +35,7 @@ template std::vector get_conflict_indices_cfem( - typename DoFHandler::active_cell_iterator &it) + typename DoFHandler::active_cell_iterator const &it) { std::vector local_dof_indices(it->get_fe().dofs_per_cell); it->get_dof_indices(local_dof_indices); diff --git a/tests/base/graph_coloring_03.cc b/tests/base/graph_coloring_03.cc index 9d0a0f9d8d..255e7dedb0 100644 --- a/tests/base/graph_coloring_03.cc +++ b/tests/base/graph_coloring_03.cc @@ -34,7 +34,7 @@ template std::vector get_conflict_indices_cfem( - typename hp::DoFHandler::active_cell_iterator &it) + typename hp::DoFHandler::active_cell_iterator const &it) { std::vector local_dof_indices(it->get_fe().dofs_per_cell); it->get_dof_indices(local_dof_indices);