_aggregator_impl.h 7.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180
  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__aggregator_impl_H
  14. #define __TBB__aggregator_impl_H
  15. #include "../atomic.h"
  16. #if !__TBBMALLOC_BUILD
  17. #include "../tbb_profiling.h"
  18. #endif
  19. namespace tbb {
  20. namespace interface6 {
  21. namespace internal {
  22. using namespace tbb::internal;
  23. //! aggregated_operation base class
  24. template <typename Derived>
  25. class aggregated_operation {
  26. public:
  27. //! Zero value means "wait" status, all other values are "user" specified values and are defined into the scope of a class which uses "status".
  28. uintptr_t status;
  29. Derived *next;
  30. aggregated_operation() : status(0), next(NULL) {}
  31. };
  32. //! Aggregator base class
  33. /** An aggregator for collecting operations coming from multiple sources and executing
  34. them serially on a single thread. operation_type must be derived from
  35. aggregated_operation. The parameter handler_type is a functor that will be passed the
  36. list of operations and is expected to handle each operation appropriately, setting the
  37. status of each operation to non-zero.*/
  38. template < typename operation_type >
  39. class aggregator_generic {
  40. public:
  41. aggregator_generic() : handler_busy(false) { pending_operations = NULL; }
  42. //! Execute an operation
  43. /** Places an operation into the waitlist (pending_operations), and either handles the list,
  44. or waits for the operation to complete, or returns.
  45. The long_life_time parameter specifies the life time of the given operation object.
  46. Operations with long_life_time == true may be accessed after execution.
  47. A "short" life time operation (long_life_time == false) can be destroyed
  48. during execution, and so any access to it after it was put into the waitlist,
  49. including status check, is invalid. As a consequence, waiting for completion
  50. of such operation causes undefined behavior.
  51. */
  52. template < typename handler_type >
  53. void execute(operation_type *op, handler_type &handle_operations, bool long_life_time = true) {
  54. operation_type *res;
  55. // op->status should be read before inserting the operation into the
  56. // aggregator waitlist since it can become invalid after executing a
  57. // handler (if the operation has 'short' life time.)
  58. const uintptr_t status = op->status;
  59. // ITT note: &(op->status) tag is used to cover accesses to this op node. This
  60. // thread has created the operation, and now releases it so that the handler
  61. // thread may handle the associated operation w/o triggering a race condition;
  62. // thus this tag will be acquired just before the operation is handled in the
  63. // handle_operations functor.
  64. call_itt_notify(releasing, &(op->status));
  65. // insert the operation in the queue.
  66. do {
  67. // Tools may flag the following line as a race; it is a false positive:
  68. // This is an atomic read; we don't provide itt_hide_load_word for atomics
  69. op->next = res = pending_operations; // NOT A RACE
  70. } while (pending_operations.compare_and_swap(op, res) != res);
  71. if (!res) { // first in the list; handle the operations.
  72. // ITT note: &pending_operations tag covers access to the handler_busy flag,
  73. // which this waiting handler thread will try to set before entering
  74. // handle_operations.
  75. call_itt_notify(acquired, &pending_operations);
  76. start_handle_operations(handle_operations);
  77. // The operation with 'short' life time can already be destroyed.
  78. if (long_life_time)
  79. __TBB_ASSERT(op->status, NULL);
  80. }
  81. // not first; wait for op to be ready.
  82. else if (!status) { // operation is blocking here.
  83. __TBB_ASSERT(long_life_time, "Waiting for an operation object that might be destroyed during processing.");
  84. call_itt_notify(prepare, &(op->status));
  85. spin_wait_while_eq(op->status, uintptr_t(0));
  86. itt_load_word_with_acquire(op->status);
  87. }
  88. }
  89. private:
  90. //! An atomically updated list (aka mailbox) of pending operations
  91. atomic<operation_type *> pending_operations;
  92. //! Controls thread access to handle_operations
  93. uintptr_t handler_busy;
  94. //! Trigger the handling of operations when the handler is free
  95. template < typename handler_type >
  96. void start_handle_operations( handler_type &handle_operations ) {
  97. operation_type *op_list;
  98. // ITT note: &handler_busy tag covers access to pending_operations as it is passed
  99. // between active and waiting handlers. Below, the waiting handler waits until
  100. // the active handler releases, and the waiting handler acquires &handler_busy as
  101. // it becomes the active_handler. The release point is at the end of this
  102. // function, when all operations in pending_operations have been handled by the
  103. // owner of this aggregator.
  104. call_itt_notify(prepare, &handler_busy);
  105. // get the handler_busy:
  106. // only one thread can possibly spin here at a time
  107. spin_wait_until_eq(handler_busy, uintptr_t(0));
  108. call_itt_notify(acquired, &handler_busy);
  109. // acquire fence not necessary here due to causality rule and surrounding atomics
  110. __TBB_store_with_release(handler_busy, uintptr_t(1));
  111. // ITT note: &pending_operations tag covers access to the handler_busy flag
  112. // itself. Capturing the state of the pending_operations signifies that
  113. // handler_busy has been set and a new active handler will now process that list's
  114. // operations.
  115. call_itt_notify(releasing, &pending_operations);
  116. // grab pending_operations
  117. op_list = pending_operations.fetch_and_store(NULL);
  118. // handle all the operations
  119. handle_operations(op_list);
  120. // release the handler
  121. itt_store_word_with_release(handler_busy, uintptr_t(0));
  122. }
  123. };
  124. template < typename handler_type, typename operation_type >
  125. class aggregator : public aggregator_generic<operation_type> {
  126. handler_type handle_operations;
  127. public:
  128. aggregator() {}
  129. explicit aggregator(handler_type h) : handle_operations(h) {}
  130. void initialize_handler(handler_type h) { handle_operations = h; }
  131. void execute(operation_type *op) {
  132. aggregator_generic<operation_type>::execute(op, handle_operations);
  133. }
  134. };
  135. // the most-compatible friend declaration (vs, gcc, icc) is
  136. // template<class U, class V> friend class aggregating_functor;
  137. template<typename aggregating_class, typename operation_list>
  138. class aggregating_functor {
  139. aggregating_class *fi;
  140. public:
  141. aggregating_functor() : fi() {}
  142. aggregating_functor(aggregating_class *fi_) : fi(fi_) {}
  143. void operator()(operation_list* op_list) { fi->handle_operations(op_list); }
  144. };
  145. } // namespace internal
  146. } // namespace interface6
  147. namespace internal {
  148. using interface6::internal::aggregated_operation;
  149. using interface6::internal::aggregator_generic;
  150. using interface6::internal::aggregator;
  151. using interface6::internal::aggregating_functor;
  152. } // namespace internal
  153. } // namespace tbb
  154. #endif // __TBB__aggregator_impl_H