1 /*
2 * Copyright (c) 2021, 2025, Oracle and/or its affiliates. All rights reserved.
3 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
4 *
5 * This code is free software; you can redistribute it and/or modify it
6 * under the terms of the GNU General Public License version 2 only, as
7 * published by the Free Software Foundation.
8 *
9 * This code is distributed in the hope that it will be useful, but WITHOUT
10 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
11 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
12 * version 2 for more details (a copy is included in the LICENSE file that
13 * accompanied this code).
14 *
15 * You should have received a copy of the GNU General Public License version
16 * 2 along with this work; if not, write to the Free Software Foundation,
17 * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
18 *
19 * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
20 * or visit www.oracle.com if you need additional information or have any
21 * questions.
22 */
23
24 #include "memory/allocation.inline.hpp"
25 #include "runtime/atomicAccess.hpp"
26 #include "utilities/globalDefinitions.hpp"
27 #include "utilities/nonblockingQueue.inline.hpp"
28 #include "utilities/pair.hpp"
29 #include "threadHelper.inline.hpp"
30 #include "unittest.hpp"
31
32 #include <new>
33
34 class NonblockingQueueTestElement {
35 typedef NonblockingQueueTestElement Element;
36
37 Element* volatile _entry;
38 Element* volatile _entry1;
39 size_t _id;
40
41 static Element* volatile* entry_ptr(Element& e) { return &e._entry; }
42 static Element* volatile* entry1_ptr(Element& e) { return &e._entry1; }
43
44 public:
45 using TestQueue = NonblockingQueue<Element, &entry_ptr>;
46 using TestQueue1 = NonblockingQueue<Element, &entry1_ptr>;
47
48 NonblockingQueueTestElement(size_t id = 0) : _entry(), _entry1(), _id(id) {}
49 size_t id() const { return _id; }
50 void set_id(size_t value) { _id = value; }
51 Element* next() { return _entry; }
52 Element* next1() { return _entry1; }
53 };
54
55 typedef NonblockingQueueTestElement Element;
56 typedef Element::TestQueue TestQueue;
57 typedef Element::TestQueue1 TestQueue1;
58
59 static void initialize(Element* elements, size_t size, TestQueue* queue) {
60 for (size_t i = 0; i < size; ++i) {
61 elements[i].set_id(i);
62 }
63 ASSERT_TRUE(queue->empty());
64 ASSERT_EQ(0u, queue->length());
65 ASSERT_TRUE(queue->is_end(queue->first()));
66 ASSERT_TRUE(queue->pop() == nullptr);
67
68 for (size_t id = 0; id < size; ++id) {
69 ASSERT_EQ(id, queue->length());
70 Element* e = &elements[id];
71 ASSERT_EQ(id, e->id());
72 queue->push(*e);
73 ASSERT_FALSE(queue->empty());
74 // first() is always the oldest element.
75 ASSERT_EQ(&elements[0], queue->first());
76 }
77 }
78
79 class NonblockingQueueTestBasics : public ::testing::Test {
80 public:
81 NonblockingQueueTestBasics();
82
83 static const size_t nelements = 10;
84 Element elements[nelements];
85 TestQueue queue;
86 };
87
88 const size_t NonblockingQueueTestBasics::nelements;
89
90 NonblockingQueueTestBasics::NonblockingQueueTestBasics() : queue() {
91 initialize(elements, nelements, &queue);
92 }
93
94 TEST_F(NonblockingQueueTestBasics, pop) {
95 for (size_t i = 0; i < nelements; ++i) {
96 ASSERT_FALSE(queue.empty());
97 ASSERT_EQ(nelements - i, queue.length());
98 Element* e = queue.pop();
99 ASSERT_TRUE(e != nullptr);
100 ASSERT_EQ(&elements[i], e);
101 ASSERT_EQ(i, e->id());
102 }
103 ASSERT_TRUE(queue.empty());
104 ASSERT_EQ(0u, queue.length());
105 ASSERT_TRUE(queue.pop() == nullptr);
106 }
107
108 TEST_F(NonblockingQueueTestBasics, append) {
109 TestQueue other_queue;
110 ASSERT_TRUE(other_queue.empty());
111 ASSERT_EQ(0u, other_queue.length());
112 ASSERT_TRUE(other_queue.is_end(other_queue.first()));
113 ASSERT_TRUE(other_queue.pop() == nullptr);
114
115 Pair<Element*, Element*> pair = queue.take_all();
116 other_queue.append(*pair.first, *pair.second);
117 ASSERT_EQ(nelements, other_queue.length());
118 ASSERT_TRUE(queue.empty());
119 ASSERT_EQ(0u, queue.length());
120 ASSERT_TRUE(queue.is_end(queue.first()));
121 ASSERT_TRUE(queue.pop() == nullptr);
122
123 for (size_t i = 0; i < nelements; ++i) {
124 ASSERT_EQ(nelements - i, other_queue.length());
125 Element* e = other_queue.pop();
126 ASSERT_TRUE(e != nullptr);
127 ASSERT_EQ(&elements[i], e);
128 ASSERT_EQ(i, e->id());
129 }
130 ASSERT_EQ(0u, other_queue.length());
131 ASSERT_TRUE(other_queue.pop() == nullptr);
132 }
133
134 TEST_F(NonblockingQueueTestBasics, two_queues) {
135 TestQueue1 queue1;
136 ASSERT_TRUE(queue1.pop() == nullptr);
137
138 for (size_t id = 0; id < nelements; ++id) {
139 queue1.push(elements[id]);
140 }
141 ASSERT_EQ(nelements, queue1.length());
142 Element* e0 = queue.first();
143 Element* e1 = queue1.first();
144 ASSERT_TRUE(e0 != nullptr);
145 ASSERT_TRUE(e1 != nullptr);
146 ASSERT_FALSE(queue.is_end(e0));
147 ASSERT_FALSE(queue1.is_end(e1));
148 while (!queue.is_end(e0) && !queue1.is_end(e1)) {
149 ASSERT_EQ(e0, e1);
150 e0 = e0->next();
151 e1 = e1->next1();
152 }
153 ASSERT_TRUE(queue.is_end(e0));
154 ASSERT_TRUE(queue1.is_end(e1));
155
156 for (size_t i = 0; i < nelements; ++i) {
157 ASSERT_EQ(nelements - i, queue.length());
158 ASSERT_EQ(nelements - i, queue1.length());
159
160 Element* e = queue.pop();
161 ASSERT_TRUE(e != nullptr);
162 ASSERT_EQ(&elements[i], e);
163 ASSERT_EQ(i, e->id());
164
165 Element* e1 = queue1.pop();
166 ASSERT_TRUE(e1 != nullptr);
167 ASSERT_EQ(&elements[i], e1);
168 ASSERT_EQ(i, e1->id());
169
170 ASSERT_EQ(e, e1);
171 }
172 ASSERT_EQ(0u, queue.length());
173 ASSERT_EQ(0u, queue1.length());
174 ASSERT_TRUE(queue.pop() == nullptr);
175 ASSERT_TRUE(queue1.pop() == nullptr);
176 }
177
178 class NonblockingQueueTestThread : public JavaTestThread {
179 uint _id;
180 TestQueue* _from;
181 TestQueue* _to;
182 volatile size_t* _processed;
183 size_t _process_limit;
184 size_t _local_processed;
185 volatile bool _ready;
186
187 public:
188 NonblockingQueueTestThread(Semaphore* post,
189 uint id,
190 TestQueue* from,
191 TestQueue* to,
192 volatile size_t* processed,
193 size_t process_limit) :
194 JavaTestThread(post),
195 _id(id),
196 _from(from),
197 _to(to),
198 _processed(processed),
199 _process_limit(process_limit),
200 _local_processed(0),
201 _ready(false)
202 {}
203
204 virtual void main_run() {
205 AtomicAccess::release_store_fence(&_ready, true);
206 while (true) {
207 Element* e = _from->pop();
208 if (e != nullptr) {
209 _to->push(*e);
210 AtomicAccess::inc(_processed);
211 ++_local_processed;
212 } else if (AtomicAccess::load_acquire(_processed) == _process_limit) {
213 tty->print_cr("thread %u processed %zu", _id, _local_processed);
214 return;
215 }
216 }
217 }
218
219 bool ready() const { return AtomicAccess::load_acquire(&_ready); }
220 };
221
222 TEST_VM(NonblockingQueueTest, stress) {
223 Semaphore post;
224 TestQueue initial_queue;
225 TestQueue start_queue;
226 TestQueue middle_queue;
227 TestQueue final_queue;
228 volatile size_t stage1_processed = 0;
229 volatile size_t stage2_processed = 0;
230
231 const size_t nelements = 10000;
232 Element* elements = NEW_C_HEAP_ARRAY(Element, nelements, mtOther);
233 for (size_t id = 0; id < nelements; ++id) {
234 ::new (&elements[id]) Element(id);
235 initial_queue.push(elements[id]);
236 }
237 ASSERT_EQ(nelements, initial_queue.length());
238
239 // - stage1 threads pop from start_queue and push to middle_queue.
240 // - stage2 threads pop from middle_queue and push to final_queue.
241 // - all threads in a stage count the number of elements processed in
242 // their corresponding stageN_processed counter.
243
244 const uint stage1_threads = 2;
245 const uint stage2_threads = 2;
246 const uint nthreads = stage1_threads + stage2_threads;
247 NonblockingQueueTestThread* threads[nthreads] = {};
248
249 for (uint i = 0; i < ARRAY_SIZE(threads); ++i) {
250 TestQueue* from = &start_queue;
251 TestQueue* to = &middle_queue;
252 volatile size_t* processed = &stage1_processed;
253 if (i >= stage1_threads) {
254 from = &middle_queue;
255 to = &final_queue;
256 processed = &stage2_processed;
257 }
258 threads[i] =
259 new NonblockingQueueTestThread(&post, i, from, to, processed, nelements);
260 threads[i]->doit();
261 while (!threads[i]->ready()) {} // Wait until ready to start test.
262 }
263
264 // Transfer elements to start_queue to start test.
265 Pair<Element*, Element*> pair = initial_queue.take_all();
266 start_queue.append(*pair.first, *pair.second);
267
268 // Wait for all threads to complete.
269 for (uint i = 0; i < nthreads; ++i) {
270 post.wait();
271 }
272
273 // Verify expected state.
274 ASSERT_EQ(nelements, stage1_processed);
275 ASSERT_EQ(nelements, stage2_processed);
276 ASSERT_EQ(0u, initial_queue.length());
277 ASSERT_EQ(0u, start_queue.length());
278 ASSERT_EQ(0u, middle_queue.length());
279 ASSERT_EQ(nelements, final_queue.length());
280 while (final_queue.pop() != nullptr) {}
281
282 FREE_C_HEAP_ARRAY(Element, elements);
283 }