1 /*
2 Copyright (c) 2020-2021 Intel Corporation
3
4 Licensed under the Apache License, Version 2.0 (the "License");
5 you may not use this file except in compliance with the License.
6 You may obtain a copy of the License at
7
8 http://www.apache.org/licenses/LICENSE-2.0
9
10 Unless required by applicable law or agreed to in writing, software
11 distributed under the License is distributed on an "AS IS" BASIS,
12 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13 See the License for the specific language governing permissions and
14 limitations under the License.
15 */
16
17 #include "oneapi/tbb/detail/_utils.h"
18 #include "oneapi/tbb/concurrent_queue.h"
19 #include "oneapi/tbb/cache_aligned_allocator.h"
20 #include "concurrent_monitor.h"
21
22 namespace tbb {
23 namespace detail {
24 namespace r1 {
25
26 static constexpr std::size_t monitors_number = 2;
27
allocate_bounded_queue_rep(std::size_t queue_rep_size)28 std::uint8_t* __TBB_EXPORTED_FUNC allocate_bounded_queue_rep( std::size_t queue_rep_size )
29 {
30 std::size_t monitors_mem_size = sizeof(concurrent_monitor) * monitors_number;
31 std::uint8_t* mem = static_cast<std::uint8_t*>(cache_aligned_allocate(queue_rep_size + monitors_mem_size));
32
33 concurrent_monitor* monitors = reinterpret_cast<concurrent_monitor*>(mem + queue_rep_size);
34 for (std::size_t i = 0; i < monitors_number; ++i) {
35 new (monitors + i) concurrent_monitor();
36 }
37
38 return mem;
39 }
40
deallocate_bounded_queue_rep(std::uint8_t * mem,std::size_t queue_rep_size)41 void __TBB_EXPORTED_FUNC deallocate_bounded_queue_rep( std::uint8_t* mem, std::size_t queue_rep_size )
42 {
43 concurrent_monitor* monitors = reinterpret_cast<concurrent_monitor*>(mem + queue_rep_size);
44 for (std::size_t i = 0; i < monitors_number; ++i) {
45 monitors[i].~concurrent_monitor();
46 }
47
48 cache_aligned_deallocate(mem);
49 }
50
wait_bounded_queue_monitor(concurrent_monitor * monitors,std::size_t monitor_tag,std::ptrdiff_t target,d1::delegate_base & predicate)51 void __TBB_EXPORTED_FUNC wait_bounded_queue_monitor( concurrent_monitor* monitors, std::size_t monitor_tag,
52 std::ptrdiff_t target, d1::delegate_base& predicate )
53 {
54 __TBB_ASSERT(monitor_tag < monitors_number, nullptr);
55 concurrent_monitor& monitor = monitors[monitor_tag];
56
57 monitor.wait<concurrent_monitor::thread_context>([&] { return !predicate(); }, std::uintptr_t(target));
58 }
59
abort_bounded_queue_monitors(concurrent_monitor * monitors)60 void __TBB_EXPORTED_FUNC abort_bounded_queue_monitors( concurrent_monitor* monitors ) {
61 concurrent_monitor& items_avail = monitors[d2::cbq_items_avail_tag];
62 concurrent_monitor& slots_avail = monitors[d2::cbq_slots_avail_tag];
63
64 items_avail.abort_all();
65 slots_avail.abort_all();
66 }
67
68 struct predicate_leq {
69 std::size_t my_ticket;
predicate_leqtbb::detail::r1::predicate_leq70 predicate_leq( std::size_t ticket ) : my_ticket(ticket) {}
operator ()tbb::detail::r1::predicate_leq71 bool operator() ( std::uintptr_t ticket ) const { return static_cast<std::size_t>(ticket) <= my_ticket; }
72 };
73
notify_bounded_queue_monitor(concurrent_monitor * monitors,std::size_t monitor_tag,std::size_t ticket)74 void __TBB_EXPORTED_FUNC notify_bounded_queue_monitor( concurrent_monitor* monitors,
75 std::size_t monitor_tag, std::size_t ticket)
76 {
77 __TBB_ASSERT(monitor_tag < monitors_number, nullptr);
78 concurrent_monitor& monitor = monitors[monitor_tag];
79 monitor.notify(predicate_leq(ticket));
80 }
81
82 } // namespace r1
83 } // namespace detail
84 } // namespace tbb
85