oneapi/tbb/parallel_pipeline.h

0001 /*
0002     Copyright (c) 2005-2021 Intel Corporation
0003
0004     Licensed under the Apache License, Version 2.0 (the "License");
0005     you may not use this file except in compliance with the License.
0006     You may obtain a copy of the License at
0007
0008         http://www.apache.org/licenses/LICENSE-2.0
0009
0010     Unless required by applicable law or agreed to in writing, software
0011     distributed under the License is distributed on an "AS IS" BASIS,
0012     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
0013     See the License for the specific language governing permissions and
0014     limitations under the License.
0015 */
0016
0017 #ifndef __TBB_parallel_pipeline_H
0018 #define __TBB_parallel_pipeline_H
0019
0020 #include "detail/_pipeline_filters.h"
0021 #include "detail/_config.h"
0022 #include "detail/_namespace_injection.h"
0023 #include "task_group.h"
0024
0025 #include <cstddef>
0026 #include <atomic>
0027 #include <type_traits>
0028
0029 namespace tbb {
0030 namespace detail {
0031
0032 namespace r1 {
0033 TBB_EXPORT void __TBB_EXPORTED_FUNC parallel_pipeline(task_group_context&, std::size_t, const d1::filter_node&);
0034 }
0035
0036 namespace d1 {
0037
0038 enum class filter_mode : unsigned int
0039 {
0040     //! processes multiple items in parallel and in no particular order
0041     parallel = base_filter::filter_is_out_of_order,
0042     //! processes items one at a time; all such filters process items in the same order
0043     serial_in_order =  base_filter::filter_is_serial,
0044     //! processes items one at a time and in no particular order
0045     serial_out_of_order = base_filter::filter_is_serial | base_filter::filter_is_out_of_order
0046 };
0047 //! Class representing a chain of type-safe pipeline filters
0048 /** @ingroup algorithms */
0049 template<typename InputType, typename OutputType>
0050 class filter {
0051     filter_node_ptr my_root;
0052     filter( filter_node_ptr root ) : my_root(root) {}
0053     friend void parallel_pipeline( size_t, const filter<void,void>&, task_group_context& );
0054     template<typename T_, typename U_, typename Body>
0055     friend filter<T_,U_> make_filter( filter_mode, const Body& );
0056     template<typename T_, typename V_, typename U_>
0057     friend filter<T_,U_> operator&( const filter<T_,V_>&, const filter<V_,U_>& );
0058 public:
0059     filter() = default;
0060     filter( const filter& rhs ) : my_root(rhs.my_root) {}
0061     filter( filter&& rhs ) : my_root(std::move(rhs.my_root)) {}
0062
0063     void operator=(const filter& rhs) {
0064         my_root = rhs.my_root;
0065     }
0066     void operator=( filter&& rhs ) {
0067         my_root = std::move(rhs.my_root);
0068     }
0069
0070     template<typename Body>
0071     filter( filter_mode mode, const Body& body ) :
0072         my_root( new(r1::allocate_memory(sizeof(filter_node_leaf<InputType, OutputType, Body>)))
0073                     filter_node_leaf<InputType, OutputType, Body>(static_cast<unsigned int>(mode), body) ) {
0074     }
0075
0076     filter& operator&=( const filter<OutputType,OutputType>& right ) {
0077         *this = *this & right;
0078         return *this;
0079     }
0080
0081     void clear() {
0082         // Like operator= with filter() on right side.
0083         my_root = nullptr;
0084     }
0085 };
0086
0087 //! Create a filter to participate in parallel_pipeline
0088 /** @ingroup algorithms */
0089 template<typename InputType, typename OutputType, typename Body>
0090 filter<InputType, OutputType> make_filter( filter_mode mode, const Body& body ) {
0091     return filter_node_ptr( new(r1::allocate_memory(sizeof(filter_node_leaf<InputType, OutputType, Body>)))
0092                                 filter_node_leaf<InputType, OutputType, Body>(static_cast<unsigned int>(mode), body) );
0093 }
0094
0095 //! Create a filter to participate in parallel_pipeline
0096 /** @ingroup algorithms */
0097 template<typename Body>
0098 filter<filter_input<Body>, filter_output<Body>> make_filter( filter_mode mode, const Body& body ) {
0099     return make_filter<filter_input<Body>, filter_output<Body>>(mode, body);
0100 }
0101
0102 //! Composition of filters left and right.
0103 /** @ingroup algorithms */
0104 template<typename T, typename V, typename U>
0105 filter<T,U> operator&( const filter<T,V>& left, const filter<V,U>& right ) {
0106     __TBB_ASSERT(left.my_root,"cannot use default-constructed filter as left argument of '&'");
0107     __TBB_ASSERT(right.my_root,"cannot use default-constructed filter as right argument of '&'");
0108     return filter_node_ptr( new (r1::allocate_memory(sizeof(filter_node))) filter_node(left.my_root,right.my_root) );
0109 }
0110
0111 #if __TBB_CPP17_DEDUCTION_GUIDES_PRESENT
0112 template<typename Body>
0113 filter(filter_mode, Body)
0114 ->filter<filter_input<Body>, filter_output<Body>>;
0115 #endif // __TBB_CPP17_DEDUCTION_GUIDES_PRESENT
0116
0117 //! Parallel pipeline over chain of filters with user-supplied context.
0118 /** @ingroup algorithms **/
0119 inline void parallel_pipeline(size_t max_number_of_live_tokens, const filter<void,void>& filter_chain, task_group_context& context) {
0120     r1::parallel_pipeline(context, max_number_of_live_tokens, *filter_chain.my_root);
0121 }
0122
0123 //! Parallel pipeline over chain of filters.
0124 /** @ingroup algorithms **/
0125 inline void parallel_pipeline(size_t max_number_of_live_tokens, const filter<void,void>& filter_chain) {
0126     task_group_context context;
0127     parallel_pipeline(max_number_of_live_tokens, filter_chain, context);
0128 }
0129
0130 //! Parallel pipeline over sequence of filters.
0131 /** @ingroup algorithms **/
0132 template<typename F1, typename F2, typename... FiltersContext>
0133 void parallel_pipeline(size_t max_number_of_live_tokens,
0134                               const F1& filter1,
0135                               const F2& filter2,
0136                               FiltersContext&&... filters) {
0137     parallel_pipeline(max_number_of_live_tokens, filter1 & filter2, std::forward<FiltersContext>(filters)...);
0138 }
0139
0140 } // namespace d1
0141 } // namespace detail
0142
0143 inline namespace v1
0144 {
0145 using detail::d1::parallel_pipeline;
0146 using detail::d1::filter;
0147 using detail::d1::make_filter;
0148 using detail::d1::filter_mode;
0149 using detail::d1::flow_control;
0150 }
0151 } // tbb
0152
0153 #endif /* __TBB_parallel_pipeline_H */