1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
|
/*
* Copyright (C) 2020 The Android Open Source Project
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#ifndef ART_RUNTIME_BASE_MESSAGE_QUEUE_H_
#define ART_RUNTIME_BASE_MESSAGE_QUEUE_H_
#include <deque>
#include <optional>
#include <variant>
#include "base/time_utils.h"
#include "mutex.h"
#include "thread.h"
#pragma clang diagnostic push
#pragma clang diagnostic error "-Wconversion"
namespace art {
struct TimeoutExpiredMessage {};
// MessageQueue is an unbounded multiple producer, multiple consumer (MPMC) queue that can be
// specialized to send messages between threads. The queue is parameterized by a set of types that
// serve as the message types. Note that messages are passed by value, so smaller messages should be
// used when possible.
//
// Example:
//
// struct IntMessage { int value; };
// struct DoubleMessage { double value; };
//
// MessageQueue<IntMessage, DoubleMessage> queue;
//
// queue.SendMessage(IntMessage{42});
// queue.SendMessage(DoubleMessage{42.0});
//
// auto message = queue.ReceiveMessage(); // message is a std::variant of the different
// // message types.
//
// if (std::holds_alternative<IntMessage>(message)) {
// cout << "Received int message with value " << std::get<IntMessage>(message) << "\n";
// }
//
// The message queue also supports a special timeout message. This is scheduled to be sent by the
// SetTimeout method, which will cause the MessageQueue to deliver a TimeoutExpiredMessage after the
// time period has elapsed. Note that only one timeout can be active can be active at a time, and
// subsequent calls to SetTimeout will overwrite any existing timeout.
//
// Example:
//
// queue.SetTimeout(5000); // request to send TimeoutExpiredMessage in 5000ms.
//
// auto message = queue.ReceiveMessage(); // blocks for 5000ms and returns
// // TimeoutExpiredMessage
//
// Note additional messages can be sent in the meantime and a ReceiveMessage call will wake up to
// return that message. The TimeoutExpiredMessage will still be sent at the right time.
//
// Finally, MessageQueue has a SwitchReceive method that can be used to run different code depending
// on the type of message received. SwitchReceive takes a set of lambda expressions that take one
// argument of one of the allowed message types. An additional lambda expression that takes a single
// auto argument can be used to serve as a catch-all case.
//
// Example:
//
// queue.SwitchReceive(
// [&](IntMessage message) {
// cout << "Received int: " << message.value << "\n";
// },
// [&](DoubleMessage message) {
// cout << "Received double: " << message.value << "\n";
// },
// [&](auto other_message) {
// // Another message was received. In this case, it's TimeoutExpiredMessage.
// }
// )
//
// For additional examples, see message_queue_test.cc.
template <typename... MessageTypes>
class MessageQueue {
public:
using Message = std::variant<TimeoutExpiredMessage, MessageTypes...>;
// Adds a message to the message queue, which can later be received with ReceiveMessage. See class
// comment for more details.
void SendMessage(Message message) {
// TimeoutExpiredMessage should not be sent manually.
DCHECK(!std::holds_alternative<TimeoutExpiredMessage>(message));
Thread* self = Thread::Current();
MutexLock lock{self, mutex_};
messages_.push_back(message);
cv_.Signal(self);
}
// Schedule a TimeoutExpiredMessage to be delivered in timeout_milliseconds. See class comment for
// more details.
void SetTimeout(uint64_t timeout_milliseconds) {
Thread* self = Thread::Current();
MutexLock lock{self, mutex_};
deadline_milliseconds_ = timeout_milliseconds + MilliTime();
cv_.Signal(self);
}
// Remove and return a message from the queue. If no message is available, ReceiveMessage will
// block until one becomes available. See class comment for more details.
Message ReceiveMessage() {
Thread* self = Thread::Current();
MutexLock lock{self, mutex_};
// Loop until we receive a message
while (true) {
uint64_t const current_time = MilliTime();
// First check if the deadline has passed.
if (deadline_milliseconds_.has_value() && deadline_milliseconds_.value() < current_time) {
deadline_milliseconds_.reset();
return TimeoutExpiredMessage{};
}
// Check if there is a message in the queue.
if (messages_.size() > 0) {
Message message = messages_.front();
messages_.pop_front();
return message;
}
// Otherwise, wait until we have a message or a timeout.
if (deadline_milliseconds_.has_value()) {
DCHECK_LE(current_time, deadline_milliseconds_.value());
int64_t timeout = static_cast<int64_t>(deadline_milliseconds_.value() - current_time);
cv_.TimedWait(self, timeout, /*ns=*/0);
} else {
cv_.Wait(self);
}
}
}
// Waits for a message and applies the appropriate function argument to the received message. See
// class comment for more details.
template <typename ReturnType = void, typename... Fn>
ReturnType SwitchReceive(Fn... case_fn) {
struct Matcher : Fn... {
using Fn::operator()...;
} matcher{case_fn...};
return std::visit(matcher, ReceiveMessage());
}
private:
Mutex mutex_{"MessageQueue Mutex"};
ConditionVariable cv_{"MessageQueue ConditionVariable", mutex_};
std::deque<Message> messages_ GUARDED_BY(mutex_);
std::optional<uint64_t> deadline_milliseconds_ GUARDED_BY(mutex_);
};
} // namespace art
#pragma clang diagnostic pop // -Wconversion
#endif // ART_RUNTIME_BASE_MESSAGE_QUEUE_H_
|
