-
Notifications
You must be signed in to change notification settings - Fork 982
/
_flow_graph_join_impl.h
1708 lines (1538 loc) · 81.1 KB
/
_flow_graph_join_impl.h
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
/*
Copyright (c) 2005-2022 Intel Corporation
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
#ifndef __TBB__flow_graph_join_impl_H
#define __TBB__flow_graph_join_impl_H
#ifndef __TBB_flow_graph_H
#error Do not #include this internal file directly; use public TBB headers instead.
#endif
// included into namespace tbb::detail::d1
struct forwarding_base : no_assign {
forwarding_base(graph &g) : graph_ref(g) {}
virtual ~forwarding_base() {}
graph& graph_ref;
};
struct queueing_forwarding_base : forwarding_base {
using forwarding_base::forwarding_base;
// decrement_port_count may create a forwarding task. If we cannot handle the task
// ourselves, ask decrement_port_count to deal with it.
virtual graph_task* decrement_port_count(bool handle_task) = 0;
};
struct reserving_forwarding_base : forwarding_base {
using forwarding_base::forwarding_base;
// decrement_port_count may create a forwarding task. If we cannot handle the task
// ourselves, ask decrement_port_count to deal with it.
virtual graph_task* decrement_port_count() = 0;
virtual void increment_port_count() = 0;
};
// specialization that lets us keep a copy of the current_key for building results.
// KeyType can be a reference type.
template<typename KeyType>
struct matching_forwarding_base : public forwarding_base {
typedef typename std::decay<KeyType>::type current_key_type;
matching_forwarding_base(graph &g) : forwarding_base(g) { }
virtual graph_task* increment_key_count(current_key_type const & /*t*/) = 0;
current_key_type current_key; // so ports can refer to FE's desired items
};
template< int N >
struct join_helper {
template< typename TupleType, typename PortType >
static inline void set_join_node_pointer(TupleType &my_input, PortType *port) {
std::get<N-1>( my_input ).set_join_node_pointer(port);
join_helper<N-1>::set_join_node_pointer( my_input, port );
}
template< typename TupleType >
static inline void consume_reservations( TupleType &my_input ) {
std::get<N-1>( my_input ).consume();
join_helper<N-1>::consume_reservations( my_input );
}
template< typename TupleType >
static inline void release_my_reservation( TupleType &my_input ) {
std::get<N-1>( my_input ).release();
}
template <typename TupleType>
static inline void release_reservations( TupleType &my_input) {
join_helper<N-1>::release_reservations(my_input);
release_my_reservation(my_input);
}
template< typename InputTuple, typename OutputTuple >
static inline bool reserve( InputTuple &my_input, OutputTuple &out) {
if ( !std::get<N-1>( my_input ).reserve( std::get<N-1>( out ) ) ) return false;
if ( !join_helper<N-1>::reserve( my_input, out ) ) {
release_my_reservation( my_input );
return false;
}
return true;
}
template<typename InputTuple, typename OutputTuple>
static inline bool get_my_item( InputTuple &my_input, OutputTuple &out) {
bool res = std::get<N-1>(my_input).get_item(std::get<N-1>(out) ); // may fail
return join_helper<N-1>::get_my_item(my_input, out) && res; // do get on other inputs before returning
}
template<typename InputTuple, typename OutputTuple>
static inline bool get_items(InputTuple &my_input, OutputTuple &out) {
return get_my_item(my_input, out);
}
template<typename InputTuple>
static inline void reset_my_port(InputTuple &my_input) {
join_helper<N-1>::reset_my_port(my_input);
std::get<N-1>(my_input).reset_port();
}
template<typename InputTuple>
static inline void reset_ports(InputTuple& my_input) {
reset_my_port(my_input);
}
template<typename InputTuple, typename KeyFuncTuple>
static inline void set_key_functors(InputTuple &my_input, KeyFuncTuple &my_key_funcs) {
std::get<N-1>(my_input).set_my_key_func(std::get<N-1>(my_key_funcs));
std::get<N-1>(my_key_funcs) = nullptr;
join_helper<N-1>::set_key_functors(my_input, my_key_funcs);
}
template< typename KeyFuncTuple>
static inline void copy_key_functors(KeyFuncTuple &my_inputs, KeyFuncTuple &other_inputs) {
__TBB_ASSERT(
std::get<N-1>(other_inputs).get_my_key_func(),
"key matching join node should not be instantiated without functors."
);
std::get<N-1>(my_inputs).set_my_key_func(std::get<N-1>(other_inputs).get_my_key_func()->clone());
join_helper<N-1>::copy_key_functors(my_inputs, other_inputs);
}
template<typename InputTuple>
static inline void reset_inputs(InputTuple &my_input, reset_flags f) {
join_helper<N-1>::reset_inputs(my_input, f);
std::get<N-1>(my_input).reset_receiver(f);
}
}; // join_helper<N>
template< >
struct join_helper<1> {
template< typename TupleType, typename PortType >
static inline void set_join_node_pointer(TupleType &my_input, PortType *port) {
std::get<0>( my_input ).set_join_node_pointer(port);
}
template< typename TupleType >
static inline void consume_reservations( TupleType &my_input ) {
std::get<0>( my_input ).consume();
}
template< typename TupleType >
static inline void release_my_reservation( TupleType &my_input ) {
std::get<0>( my_input ).release();
}
template<typename TupleType>
static inline void release_reservations( TupleType &my_input) {
release_my_reservation(my_input);
}
template< typename InputTuple, typename OutputTuple >
static inline bool reserve( InputTuple &my_input, OutputTuple &out) {
return std::get<0>( my_input ).reserve( std::get<0>( out ) );
}
template<typename InputTuple, typename OutputTuple>
static inline bool get_my_item( InputTuple &my_input, OutputTuple &out) {
return std::get<0>(my_input).get_item(std::get<0>(out));
}
template<typename InputTuple, typename OutputTuple>
static inline bool get_items(InputTuple &my_input, OutputTuple &out) {
return get_my_item(my_input, out);
}
template<typename InputTuple>
static inline void reset_my_port(InputTuple &my_input) {
std::get<0>(my_input).reset_port();
}
template<typename InputTuple>
static inline void reset_ports(InputTuple& my_input) {
reset_my_port(my_input);
}
template<typename InputTuple, typename KeyFuncTuple>
static inline void set_key_functors(InputTuple &my_input, KeyFuncTuple &my_key_funcs) {
std::get<0>(my_input).set_my_key_func(std::get<0>(my_key_funcs));
std::get<0>(my_key_funcs) = nullptr;
}
template< typename KeyFuncTuple>
static inline void copy_key_functors(KeyFuncTuple &my_inputs, KeyFuncTuple &other_inputs) {
__TBB_ASSERT(
std::get<0>(other_inputs).get_my_key_func(),
"key matching join node should not be instantiated without functors."
);
std::get<0>(my_inputs).set_my_key_func(std::get<0>(other_inputs).get_my_key_func()->clone());
}
template<typename InputTuple>
static inline void reset_inputs(InputTuple &my_input, reset_flags f) {
std::get<0>(my_input).reset_receiver(f);
}
}; // join_helper<1>
//! The two-phase join port
template< typename T >
class reserving_port : public receiver<T> {
public:
typedef T input_type;
typedef typename receiver<input_type>::predecessor_type predecessor_type;
private:
// ----------- Aggregator ------------
enum op_type { reg_pred, rem_pred, res_item, rel_res, con_res
};
typedef reserving_port<T> class_type;
class reserving_port_operation : public aggregated_operation<reserving_port_operation> {
public:
char type;
union {
T *my_arg;
predecessor_type *my_pred;
};
reserving_port_operation(const T& e, op_type t) :
type(char(t)), my_arg(const_cast<T*>(&e)) {}
reserving_port_operation(const predecessor_type &s, op_type t) : type(char(t)),
my_pred(const_cast<predecessor_type *>(&s)) {}
reserving_port_operation(op_type t) : type(char(t)) {}
};
typedef aggregating_functor<class_type, reserving_port_operation> handler_type;
friend class aggregating_functor<class_type, reserving_port_operation>;
aggregator<handler_type, reserving_port_operation> my_aggregator;
void handle_operations(reserving_port_operation* op_list) {
reserving_port_operation *current;
bool was_missing_predecessors = false;
while(op_list) {
current = op_list;
op_list = op_list->next;
switch(current->type) {
case reg_pred:
was_missing_predecessors = my_predecessors.empty();
my_predecessors.add(*(current->my_pred));
if ( was_missing_predecessors ) {
(void) my_join->decrement_port_count(); // may try to forward
}
current->status.store( SUCCEEDED, std::memory_order_release);
break;
case rem_pred:
if ( !my_predecessors.empty() ) {
my_predecessors.remove(*(current->my_pred));
if ( my_predecessors.empty() ) // was the last predecessor
my_join->increment_port_count();
}
// TODO: consider returning failure if there were no predecessors to remove
current->status.store( SUCCEEDED, std::memory_order_release );
break;
case res_item:
if ( reserved ) {
current->status.store( FAILED, std::memory_order_release);
}
else if ( my_predecessors.try_reserve( *(current->my_arg) ) ) {
reserved = true;
current->status.store( SUCCEEDED, std::memory_order_release);
} else {
if ( my_predecessors.empty() ) {
my_join->increment_port_count();
}
current->status.store( FAILED, std::memory_order_release);
}
break;
case rel_res:
reserved = false;
my_predecessors.try_release( );
current->status.store( SUCCEEDED, std::memory_order_release);
break;
case con_res:
reserved = false;
my_predecessors.try_consume( );
current->status.store( SUCCEEDED, std::memory_order_release);
break;
}
}
}
protected:
template< typename R, typename B > friend class run_and_put_task;
template<typename X, typename Y> friend class broadcast_cache;
template<typename X, typename Y> friend class round_robin_cache;
graph_task* try_put_task( const T & ) override {
return nullptr;
}
graph& graph_reference() const override {
return my_join->graph_ref;
}
public:
//! Constructor
reserving_port() : my_join(nullptr), my_predecessors(this), reserved(false) {
my_aggregator.initialize_handler(handler_type(this));
}
// copy constructor
reserving_port(const reserving_port& /* other */) = delete;
void set_join_node_pointer(reserving_forwarding_base *join) {
my_join = join;
}
//! Add a predecessor
bool register_predecessor( predecessor_type &src ) override {
reserving_port_operation op_data(src, reg_pred);
my_aggregator.execute(&op_data);
return op_data.status == SUCCEEDED;
}
//! Remove a predecessor
bool remove_predecessor( predecessor_type &src ) override {
reserving_port_operation op_data(src, rem_pred);
my_aggregator.execute(&op_data);
return op_data.status == SUCCEEDED;
}
//! Reserve an item from the port
bool reserve( T &v ) {
reserving_port_operation op_data(v, res_item);
my_aggregator.execute(&op_data);
return op_data.status == SUCCEEDED;
}
//! Release the port
void release( ) {
reserving_port_operation op_data(rel_res);
my_aggregator.execute(&op_data);
}
//! Complete use of the port
void consume( ) {
reserving_port_operation op_data(con_res);
my_aggregator.execute(&op_data);
}
void reset_receiver( reset_flags f) {
if(f & rf_clear_edges) my_predecessors.clear();
else
my_predecessors.reset();
reserved = false;
__TBB_ASSERT(!(f&rf_clear_edges) || my_predecessors.empty(), "port edges not removed");
}
private:
#if __TBB_PREVIEW_FLOW_GRAPH_NODE_SET
friend class get_graph_helper;
#endif
reserving_forwarding_base *my_join;
reservable_predecessor_cache< T, null_mutex > my_predecessors;
bool reserved;
}; // reserving_port
//! queueing join_port
template<typename T>
class queueing_port : public receiver<T>, public item_buffer<T> {
public:
typedef T input_type;
typedef typename receiver<input_type>::predecessor_type predecessor_type;
typedef queueing_port<T> class_type;
// ----------- Aggregator ------------
private:
enum op_type { get__item, res_port, try__put_task
};
class queueing_port_operation : public aggregated_operation<queueing_port_operation> {
public:
char type;
T my_val;
T* my_arg;
graph_task* bypass_t;
// constructor for value parameter
queueing_port_operation(const T& e, op_type t) :
type(char(t)), my_val(e), my_arg(nullptr)
, bypass_t(nullptr)
{}
// constructor for pointer parameter
queueing_port_operation(const T* p, op_type t) :
type(char(t)), my_arg(const_cast<T*>(p))
, bypass_t(nullptr)
{}
// constructor with no parameter
queueing_port_operation(op_type t) : type(char(t)), my_arg(nullptr)
, bypass_t(nullptr)
{}
};
typedef aggregating_functor<class_type, queueing_port_operation> handler_type;
friend class aggregating_functor<class_type, queueing_port_operation>;
aggregator<handler_type, queueing_port_operation> my_aggregator;
void handle_operations(queueing_port_operation* op_list) {
queueing_port_operation *current;
bool was_empty;
while(op_list) {
current = op_list;
op_list = op_list->next;
switch(current->type) {
case try__put_task: {
graph_task* rtask = nullptr;
was_empty = this->buffer_empty();
this->push_back(current->my_val);
if (was_empty) rtask = my_join->decrement_port_count(false);
else
rtask = SUCCESSFULLY_ENQUEUED;
current->bypass_t = rtask;
current->status.store( SUCCEEDED, std::memory_order_release);
}
break;
case get__item:
if(!this->buffer_empty()) {
__TBB_ASSERT(current->my_arg, nullptr);
*(current->my_arg) = this->front();
current->status.store( SUCCEEDED, std::memory_order_release);
}
else {
current->status.store( FAILED, std::memory_order_release);
}
break;
case res_port:
__TBB_ASSERT(this->my_item_valid(this->my_head), "No item to reset");
this->destroy_front();
if(this->my_item_valid(this->my_head)) {
(void)my_join->decrement_port_count(true);
}
current->status.store( SUCCEEDED, std::memory_order_release);
break;
}
}
}
// ------------ End Aggregator ---------------
protected:
template< typename R, typename B > friend class run_and_put_task;
template<typename X, typename Y> friend class broadcast_cache;
template<typename X, typename Y> friend class round_robin_cache;
graph_task* try_put_task(const T &v) override {
queueing_port_operation op_data(v, try__put_task);
my_aggregator.execute(&op_data);
__TBB_ASSERT(op_data.status == SUCCEEDED || !op_data.bypass_t, "inconsistent return from aggregator");
if(!op_data.bypass_t) return SUCCESSFULLY_ENQUEUED;
return op_data.bypass_t;
}
graph& graph_reference() const override {
return my_join->graph_ref;
}
public:
//! Constructor
queueing_port() : item_buffer<T>() {
my_join = nullptr;
my_aggregator.initialize_handler(handler_type(this));
}
//! copy constructor
queueing_port(const queueing_port& /* other */) = delete;
//! record parent for tallying available items
void set_join_node_pointer(queueing_forwarding_base *join) {
my_join = join;
}
bool get_item( T &v ) {
queueing_port_operation op_data(&v, get__item);
my_aggregator.execute(&op_data);
return op_data.status == SUCCEEDED;
}
// reset_port is called when item is accepted by successor, but
// is initiated by join_node.
void reset_port() {
queueing_port_operation op_data(res_port);
my_aggregator.execute(&op_data);
return;
}
void reset_receiver(reset_flags) {
item_buffer<T>::reset();
}
private:
#if __TBB_PREVIEW_FLOW_GRAPH_NODE_SET
friend class get_graph_helper;
#endif
queueing_forwarding_base *my_join;
}; // queueing_port
#include "_flow_graph_tagged_buffer_impl.h"
template<typename K>
struct count_element {
K my_key;
size_t my_value;
};
// method to access the key in the counting table
// the ref has already been removed from K
template< typename K >
struct key_to_count_functor {
typedef count_element<K> table_item_type;
const K& operator()(const table_item_type& v) { return v.my_key; }
};
// the ports can have only one template parameter. We wrap the types needed in
// a traits type
template< class TraitsType >
class key_matching_port :
public receiver<typename TraitsType::T>,
public hash_buffer< typename TraitsType::K, typename TraitsType::T, typename TraitsType::TtoK,
typename TraitsType::KHash > {
public:
typedef TraitsType traits;
typedef key_matching_port<traits> class_type;
typedef typename TraitsType::T input_type;
typedef typename TraitsType::K key_type;
typedef typename std::decay<key_type>::type noref_key_type;
typedef typename receiver<input_type>::predecessor_type predecessor_type;
typedef typename TraitsType::TtoK type_to_key_func_type;
typedef typename TraitsType::KHash hash_compare_type;
typedef hash_buffer< key_type, input_type, type_to_key_func_type, hash_compare_type > buffer_type;
private:
// ----------- Aggregator ------------
private:
enum op_type { try__put, get__item, res_port
};
class key_matching_port_operation : public aggregated_operation<key_matching_port_operation> {
public:
char type;
input_type my_val;
input_type *my_arg;
// constructor for value parameter
key_matching_port_operation(const input_type& e, op_type t) :
type(char(t)), my_val(e), my_arg(nullptr) {}
// constructor for pointer parameter
key_matching_port_operation(const input_type* p, op_type t) :
type(char(t)), my_arg(const_cast<input_type*>(p)) {}
// constructor with no parameter
key_matching_port_operation(op_type t) : type(char(t)), my_arg(nullptr) {}
};
typedef aggregating_functor<class_type, key_matching_port_operation> handler_type;
friend class aggregating_functor<class_type, key_matching_port_operation>;
aggregator<handler_type, key_matching_port_operation> my_aggregator;
void handle_operations(key_matching_port_operation* op_list) {
key_matching_port_operation *current;
while(op_list) {
current = op_list;
op_list = op_list->next;
switch(current->type) {
case try__put: {
bool was_inserted = this->insert_with_key(current->my_val);
// return failure if a duplicate insertion occurs
current->status.store( was_inserted ? SUCCEEDED : FAILED, std::memory_order_release);
}
break;
case get__item:
// use current_key from FE for item
__TBB_ASSERT(current->my_arg, nullptr);
if(!this->find_with_key(my_join->current_key, *(current->my_arg))) {
__TBB_ASSERT(false, "Failed to find item corresponding to current_key.");
}
current->status.store( SUCCEEDED, std::memory_order_release);
break;
case res_port:
// use current_key from FE for item
this->delete_with_key(my_join->current_key);
current->status.store( SUCCEEDED, std::memory_order_release);
break;
}
}
}
// ------------ End Aggregator ---------------
protected:
template< typename R, typename B > friend class run_and_put_task;
template<typename X, typename Y> friend class broadcast_cache;
template<typename X, typename Y> friend class round_robin_cache;
graph_task* try_put_task(const input_type& v) override {
key_matching_port_operation op_data(v, try__put);
graph_task* rtask = nullptr;
my_aggregator.execute(&op_data);
if(op_data.status == SUCCEEDED) {
rtask = my_join->increment_key_count((*(this->get_key_func()))(v)); // may spawn
// rtask has to reflect the return status of the try_put
if(!rtask) rtask = SUCCESSFULLY_ENQUEUED;
}
return rtask;
}
graph& graph_reference() const override {
return my_join->graph_ref;
}
public:
key_matching_port() : receiver<input_type>(), buffer_type() {
my_join = nullptr;
my_aggregator.initialize_handler(handler_type(this));
}
// copy constructor
key_matching_port(const key_matching_port& /*other*/) = delete;
#if __INTEL_COMPILER <= 2021
// Suppress superfluous diagnostic about virtual keyword absence in a destructor of an inherited
// class while the parent class has the virtual keyword for the destrocutor.
virtual
#endif
~key_matching_port() { }
void set_join_node_pointer(forwarding_base *join) {
my_join = dynamic_cast<matching_forwarding_base<key_type>*>(join);
}
void set_my_key_func(type_to_key_func_type *f) { this->set_key_func(f); }
type_to_key_func_type* get_my_key_func() { return this->get_key_func(); }
bool get_item( input_type &v ) {
// aggregator uses current_key from FE for Key
key_matching_port_operation op_data(&v, get__item);
my_aggregator.execute(&op_data);
return op_data.status == SUCCEEDED;
}
// reset_port is called when item is accepted by successor, but
// is initiated by join_node.
void reset_port() {
key_matching_port_operation op_data(res_port);
my_aggregator.execute(&op_data);
return;
}
void reset_receiver(reset_flags ) {
buffer_type::reset();
}
private:
// my_join forwarding base used to count number of inputs that
// received key.
matching_forwarding_base<key_type> *my_join;
}; // key_matching_port
using namespace graph_policy_namespace;
template<typename JP, typename InputTuple, typename OutputTuple>
class join_node_base;
//! join_node_FE : implements input port policy
template<typename JP, typename InputTuple, typename OutputTuple>
class join_node_FE;
template<typename InputTuple, typename OutputTuple>
class join_node_FE<reserving, InputTuple, OutputTuple> : public reserving_forwarding_base {
private:
static const int N = std::tuple_size<OutputTuple>::value;
typedef OutputTuple output_type;
typedef InputTuple input_type;
typedef join_node_base<reserving, InputTuple, OutputTuple> base_node_type; // for forwarding
public:
join_node_FE(graph &g) : reserving_forwarding_base(g), my_node(nullptr) {
ports_with_no_inputs = N;
join_helper<N>::set_join_node_pointer(my_inputs, this);
}
join_node_FE(const join_node_FE& other) : reserving_forwarding_base((other.reserving_forwarding_base::graph_ref)), my_node(nullptr) {
ports_with_no_inputs = N;
join_helper<N>::set_join_node_pointer(my_inputs, this);
}
void set_my_node(base_node_type *new_my_node) { my_node = new_my_node; }
void increment_port_count() override {
++ports_with_no_inputs;
}
// if all input_ports have predecessors, spawn forward to try and consume tuples
graph_task* decrement_port_count() override {
if(ports_with_no_inputs.fetch_sub(1) == 1) {
if(is_graph_active(this->graph_ref)) {
small_object_allocator allocator{};
typedef forward_task_bypass<base_node_type> task_type;
graph_task* t = allocator.new_object<task_type>(graph_ref, allocator, *my_node);
graph_ref.reserve_wait();
spawn_in_graph_arena(this->graph_ref, *t);
}
}
return nullptr;
}
input_type &input_ports() { return my_inputs; }
protected:
void reset( reset_flags f) {
// called outside of parallel contexts
ports_with_no_inputs = N;
join_helper<N>::reset_inputs(my_inputs, f);
}
// all methods on input ports should be called under mutual exclusion from join_node_base.
bool tuple_build_may_succeed() {
return !ports_with_no_inputs;
}
bool try_to_make_tuple(output_type &out) {
if(ports_with_no_inputs) return false;
return join_helper<N>::reserve(my_inputs, out);
}
void tuple_accepted() {
join_helper<N>::consume_reservations(my_inputs);
}
void tuple_rejected() {
join_helper<N>::release_reservations(my_inputs);
}
input_type my_inputs;
base_node_type *my_node;
std::atomic<std::size_t> ports_with_no_inputs;
}; // join_node_FE<reserving, ... >
template<typename InputTuple, typename OutputTuple>
class join_node_FE<queueing, InputTuple, OutputTuple> : public queueing_forwarding_base {
public:
static const int N = std::tuple_size<OutputTuple>::value;
typedef OutputTuple output_type;
typedef InputTuple input_type;
typedef join_node_base<queueing, InputTuple, OutputTuple> base_node_type; // for forwarding
join_node_FE(graph &g) : queueing_forwarding_base(g), my_node(nullptr) {
ports_with_no_items = N;
join_helper<N>::set_join_node_pointer(my_inputs, this);
}
join_node_FE(const join_node_FE& other) : queueing_forwarding_base((other.queueing_forwarding_base::graph_ref)), my_node(nullptr) {
ports_with_no_items = N;
join_helper<N>::set_join_node_pointer(my_inputs, this);
}
// needed for forwarding
void set_my_node(base_node_type *new_my_node) { my_node = new_my_node; }
void reset_port_count() {
ports_with_no_items = N;
}
// if all input_ports have items, spawn forward to try and consume tuples
graph_task* decrement_port_count(bool handle_task) override
{
if(ports_with_no_items.fetch_sub(1) == 1) {
if(is_graph_active(this->graph_ref)) {
small_object_allocator allocator{};
typedef forward_task_bypass<base_node_type> task_type;
graph_task* t = allocator.new_object<task_type>(graph_ref, allocator, *my_node);
graph_ref.reserve_wait();
if( !handle_task )
return t;
spawn_in_graph_arena(this->graph_ref, *t);
}
}
return nullptr;
}
input_type &input_ports() { return my_inputs; }
protected:
void reset( reset_flags f) {
reset_port_count();
join_helper<N>::reset_inputs(my_inputs, f );
}
// all methods on input ports should be called under mutual exclusion from join_node_base.
bool tuple_build_may_succeed() {
return !ports_with_no_items;
}
bool try_to_make_tuple(output_type &out) {
if(ports_with_no_items) return false;
return join_helper<N>::get_items(my_inputs, out);
}
void tuple_accepted() {
reset_port_count();
join_helper<N>::reset_ports(my_inputs);
}
void tuple_rejected() {
// nothing to do.
}
input_type my_inputs;
base_node_type *my_node;
std::atomic<std::size_t> ports_with_no_items;
}; // join_node_FE<queueing, ...>
// key_matching join front-end.
template<typename InputTuple, typename OutputTuple, typename K, typename KHash>
class join_node_FE<key_matching<K,KHash>, InputTuple, OutputTuple> : public matching_forwarding_base<K>,
// buffer of key value counts
public hash_buffer< // typedefed below to key_to_count_buffer_type
typename std::decay<K>::type&, // force ref type on K
count_element<typename std::decay<K>::type>,
type_to_key_function_body<
count_element<typename std::decay<K>::type>,
typename std::decay<K>::type& >,
KHash >,
// buffer of output items
public item_buffer<OutputTuple> {
public:
static const int N = std::tuple_size<OutputTuple>::value;
typedef OutputTuple output_type;
typedef InputTuple input_type;
typedef K key_type;
typedef typename std::decay<key_type>::type unref_key_type;
typedef KHash key_hash_compare;
// must use K without ref.
typedef count_element<unref_key_type> count_element_type;
// method that lets us refer to the key of this type.
typedef key_to_count_functor<unref_key_type> key_to_count_func;
typedef type_to_key_function_body< count_element_type, unref_key_type&> TtoK_function_body_type;
typedef type_to_key_function_body_leaf<count_element_type, unref_key_type&, key_to_count_func> TtoK_function_body_leaf_type;
// this is the type of the special table that keeps track of the number of discrete
// elements corresponding to each key that we've seen.
typedef hash_buffer< unref_key_type&, count_element_type, TtoK_function_body_type, key_hash_compare >
key_to_count_buffer_type;
typedef item_buffer<output_type> output_buffer_type;
typedef join_node_base<key_matching<key_type,key_hash_compare>, InputTuple, OutputTuple> base_node_type; // for forwarding
typedef matching_forwarding_base<key_type> forwarding_base_type;
// ----------- Aggregator ------------
// the aggregator is only needed to serialize the access to the hash table.
// and the output_buffer_type base class
private:
enum op_type { res_count, inc_count, may_succeed, try_make };
typedef join_node_FE<key_matching<key_type,key_hash_compare>, InputTuple, OutputTuple> class_type;
class key_matching_FE_operation : public aggregated_operation<key_matching_FE_operation> {
public:
char type;
unref_key_type my_val;
output_type* my_output;
graph_task* bypass_t;
// constructor for value parameter
key_matching_FE_operation(const unref_key_type& e , op_type t) : type(char(t)), my_val(e),
my_output(nullptr), bypass_t(nullptr) {}
key_matching_FE_operation(output_type *p, op_type t) : type(char(t)), my_output(p), bypass_t(nullptr) {}
// constructor with no parameter
key_matching_FE_operation(op_type t) : type(char(t)), my_output(nullptr), bypass_t(nullptr) {}
};
typedef aggregating_functor<class_type, key_matching_FE_operation> handler_type;
friend class aggregating_functor<class_type, key_matching_FE_operation>;
aggregator<handler_type, key_matching_FE_operation> my_aggregator;
// called from aggregator, so serialized
// returns a task pointer if the a task would have been enqueued but we asked that
// it be returned. Otherwise returns nullptr.
graph_task* fill_output_buffer(unref_key_type &t) {
output_type l_out;
graph_task* rtask = nullptr;
bool do_fwd = this->buffer_empty() && is_graph_active(this->graph_ref);
this->current_key = t;
this->delete_with_key(this->current_key); // remove the key
if(join_helper<N>::get_items(my_inputs, l_out)) { // <== call back
this->push_back(l_out);
if(do_fwd) { // we enqueue if receiving an item from predecessor, not if successor asks for item
small_object_allocator allocator{};
typedef forward_task_bypass<base_node_type> task_type;
rtask = allocator.new_object<task_type>(this->graph_ref, allocator, *my_node);
this->graph_ref.reserve_wait();
do_fwd = false;
}
// retire the input values
join_helper<N>::reset_ports(my_inputs); // <== call back
}
else {
__TBB_ASSERT(false, "should have had something to push");
}
return rtask;
}
void handle_operations(key_matching_FE_operation* op_list) {
key_matching_FE_operation *current;
while(op_list) {
current = op_list;
op_list = op_list->next;
switch(current->type) {
case res_count: // called from BE
{
this->destroy_front();
current->status.store( SUCCEEDED, std::memory_order_release);
}
break;
case inc_count: { // called from input ports
count_element_type *p = nullptr;
unref_key_type &t = current->my_val;
if(!(this->find_ref_with_key(t,p))) {
count_element_type ev;
ev.my_key = t;
ev.my_value = 0;
this->insert_with_key(ev);
bool found = this->find_ref_with_key(t, p);
__TBB_ASSERT_EX(found, "should find key after inserting it");
}
if(++(p->my_value) == size_t(N)) {
current->bypass_t = fill_output_buffer(t);
}
}
current->status.store( SUCCEEDED, std::memory_order_release);
break;
case may_succeed: // called from BE
current->status.store( this->buffer_empty() ? FAILED : SUCCEEDED, std::memory_order_release);
break;
case try_make: // called from BE
if(this->buffer_empty()) {
current->status.store( FAILED, std::memory_order_release);
}
else {
*(current->my_output) = this->front();
current->status.store( SUCCEEDED, std::memory_order_release);
}
break;
}
}
}
// ------------ End Aggregator ---------------
public:
template<typename FunctionTuple>
join_node_FE(graph &g, FunctionTuple &TtoK_funcs) : forwarding_base_type(g), my_node(nullptr) {
join_helper<N>::set_join_node_pointer(my_inputs, this);
join_helper<N>::set_key_functors(my_inputs, TtoK_funcs);
my_aggregator.initialize_handler(handler_type(this));
TtoK_function_body_type *cfb = new TtoK_function_body_leaf_type(key_to_count_func());
this->set_key_func(cfb);
}
join_node_FE(const join_node_FE& other) : forwarding_base_type((other.forwarding_base_type::graph_ref)), key_to_count_buffer_type(),
output_buffer_type() {
my_node = nullptr;
join_helper<N>::set_join_node_pointer(my_inputs, this);
join_helper<N>::copy_key_functors(my_inputs, const_cast<input_type &>(other.my_inputs));
my_aggregator.initialize_handler(handler_type(this));
TtoK_function_body_type *cfb = new TtoK_function_body_leaf_type(key_to_count_func());
this->set_key_func(cfb);
}
// needed for forwarding
void set_my_node(base_node_type *new_my_node) { my_node = new_my_node; }
void reset_port_count() { // called from BE
key_matching_FE_operation op_data(res_count);
my_aggregator.execute(&op_data);
return;
}
// if all input_ports have items, spawn forward to try and consume tuples
// return a task if we are asked and did create one.
graph_task *increment_key_count(unref_key_type const & t) override { // called from input_ports
key_matching_FE_operation op_data(t, inc_count);
my_aggregator.execute(&op_data);
return op_data.bypass_t;
}
input_type &input_ports() { return my_inputs; }
protected:
void reset( reset_flags f ) {
// called outside of parallel contexts
join_helper<N>::reset_inputs(my_inputs, f);
key_to_count_buffer_type::reset();
output_buffer_type::reset();
}
// all methods on input ports should be called under mutual exclusion from join_node_base.
bool tuple_build_may_succeed() { // called from back-end
key_matching_FE_operation op_data(may_succeed);