_flow_graph_join_impl.h 95 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002
  1. /*
  2. Copyright (c) 2005-2020 Intel Corporation
  3. Licensed under the Apache License, Version 2.0 (the "License");
  4. you may not use this file except in compliance with the License.
  5. You may obtain a copy of the License at
  6. http://www.apache.org/licenses/LICENSE-2.0
  7. Unless required by applicable law or agreed to in writing, software
  8. distributed under the License is distributed on an "AS IS" BASIS,
  9. WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  10. See the License for the specific language governing permissions and
  11. limitations under the License.
  12. */
  13. #ifndef __TBB__flow_graph_join_impl_H
  14. #define __TBB__flow_graph_join_impl_H
  15. #ifndef __TBB_flow_graph_H
  16. #error Do not #include this internal file directly; use public TBB headers instead.
  17. #endif
  18. namespace internal {
  19. struct forwarding_base : tbb::internal::no_assign {
  20. forwarding_base(graph &g) : graph_ref(g) {}
  21. virtual ~forwarding_base() {}
  22. // decrement_port_count may create a forwarding task. If we cannot handle the task
  23. // ourselves, ask decrement_port_count to deal with it.
  24. virtual task * decrement_port_count(bool handle_task) = 0;
  25. virtual void increment_port_count() = 0;
  26. // moved here so input ports can queue tasks
  27. graph& graph_ref;
  28. };
  29. // specialization that lets us keep a copy of the current_key for building results.
  30. // KeyType can be a reference type.
  31. template<typename KeyType>
  32. struct matching_forwarding_base : public forwarding_base {
  33. typedef typename tbb::internal::strip<KeyType>::type current_key_type;
  34. matching_forwarding_base(graph &g) : forwarding_base(g) { }
  35. virtual task * increment_key_count(current_key_type const & /*t*/, bool /*handle_task*/) = 0; // {return NULL;}
  36. current_key_type current_key; // so ports can refer to FE's desired items
  37. };
  38. template< int N >
  39. struct join_helper {
  40. template< typename TupleType, typename PortType >
  41. static inline void set_join_node_pointer(TupleType &my_input, PortType *port) {
  42. tbb::flow::get<N-1>( my_input ).set_join_node_pointer(port);
  43. join_helper<N-1>::set_join_node_pointer( my_input, port );
  44. }
  45. template< typename TupleType >
  46. static inline void consume_reservations( TupleType &my_input ) {
  47. tbb::flow::get<N-1>( my_input ).consume();
  48. join_helper<N-1>::consume_reservations( my_input );
  49. }
  50. template< typename TupleType >
  51. static inline void release_my_reservation( TupleType &my_input ) {
  52. tbb::flow::get<N-1>( my_input ).release();
  53. }
  54. template <typename TupleType>
  55. static inline void release_reservations( TupleType &my_input) {
  56. join_helper<N-1>::release_reservations(my_input);
  57. release_my_reservation(my_input);
  58. }
  59. template< typename InputTuple, typename OutputTuple >
  60. static inline bool reserve( InputTuple &my_input, OutputTuple &out) {
  61. if ( !tbb::flow::get<N-1>( my_input ).reserve( tbb::flow::get<N-1>( out ) ) ) return false;
  62. if ( !join_helper<N-1>::reserve( my_input, out ) ) {
  63. release_my_reservation( my_input );
  64. return false;
  65. }
  66. return true;
  67. }
  68. template<typename InputTuple, typename OutputTuple>
  69. static inline bool get_my_item( InputTuple &my_input, OutputTuple &out) {
  70. bool res = tbb::flow::get<N-1>(my_input).get_item(tbb::flow::get<N-1>(out) ); // may fail
  71. return join_helper<N-1>::get_my_item(my_input, out) && res; // do get on other inputs before returning
  72. }
  73. template<typename InputTuple, typename OutputTuple>
  74. static inline bool get_items(InputTuple &my_input, OutputTuple &out) {
  75. return get_my_item(my_input, out);
  76. }
  77. template<typename InputTuple>
  78. static inline void reset_my_port(InputTuple &my_input) {
  79. join_helper<N-1>::reset_my_port(my_input);
  80. tbb::flow::get<N-1>(my_input).reset_port();
  81. }
  82. template<typename InputTuple>
  83. static inline void reset_ports(InputTuple& my_input) {
  84. reset_my_port(my_input);
  85. }
  86. template<typename InputTuple, typename KeyFuncTuple>
  87. static inline void set_key_functors(InputTuple &my_input, KeyFuncTuple &my_key_funcs) {
  88. tbb::flow::get<N-1>(my_input).set_my_key_func(tbb::flow::get<N-1>(my_key_funcs));
  89. tbb::flow::get<N-1>(my_key_funcs) = NULL;
  90. join_helper<N-1>::set_key_functors(my_input, my_key_funcs);
  91. }
  92. template< typename KeyFuncTuple>
  93. static inline void copy_key_functors(KeyFuncTuple &my_inputs, KeyFuncTuple &other_inputs) {
  94. if(tbb::flow::get<N-1>(other_inputs).get_my_key_func()) {
  95. tbb::flow::get<N-1>(my_inputs).set_my_key_func(tbb::flow::get<N-1>(other_inputs).get_my_key_func()->clone());
  96. }
  97. join_helper<N-1>::copy_key_functors(my_inputs, other_inputs);
  98. }
  99. template<typename InputTuple>
  100. static inline void reset_inputs(InputTuple &my_input, reset_flags f) {
  101. join_helper<N-1>::reset_inputs(my_input, f);
  102. tbb::flow::get<N-1>(my_input).reset_receiver(f);
  103. }
  104. #if TBB_DEPRECATED_FLOW_NODE_EXTRACTION
  105. template<typename InputTuple>
  106. static inline void extract_inputs(InputTuple &my_input) {
  107. join_helper<N-1>::extract_inputs(my_input);
  108. tbb::flow::get<N-1>(my_input).extract_receiver();
  109. }
  110. #endif
  111. }; // join_helper<N>
  112. template< >
  113. struct join_helper<1> {
  114. template< typename TupleType, typename PortType >
  115. static inline void set_join_node_pointer(TupleType &my_input, PortType *port) {
  116. tbb::flow::get<0>( my_input ).set_join_node_pointer(port);
  117. }
  118. template< typename TupleType >
  119. static inline void consume_reservations( TupleType &my_input ) {
  120. tbb::flow::get<0>( my_input ).consume();
  121. }
  122. template< typename TupleType >
  123. static inline void release_my_reservation( TupleType &my_input ) {
  124. tbb::flow::get<0>( my_input ).release();
  125. }
  126. template<typename TupleType>
  127. static inline void release_reservations( TupleType &my_input) {
  128. release_my_reservation(my_input);
  129. }
  130. template< typename InputTuple, typename OutputTuple >
  131. static inline bool reserve( InputTuple &my_input, OutputTuple &out) {
  132. return tbb::flow::get<0>( my_input ).reserve( tbb::flow::get<0>( out ) );
  133. }
  134. template<typename InputTuple, typename OutputTuple>
  135. static inline bool get_my_item( InputTuple &my_input, OutputTuple &out) {
  136. return tbb::flow::get<0>(my_input).get_item(tbb::flow::get<0>(out));
  137. }
  138. template<typename InputTuple, typename OutputTuple>
  139. static inline bool get_items(InputTuple &my_input, OutputTuple &out) {
  140. return get_my_item(my_input, out);
  141. }
  142. template<typename InputTuple>
  143. static inline void reset_my_port(InputTuple &my_input) {
  144. tbb::flow::get<0>(my_input).reset_port();
  145. }
  146. template<typename InputTuple>
  147. static inline void reset_ports(InputTuple& my_input) {
  148. reset_my_port(my_input);
  149. }
  150. template<typename InputTuple, typename KeyFuncTuple>
  151. static inline void set_key_functors(InputTuple &my_input, KeyFuncTuple &my_key_funcs) {
  152. tbb::flow::get<0>(my_input).set_my_key_func(tbb::flow::get<0>(my_key_funcs));
  153. tbb::flow::get<0>(my_key_funcs) = NULL;
  154. }
  155. template< typename KeyFuncTuple>
  156. static inline void copy_key_functors(KeyFuncTuple &my_inputs, KeyFuncTuple &other_inputs) {
  157. if(tbb::flow::get<0>(other_inputs).get_my_key_func()) {
  158. tbb::flow::get<0>(my_inputs).set_my_key_func(tbb::flow::get<0>(other_inputs).get_my_key_func()->clone());
  159. }
  160. }
  161. template<typename InputTuple>
  162. static inline void reset_inputs(InputTuple &my_input, reset_flags f) {
  163. tbb::flow::get<0>(my_input).reset_receiver(f);
  164. }
  165. #if TBB_DEPRECATED_FLOW_NODE_EXTRACTION
  166. template<typename InputTuple>
  167. static inline void extract_inputs(InputTuple &my_input) {
  168. tbb::flow::get<0>(my_input).extract_receiver();
  169. }
  170. #endif
  171. }; // join_helper<1>
  172. //! The two-phase join port
  173. template< typename T >
  174. class reserving_port : public receiver<T> {
  175. public:
  176. typedef T input_type;
  177. typedef typename receiver<input_type>::predecessor_type predecessor_type;
  178. #if TBB_DEPRECATED_FLOW_NODE_EXTRACTION
  179. typedef typename receiver<input_type>::predecessor_list_type predecessor_list_type;
  180. typedef typename receiver<input_type>::built_predecessors_type built_predecessors_type;
  181. #endif
  182. private:
  183. // ----------- Aggregator ------------
  184. enum op_type { reg_pred, rem_pred, res_item, rel_res, con_res
  185. #if TBB_DEPRECATED_FLOW_NODE_EXTRACTION
  186. , add_blt_pred, del_blt_pred, blt_pred_cnt, blt_pred_cpy
  187. #endif
  188. };
  189. typedef reserving_port<T> class_type;
  190. class reserving_port_operation : public aggregated_operation<reserving_port_operation> {
  191. public:
  192. char type;
  193. union {
  194. T *my_arg;
  195. predecessor_type *my_pred;
  196. #if TBB_DEPRECATED_FLOW_NODE_EXTRACTION
  197. size_t cnt_val;
  198. predecessor_list_type *plist;
  199. #endif
  200. };
  201. reserving_port_operation(const T& e, op_type t) :
  202. type(char(t)), my_arg(const_cast<T*>(&e)) {}
  203. reserving_port_operation(const predecessor_type &s, op_type t) : type(char(t)),
  204. my_pred(const_cast<predecessor_type *>(&s)) {}
  205. reserving_port_operation(op_type t) : type(char(t)) {}
  206. };
  207. typedef internal::aggregating_functor<class_type, reserving_port_operation> handler_type;
  208. friend class internal::aggregating_functor<class_type, reserving_port_operation>;
  209. aggregator<handler_type, reserving_port_operation> my_aggregator;
  210. void handle_operations(reserving_port_operation* op_list) {
  211. reserving_port_operation *current;
  212. bool no_predecessors;
  213. while(op_list) {
  214. current = op_list;
  215. op_list = op_list->next;
  216. switch(current->type) {
  217. case reg_pred:
  218. no_predecessors = my_predecessors.empty();
  219. my_predecessors.add(*(current->my_pred));
  220. if ( no_predecessors ) {
  221. (void) my_join->decrement_port_count(true); // may try to forward
  222. }
  223. __TBB_store_with_release(current->status, SUCCEEDED);
  224. break;
  225. case rem_pred:
  226. my_predecessors.remove(*(current->my_pred));
  227. if(my_predecessors.empty()) my_join->increment_port_count();
  228. __TBB_store_with_release(current->status, SUCCEEDED);
  229. break;
  230. case res_item:
  231. if ( reserved ) {
  232. __TBB_store_with_release(current->status, FAILED);
  233. }
  234. else if ( my_predecessors.try_reserve( *(current->my_arg) ) ) {
  235. reserved = true;
  236. __TBB_store_with_release(current->status, SUCCEEDED);
  237. } else {
  238. if ( my_predecessors.empty() ) {
  239. my_join->increment_port_count();
  240. }
  241. __TBB_store_with_release(current->status, FAILED);
  242. }
  243. break;
  244. case rel_res:
  245. reserved = false;
  246. my_predecessors.try_release( );
  247. __TBB_store_with_release(current->status, SUCCEEDED);
  248. break;
  249. case con_res:
  250. reserved = false;
  251. my_predecessors.try_consume( );
  252. __TBB_store_with_release(current->status, SUCCEEDED);
  253. break;
  254. #if TBB_DEPRECATED_FLOW_NODE_EXTRACTION
  255. case add_blt_pred:
  256. my_predecessors.internal_add_built_predecessor(*(current->my_pred));
  257. __TBB_store_with_release(current->status, SUCCEEDED);
  258. break;
  259. case del_blt_pred:
  260. my_predecessors.internal_delete_built_predecessor(*(current->my_pred));
  261. __TBB_store_with_release(current->status, SUCCEEDED);
  262. break;
  263. case blt_pred_cnt:
  264. current->cnt_val = my_predecessors.predecessor_count();
  265. __TBB_store_with_release(current->status, SUCCEEDED);
  266. break;
  267. case blt_pred_cpy:
  268. my_predecessors.copy_predecessors(*(current->plist));
  269. __TBB_store_with_release(current->status, SUCCEEDED);
  270. break;
  271. #endif /* TBB_DEPRECATED_FLOW_NODE_EXTRACTION */
  272. }
  273. }
  274. }
  275. protected:
  276. template< typename R, typename B > friend class run_and_put_task;
  277. template<typename X, typename Y> friend class internal::broadcast_cache;
  278. template<typename X, typename Y> friend class internal::round_robin_cache;
  279. task *try_put_task( const T & ) __TBB_override {
  280. return NULL;
  281. }
  282. graph& graph_reference() const __TBB_override {
  283. return my_join->graph_ref;
  284. }
  285. public:
  286. //! Constructor
  287. reserving_port() : reserved(false) {
  288. my_join = NULL;
  289. my_predecessors.set_owner( this );
  290. my_aggregator.initialize_handler(handler_type(this));
  291. }
  292. // copy constructor
  293. reserving_port(const reserving_port& /* other */) : receiver<T>() {
  294. reserved = false;
  295. my_join = NULL;
  296. my_predecessors.set_owner( this );
  297. my_aggregator.initialize_handler(handler_type(this));
  298. }
  299. void set_join_node_pointer(forwarding_base *join) {
  300. my_join = join;
  301. }
  302. //! Add a predecessor
  303. bool register_predecessor( predecessor_type &src ) __TBB_override {
  304. reserving_port_operation op_data(src, reg_pred);
  305. my_aggregator.execute(&op_data);
  306. return op_data.status == SUCCEEDED;
  307. }
  308. //! Remove a predecessor
  309. bool remove_predecessor( predecessor_type &src ) __TBB_override {
  310. reserving_port_operation op_data(src, rem_pred);
  311. my_aggregator.execute(&op_data);
  312. return op_data.status == SUCCEEDED;
  313. }
  314. //! Reserve an item from the port
  315. bool reserve( T &v ) {
  316. reserving_port_operation op_data(v, res_item);
  317. my_aggregator.execute(&op_data);
  318. return op_data.status == SUCCEEDED;
  319. }
  320. //! Release the port
  321. void release( ) {
  322. reserving_port_operation op_data(rel_res);
  323. my_aggregator.execute(&op_data);
  324. }
  325. //! Complete use of the port
  326. void consume( ) {
  327. reserving_port_operation op_data(con_res);
  328. my_aggregator.execute(&op_data);
  329. }
  330. #if TBB_DEPRECATED_FLOW_NODE_EXTRACTION
  331. built_predecessors_type &built_predecessors() __TBB_override { return my_predecessors.built_predecessors(); }
  332. void internal_add_built_predecessor(predecessor_type &src) __TBB_override {
  333. reserving_port_operation op_data(src, add_blt_pred);
  334. my_aggregator.execute(&op_data);
  335. }
  336. void internal_delete_built_predecessor(predecessor_type &src) __TBB_override {
  337. reserving_port_operation op_data(src, del_blt_pred);
  338. my_aggregator.execute(&op_data);
  339. }
  340. size_t predecessor_count() __TBB_override {
  341. reserving_port_operation op_data(blt_pred_cnt);
  342. my_aggregator.execute(&op_data);
  343. return op_data.cnt_val;
  344. }
  345. void copy_predecessors(predecessor_list_type &l) __TBB_override {
  346. reserving_port_operation op_data(blt_pred_cpy);
  347. op_data.plist = &l;
  348. my_aggregator.execute(&op_data);
  349. }
  350. void extract_receiver() {
  351. my_predecessors.built_predecessors().receiver_extract(*this);
  352. }
  353. #endif /* TBB_DEPRECATED_FLOW_NODE_EXTRACTION */
  354. void reset_receiver( reset_flags f) __TBB_override {
  355. if(f & rf_clear_edges) my_predecessors.clear();
  356. else
  357. my_predecessors.reset();
  358. reserved = false;
  359. __TBB_ASSERT(!(f&rf_clear_edges) || my_predecessors.empty(), "port edges not removed");
  360. }
  361. private:
  362. #if __TBB_PREVIEW_FLOW_GRAPH_NODE_SET
  363. friend class get_graph_helper;
  364. #endif
  365. forwarding_base *my_join;
  366. reservable_predecessor_cache< T, null_mutex > my_predecessors;
  367. bool reserved;
  368. }; // reserving_port
  369. //! queueing join_port
  370. template<typename T>
  371. class queueing_port : public receiver<T>, public item_buffer<T> {
  372. public:
  373. typedef T input_type;
  374. typedef typename receiver<input_type>::predecessor_type predecessor_type;
  375. typedef queueing_port<T> class_type;
  376. #if TBB_DEPRECATED_FLOW_NODE_EXTRACTION
  377. typedef typename receiver<input_type>::built_predecessors_type built_predecessors_type;
  378. typedef typename receiver<input_type>::predecessor_list_type predecessor_list_type;
  379. #endif
  380. // ----------- Aggregator ------------
  381. private:
  382. enum op_type { get__item, res_port, try__put_task
  383. #if TBB_DEPRECATED_FLOW_NODE_EXTRACTION
  384. , add_blt_pred, del_blt_pred, blt_pred_cnt, blt_pred_cpy
  385. #endif
  386. };
  387. class queueing_port_operation : public aggregated_operation<queueing_port_operation> {
  388. public:
  389. char type;
  390. T my_val;
  391. T *my_arg;
  392. #if TBB_DEPRECATED_FLOW_NODE_EXTRACTION
  393. predecessor_type *pred;
  394. size_t cnt_val;
  395. predecessor_list_type *plist;
  396. #endif
  397. task * bypass_t;
  398. // constructor for value parameter
  399. queueing_port_operation(const T& e, op_type t) :
  400. type(char(t)), my_val(e)
  401. , bypass_t(NULL)
  402. {}
  403. // constructor for pointer parameter
  404. queueing_port_operation(const T* p, op_type t) :
  405. type(char(t)), my_arg(const_cast<T*>(p))
  406. , bypass_t(NULL)
  407. {}
  408. // constructor with no parameter
  409. queueing_port_operation(op_type t) : type(char(t))
  410. , bypass_t(NULL)
  411. {}
  412. };
  413. typedef internal::aggregating_functor<class_type, queueing_port_operation> handler_type;
  414. friend class internal::aggregating_functor<class_type, queueing_port_operation>;
  415. aggregator<handler_type, queueing_port_operation> my_aggregator;
  416. void handle_operations(queueing_port_operation* op_list) {
  417. queueing_port_operation *current;
  418. bool was_empty;
  419. while(op_list) {
  420. current = op_list;
  421. op_list = op_list->next;
  422. switch(current->type) {
  423. case try__put_task: {
  424. task *rtask = NULL;
  425. was_empty = this->buffer_empty();
  426. this->push_back(current->my_val);
  427. if (was_empty) rtask = my_join->decrement_port_count(false);
  428. else
  429. rtask = SUCCESSFULLY_ENQUEUED;
  430. current->bypass_t = rtask;
  431. __TBB_store_with_release(current->status, SUCCEEDED);
  432. }
  433. break;
  434. case get__item:
  435. if(!this->buffer_empty()) {
  436. *(current->my_arg) = this->front();
  437. __TBB_store_with_release(current->status, SUCCEEDED);
  438. }
  439. else {
  440. __TBB_store_with_release(current->status, FAILED);
  441. }
  442. break;
  443. case res_port:
  444. __TBB_ASSERT(this->my_item_valid(this->my_head), "No item to reset");
  445. this->destroy_front();
  446. if(this->my_item_valid(this->my_head)) {
  447. (void)my_join->decrement_port_count(true);
  448. }
  449. __TBB_store_with_release(current->status, SUCCEEDED);
  450. break;
  451. #if TBB_DEPRECATED_FLOW_NODE_EXTRACTION
  452. case add_blt_pred:
  453. my_built_predecessors.add_edge(*(current->pred));
  454. __TBB_store_with_release(current->status, SUCCEEDED);
  455. break;
  456. case del_blt_pred:
  457. my_built_predecessors.delete_edge(*(current->pred));
  458. __TBB_store_with_release(current->status, SUCCEEDED);
  459. break;
  460. case blt_pred_cnt:
  461. current->cnt_val = my_built_predecessors.edge_count();
  462. __TBB_store_with_release(current->status, SUCCEEDED);
  463. break;
  464. case blt_pred_cpy:
  465. my_built_predecessors.copy_edges(*(current->plist));
  466. __TBB_store_with_release(current->status, SUCCEEDED);
  467. break;
  468. #endif /* TBB_DEPRECATED_FLOW_NODE_EXTRACTION */
  469. }
  470. }
  471. }
  472. // ------------ End Aggregator ---------------
  473. protected:
  474. template< typename R, typename B > friend class run_and_put_task;
  475. template<typename X, typename Y> friend class internal::broadcast_cache;
  476. template<typename X, typename Y> friend class internal::round_robin_cache;
  477. task *try_put_task(const T &v) __TBB_override {
  478. queueing_port_operation op_data(v, try__put_task);
  479. my_aggregator.execute(&op_data);
  480. __TBB_ASSERT(op_data.status == SUCCEEDED || !op_data.bypass_t, "inconsistent return from aggregator");
  481. if(!op_data.bypass_t) return SUCCESSFULLY_ENQUEUED;
  482. return op_data.bypass_t;
  483. }
  484. graph& graph_reference() const __TBB_override {
  485. return my_join->graph_ref;
  486. }
  487. public:
  488. //! Constructor
  489. queueing_port() : item_buffer<T>() {
  490. my_join = NULL;
  491. my_aggregator.initialize_handler(handler_type(this));
  492. }
  493. //! copy constructor
  494. queueing_port(const queueing_port& /* other */) : receiver<T>(), item_buffer<T>() {
  495. my_join = NULL;
  496. my_aggregator.initialize_handler(handler_type(this));
  497. }
  498. //! record parent for tallying available items
  499. void set_join_node_pointer(forwarding_base *join) {
  500. my_join = join;
  501. }
  502. bool get_item( T &v ) {
  503. queueing_port_operation op_data(&v, get__item);
  504. my_aggregator.execute(&op_data);
  505. return op_data.status == SUCCEEDED;
  506. }
  507. // reset_port is called when item is accepted by successor, but
  508. // is initiated by join_node.
  509. void reset_port() {
  510. queueing_port_operation op_data(res_port);
  511. my_aggregator.execute(&op_data);
  512. return;
  513. }
  514. #if TBB_DEPRECATED_FLOW_NODE_EXTRACTION
  515. built_predecessors_type &built_predecessors() __TBB_override { return my_built_predecessors; }
  516. void internal_add_built_predecessor(predecessor_type &p) __TBB_override {
  517. queueing_port_operation op_data(add_blt_pred);
  518. op_data.pred = &p;
  519. my_aggregator.execute(&op_data);
  520. }
  521. void internal_delete_built_predecessor(predecessor_type &p) __TBB_override {
  522. queueing_port_operation op_data(del_blt_pred);
  523. op_data.pred = &p;
  524. my_aggregator.execute(&op_data);
  525. }
  526. size_t predecessor_count() __TBB_override {
  527. queueing_port_operation op_data(blt_pred_cnt);
  528. my_aggregator.execute(&op_data);
  529. return op_data.cnt_val;
  530. }
  531. void copy_predecessors(predecessor_list_type &l) __TBB_override {
  532. queueing_port_operation op_data(blt_pred_cpy);
  533. op_data.plist = &l;
  534. my_aggregator.execute(&op_data);
  535. }
  536. void extract_receiver() {
  537. item_buffer<T>::reset();
  538. my_built_predecessors.receiver_extract(*this);
  539. }
  540. #endif /* TBB_DEPRECATED_FLOW_NODE_EXTRACTION */
  541. void reset_receiver(reset_flags f) __TBB_override {
  542. tbb::internal::suppress_unused_warning(f);
  543. item_buffer<T>::reset();
  544. #if TBB_DEPRECATED_FLOW_NODE_EXTRACTION
  545. if (f & rf_clear_edges)
  546. my_built_predecessors.clear();
  547. #endif
  548. }
  549. private:
  550. #if __TBB_PREVIEW_FLOW_GRAPH_NODE_SET
  551. friend class get_graph_helper;
  552. #endif
  553. forwarding_base *my_join;
  554. #if TBB_DEPRECATED_FLOW_NODE_EXTRACTION
  555. edge_container<predecessor_type> my_built_predecessors;
  556. #endif
  557. }; // queueing_port
  558. #include "_flow_graph_tagged_buffer_impl.h"
  559. template<typename K>
  560. struct count_element {
  561. K my_key;
  562. size_t my_value;
  563. };
  564. // method to access the key in the counting table
  565. // the ref has already been removed from K
  566. template< typename K >
  567. struct key_to_count_functor {
  568. typedef count_element<K> table_item_type;
  569. const K& operator()(const table_item_type& v) { return v.my_key; }
  570. };
  571. // the ports can have only one template parameter. We wrap the types needed in
  572. // a traits type
  573. template< class TraitsType >
  574. class key_matching_port :
  575. public receiver<typename TraitsType::T>,
  576. public hash_buffer< typename TraitsType::K, typename TraitsType::T, typename TraitsType::TtoK,
  577. typename TraitsType::KHash > {
  578. public:
  579. typedef TraitsType traits;
  580. typedef key_matching_port<traits> class_type;
  581. typedef typename TraitsType::T input_type;
  582. typedef typename TraitsType::K key_type;
  583. typedef typename tbb::internal::strip<key_type>::type noref_key_type;
  584. typedef typename receiver<input_type>::predecessor_type predecessor_type;
  585. typedef typename TraitsType::TtoK type_to_key_func_type;
  586. typedef typename TraitsType::KHash hash_compare_type;
  587. typedef hash_buffer< key_type, input_type, type_to_key_func_type, hash_compare_type > buffer_type;
  588. #if TBB_DEPRECATED_FLOW_NODE_EXTRACTION
  589. typedef typename receiver<input_type>::built_predecessors_type built_predecessors_type;
  590. typedef typename receiver<input_type>::predecessor_list_type predecessor_list_type;
  591. #endif
  592. private:
  593. // ----------- Aggregator ------------
  594. private:
  595. enum op_type { try__put, get__item, res_port
  596. #if TBB_DEPRECATED_FLOW_NODE_EXTRACTION
  597. , add_blt_pred, del_blt_pred, blt_pred_cnt, blt_pred_cpy
  598. #endif
  599. };
  600. class key_matching_port_operation : public aggregated_operation<key_matching_port_operation> {
  601. public:
  602. char type;
  603. input_type my_val;
  604. input_type *my_arg;
  605. #if TBB_DEPRECATED_FLOW_NODE_EXTRACTION
  606. predecessor_type *pred;
  607. size_t cnt_val;
  608. predecessor_list_type *plist;
  609. #endif
  610. // constructor for value parameter
  611. key_matching_port_operation(const input_type& e, op_type t) :
  612. type(char(t)), my_val(e) {}
  613. // constructor for pointer parameter
  614. key_matching_port_operation(const input_type* p, op_type t) :
  615. type(char(t)), my_arg(const_cast<input_type*>(p)) {}
  616. // constructor with no parameter
  617. key_matching_port_operation(op_type t) : type(char(t)) {}
  618. };
  619. typedef internal::aggregating_functor<class_type, key_matching_port_operation> handler_type;
  620. friend class internal::aggregating_functor<class_type, key_matching_port_operation>;
  621. aggregator<handler_type, key_matching_port_operation> my_aggregator;
  622. void handle_operations(key_matching_port_operation* op_list) {
  623. key_matching_port_operation *current;
  624. while(op_list) {
  625. current = op_list;
  626. op_list = op_list->next;
  627. switch(current->type) {
  628. case try__put: {
  629. bool was_inserted = this->insert_with_key(current->my_val);
  630. // return failure if a duplicate insertion occurs
  631. __TBB_store_with_release(current->status, was_inserted ? SUCCEEDED : FAILED);
  632. }
  633. break;
  634. case get__item:
  635. // use current_key from FE for item
  636. if(!this->find_with_key(my_join->current_key, *(current->my_arg))) {
  637. __TBB_ASSERT(false, "Failed to find item corresponding to current_key.");
  638. }
  639. __TBB_store_with_release(current->status, SUCCEEDED);
  640. break;
  641. case res_port:
  642. // use current_key from FE for item
  643. this->delete_with_key(my_join->current_key);
  644. __TBB_store_with_release(current->status, SUCCEEDED);
  645. break;
  646. #if TBB_DEPRECATED_FLOW_NODE_EXTRACTION
  647. case add_blt_pred:
  648. my_built_predecessors.add_edge(*(current->pred));
  649. __TBB_store_with_release(current->status, SUCCEEDED);
  650. break;
  651. case del_blt_pred:
  652. my_built_predecessors.delete_edge(*(current->pred));
  653. __TBB_store_with_release(current->status, SUCCEEDED);
  654. break;
  655. case blt_pred_cnt:
  656. current->cnt_val = my_built_predecessors.edge_count();
  657. __TBB_store_with_release(current->status, SUCCEEDED);
  658. break;
  659. case blt_pred_cpy:
  660. my_built_predecessors.copy_edges(*(current->plist));
  661. __TBB_store_with_release(current->status, SUCCEEDED);
  662. break;
  663. #endif
  664. }
  665. }
  666. }
  667. // ------------ End Aggregator ---------------
  668. protected:
  669. template< typename R, typename B > friend class run_and_put_task;
  670. template<typename X, typename Y> friend class internal::broadcast_cache;
  671. template<typename X, typename Y> friend class internal::round_robin_cache;
  672. task *try_put_task(const input_type& v) __TBB_override {
  673. key_matching_port_operation op_data(v, try__put);
  674. task *rtask = NULL;
  675. my_aggregator.execute(&op_data);
  676. if(op_data.status == SUCCEEDED) {
  677. rtask = my_join->increment_key_count((*(this->get_key_func()))(v), false); // may spawn
  678. // rtask has to reflect the return status of the try_put
  679. if(!rtask) rtask = SUCCESSFULLY_ENQUEUED;
  680. }
  681. return rtask;
  682. }
  683. graph& graph_reference() const __TBB_override {
  684. return my_join->graph_ref;
  685. }
  686. public:
  687. key_matching_port() : receiver<input_type>(), buffer_type() {
  688. my_join = NULL;
  689. my_aggregator.initialize_handler(handler_type(this));
  690. }
  691. // copy constructor
  692. key_matching_port(const key_matching_port& /*other*/) : receiver<input_type>(), buffer_type() {
  693. my_join = NULL;
  694. my_aggregator.initialize_handler(handler_type(this));
  695. }
  696. ~key_matching_port() { }
  697. void set_join_node_pointer(forwarding_base *join) {
  698. my_join = dynamic_cast<matching_forwarding_base<key_type>*>(join);
  699. }
  700. void set_my_key_func(type_to_key_func_type *f) { this->set_key_func(f); }
  701. type_to_key_func_type* get_my_key_func() { return this->get_key_func(); }
  702. bool get_item( input_type &v ) {
  703. // aggregator uses current_key from FE for Key
  704. key_matching_port_operation op_data(&v, get__item);
  705. my_aggregator.execute(&op_data);
  706. return op_data.status == SUCCEEDED;
  707. }
  708. #if TBB_DEPRECATED_FLOW_NODE_EXTRACTION
  709. built_predecessors_type &built_predecessors() __TBB_override { return my_built_predecessors; }
  710. void internal_add_built_predecessor(predecessor_type &p) __TBB_override {
  711. key_matching_port_operation op_data(add_blt_pred);
  712. op_data.pred = &p;
  713. my_aggregator.execute(&op_data);
  714. }
  715. void internal_delete_built_predecessor(predecessor_type &p) __TBB_override {
  716. key_matching_port_operation op_data(del_blt_pred);
  717. op_data.pred = &p;
  718. my_aggregator.execute(&op_data);
  719. }
  720. size_t predecessor_count() __TBB_override {
  721. key_matching_port_operation op_data(blt_pred_cnt);
  722. my_aggregator.execute(&op_data);
  723. return op_data.cnt_val;
  724. }
  725. void copy_predecessors(predecessor_list_type &l) __TBB_override {
  726. key_matching_port_operation op_data(blt_pred_cpy);
  727. op_data.plist = &l;
  728. my_aggregator.execute(&op_data);
  729. }
  730. #endif
  731. // reset_port is called when item is accepted by successor, but
  732. // is initiated by join_node.
  733. void reset_port() {
  734. key_matching_port_operation op_data(res_port);
  735. my_aggregator.execute(&op_data);
  736. return;
  737. }
  738. #if TBB_DEPRECATED_FLOW_NODE_EXTRACTION
  739. void extract_receiver() {
  740. buffer_type::reset();
  741. my_built_predecessors.receiver_extract(*this);
  742. }
  743. #endif
  744. void reset_receiver(reset_flags f ) __TBB_override {
  745. tbb::internal::suppress_unused_warning(f);
  746. buffer_type::reset();
  747. #if TBB_DEPRECATED_FLOW_NODE_EXTRACTION
  748. if (f & rf_clear_edges)
  749. my_built_predecessors.clear();
  750. #endif
  751. }
  752. private:
  753. // my_join forwarding base used to count number of inputs that
  754. // received key.
  755. matching_forwarding_base<key_type> *my_join;
  756. #if TBB_DEPRECATED_FLOW_NODE_EXTRACTION
  757. edge_container<predecessor_type> my_built_predecessors;
  758. #endif
  759. }; // key_matching_port
  760. using namespace graph_policy_namespace;
  761. template<typename JP, typename InputTuple, typename OutputTuple>
  762. class join_node_base;
  763. //! join_node_FE : implements input port policy
  764. template<typename JP, typename InputTuple, typename OutputTuple>
  765. class join_node_FE;
  766. template<typename InputTuple, typename OutputTuple>
  767. class join_node_FE<reserving, InputTuple, OutputTuple> : public forwarding_base {
  768. public:
  769. static const int N = tbb::flow::tuple_size<OutputTuple>::value;
  770. typedef OutputTuple output_type;
  771. typedef InputTuple input_type;
  772. typedef join_node_base<reserving, InputTuple, OutputTuple> base_node_type; // for forwarding
  773. join_node_FE(graph &g) : forwarding_base(g), my_node(NULL) {
  774. ports_with_no_inputs = N;
  775. join_helper<N>::set_join_node_pointer(my_inputs, this);
  776. }
  777. join_node_FE(const join_node_FE& other) : forwarding_base((other.forwarding_base::graph_ref)), my_node(NULL) {
  778. ports_with_no_inputs = N;
  779. join_helper<N>::set_join_node_pointer(my_inputs, this);
  780. }
  781. void set_my_node(base_node_type *new_my_node) { my_node = new_my_node; }
  782. void increment_port_count() __TBB_override {
  783. ++ports_with_no_inputs;
  784. }
  785. // if all input_ports have predecessors, spawn forward to try and consume tuples
  786. task * decrement_port_count(bool handle_task) __TBB_override {
  787. if(ports_with_no_inputs.fetch_and_decrement() == 1) {
  788. if(internal::is_graph_active(this->graph_ref)) {
  789. task *rtask = new ( task::allocate_additional_child_of( *(this->graph_ref.root_task()) ) )
  790. forward_task_bypass<base_node_type>(*my_node);
  791. if(!handle_task) return rtask;
  792. internal::spawn_in_graph_arena(this->graph_ref, *rtask);
  793. }
  794. }
  795. return NULL;
  796. }
  797. input_type &input_ports() { return my_inputs; }
  798. protected:
  799. void reset( reset_flags f) {
  800. // called outside of parallel contexts
  801. ports_with_no_inputs = N;
  802. join_helper<N>::reset_inputs(my_inputs, f);
  803. }
  804. #if TBB_DEPRECATED_FLOW_NODE_EXTRACTION
  805. void extract( ) {
  806. // called outside of parallel contexts
  807. ports_with_no_inputs = N;
  808. join_helper<N>::extract_inputs(my_inputs);
  809. }
  810. #endif
  811. // all methods on input ports should be called under mutual exclusion from join_node_base.
  812. bool tuple_build_may_succeed() {
  813. return !ports_with_no_inputs;
  814. }
  815. bool try_to_make_tuple(output_type &out) {
  816. if(ports_with_no_inputs) return false;
  817. return join_helper<N>::reserve(my_inputs, out);
  818. }
  819. void tuple_accepted() {
  820. join_helper<N>::consume_reservations(my_inputs);
  821. }
  822. void tuple_rejected() {
  823. join_helper<N>::release_reservations(my_inputs);
  824. }
  825. input_type my_inputs;
  826. base_node_type *my_node;
  827. atomic<size_t> ports_with_no_inputs;
  828. }; // join_node_FE<reserving, ... >
  829. template<typename InputTuple, typename OutputTuple>
  830. class join_node_FE<queueing, InputTuple, OutputTuple> : public forwarding_base {
  831. public:
  832. static const int N = tbb::flow::tuple_size<OutputTuple>::value;
  833. typedef OutputTuple output_type;
  834. typedef InputTuple input_type;
  835. typedef join_node_base<queueing, InputTuple, OutputTuple> base_node_type; // for forwarding
  836. join_node_FE(graph &g) : forwarding_base(g), my_node(NULL) {
  837. ports_with_no_items = N;
  838. join_helper<N>::set_join_node_pointer(my_inputs, this);
  839. }
  840. join_node_FE(const join_node_FE& other) : forwarding_base((other.forwarding_base::graph_ref)), my_node(NULL) {
  841. ports_with_no_items = N;
  842. join_helper<N>::set_join_node_pointer(my_inputs, this);
  843. }
  844. // needed for forwarding
  845. void set_my_node(base_node_type *new_my_node) { my_node = new_my_node; }
  846. void reset_port_count() {
  847. ports_with_no_items = N;
  848. }
  849. // if all input_ports have items, spawn forward to try and consume tuples
  850. task * decrement_port_count(bool handle_task) __TBB_override
  851. {
  852. if(ports_with_no_items.fetch_and_decrement() == 1) {
  853. if(internal::is_graph_active(this->graph_ref)) {
  854. task *rtask = new ( task::allocate_additional_child_of( *(this->graph_ref.root_task()) ) )
  855. forward_task_bypass <base_node_type>(*my_node);
  856. if(!handle_task) return rtask;
  857. internal::spawn_in_graph_arena(this->graph_ref, *rtask);
  858. }
  859. }
  860. return NULL;
  861. }
  862. void increment_port_count() __TBB_override { __TBB_ASSERT(false, NULL); } // should never be called
  863. input_type &input_ports() { return my_inputs; }
  864. protected:
  865. void reset( reset_flags f) {
  866. reset_port_count();
  867. join_helper<N>::reset_inputs(my_inputs, f );
  868. }
  869. #if TBB_DEPRECATED_FLOW_NODE_EXTRACTION
  870. void extract() {
  871. reset_port_count();
  872. join_helper<N>::extract_inputs(my_inputs);
  873. }
  874. #endif
  875. // all methods on input ports should be called under mutual exclusion from join_node_base.
  876. bool tuple_build_may_succeed() {
  877. return !ports_with_no_items;
  878. }
  879. bool try_to_make_tuple(output_type &out) {
  880. if(ports_with_no_items) return false;
  881. return join_helper<N>::get_items(my_inputs, out);
  882. }
  883. void tuple_accepted() {
  884. reset_port_count();
  885. join_helper<N>::reset_ports(my_inputs);
  886. }
  887. void tuple_rejected() {
  888. // nothing to do.
  889. }
  890. input_type my_inputs;
  891. base_node_type *my_node;
  892. atomic<size_t> ports_with_no_items;
  893. }; // join_node_FE<queueing, ...>
  894. // key_matching join front-end.
  895. template<typename InputTuple, typename OutputTuple, typename K, typename KHash>
  896. class join_node_FE<key_matching<K,KHash>, InputTuple, OutputTuple> : public matching_forwarding_base<K>,
  897. // buffer of key value counts
  898. public hash_buffer< // typedefed below to key_to_count_buffer_type
  899. typename tbb::internal::strip<K>::type&, // force ref type on K
  900. count_element<typename tbb::internal::strip<K>::type>,
  901. internal::type_to_key_function_body<
  902. count_element<typename tbb::internal::strip<K>::type>,
  903. typename tbb::internal::strip<K>::type& >,
  904. KHash >,
  905. // buffer of output items
  906. public item_buffer<OutputTuple> {
  907. public:
  908. static const int N = tbb::flow::tuple_size<OutputTuple>::value;
  909. typedef OutputTuple output_type;
  910. typedef InputTuple input_type;
  911. typedef K key_type;
  912. typedef typename tbb::internal::strip<key_type>::type unref_key_type;
  913. typedef KHash key_hash_compare;
  914. // must use K without ref.
  915. typedef count_element<unref_key_type> count_element_type;
  916. // method that lets us refer to the key of this type.
  917. typedef key_to_count_functor<unref_key_type> key_to_count_func;
  918. typedef internal::type_to_key_function_body< count_element_type, unref_key_type&> TtoK_function_body_type;
  919. typedef internal::type_to_key_function_body_leaf<count_element_type, unref_key_type&, key_to_count_func> TtoK_function_body_leaf_type;
  920. // this is the type of the special table that keeps track of the number of discrete
  921. // elements corresponding to each key that we've seen.
  922. typedef hash_buffer< unref_key_type&, count_element_type, TtoK_function_body_type, key_hash_compare >
  923. key_to_count_buffer_type;
  924. typedef item_buffer<output_type> output_buffer_type;
  925. typedef join_node_base<key_matching<key_type,key_hash_compare>, InputTuple, OutputTuple> base_node_type; // for forwarding
  926. typedef matching_forwarding_base<key_type> forwarding_base_type;
  927. // ----------- Aggregator ------------
  928. // the aggregator is only needed to serialize the access to the hash table.
  929. // and the output_buffer_type base class
  930. private:
  931. enum op_type { res_count, inc_count, may_succeed, try_make };
  932. typedef join_node_FE<key_matching<key_type,key_hash_compare>, InputTuple, OutputTuple> class_type;
  933. class key_matching_FE_operation : public aggregated_operation<key_matching_FE_operation> {
  934. public:
  935. char type;
  936. unref_key_type my_val;
  937. output_type* my_output;
  938. task *bypass_t;
  939. bool enqueue_task;
  940. // constructor for value parameter
  941. key_matching_FE_operation(const unref_key_type& e , bool q_task , op_type t) : type(char(t)), my_val(e),
  942. my_output(NULL), bypass_t(NULL), enqueue_task(q_task) {}
  943. key_matching_FE_operation(output_type *p, op_type t) : type(char(t)), my_output(p), bypass_t(NULL),
  944. enqueue_task(true) {}
  945. // constructor with no parameter
  946. key_matching_FE_operation(op_type t) : type(char(t)), my_output(NULL), bypass_t(NULL), enqueue_task(true) {}
  947. };
  948. typedef internal::aggregating_functor<class_type, key_matching_FE_operation> handler_type;
  949. friend class internal::aggregating_functor<class_type, key_matching_FE_operation>;
  950. aggregator<handler_type, key_matching_FE_operation> my_aggregator;
  951. // called from aggregator, so serialized
  952. // returns a task pointer if the a task would have been enqueued but we asked that
  953. // it be returned. Otherwise returns NULL.
  954. task * fill_output_buffer(unref_key_type &t, bool should_enqueue, bool handle_task) {
  955. output_type l_out;
  956. task *rtask = NULL;
  957. bool do_fwd = should_enqueue && this->buffer_empty() && internal::is_graph_active(this->graph_ref);
  958. this->current_key = t;
  959. this->delete_with_key(this->current_key); // remove the key
  960. if(join_helper<N>::get_items(my_inputs, l_out)) { // <== call back
  961. this->push_back(l_out);
  962. if(do_fwd) { // we enqueue if receiving an item from predecessor, not if successor asks for item
  963. rtask = new ( task::allocate_additional_child_of( *(this->graph_ref.root_task()) ) )
  964. forward_task_bypass<base_node_type>(*my_node);
  965. if(handle_task) {
  966. internal::spawn_in_graph_arena(this->graph_ref, *rtask);
  967. rtask = NULL;
  968. }
  969. do_fwd = false;
  970. }
  971. // retire the input values
  972. join_helper<N>::reset_ports(my_inputs); // <== call back
  973. }
  974. else {
  975. __TBB_ASSERT(false, "should have had something to push");
  976. }
  977. return rtask;
  978. }
  979. void handle_operations(key_matching_FE_operation* op_list) {
  980. key_matching_FE_operation *current;
  981. while(op_list) {
  982. current = op_list;
  983. op_list = op_list->next;
  984. switch(current->type) {
  985. case res_count: // called from BE
  986. {
  987. this->destroy_front();
  988. __TBB_store_with_release(current->status, SUCCEEDED);
  989. }
  990. break;
  991. case inc_count: { // called from input ports
  992. count_element_type *p = 0;
  993. unref_key_type &t = current->my_val;
  994. bool do_enqueue = current->enqueue_task;
  995. if(!(this->find_ref_with_key(t,p))) {
  996. count_element_type ev;
  997. ev.my_key = t;
  998. ev.my_value = 0;
  999. this->insert_with_key(ev);
  1000. if(!(this->find_ref_with_key(t,p))) {
  1001. __TBB_ASSERT(false, "should find key after inserting it");
  1002. }
  1003. }
  1004. if(++(p->my_value) == size_t(N)) {
  1005. task *rtask = fill_output_buffer(t, true, do_enqueue);
  1006. __TBB_ASSERT(!rtask || !do_enqueue, "task should not be returned");
  1007. current->bypass_t = rtask;
  1008. }
  1009. }
  1010. __TBB_store_with_release(current->status, SUCCEEDED);
  1011. break;
  1012. case may_succeed: // called from BE
  1013. __TBB_store_with_release(current->status, this->buffer_empty() ? FAILED : SUCCEEDED);
  1014. break;
  1015. case try_make: // called from BE
  1016. if(this->buffer_empty()) {
  1017. __TBB_store_with_release(current->status, FAILED);
  1018. }
  1019. else {
  1020. *(current->my_output) = this->front();
  1021. __TBB_store_with_release(current->status, SUCCEEDED);
  1022. }
  1023. break;
  1024. }
  1025. }
  1026. }
  1027. // ------------ End Aggregator ---------------
  1028. public:
  1029. template<typename FunctionTuple>
  1030. join_node_FE(graph &g, FunctionTuple &TtoK_funcs) : forwarding_base_type(g), my_node(NULL) {
  1031. join_helper<N>::set_join_node_pointer(my_inputs, this);
  1032. join_helper<N>::set_key_functors(my_inputs, TtoK_funcs);
  1033. my_aggregator.initialize_handler(handler_type(this));
  1034. TtoK_function_body_type *cfb = new TtoK_function_body_leaf_type(key_to_count_func());
  1035. this->set_key_func(cfb);
  1036. }
  1037. join_node_FE(const join_node_FE& other) : forwarding_base_type((other.forwarding_base_type::graph_ref)), key_to_count_buffer_type(),
  1038. output_buffer_type() {
  1039. my_node = NULL;
  1040. join_helper<N>::set_join_node_pointer(my_inputs, this);
  1041. join_helper<N>::copy_key_functors(my_inputs, const_cast<input_type &>(other.my_inputs));
  1042. my_aggregator.initialize_handler(handler_type(this));
  1043. TtoK_function_body_type *cfb = new TtoK_function_body_leaf_type(key_to_count_func());
  1044. this->set_key_func(cfb);
  1045. }
  1046. // needed for forwarding
  1047. void set_my_node(base_node_type *new_my_node) { my_node = new_my_node; }
  1048. void reset_port_count() { // called from BE
  1049. key_matching_FE_operation op_data(res_count);
  1050. my_aggregator.execute(&op_data);
  1051. return;
  1052. }
  1053. // if all input_ports have items, spawn forward to try and consume tuples
  1054. // return a task if we are asked and did create one.
  1055. task *increment_key_count(unref_key_type const & t, bool handle_task) __TBB_override { // called from input_ports
  1056. key_matching_FE_operation op_data(t, handle_task, inc_count);
  1057. my_aggregator.execute(&op_data);
  1058. return op_data.bypass_t;
  1059. }
  1060. task *decrement_port_count(bool /*handle_task*/) __TBB_override { __TBB_ASSERT(false, NULL); return NULL; }
  1061. void increment_port_count() __TBB_override { __TBB_ASSERT(false, NULL); } // should never be called
  1062. input_type &input_ports() { return my_inputs; }
  1063. protected:
  1064. void reset( reset_flags f ) {
  1065. // called outside of parallel contexts
  1066. join_helper<N>::reset_inputs(my_inputs, f);
  1067. key_to_count_buffer_type::reset();
  1068. output_buffer_type::reset();
  1069. }
  1070. #if TBB_DEPRECATED_FLOW_NODE_EXTRACTION
  1071. void extract() {
  1072. // called outside of parallel contexts
  1073. join_helper<N>::extract_inputs(my_inputs);
  1074. key_to_count_buffer_type::reset(); // have to reset the tag counts
  1075. output_buffer_type::reset(); // also the queue of outputs
  1076. // my_node->current_tag = NO_TAG;
  1077. }
  1078. #endif
  1079. // all methods on input ports should be called under mutual exclusion from join_node_base.
  1080. bool tuple_build_may_succeed() { // called from back-end
  1081. key_matching_FE_operation op_data(may_succeed);
  1082. my_aggregator.execute(&op_data);
  1083. return op_data.status == SUCCEEDED;
  1084. }
  1085. // cannot lock while calling back to input_ports. current_key will only be set
  1086. // and reset under the aggregator, so it will remain consistent.
  1087. bool try_to_make_tuple(output_type &out) {
  1088. key_matching_FE_operation op_data(&out,try_make);
  1089. my_aggregator.execute(&op_data);
  1090. return op_data.status == SUCCEEDED;
  1091. }
  1092. void tuple_accepted() {
  1093. reset_port_count(); // reset current_key after ports reset.
  1094. }
  1095. void tuple_rejected() {
  1096. // nothing to do.
  1097. }
  1098. input_type my_inputs; // input ports
  1099. base_node_type *my_node;
  1100. }; // join_node_FE<key_matching<K,KHash>, InputTuple, OutputTuple>
  1101. //! join_node_base
  1102. template<typename JP, typename InputTuple, typename OutputTuple>
  1103. class join_node_base : public graph_node, public join_node_FE<JP, InputTuple, OutputTuple>,
  1104. public sender<OutputTuple> {
  1105. protected:
  1106. using graph_node::my_graph;
  1107. public:
  1108. typedef OutputTuple output_type;
  1109. typedef typename sender<output_type>::successor_type successor_type;
  1110. typedef join_node_FE<JP, InputTuple, OutputTuple> input_ports_type;
  1111. using input_ports_type::tuple_build_may_succeed;
  1112. using input_ports_type::try_to_make_tuple;
  1113. using input_ports_type::tuple_accepted;
  1114. using input_ports_type::tuple_rejected;
  1115. #if TBB_DEPRECATED_FLOW_NODE_EXTRACTION
  1116. typedef typename sender<output_type>::built_successors_type built_successors_type;
  1117. typedef typename sender<output_type>::successor_list_type successor_list_type;
  1118. #endif
  1119. private:
  1120. // ----------- Aggregator ------------
  1121. enum op_type { reg_succ, rem_succ, try__get, do_fwrd, do_fwrd_bypass
  1122. #if TBB_DEPRECATED_FLOW_NODE_EXTRACTION
  1123. , add_blt_succ, del_blt_succ, blt_succ_cnt, blt_succ_cpy
  1124. #endif
  1125. };
  1126. typedef join_node_base<JP,InputTuple,OutputTuple> class_type;
  1127. class join_node_base_operation : public aggregated_operation<join_node_base_operation> {
  1128. public:
  1129. char type;
  1130. union {
  1131. output_type *my_arg;
  1132. successor_type *my_succ;
  1133. #if TBB_DEPRECATED_FLOW_NODE_EXTRACTION
  1134. size_t cnt_val;
  1135. successor_list_type *slist;
  1136. #endif
  1137. };
  1138. task *bypass_t;
  1139. join_node_base_operation(const output_type& e, op_type t) : type(char(t)),
  1140. my_arg(const_cast<output_type*>(&e)), bypass_t(NULL) {}
  1141. join_node_base_operation(const successor_type &s, op_type t) : type(char(t)),
  1142. my_succ(const_cast<successor_type *>(&s)), bypass_t(NULL) {}
  1143. join_node_base_operation(op_type t) : type(char(t)), bypass_t(NULL) {}
  1144. };
  1145. typedef internal::aggregating_functor<class_type, join_node_base_operation> handler_type;
  1146. friend class internal::aggregating_functor<class_type, join_node_base_operation>;
  1147. bool forwarder_busy;
  1148. aggregator<handler_type, join_node_base_operation> my_aggregator;
  1149. void handle_operations(join_node_base_operation* op_list) {
  1150. join_node_base_operation *current;
  1151. while(op_list) {
  1152. current = op_list;
  1153. op_list = op_list->next;
  1154. switch(current->type) {
  1155. case reg_succ: {
  1156. my_successors.register_successor(*(current->my_succ));
  1157. if(tuple_build_may_succeed() && !forwarder_busy && internal::is_graph_active(my_graph)) {
  1158. task *rtask = new ( task::allocate_additional_child_of(*(my_graph.root_task())) )
  1159. forward_task_bypass
  1160. <join_node_base<JP,InputTuple,OutputTuple> >(*this);
  1161. internal::spawn_in_graph_arena(my_graph, *rtask);
  1162. forwarder_busy = true;
  1163. }
  1164. __TBB_store_with_release(current->status, SUCCEEDED);
  1165. }
  1166. break;
  1167. case rem_succ:
  1168. my_successors.remove_successor(*(current->my_succ));
  1169. __TBB_store_with_release(current->status, SUCCEEDED);
  1170. break;
  1171. case try__get:
  1172. if(tuple_build_may_succeed()) {
  1173. if(try_to_make_tuple(*(current->my_arg))) {
  1174. tuple_accepted();
  1175. __TBB_store_with_release(current->status, SUCCEEDED);
  1176. }
  1177. else __TBB_store_with_release(current->status, FAILED);
  1178. }
  1179. else __TBB_store_with_release(current->status, FAILED);
  1180. break;
  1181. case do_fwrd_bypass: {
  1182. bool build_succeeded;
  1183. task *last_task = NULL;
  1184. output_type out;
  1185. if(tuple_build_may_succeed()) { // checks output queue of FE
  1186. do {
  1187. build_succeeded = try_to_make_tuple(out); // fetch front_end of queue
  1188. if(build_succeeded) {
  1189. task *new_task = my_successors.try_put_task(out);
  1190. last_task = combine_tasks(my_graph, last_task, new_task);
  1191. if(new_task) {
  1192. tuple_accepted();
  1193. }
  1194. else {
  1195. tuple_rejected();
  1196. build_succeeded = false;
  1197. }
  1198. }
  1199. } while(build_succeeded);
  1200. }
  1201. current->bypass_t = last_task;
  1202. __TBB_store_with_release(current->status, SUCCEEDED);
  1203. forwarder_busy = false;
  1204. }
  1205. break;
  1206. #if TBB_DEPRECATED_FLOW_NODE_EXTRACTION
  1207. case add_blt_succ:
  1208. my_successors.internal_add_built_successor(*(current->my_succ));
  1209. __TBB_store_with_release(current->status, SUCCEEDED);
  1210. break;
  1211. case del_blt_succ:
  1212. my_successors.internal_delete_built_successor(*(current->my_succ));
  1213. __TBB_store_with_release(current->status, SUCCEEDED);
  1214. break;
  1215. case blt_succ_cnt:
  1216. current->cnt_val = my_successors.successor_count();
  1217. __TBB_store_with_release(current->status, SUCCEEDED);
  1218. break;
  1219. case blt_succ_cpy:
  1220. my_successors.copy_successors(*(current->slist));
  1221. __TBB_store_with_release(current->status, SUCCEEDED);
  1222. break;
  1223. #endif /* TBB_DEPRECATED_FLOW_NODE_EXTRACTION */
  1224. }
  1225. }
  1226. }
  1227. // ---------- end aggregator -----------
  1228. public:
  1229. join_node_base(graph &g) : graph_node(g), input_ports_type(g), forwarder_busy(false) {
  1230. my_successors.set_owner(this);
  1231. input_ports_type::set_my_node(this);
  1232. my_aggregator.initialize_handler(handler_type(this));
  1233. }
  1234. join_node_base(const join_node_base& other) :
  1235. graph_node(other.graph_node::my_graph), input_ports_type(other),
  1236. sender<OutputTuple>(), forwarder_busy(false), my_successors() {
  1237. my_successors.set_owner(this);
  1238. input_ports_type::set_my_node(this);
  1239. my_aggregator.initialize_handler(handler_type(this));
  1240. }
  1241. template<typename FunctionTuple>
  1242. join_node_base(graph &g, FunctionTuple f) : graph_node(g), input_ports_type(g, f), forwarder_busy(false) {
  1243. my_successors.set_owner(this);
  1244. input_ports_type::set_my_node(this);
  1245. my_aggregator.initialize_handler(handler_type(this));
  1246. }
  1247. bool register_successor(successor_type &r) __TBB_override {
  1248. join_node_base_operation op_data(r, reg_succ);
  1249. my_aggregator.execute(&op_data);
  1250. return op_data.status == SUCCEEDED;
  1251. }
  1252. bool remove_successor( successor_type &r) __TBB_override {
  1253. join_node_base_operation op_data(r, rem_succ);
  1254. my_aggregator.execute(&op_data);
  1255. return op_data.status == SUCCEEDED;
  1256. }
  1257. bool try_get( output_type &v) __TBB_override {
  1258. join_node_base_operation op_data(v, try__get);
  1259. my_aggregator.execute(&op_data);
  1260. return op_data.status == SUCCEEDED;
  1261. }
  1262. #if TBB_DEPRECATED_FLOW_NODE_EXTRACTION
  1263. built_successors_type &built_successors() __TBB_override { return my_successors.built_successors(); }
  1264. void internal_add_built_successor( successor_type &r) __TBB_override {
  1265. join_node_base_operation op_data(r, add_blt_succ);
  1266. my_aggregator.execute(&op_data);
  1267. }
  1268. void internal_delete_built_successor( successor_type &r) __TBB_override {
  1269. join_node_base_operation op_data(r, del_blt_succ);
  1270. my_aggregator.execute(&op_data);
  1271. }
  1272. size_t successor_count() __TBB_override {
  1273. join_node_base_operation op_data(blt_succ_cnt);
  1274. my_aggregator.execute(&op_data);
  1275. return op_data.cnt_val;
  1276. }
  1277. void copy_successors(successor_list_type &l) __TBB_override {
  1278. join_node_base_operation op_data(blt_succ_cpy);
  1279. op_data.slist = &l;
  1280. my_aggregator.execute(&op_data);
  1281. }
  1282. #endif /* TBB_DEPRECATED_FLOW_NODE_EXTRACTION */
  1283. #if TBB_DEPRECATED_FLOW_NODE_EXTRACTION
  1284. void extract() __TBB_override {
  1285. input_ports_type::extract();
  1286. my_successors.built_successors().sender_extract(*this);
  1287. }
  1288. #endif
  1289. protected:
  1290. void reset_node(reset_flags f) __TBB_override {
  1291. input_ports_type::reset(f);
  1292. if(f & rf_clear_edges) my_successors.clear();
  1293. }
  1294. private:
  1295. broadcast_cache<output_type, null_rw_mutex> my_successors;
  1296. friend class forward_task_bypass< join_node_base<JP, InputTuple, OutputTuple> >;
  1297. task *forward_task() {
  1298. join_node_base_operation op_data(do_fwrd_bypass);
  1299. my_aggregator.execute(&op_data);
  1300. return op_data.bypass_t;
  1301. }
  1302. }; // join_node_base
  1303. // join base class type generator
  1304. template<int N, template<class> class PT, typename OutputTuple, typename JP>
  1305. struct join_base {
  1306. typedef typename internal::join_node_base<JP, typename wrap_tuple_elements<N,PT,OutputTuple>::type, OutputTuple> type;
  1307. };
  1308. template<int N, typename OutputTuple, typename K, typename KHash>
  1309. struct join_base<N, key_matching_port, OutputTuple, key_matching<K,KHash> > {
  1310. typedef key_matching<K, KHash> key_traits_type;
  1311. typedef K key_type;
  1312. typedef KHash key_hash_compare;
  1313. typedef typename internal::join_node_base< key_traits_type,
  1314. // ports type
  1315. typename wrap_key_tuple_elements<N,key_matching_port,key_traits_type,OutputTuple>::type,
  1316. OutputTuple > type;
  1317. };
  1318. //! unfolded_join_node : passes input_ports_type to join_node_base. We build the input port type
  1319. // using tuple_element. The class PT is the port type (reserving_port, queueing_port, key_matching_port)
  1320. // and should match the typename.
  1321. template<int N, template<class> class PT, typename OutputTuple, typename JP>
  1322. class unfolded_join_node : public join_base<N,PT,OutputTuple,JP>::type {
  1323. public:
  1324. typedef typename wrap_tuple_elements<N, PT, OutputTuple>::type input_ports_type;
  1325. typedef OutputTuple output_type;
  1326. private:
  1327. typedef join_node_base<JP, input_ports_type, output_type > base_type;
  1328. public:
  1329. unfolded_join_node(graph &g) : base_type(g) {}
  1330. unfolded_join_node(const unfolded_join_node &other) : base_type(other) {}
  1331. };
  1332. #if __TBB_PREVIEW_MESSAGE_BASED_KEY_MATCHING
  1333. template <typename K, typename T>
  1334. struct key_from_message_body {
  1335. K operator()(const T& t) const {
  1336. using tbb::flow::key_from_message;
  1337. return key_from_message<K>(t);
  1338. }
  1339. };
  1340. // Adds const to reference type
  1341. template <typename K, typename T>
  1342. struct key_from_message_body<K&,T> {
  1343. const K& operator()(const T& t) const {
  1344. using tbb::flow::key_from_message;
  1345. return key_from_message<const K&>(t);
  1346. }
  1347. };
  1348. #endif /* __TBB_PREVIEW_MESSAGE_BASED_KEY_MATCHING */
  1349. // key_matching unfolded_join_node. This must be a separate specialization because the constructors
  1350. // differ.
  1351. template<typename OutputTuple, typename K, typename KHash>
  1352. class unfolded_join_node<2,key_matching_port,OutputTuple,key_matching<K,KHash> > : public
  1353. join_base<2,key_matching_port,OutputTuple,key_matching<K,KHash> >::type {
  1354. typedef typename tbb::flow::tuple_element<0, OutputTuple>::type T0;
  1355. typedef typename tbb::flow::tuple_element<1, OutputTuple>::type T1;
  1356. public:
  1357. typedef typename wrap_key_tuple_elements<2,key_matching_port,key_matching<K,KHash>,OutputTuple>::type input_ports_type;
  1358. typedef OutputTuple output_type;
  1359. private:
  1360. typedef join_node_base<key_matching<K,KHash>, input_ports_type, output_type > base_type;
  1361. typedef typename internal::type_to_key_function_body<T0, K> *f0_p;
  1362. typedef typename internal::type_to_key_function_body<T1, K> *f1_p;
  1363. typedef typename tbb::flow::tuple< f0_p, f1_p > func_initializer_type;
  1364. public:
  1365. #if __TBB_PREVIEW_MESSAGE_BASED_KEY_MATCHING
  1366. unfolded_join_node(graph &g) : base_type(g,
  1367. func_initializer_type(
  1368. new internal::type_to_key_function_body_leaf<T0, K, key_from_message_body<K,T0> >(key_from_message_body<K,T0>()),
  1369. new internal::type_to_key_function_body_leaf<T1, K, key_from_message_body<K,T1> >(key_from_message_body<K,T1>())
  1370. ) ) {
  1371. }
  1372. #endif /* __TBB_PREVIEW_MESSAGE_BASED_KEY_MATCHING */
  1373. template<typename Body0, typename Body1>
  1374. unfolded_join_node(graph &g, Body0 body0, Body1 body1) : base_type(g,
  1375. func_initializer_type(
  1376. new internal::type_to_key_function_body_leaf<T0, K, Body0>(body0),
  1377. new internal::type_to_key_function_body_leaf<T1, K, Body1>(body1)
  1378. ) ) {
  1379. __TBB_STATIC_ASSERT(tbb::flow::tuple_size<OutputTuple>::value == 2, "wrong number of body initializers");
  1380. }
  1381. unfolded_join_node(const unfolded_join_node &other) : base_type(other) {}
  1382. };
  1383. template<typename OutputTuple, typename K, typename KHash>
  1384. class unfolded_join_node<3,key_matching_port,OutputTuple,key_matching<K,KHash> > : public
  1385. join_base<3,key_matching_port,OutputTuple,key_matching<K,KHash> >::type {
  1386. typedef typename tbb::flow::tuple_element<0, OutputTuple>::type T0;
  1387. typedef typename tbb::flow::tuple_element<1, OutputTuple>::type T1;
  1388. typedef typename tbb::flow::tuple_element<2, OutputTuple>::type T2;
  1389. public:
  1390. typedef typename wrap_key_tuple_elements<3,key_matching_port,key_matching<K,KHash>,OutputTuple>::type input_ports_type;
  1391. typedef OutputTuple output_type;
  1392. private:
  1393. typedef join_node_base<key_matching<K,KHash>, input_ports_type, output_type > base_type;
  1394. typedef typename internal::type_to_key_function_body<T0, K> *f0_p;
  1395. typedef typename internal::type_to_key_function_body<T1, K> *f1_p;
  1396. typedef typename internal::type_to_key_function_body<T2, K> *f2_p;
  1397. typedef typename tbb::flow::tuple< f0_p, f1_p, f2_p > func_initializer_type;
  1398. public:
  1399. #if __TBB_PREVIEW_MESSAGE_BASED_KEY_MATCHING
  1400. unfolded_join_node(graph &g) : base_type(g,
  1401. func_initializer_type(
  1402. new internal::type_to_key_function_body_leaf<T0, K, key_from_message_body<K,T0> >(key_from_message_body<K,T0>()),
  1403. new internal::type_to_key_function_body_leaf<T1, K, key_from_message_body<K,T1> >(key_from_message_body<K,T1>()),
  1404. new internal::type_to_key_function_body_leaf<T2, K, key_from_message_body<K,T2> >(key_from_message_body<K,T2>())
  1405. ) ) {
  1406. }
  1407. #endif /* __TBB_PREVIEW_MESSAGE_BASED_KEY_MATCHING */
  1408. template<typename Body0, typename Body1, typename Body2>
  1409. unfolded_join_node(graph &g, Body0 body0, Body1 body1, Body2 body2) : base_type(g,
  1410. func_initializer_type(
  1411. new internal::type_to_key_function_body_leaf<T0, K, Body0>(body0),
  1412. new internal::type_to_key_function_body_leaf<T1, K, Body1>(body1),
  1413. new internal::type_to_key_function_body_leaf<T2, K, Body2>(body2)
  1414. ) ) {
  1415. __TBB_STATIC_ASSERT(tbb::flow::tuple_size<OutputTuple>::value == 3, "wrong number of body initializers");
  1416. }
  1417. unfolded_join_node(const unfolded_join_node &other) : base_type(other) {}
  1418. };
  1419. template<typename OutputTuple, typename K, typename KHash>
  1420. class unfolded_join_node<4,key_matching_port,OutputTuple,key_matching<K,KHash> > : public
  1421. join_base<4,key_matching_port,OutputTuple,key_matching<K,KHash> >::type {
  1422. typedef typename tbb::flow::tuple_element<0, OutputTuple>::type T0;
  1423. typedef typename tbb::flow::tuple_element<1, OutputTuple>::type T1;
  1424. typedef typename tbb::flow::tuple_element<2, OutputTuple>::type T2;
  1425. typedef typename tbb::flow::tuple_element<3, OutputTuple>::type T3;
  1426. public:
  1427. typedef typename wrap_key_tuple_elements<4,key_matching_port,key_matching<K,KHash>,OutputTuple>::type input_ports_type;
  1428. typedef OutputTuple output_type;
  1429. private:
  1430. typedef join_node_base<key_matching<K,KHash>, input_ports_type, output_type > base_type;
  1431. typedef typename internal::type_to_key_function_body<T0, K> *f0_p;
  1432. typedef typename internal::type_to_key_function_body<T1, K> *f1_p;
  1433. typedef typename internal::type_to_key_function_body<T2, K> *f2_p;
  1434. typedef typename internal::type_to_key_function_body<T3, K> *f3_p;
  1435. typedef typename tbb::flow::tuple< f0_p, f1_p, f2_p, f3_p > func_initializer_type;
  1436. public:
  1437. #if __TBB_PREVIEW_MESSAGE_BASED_KEY_MATCHING
  1438. unfolded_join_node(graph &g) : base_type(g,
  1439. func_initializer_type(
  1440. new internal::type_to_key_function_body_leaf<T0, K, key_from_message_body<K,T0> >(key_from_message_body<K,T0>()),
  1441. new internal::type_to_key_function_body_leaf<T1, K, key_from_message_body<K,T1> >(key_from_message_body<K,T1>()),
  1442. new internal::type_to_key_function_body_leaf<T2, K, key_from_message_body<K,T2> >(key_from_message_body<K,T2>()),
  1443. new internal::type_to_key_function_body_leaf<T3, K, key_from_message_body<K,T3> >(key_from_message_body<K,T3>())
  1444. ) ) {
  1445. }
  1446. #endif /* __TBB_PREVIEW_MESSAGE_BASED_KEY_MATCHING */
  1447. template<typename Body0, typename Body1, typename Body2, typename Body3>
  1448. unfolded_join_node(graph &g, Body0 body0, Body1 body1, Body2 body2, Body3 body3) : base_type(g,
  1449. func_initializer_type(
  1450. new internal::type_to_key_function_body_leaf<T0, K, Body0>(body0),
  1451. new internal::type_to_key_function_body_leaf<T1, K, Body1>(body1),
  1452. new internal::type_to_key_function_body_leaf<T2, K, Body2>(body2),
  1453. new internal::type_to_key_function_body_leaf<T3, K, Body3>(body3)
  1454. ) ) {
  1455. __TBB_STATIC_ASSERT(tbb::flow::tuple_size<OutputTuple>::value == 4, "wrong number of body initializers");
  1456. }
  1457. unfolded_join_node(const unfolded_join_node &other) : base_type(other) {}
  1458. };
  1459. template<typename OutputTuple, typename K, typename KHash>
  1460. class unfolded_join_node<5,key_matching_port,OutputTuple,key_matching<K,KHash> > : public
  1461. join_base<5,key_matching_port,OutputTuple,key_matching<K,KHash> >::type {
  1462. typedef typename tbb::flow::tuple_element<0, OutputTuple>::type T0;
  1463. typedef typename tbb::flow::tuple_element<1, OutputTuple>::type T1;
  1464. typedef typename tbb::flow::tuple_element<2, OutputTuple>::type T2;
  1465. typedef typename tbb::flow::tuple_element<3, OutputTuple>::type T3;
  1466. typedef typename tbb::flow::tuple_element<4, OutputTuple>::type T4;
  1467. public:
  1468. typedef typename wrap_key_tuple_elements<5,key_matching_port,key_matching<K,KHash>,OutputTuple>::type input_ports_type;
  1469. typedef OutputTuple output_type;
  1470. private:
  1471. typedef join_node_base<key_matching<K,KHash> , input_ports_type, output_type > base_type;
  1472. typedef typename internal::type_to_key_function_body<T0, K> *f0_p;
  1473. typedef typename internal::type_to_key_function_body<T1, K> *f1_p;
  1474. typedef typename internal::type_to_key_function_body<T2, K> *f2_p;
  1475. typedef typename internal::type_to_key_function_body<T3, K> *f3_p;
  1476. typedef typename internal::type_to_key_function_body<T4, K> *f4_p;
  1477. typedef typename tbb::flow::tuple< f0_p, f1_p, f2_p, f3_p, f4_p > func_initializer_type;
  1478. public:
  1479. #if __TBB_PREVIEW_MESSAGE_BASED_KEY_MATCHING
  1480. unfolded_join_node(graph &g) : base_type(g,
  1481. func_initializer_type(
  1482. new internal::type_to_key_function_body_leaf<T0, K, key_from_message_body<K,T0> >(key_from_message_body<K,T0>()),
  1483. new internal::type_to_key_function_body_leaf<T1, K, key_from_message_body<K,T1> >(key_from_message_body<K,T1>()),
  1484. new internal::type_to_key_function_body_leaf<T2, K, key_from_message_body<K,T2> >(key_from_message_body<K,T2>()),
  1485. new internal::type_to_key_function_body_leaf<T3, K, key_from_message_body<K,T3> >(key_from_message_body<K,T3>()),
  1486. new internal::type_to_key_function_body_leaf<T4, K, key_from_message_body<K,T4> >(key_from_message_body<K,T4>())
  1487. ) ) {
  1488. }
  1489. #endif /* __TBB_PREVIEW_MESSAGE_BASED_KEY_MATCHING */
  1490. template<typename Body0, typename Body1, typename Body2, typename Body3, typename Body4>
  1491. unfolded_join_node(graph &g, Body0 body0, Body1 body1, Body2 body2, Body3 body3, Body4 body4) : base_type(g,
  1492. func_initializer_type(
  1493. new internal::type_to_key_function_body_leaf<T0, K, Body0>(body0),
  1494. new internal::type_to_key_function_body_leaf<T1, K, Body1>(body1),
  1495. new internal::type_to_key_function_body_leaf<T2, K, Body2>(body2),
  1496. new internal::type_to_key_function_body_leaf<T3, K, Body3>(body3),
  1497. new internal::type_to_key_function_body_leaf<T4, K, Body4>(body4)
  1498. ) ) {
  1499. __TBB_STATIC_ASSERT(tbb::flow::tuple_size<OutputTuple>::value == 5, "wrong number of body initializers");
  1500. }
  1501. unfolded_join_node(const unfolded_join_node &other) : base_type(other) {}
  1502. };
  1503. #if __TBB_VARIADIC_MAX >= 6
  1504. template<typename OutputTuple, typename K, typename KHash>
  1505. class unfolded_join_node<6,key_matching_port,OutputTuple,key_matching<K,KHash> > : public
  1506. join_base<6,key_matching_port,OutputTuple,key_matching<K,KHash> >::type {
  1507. typedef typename tbb::flow::tuple_element<0, OutputTuple>::type T0;
  1508. typedef typename tbb::flow::tuple_element<1, OutputTuple>::type T1;
  1509. typedef typename tbb::flow::tuple_element<2, OutputTuple>::type T2;
  1510. typedef typename tbb::flow::tuple_element<3, OutputTuple>::type T3;
  1511. typedef typename tbb::flow::tuple_element<4, OutputTuple>::type T4;
  1512. typedef typename tbb::flow::tuple_element<5, OutputTuple>::type T5;
  1513. public:
  1514. typedef typename wrap_key_tuple_elements<6,key_matching_port,key_matching<K,KHash>,OutputTuple>::type input_ports_type;
  1515. typedef OutputTuple output_type;
  1516. private:
  1517. typedef join_node_base<key_matching<K,KHash> , input_ports_type, output_type > base_type;
  1518. typedef typename internal::type_to_key_function_body<T0, K> *f0_p;
  1519. typedef typename internal::type_to_key_function_body<T1, K> *f1_p;
  1520. typedef typename internal::type_to_key_function_body<T2, K> *f2_p;
  1521. typedef typename internal::type_to_key_function_body<T3, K> *f3_p;
  1522. typedef typename internal::type_to_key_function_body<T4, K> *f4_p;
  1523. typedef typename internal::type_to_key_function_body<T5, K> *f5_p;
  1524. typedef typename tbb::flow::tuple< f0_p, f1_p, f2_p, f3_p, f4_p, f5_p > func_initializer_type;
  1525. public:
  1526. #if __TBB_PREVIEW_MESSAGE_BASED_KEY_MATCHING
  1527. unfolded_join_node(graph &g) : base_type(g,
  1528. func_initializer_type(
  1529. new internal::type_to_key_function_body_leaf<T0, K, key_from_message_body<K,T0> >(key_from_message_body<K,T0>()),
  1530. new internal::type_to_key_function_body_leaf<T1, K, key_from_message_body<K,T1> >(key_from_message_body<K,T1>()),
  1531. new internal::type_to_key_function_body_leaf<T2, K, key_from_message_body<K,T2> >(key_from_message_body<K,T2>()),
  1532. new internal::type_to_key_function_body_leaf<T3, K, key_from_message_body<K,T3> >(key_from_message_body<K,T3>()),
  1533. new internal::type_to_key_function_body_leaf<T4, K, key_from_message_body<K,T4> >(key_from_message_body<K,T4>()),
  1534. new internal::type_to_key_function_body_leaf<T5, K, key_from_message_body<K,T5> >(key_from_message_body<K,T5>())
  1535. ) ) {
  1536. }
  1537. #endif /* __TBB_PREVIEW_MESSAGE_BASED_KEY_MATCHING */
  1538. template<typename Body0, typename Body1, typename Body2, typename Body3, typename Body4, typename Body5>
  1539. unfolded_join_node(graph &g, Body0 body0, Body1 body1, Body2 body2, Body3 body3, Body4 body4, Body5 body5)
  1540. : base_type(g, func_initializer_type(
  1541. new internal::type_to_key_function_body_leaf<T0, K, Body0>(body0),
  1542. new internal::type_to_key_function_body_leaf<T1, K, Body1>(body1),
  1543. new internal::type_to_key_function_body_leaf<T2, K, Body2>(body2),
  1544. new internal::type_to_key_function_body_leaf<T3, K, Body3>(body3),
  1545. new internal::type_to_key_function_body_leaf<T4, K, Body4>(body4),
  1546. new internal::type_to_key_function_body_leaf<T5, K, Body5>(body5)
  1547. ) ) {
  1548. __TBB_STATIC_ASSERT(tbb::flow::tuple_size<OutputTuple>::value == 6, "wrong number of body initializers");
  1549. }
  1550. unfolded_join_node(const unfolded_join_node &other) : base_type(other) {}
  1551. };
  1552. #endif
  1553. #if __TBB_VARIADIC_MAX >= 7
  1554. template<typename OutputTuple, typename K, typename KHash>
  1555. class unfolded_join_node<7,key_matching_port,OutputTuple,key_matching<K,KHash> > : public
  1556. join_base<7,key_matching_port,OutputTuple,key_matching<K,KHash> >::type {
  1557. typedef typename tbb::flow::tuple_element<0, OutputTuple>::type T0;
  1558. typedef typename tbb::flow::tuple_element<1, OutputTuple>::type T1;
  1559. typedef typename tbb::flow::tuple_element<2, OutputTuple>::type T2;
  1560. typedef typename tbb::flow::tuple_element<3, OutputTuple>::type T3;
  1561. typedef typename tbb::flow::tuple_element<4, OutputTuple>::type T4;
  1562. typedef typename tbb::flow::tuple_element<5, OutputTuple>::type T5;
  1563. typedef typename tbb::flow::tuple_element<6, OutputTuple>::type T6;
  1564. public:
  1565. typedef typename wrap_key_tuple_elements<7,key_matching_port,key_matching<K,KHash>,OutputTuple>::type input_ports_type;
  1566. typedef OutputTuple output_type;
  1567. private:
  1568. typedef join_node_base<key_matching<K,KHash> , input_ports_type, output_type > base_type;
  1569. typedef typename internal::type_to_key_function_body<T0, K> *f0_p;
  1570. typedef typename internal::type_to_key_function_body<T1, K> *f1_p;
  1571. typedef typename internal::type_to_key_function_body<T2, K> *f2_p;
  1572. typedef typename internal::type_to_key_function_body<T3, K> *f3_p;
  1573. typedef typename internal::type_to_key_function_body<T4, K> *f4_p;
  1574. typedef typename internal::type_to_key_function_body<T5, K> *f5_p;
  1575. typedef typename internal::type_to_key_function_body<T6, K> *f6_p;
  1576. typedef typename tbb::flow::tuple< f0_p, f1_p, f2_p, f3_p, f4_p, f5_p, f6_p > func_initializer_type;
  1577. public:
  1578. #if __TBB_PREVIEW_MESSAGE_BASED_KEY_MATCHING
  1579. unfolded_join_node(graph &g) : base_type(g,
  1580. func_initializer_type(
  1581. new internal::type_to_key_function_body_leaf<T0, K, key_from_message_body<K,T0> >(key_from_message_body<K,T0>()),
  1582. new internal::type_to_key_function_body_leaf<T1, K, key_from_message_body<K,T1> >(key_from_message_body<K,T1>()),
  1583. new internal::type_to_key_function_body_leaf<T2, K, key_from_message_body<K,T2> >(key_from_message_body<K,T2>()),
  1584. new internal::type_to_key_function_body_leaf<T3, K, key_from_message_body<K,T3> >(key_from_message_body<K,T3>()),
  1585. new internal::type_to_key_function_body_leaf<T4, K, key_from_message_body<K,T4> >(key_from_message_body<K,T4>()),
  1586. new internal::type_to_key_function_body_leaf<T5, K, key_from_message_body<K,T5> >(key_from_message_body<K,T5>()),
  1587. new internal::type_to_key_function_body_leaf<T6, K, key_from_message_body<K,T6> >(key_from_message_body<K,T6>())
  1588. ) ) {
  1589. }
  1590. #endif /* __TBB_PREVIEW_MESSAGE_BASED_KEY_MATCHING */
  1591. template<typename Body0, typename Body1, typename Body2, typename Body3, typename Body4,
  1592. typename Body5, typename Body6>
  1593. unfolded_join_node(graph &g, Body0 body0, Body1 body1, Body2 body2, Body3 body3, Body4 body4,
  1594. Body5 body5, Body6 body6) : base_type(g, func_initializer_type(
  1595. new internal::type_to_key_function_body_leaf<T0, K, Body0>(body0),
  1596. new internal::type_to_key_function_body_leaf<T1, K, Body1>(body1),
  1597. new internal::type_to_key_function_body_leaf<T2, K, Body2>(body2),
  1598. new internal::type_to_key_function_body_leaf<T3, K, Body3>(body3),
  1599. new internal::type_to_key_function_body_leaf<T4, K, Body4>(body4),
  1600. new internal::type_to_key_function_body_leaf<T5, K, Body5>(body5),
  1601. new internal::type_to_key_function_body_leaf<T6, K, Body6>(body6)
  1602. ) ) {
  1603. __TBB_STATIC_ASSERT(tbb::flow::tuple_size<OutputTuple>::value == 7, "wrong number of body initializers");
  1604. }
  1605. unfolded_join_node(const unfolded_join_node &other) : base_type(other) {}
  1606. };
  1607. #endif
  1608. #if __TBB_VARIADIC_MAX >= 8
  1609. template<typename OutputTuple, typename K, typename KHash>
  1610. class unfolded_join_node<8,key_matching_port,OutputTuple,key_matching<K,KHash> > : public
  1611. join_base<8,key_matching_port,OutputTuple,key_matching<K,KHash> >::type {
  1612. typedef typename tbb::flow::tuple_element<0, OutputTuple>::type T0;
  1613. typedef typename tbb::flow::tuple_element<1, OutputTuple>::type T1;
  1614. typedef typename tbb::flow::tuple_element<2, OutputTuple>::type T2;
  1615. typedef typename tbb::flow::tuple_element<3, OutputTuple>::type T3;
  1616. typedef typename tbb::flow::tuple_element<4, OutputTuple>::type T4;
  1617. typedef typename tbb::flow::tuple_element<5, OutputTuple>::type T5;
  1618. typedef typename tbb::flow::tuple_element<6, OutputTuple>::type T6;
  1619. typedef typename tbb::flow::tuple_element<7, OutputTuple>::type T7;
  1620. public:
  1621. typedef typename wrap_key_tuple_elements<8,key_matching_port,key_matching<K,KHash>,OutputTuple>::type input_ports_type;
  1622. typedef OutputTuple output_type;
  1623. private:
  1624. typedef join_node_base<key_matching<K,KHash> , input_ports_type, output_type > base_type;
  1625. typedef typename internal::type_to_key_function_body<T0, K> *f0_p;
  1626. typedef typename internal::type_to_key_function_body<T1, K> *f1_p;
  1627. typedef typename internal::type_to_key_function_body<T2, K> *f2_p;
  1628. typedef typename internal::type_to_key_function_body<T3, K> *f3_p;
  1629. typedef typename internal::type_to_key_function_body<T4, K> *f4_p;
  1630. typedef typename internal::type_to_key_function_body<T5, K> *f5_p;
  1631. typedef typename internal::type_to_key_function_body<T6, K> *f6_p;
  1632. typedef typename internal::type_to_key_function_body<T7, K> *f7_p;
  1633. typedef typename tbb::flow::tuple< f0_p, f1_p, f2_p, f3_p, f4_p, f5_p, f6_p, f7_p > func_initializer_type;
  1634. public:
  1635. #if __TBB_PREVIEW_MESSAGE_BASED_KEY_MATCHING
  1636. unfolded_join_node(graph &g) : base_type(g,
  1637. func_initializer_type(
  1638. new internal::type_to_key_function_body_leaf<T0, K, key_from_message_body<K,T0> >(key_from_message_body<K,T0>()),
  1639. new internal::type_to_key_function_body_leaf<T1, K, key_from_message_body<K,T1> >(key_from_message_body<K,T1>()),
  1640. new internal::type_to_key_function_body_leaf<T2, K, key_from_message_body<K,T2> >(key_from_message_body<K,T2>()),
  1641. new internal::type_to_key_function_body_leaf<T3, K, key_from_message_body<K,T3> >(key_from_message_body<K,T3>()),
  1642. new internal::type_to_key_function_body_leaf<T4, K, key_from_message_body<K,T4> >(key_from_message_body<K,T4>()),
  1643. new internal::type_to_key_function_body_leaf<T5, K, key_from_message_body<K,T5> >(key_from_message_body<K,T5>()),
  1644. new internal::type_to_key_function_body_leaf<T6, K, key_from_message_body<K,T6> >(key_from_message_body<K,T6>()),
  1645. new internal::type_to_key_function_body_leaf<T7, K, key_from_message_body<K,T7> >(key_from_message_body<K,T7>())
  1646. ) ) {
  1647. }
  1648. #endif /* __TBB_PREVIEW_MESSAGE_BASED_KEY_MATCHING */
  1649. template<typename Body0, typename Body1, typename Body2, typename Body3, typename Body4,
  1650. typename Body5, typename Body6, typename Body7>
  1651. unfolded_join_node(graph &g, Body0 body0, Body1 body1, Body2 body2, Body3 body3, Body4 body4,
  1652. Body5 body5, Body6 body6, Body7 body7) : base_type(g, func_initializer_type(
  1653. new internal::type_to_key_function_body_leaf<T0, K, Body0>(body0),
  1654. new internal::type_to_key_function_body_leaf<T1, K, Body1>(body1),
  1655. new internal::type_to_key_function_body_leaf<T2, K, Body2>(body2),
  1656. new internal::type_to_key_function_body_leaf<T3, K, Body3>(body3),
  1657. new internal::type_to_key_function_body_leaf<T4, K, Body4>(body4),
  1658. new internal::type_to_key_function_body_leaf<T5, K, Body5>(body5),
  1659. new internal::type_to_key_function_body_leaf<T6, K, Body6>(body6),
  1660. new internal::type_to_key_function_body_leaf<T7, K, Body7>(body7)
  1661. ) ) {
  1662. __TBB_STATIC_ASSERT(tbb::flow::tuple_size<OutputTuple>::value == 8, "wrong number of body initializers");
  1663. }
  1664. unfolded_join_node(const unfolded_join_node &other) : base_type(other) {}
  1665. };
  1666. #endif
  1667. #if __TBB_VARIADIC_MAX >= 9
  1668. template<typename OutputTuple, typename K, typename KHash>
  1669. class unfolded_join_node<9,key_matching_port,OutputTuple,key_matching<K,KHash> > : public
  1670. join_base<9,key_matching_port,OutputTuple,key_matching<K,KHash> >::type {
  1671. typedef typename tbb::flow::tuple_element<0, OutputTuple>::type T0;
  1672. typedef typename tbb::flow::tuple_element<1, OutputTuple>::type T1;
  1673. typedef typename tbb::flow::tuple_element<2, OutputTuple>::type T2;
  1674. typedef typename tbb::flow::tuple_element<3, OutputTuple>::type T3;
  1675. typedef typename tbb::flow::tuple_element<4, OutputTuple>::type T4;
  1676. typedef typename tbb::flow::tuple_element<5, OutputTuple>::type T5;
  1677. typedef typename tbb::flow::tuple_element<6, OutputTuple>::type T6;
  1678. typedef typename tbb::flow::tuple_element<7, OutputTuple>::type T7;
  1679. typedef typename tbb::flow::tuple_element<8, OutputTuple>::type T8;
  1680. public:
  1681. typedef typename wrap_key_tuple_elements<9,key_matching_port,key_matching<K,KHash>,OutputTuple>::type input_ports_type;
  1682. typedef OutputTuple output_type;
  1683. private:
  1684. typedef join_node_base<key_matching<K,KHash> , input_ports_type, output_type > base_type;
  1685. typedef typename internal::type_to_key_function_body<T0, K> *f0_p;
  1686. typedef typename internal::type_to_key_function_body<T1, K> *f1_p;
  1687. typedef typename internal::type_to_key_function_body<T2, K> *f2_p;
  1688. typedef typename internal::type_to_key_function_body<T3, K> *f3_p;
  1689. typedef typename internal::type_to_key_function_body<T4, K> *f4_p;
  1690. typedef typename internal::type_to_key_function_body<T5, K> *f5_p;
  1691. typedef typename internal::type_to_key_function_body<T6, K> *f6_p;
  1692. typedef typename internal::type_to_key_function_body<T7, K> *f7_p;
  1693. typedef typename internal::type_to_key_function_body<T8, K> *f8_p;
  1694. typedef typename tbb::flow::tuple< f0_p, f1_p, f2_p, f3_p, f4_p, f5_p, f6_p, f7_p, f8_p > func_initializer_type;
  1695. public:
  1696. #if __TBB_PREVIEW_MESSAGE_BASED_KEY_MATCHING
  1697. unfolded_join_node(graph &g) : base_type(g,
  1698. func_initializer_type(
  1699. new internal::type_to_key_function_body_leaf<T0, K, key_from_message_body<K,T0> >(key_from_message_body<K,T0>()),
  1700. new internal::type_to_key_function_body_leaf<T1, K, key_from_message_body<K,T1> >(key_from_message_body<K,T1>()),
  1701. new internal::type_to_key_function_body_leaf<T2, K, key_from_message_body<K,T2> >(key_from_message_body<K,T2>()),
  1702. new internal::type_to_key_function_body_leaf<T3, K, key_from_message_body<K,T3> >(key_from_message_body<K,T3>()),
  1703. new internal::type_to_key_function_body_leaf<T4, K, key_from_message_body<K,T4> >(key_from_message_body<K,T4>()),
  1704. new internal::type_to_key_function_body_leaf<T5, K, key_from_message_body<K,T5> >(key_from_message_body<K,T5>()),
  1705. new internal::type_to_key_function_body_leaf<T6, K, key_from_message_body<K,T6> >(key_from_message_body<K,T6>()),
  1706. new internal::type_to_key_function_body_leaf<T7, K, key_from_message_body<K,T7> >(key_from_message_body<K,T7>()),
  1707. new internal::type_to_key_function_body_leaf<T8, K, key_from_message_body<K,T8> >(key_from_message_body<K,T8>())
  1708. ) ) {
  1709. }
  1710. #endif /* __TBB_PREVIEW_MESSAGE_BASED_KEY_MATCHING */
  1711. template<typename Body0, typename Body1, typename Body2, typename Body3, typename Body4,
  1712. typename Body5, typename Body6, typename Body7, typename Body8>
  1713. unfolded_join_node(graph &g, Body0 body0, Body1 body1, Body2 body2, Body3 body3, Body4 body4,
  1714. Body5 body5, Body6 body6, Body7 body7, Body8 body8) : base_type(g, func_initializer_type(
  1715. new internal::type_to_key_function_body_leaf<T0, K, Body0>(body0),
  1716. new internal::type_to_key_function_body_leaf<T1, K, Body1>(body1),
  1717. new internal::type_to_key_function_body_leaf<T2, K, Body2>(body2),
  1718. new internal::type_to_key_function_body_leaf<T3, K, Body3>(body3),
  1719. new internal::type_to_key_function_body_leaf<T4, K, Body4>(body4),
  1720. new internal::type_to_key_function_body_leaf<T5, K, Body5>(body5),
  1721. new internal::type_to_key_function_body_leaf<T6, K, Body6>(body6),
  1722. new internal::type_to_key_function_body_leaf<T7, K, Body7>(body7),
  1723. new internal::type_to_key_function_body_leaf<T8, K, Body8>(body8)
  1724. ) ) {
  1725. __TBB_STATIC_ASSERT(tbb::flow::tuple_size<OutputTuple>::value == 9, "wrong number of body initializers");
  1726. }
  1727. unfolded_join_node(const unfolded_join_node &other) : base_type(other) {}
  1728. };
  1729. #endif
  1730. #if __TBB_VARIADIC_MAX >= 10
  1731. template<typename OutputTuple, typename K, typename KHash>
  1732. class unfolded_join_node<10,key_matching_port,OutputTuple,key_matching<K,KHash> > : public
  1733. join_base<10,key_matching_port,OutputTuple,key_matching<K,KHash> >::type {
  1734. typedef typename tbb::flow::tuple_element<0, OutputTuple>::type T0;
  1735. typedef typename tbb::flow::tuple_element<1, OutputTuple>::type T1;
  1736. typedef typename tbb::flow::tuple_element<2, OutputTuple>::type T2;
  1737. typedef typename tbb::flow::tuple_element<3, OutputTuple>::type T3;
  1738. typedef typename tbb::flow::tuple_element<4, OutputTuple>::type T4;
  1739. typedef typename tbb::flow::tuple_element<5, OutputTuple>::type T5;
  1740. typedef typename tbb::flow::tuple_element<6, OutputTuple>::type T6;
  1741. typedef typename tbb::flow::tuple_element<7, OutputTuple>::type T7;
  1742. typedef typename tbb::flow::tuple_element<8, OutputTuple>::type T8;
  1743. typedef typename tbb::flow::tuple_element<9, OutputTuple>::type T9;
  1744. public:
  1745. typedef typename wrap_key_tuple_elements<10,key_matching_port,key_matching<K,KHash>,OutputTuple>::type input_ports_type;
  1746. typedef OutputTuple output_type;
  1747. private:
  1748. typedef join_node_base<key_matching<K,KHash> , input_ports_type, output_type > base_type;
  1749. typedef typename internal::type_to_key_function_body<T0, K> *f0_p;
  1750. typedef typename internal::type_to_key_function_body<T1, K> *f1_p;
  1751. typedef typename internal::type_to_key_function_body<T2, K> *f2_p;
  1752. typedef typename internal::type_to_key_function_body<T3, K> *f3_p;
  1753. typedef typename internal::type_to_key_function_body<T4, K> *f4_p;
  1754. typedef typename internal::type_to_key_function_body<T5, K> *f5_p;
  1755. typedef typename internal::type_to_key_function_body<T6, K> *f6_p;
  1756. typedef typename internal::type_to_key_function_body<T7, K> *f7_p;
  1757. typedef typename internal::type_to_key_function_body<T8, K> *f8_p;
  1758. typedef typename internal::type_to_key_function_body<T9, K> *f9_p;
  1759. typedef typename tbb::flow::tuple< f0_p, f1_p, f2_p, f3_p, f4_p, f5_p, f6_p, f7_p, f8_p, f9_p > func_initializer_type;
  1760. public:
  1761. #if __TBB_PREVIEW_MESSAGE_BASED_KEY_MATCHING
  1762. unfolded_join_node(graph &g) : base_type(g,
  1763. func_initializer_type(
  1764. new internal::type_to_key_function_body_leaf<T0, K, key_from_message_body<K,T0> >(key_from_message_body<K,T0>()),
  1765. new internal::type_to_key_function_body_leaf<T1, K, key_from_message_body<K,T1> >(key_from_message_body<K,T1>()),
  1766. new internal::type_to_key_function_body_leaf<T2, K, key_from_message_body<K,T2> >(key_from_message_body<K,T2>()),
  1767. new internal::type_to_key_function_body_leaf<T3, K, key_from_message_body<K,T3> >(key_from_message_body<K,T3>()),
  1768. new internal::type_to_key_function_body_leaf<T4, K, key_from_message_body<K,T4> >(key_from_message_body<K,T4>()),
  1769. new internal::type_to_key_function_body_leaf<T5, K, key_from_message_body<K,T5> >(key_from_message_body<K,T5>()),
  1770. new internal::type_to_key_function_body_leaf<T6, K, key_from_message_body<K,T6> >(key_from_message_body<K,T6>()),
  1771. new internal::type_to_key_function_body_leaf<T7, K, key_from_message_body<K,T7> >(key_from_message_body<K,T7>()),
  1772. new internal::type_to_key_function_body_leaf<T8, K, key_from_message_body<K,T8> >(key_from_message_body<K,T8>()),
  1773. new internal::type_to_key_function_body_leaf<T9, K, key_from_message_body<K,T9> >(key_from_message_body<K,T9>())
  1774. ) ) {
  1775. }
  1776. #endif /* __TBB_PREVIEW_MESSAGE_BASED_KEY_MATCHING */
  1777. template<typename Body0, typename Body1, typename Body2, typename Body3, typename Body4,
  1778. typename Body5, typename Body6, typename Body7, typename Body8, typename Body9>
  1779. unfolded_join_node(graph &g, Body0 body0, Body1 body1, Body2 body2, Body3 body3, Body4 body4,
  1780. Body5 body5, Body6 body6, Body7 body7, Body8 body8, Body9 body9) : base_type(g, func_initializer_type(
  1781. new internal::type_to_key_function_body_leaf<T0, K, Body0>(body0),
  1782. new internal::type_to_key_function_body_leaf<T1, K, Body1>(body1),
  1783. new internal::type_to_key_function_body_leaf<T2, K, Body2>(body2),
  1784. new internal::type_to_key_function_body_leaf<T3, K, Body3>(body3),
  1785. new internal::type_to_key_function_body_leaf<T4, K, Body4>(body4),
  1786. new internal::type_to_key_function_body_leaf<T5, K, Body5>(body5),
  1787. new internal::type_to_key_function_body_leaf<T6, K, Body6>(body6),
  1788. new internal::type_to_key_function_body_leaf<T7, K, Body7>(body7),
  1789. new internal::type_to_key_function_body_leaf<T8, K, Body8>(body8),
  1790. new internal::type_to_key_function_body_leaf<T9, K, Body9>(body9)
  1791. ) ) {
  1792. __TBB_STATIC_ASSERT(tbb::flow::tuple_size<OutputTuple>::value == 10, "wrong number of body initializers");
  1793. }
  1794. unfolded_join_node(const unfolded_join_node &other) : base_type(other) {}
  1795. };
  1796. #endif
  1797. //! templated function to refer to input ports of the join node
  1798. template<size_t N, typename JNT>
  1799. typename tbb::flow::tuple_element<N, typename JNT::input_ports_type>::type &input_port(JNT &jn) {
  1800. return tbb::flow::get<N>(jn.input_ports());
  1801. }
  1802. }
  1803. #endif // __TBB__flow_graph_join_impl_H