OLD | NEW |
1 // Copyright (c) 2012, the Dart project authors. Please see the AUTHORS file | 1 // Copyright (c) 2012, the Dart project authors. Please see the AUTHORS file |
2 // for details. All rights reserved. Use of this source code is governed by a | 2 // for details. All rights reserved. Use of this source code is governed by a |
3 // BSD-style license that can be found in the LICENSE file. | 3 // BSD-style license that can be found in the LICENSE file. |
4 | 4 |
5 #include "platform/assert.h" | 5 #include "platform/assert.h" |
6 #include "vm/message.h" | 6 #include "vm/message.h" |
7 #include "vm/unit_test.h" | 7 #include "vm/unit_test.h" |
8 | 8 |
9 namespace dart { | 9 namespace dart { |
10 | 10 |
11 | 11 |
12 // Provide access to private members of MessageQueue for testing. | 12 // Provide access to private members of MessageQueue for testing. |
13 class MessageQueueTestPeer { | 13 class MessageQueueTestPeer { |
14 public: | 14 public: |
15 explicit MessageQueueTestPeer(MessageQueue* queue) : queue_(queue) {} | 15 explicit MessageQueueTestPeer(MessageQueue* queue) : queue_(queue) {} |
16 | 16 |
17 bool HasMessage() const { | 17 bool HasMessage() const { |
| 18 // We don't really need to grab the monitor during the unit test, |
| 19 // but it doesn't hurt. |
| 20 queue_->monitor_.Enter(); |
18 bool result = (queue_->head_[Message::kNormalPriority] != NULL || | 21 bool result = (queue_->head_[Message::kNormalPriority] != NULL || |
19 queue_->head_[Message::kOOBPriority] != NULL); | 22 queue_->head_[Message::kOOBPriority] != NULL); |
| 23 queue_->monitor_.Exit(); |
20 return result; | 24 return result; |
21 } | 25 } |
22 | 26 |
23 private: | 27 private: |
24 MessageQueue* queue_; | 28 MessageQueue* queue_; |
25 }; | 29 }; |
26 | 30 |
27 | 31 |
28 class MessageHandlerTestPeer { | |
29 public: | |
30 explicit MessageHandlerTestPeer(MessageHandler* handler) | |
31 : handler_(handler) {} | |
32 | |
33 MessageQueue* queue() { return handler_->queue_; } | |
34 | |
35 private: | |
36 MessageHandler* handler_; | |
37 }; | |
38 | |
39 | |
40 static uint8_t* AllocMsg(const char* str) { | 32 static uint8_t* AllocMsg(const char* str) { |
41 return reinterpret_cast<uint8_t*>(strdup(str)); | 33 return reinterpret_cast<uint8_t*>(strdup(str)); |
42 } | 34 } |
43 | 35 |
44 | 36 |
45 TEST_CASE(MessageQueue_BasicOperations) { | 37 TEST_CASE(MessageQueue_BasicOperations) { |
46 MessageQueue queue; | 38 MessageQueue queue; |
47 MessageQueueTestPeer queue_peer(&queue); | 39 MessageQueueTestPeer queue_peer(&queue); |
48 EXPECT(!queue_peer.HasMessage()); | 40 EXPECT(!queue_peer.HasMessage()); |
49 | 41 |
50 Dart_Port port = 1; | 42 Dart_Port port = 1; |
51 | 43 |
52 // Add two messages. | 44 // Add two messages. |
53 Message* msg1 = | 45 Message* msg1 = |
54 new Message(port, 0, AllocMsg("msg1"), Message::kNormalPriority); | 46 new Message(port, 0, AllocMsg("msg1"), Message::kNormalPriority); |
55 queue.Enqueue(msg1); | 47 queue.Enqueue(msg1); |
56 EXPECT(queue_peer.HasMessage()); | 48 EXPECT(queue_peer.HasMessage()); |
57 | 49 |
58 Message* msg2 = | 50 Message* msg2 = |
59 new Message(port, 0, AllocMsg("msg2"), Message::kNormalPriority); | 51 new Message(port, 0, AllocMsg("msg2"), Message::kNormalPriority); |
60 | 52 |
61 queue.Enqueue(msg2); | 53 queue.Enqueue(msg2); |
62 EXPECT(queue_peer.HasMessage()); | 54 EXPECT(queue_peer.HasMessage()); |
63 | 55 |
64 // Remove two messages. | 56 // Remove two messages. |
65 Message* msg = queue.Dequeue(); | 57 Message* msg = queue.Dequeue(0); |
66 EXPECT(msg != NULL); | 58 EXPECT(msg != NULL); |
67 EXPECT_STREQ("msg1", reinterpret_cast<char*>(msg->data())); | 59 EXPECT_STREQ("msg1", reinterpret_cast<char*>(msg->data())); |
68 EXPECT(queue_peer.HasMessage()); | 60 EXPECT(queue_peer.HasMessage()); |
69 | 61 |
70 msg = queue.Dequeue(); | 62 msg = queue.Dequeue(0); |
71 EXPECT(msg != NULL); | 63 EXPECT(msg != NULL); |
72 EXPECT_STREQ("msg2", reinterpret_cast<char*>(msg->data())); | 64 EXPECT_STREQ("msg2", reinterpret_cast<char*>(msg->data())); |
73 EXPECT(!queue_peer.HasMessage()); | 65 EXPECT(!queue_peer.HasMessage()); |
74 | 66 |
75 delete msg1; | 67 delete msg1; |
76 delete msg2; | 68 delete msg2; |
77 } | 69 } |
78 | 70 |
79 | 71 |
80 TEST_CASE(MessageQueue_Priorities) { | 72 TEST_CASE(MessageQueue_Priorities) { |
81 MessageQueue queue; | 73 MessageQueue queue; |
82 MessageQueueTestPeer queue_peer(&queue); | 74 MessageQueueTestPeer queue_peer(&queue); |
83 EXPECT(!queue_peer.HasMessage()); | 75 EXPECT(!queue_peer.HasMessage()); |
84 | 76 |
85 Dart_Port port = 1; | 77 Dart_Port port = 1; |
86 | 78 |
87 // Add two messages. | 79 // Add two messages. |
88 Message* msg1 = | 80 Message* msg1 = |
89 new Message(port, 0, AllocMsg("msg1"), Message::kNormalPriority); | 81 new Message(port, 0, AllocMsg("msg1"), Message::kNormalPriority); |
90 queue.Enqueue(msg1); | 82 queue.Enqueue(msg1); |
91 EXPECT(queue_peer.HasMessage()); | 83 EXPECT(queue_peer.HasMessage()); |
92 | 84 |
93 Message* msg2 = | 85 Message* msg2 = |
94 new Message(port, 0, AllocMsg("msg2"), Message::kOOBPriority); | 86 new Message(port, 0, AllocMsg("msg2"), Message::kOOBPriority); |
95 | 87 |
96 queue.Enqueue(msg2); | 88 queue.Enqueue(msg2); |
97 EXPECT(queue_peer.HasMessage()); | 89 EXPECT(queue_peer.HasMessage()); |
98 | 90 |
99 // The higher priority message is delivered first. | 91 // The higher priority message is delivered first. |
100 Message* msg = queue.Dequeue(); | 92 Message* msg = queue.Dequeue(0); |
101 EXPECT(msg != NULL); | 93 EXPECT(msg != NULL); |
102 EXPECT_STREQ("msg2", reinterpret_cast<char*>(msg->data())); | 94 EXPECT_STREQ("msg2", reinterpret_cast<char*>(msg->data())); |
103 EXPECT(queue_peer.HasMessage()); | 95 EXPECT(queue_peer.HasMessage()); |
104 | 96 |
105 msg = queue.Dequeue(); | 97 msg = queue.Dequeue(0); |
106 EXPECT(msg != NULL); | 98 EXPECT(msg != NULL); |
107 EXPECT_STREQ("msg1", reinterpret_cast<char*>(msg->data())); | 99 EXPECT_STREQ("msg1", reinterpret_cast<char*>(msg->data())); |
108 EXPECT(!queue_peer.HasMessage()); | 100 EXPECT(!queue_peer.HasMessage()); |
109 | 101 |
110 delete msg1; | 102 delete msg1; |
111 delete msg2; | 103 delete msg2; |
112 } | 104 } |
113 | 105 |
114 | 106 |
115 // A thread which receives an expected sequence of messages. | 107 // A thread which receives an expected sequence of messages. |
116 static Monitor* sync = NULL; | 108 static Monitor* sync = NULL; |
117 static MessageHandler* shared_handler = NULL; | 109 static MessageQueue* shared_queue = NULL; |
118 void MessageReceiver_start(uword unused) { | 110 void MessageReceiver_start(uword unused) { |
119 // We only need an isolate here because the MonitorLocker in the | 111 // We only need an isolate here because the MonitorLocker in the |
120 // MessageQueue expects it, we don't need to initialize the isolate | 112 // MessageQueue expects it, we don't need to initialize the isolate |
121 // as it does not run any dart code. | 113 // as it does not run any dart code. |
122 Dart::CreateIsolate(NULL); | 114 Dart::CreateIsolate(NULL); |
123 | 115 |
124 // Create a message handler and share it. | 116 // Create a message queue and share it. |
125 MessageHandler* handler = new MessageHandler(); | 117 MessageQueue* queue = new MessageQueue(); |
126 MessageHandlerTestPeer handler_peer(handler); | 118 MessageQueueTestPeer peer(queue); |
127 MessageQueueTestPeer queue_peer(handler_peer.queue()); | 119 shared_queue = queue; |
128 shared_handler = handler; | |
129 | 120 |
130 // Tell the other thread that the shared queue is ready. | 121 // Tell the other thread that the shared queue is ready. |
131 { | 122 { |
132 MonitorLocker ml(sync); | 123 MonitorLocker ml(sync); |
133 ml.Notify(); | 124 ml.Notify(); |
134 } | 125 } |
135 | 126 |
136 // Wait for the other thread to fill the queue a bit. | 127 // Wait for the other thread to fill the queue a bit. |
137 while (!queue_peer.HasMessage()) { | 128 while (!peer.HasMessage()) { |
138 MonitorLocker ml(sync); | 129 MonitorLocker ml(sync); |
139 ml.Wait(5); | 130 ml.Wait(5); |
140 } | 131 } |
141 | 132 |
142 for (int i = 0; i < 3; i++) { | 133 for (int i = 0; i < 3; i++) { |
143 Message* msg = handler->Dequeue(0); | 134 Message* msg = queue->Dequeue(0); |
144 EXPECT(msg != NULL); | 135 EXPECT(msg != NULL); |
145 EXPECT_EQ(i + 10, msg->dest_port()); | 136 EXPECT_EQ(i + 10, msg->dest_port()); |
146 EXPECT_EQ(i + 100, msg->reply_port()); | 137 EXPECT_EQ(i + 100, msg->reply_port()); |
147 EXPECT_EQ(i + 1000, *(reinterpret_cast<int*>(msg->data()))); | 138 EXPECT_EQ(i + 1000, *(reinterpret_cast<int*>(msg->data()))); |
148 delete msg; | 139 delete msg; |
149 } | 140 } |
150 for (int i = 0; i < 3; i++) { | 141 for (int i = 0; i < 3; i++) { |
151 Message* msg = handler->Dequeue(0); | 142 Message* msg = queue->Dequeue(0); |
152 EXPECT(msg != NULL); | 143 EXPECT(msg != NULL); |
153 EXPECT_EQ(i + 20, msg->dest_port()); | 144 EXPECT_EQ(i + 20, msg->dest_port()); |
154 EXPECT_EQ(i + 200, msg->reply_port()); | 145 EXPECT_EQ(i + 200, msg->reply_port()); |
155 EXPECT_EQ(i + 2000, *(reinterpret_cast<int*>(msg->data()))); | 146 EXPECT_EQ(i + 2000, *(reinterpret_cast<int*>(msg->data()))); |
156 delete msg; | 147 delete msg; |
157 } | 148 } |
158 shared_handler = NULL; | 149 shared_queue = NULL; |
159 delete handler; | 150 delete queue; |
160 Dart::ShutdownIsolate(); | 151 Dart::ShutdownIsolate(); |
161 } | 152 } |
162 | 153 |
163 | 154 |
164 TEST_CASE(MessageHandler_WaitNotify) { | 155 TEST_CASE(MessageQueue_WaitNotify) { |
165 sync = new Monitor(); | 156 sync = new Monitor(); |
166 | 157 |
167 int result = Thread::Start(MessageReceiver_start, 0); | 158 int result = Thread::Start(MessageReceiver_start, 0); |
168 EXPECT_EQ(0, result); | 159 EXPECT_EQ(0, result); |
169 | 160 |
170 // Wait for the shared handler to be created. | 161 // Wait for the shared queue to be created. |
171 while (shared_handler == NULL) { | 162 while (shared_queue == NULL) { |
172 MonitorLocker ml(sync); | 163 MonitorLocker ml(sync); |
173 ml.Wait(5); | 164 ml.Wait(5); |
174 } | 165 } |
175 ASSERT(shared_handler != NULL); | 166 ASSERT(shared_queue != NULL); |
176 | 167 |
177 // Pile up three messages before the other thread runs. | 168 // Pile up three messages before the other thread runs. |
178 for (int i = 0; i < 3; i++) { | 169 for (int i = 0; i < 3; i++) { |
179 int* data = reinterpret_cast<int*>(malloc(sizeof(*data))); | 170 int* data = reinterpret_cast<int*>(malloc(sizeof(*data))); |
180 *data = i + 1000; | 171 *data = i + 1000; |
181 Message* msg = | 172 Message* msg = |
182 new Message(i + 10, i + 100, reinterpret_cast<uint8_t*>(data), | 173 new Message(i + 10, i + 100, reinterpret_cast<uint8_t*>(data), |
183 Message::kNormalPriority); | 174 Message::kNormalPriority); |
184 shared_handler->PostMessage(msg); | 175 shared_queue->Enqueue(msg); |
185 } | 176 } |
186 | 177 |
187 // Wake the other thread and have it start consuming messages. | 178 // Wake the other thread and have it start consuming messages. |
188 { | 179 { |
189 MonitorLocker ml(sync); | 180 MonitorLocker ml(sync); |
190 ml.Notify(); | 181 ml.Notify(); |
191 } | 182 } |
192 | 183 |
193 // Add a few more messages after sleeping to allow the other thread | 184 // Add a few more messages after sleeping to allow the other thread |
194 // to potentially exercise the blocking code path in Dequeue. | 185 // to potentially exercise the blocking code path in Dequeue. |
195 OS::Sleep(5); | 186 OS::Sleep(5); |
196 for (int i = 0; i < 3; i++) { | 187 for (int i = 0; i < 3; i++) { |
197 int* data = reinterpret_cast<int*>(malloc(sizeof(*data))); | 188 int* data = reinterpret_cast<int*>(malloc(sizeof(*data))); |
198 *data = i + 2000; | 189 *data = i + 2000; |
199 Message* msg = | 190 Message* msg = |
200 new Message(i + 20, i + 200, reinterpret_cast<uint8_t*>(data), | 191 new Message(i + 20, i + 200, reinterpret_cast<uint8_t*>(data), |
201 Message::kNormalPriority); | 192 Message::kNormalPriority); |
202 shared_handler->PostMessage(msg); | 193 shared_queue->Enqueue(msg); |
203 } | 194 } |
204 | 195 |
205 sync = NULL; | 196 sync = NULL; |
206 delete sync; | 197 delete sync; |
207 | 198 |
208 // Give the spawned thread enough time to properly exit. | 199 // Give the spawned thread enough time to properly exit. |
209 OS::Sleep(20); | 200 OS::Sleep(20); |
210 } | 201 } |
211 | 202 |
212 | 203 |
(...skipping 31 matching lines...) Expand 10 before | Expand all | Expand 10 after Loading... |
244 queue.Enqueue(msg1); | 235 queue.Enqueue(msg1); |
245 Message* msg2 = | 236 Message* msg2 = |
246 new Message(port2, 0, AllocMsg("msg2"), Message::kNormalPriority); | 237 new Message(port2, 0, AllocMsg("msg2"), Message::kNormalPriority); |
247 queue.Enqueue(msg2); | 238 queue.Enqueue(msg2); |
248 EXPECT(queue_peer.HasMessage()); | 239 EXPECT(queue_peer.HasMessage()); |
249 | 240 |
250 queue.Flush(port1); | 241 queue.Flush(port1); |
251 | 242 |
252 // One message is left in the queue. | 243 // One message is left in the queue. |
253 EXPECT(queue_peer.HasMessage()); | 244 EXPECT(queue_peer.HasMessage()); |
254 Message* msg = queue.Dequeue(); | 245 Message* msg = queue.Dequeue(0); |
255 EXPECT(msg != NULL); | 246 EXPECT(msg != NULL); |
256 EXPECT_STREQ("msg2", reinterpret_cast<char*>(msg->data())); | 247 EXPECT_STREQ("msg2", reinterpret_cast<char*>(msg->data())); |
257 | 248 |
258 EXPECT(!queue_peer.HasMessage()); | 249 EXPECT(!queue_peer.HasMessage()); |
259 | 250 |
260 // msg1 is already deleted by Flush. | 251 // msg1 is already deleted by Flush. |
261 delete msg2; | 252 delete msg2; |
262 } | 253 } |
263 | 254 |
264 | 255 |
(...skipping 24 matching lines...) Expand all Loading... |
289 Dart_Port port1 = 1; | 280 Dart_Port port1 = 1; |
290 | 281 |
291 EXPECT(!queue_peer.HasMessage()); | 282 EXPECT(!queue_peer.HasMessage()); |
292 queue.Flush(port1); | 283 queue.Flush(port1); |
293 | 284 |
294 // Queue is still empty. | 285 // Queue is still empty. |
295 EXPECT(!queue_peer.HasMessage()); | 286 EXPECT(!queue_peer.HasMessage()); |
296 } | 287 } |
297 | 288 |
298 } // namespace dart | 289 } // namespace dart |
OLD | NEW |