OLD | NEW |
---|---|
1 // Copyright (c) 2011, the Dart project authors. Please see the AUTHORS file | 1 // Copyright (c) 2011, 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 "vm/message_queue.h" | 5 #include "vm/message_queue.h" |
6 | 6 |
7 namespace dart { | 7 namespace dart { |
8 | 8 |
9 MessageQueue::~MessageQueue() { | 9 MessageQueue::MessageQueue() { |
10 // Ensure that all pending messages have been released. | 10 for (int p = Message::kFirstPriority; p < Message::kNumPriorities; p++) { |
11 ASSERT(head_ == NULL); | 11 head_[p] = NULL; |
12 tail_[p] = NULL; | |
13 } | |
12 } | 14 } |
13 | 15 |
14 | 16 |
15 void MessageQueue::Enqueue(PortMessage* msg) { | 17 MessageQueue::~MessageQueue() { |
18 // Ensure that all pending messages have been released. | |
19 #if defined(DEBUG) | |
20 for (int p = Message::kFirstPriority; p < Message::kNumPriorities; p++) { | |
21 ASSERT(head_[p] == NULL); | |
22 } | |
23 #endif | |
24 } | |
25 | |
26 | |
27 void MessageQueue::Enqueue(Message* msg) { | |
16 // TODO(turnidge): Can't use MonitorLocker here because | 28 // TODO(turnidge): Can't use MonitorLocker here because |
17 // MonitorLocker is a StackResource, which requires a current | 29 // MonitorLocker is a StackResource, which requires a current |
18 // isolate. Should MonitorLocker really be a StackResource? | 30 // isolate. Should MonitorLocker really be a StackResource? |
19 monitor_.Enter(); | 31 monitor_.Enter(); |
32 | |
33 Message::Priority p = msg->priority(); | |
20 // Make sure messages are not reused. | 34 // Make sure messages are not reused. |
21 ASSERT(msg->next_ == NULL); | 35 ASSERT(msg->next_ == NULL); |
22 if (head_ == NULL) { | 36 if (head_[p] == NULL) { |
23 // Only element in the queue. | 37 // Only element in the queue. |
24 head_ = msg; | 38 head_[p] = msg; |
25 tail_ = msg; | 39 tail_[p] = msg; |
26 | 40 |
27 // We only need to notify if the queue was empty. | 41 // We only need to notify if the queue was empty. |
28 monitor_.Notify(); | 42 monitor_.Notify(); |
29 } else { | 43 } else { |
30 ASSERT(tail_ != NULL); | 44 ASSERT(tail_[p] != NULL); |
31 // Append at the tail. | 45 // Append at the tail. |
32 tail_->next_ = msg; | 46 tail_[p]->next_ = msg; |
33 tail_ = msg; | 47 tail_[p] = msg; |
34 } | 48 } |
35 | 49 |
36 monitor_.Exit(); | 50 monitor_.Exit(); |
37 } | 51 } |
38 | 52 |
39 | 53 |
40 PortMessage* MessageQueue::Dequeue(int64_t millis) { | 54 Message* MessageQueue::DequeueNoWait() { |
Anton Muhin
2012/01/12 12:58:16
you don't lock on monitor_, is it ok?
turnidge
2012/01/12 19:01:55
It is not okay. Fixed.
| |
55 // Look for the highest priority available message. | |
56 for (int p = Message::kNumPriorities-1; p >= Message::kFirstPriority; p--) { | |
57 Message* result = head_[p]; | |
58 if (result != NULL) { | |
59 head_[p] = result->next_; | |
60 // The following update to tail_ is not strictly needed. | |
Anton Muhin
2012/01/12 12:58:16
maybe move under DEBUG?
turnidge
2012/01/12 19:01:55
It's not expensive and it makes me feel happier.
| |
61 if (head_[p] == NULL) { | |
62 tail_[p] = NULL; | |
63 } | |
64 #if defined(DEBUG) | |
65 result->next_ = result; // Make sure to trigger ASSERT in Enqueue. | |
66 #endif // DEBUG | |
67 return result; | |
68 } | |
69 } | |
70 return NULL; | |
71 } | |
72 | |
73 | |
74 Message* MessageQueue::Dequeue(int64_t millis) { | |
41 ASSERT(millis >= 0); | 75 ASSERT(millis >= 0); |
42 MonitorLocker ml(&monitor_); | 76 MonitorLocker ml(&monitor_); |
43 PortMessage* result = head_; | 77 |
78 Message* result = DequeueNoWait(); | |
44 if (result == NULL) { | 79 if (result == NULL) { |
80 // No message available at any priority. | |
45 ml.Wait(millis); | 81 ml.Wait(millis); |
46 result = head_; | 82 result = DequeueNoWait(); |
47 } | |
48 if (result != NULL) { | |
49 head_ = result->next_; | |
50 // The following update to tail_ is not strictly needed. | |
51 if (head_ == NULL) { | |
52 tail_ = NULL; | |
53 } | |
54 #if defined(DEBUG) | |
55 result->next_ = result; // Make sure to trigger ASSERT in Enqueue. | |
56 #endif // DEBUG | |
57 } | 83 } |
58 return result; | 84 return result; |
59 } | 85 } |
60 | 86 |
61 | 87 |
62 void MessageQueue::Flush(Dart_Port port) { | 88 void MessageQueue::Flush(Dart_Port port) { |
63 MonitorLocker ml(&monitor_); | 89 MonitorLocker ml(&monitor_); |
64 PortMessage* cur = head_; | 90 for (int p = Message::kFirstPriority; p < Message::kNumPriorities; p++) { |
65 PortMessage* prev = NULL; | 91 Message* cur = head_[p]; |
66 while (cur != NULL) { | 92 Message* prev = NULL; |
67 PortMessage* next = cur->next_; | 93 while (cur != NULL) { |
68 // If the message matches, then remove it from the queue and delete it. | 94 Message* next = cur->next_; |
69 if (cur->dest_port() == port) { | 95 // If the message matches, then remove it from the queue and delete it. |
70 if (prev != NULL) { | 96 if (cur->dest_port() == port) { |
71 prev->next_ = next; | 97 if (prev != NULL) { |
98 prev->next_ = next; | |
99 } else { | |
100 head_[p] = next; | |
101 } | |
102 delete cur; | |
72 } else { | 103 } else { |
73 head_ = next; | 104 // Move prev forward. |
105 prev = cur; | |
74 } | 106 } |
75 delete cur; | 107 // Advance to the next message in the queue. |
76 } else { | 108 cur = next; |
77 // Move prev forward. | |
78 prev = cur; | |
79 } | 109 } |
80 // Advance to the next message in the queue. | 110 tail_[p] = prev; |
81 cur = next; | |
82 } | 111 } |
83 tail_ = prev; | |
84 } | 112 } |
85 | 113 |
86 | 114 |
87 void MessageQueue::FlushAll() { | 115 void MessageQueue::FlushAll() { |
88 MonitorLocker ml(&monitor_); | 116 MonitorLocker ml(&monitor_); |
89 PortMessage* cur = head_; | 117 for (int p = Message::kFirstPriority; p < Message::kNumPriorities; p++) { |
90 head_ = NULL; | 118 Message* cur = head_[p]; |
91 tail_ = NULL; | 119 head_[p] = NULL; |
92 while (cur != NULL) { | 120 tail_[p] = NULL; |
93 PortMessage* next = cur->next_; | 121 while (cur != NULL) { |
94 delete cur; | 122 Message* next = cur->next_; |
95 cur = next; | 123 delete cur; |
124 cur = next; | |
125 } | |
96 } | 126 } |
97 } | 127 } |
98 | 128 |
99 | 129 |
100 } // namespace dart | 130 } // namespace dart |
OLD | NEW |