Revert^2 "Add MessageQueue for cross-thread communication"

This reverts commit c7edde06dc2cfb90114898e8de82e0e1287c2885.

Reason for revert: fixing bugs.

There were two issues. The first was a logic error in
MessageQueueTest::ReceiveInOrder.  The test would allow any message to
be replaced by a timeout message, but then the sequence would be out of
sync. The change now explicitly advances the message pointer, which
means the timeout message does not effectively replace an expected
message.

The seccond issue was one where MessageQueue may try to call
ConditionVariable::TimedWait with a negative timeout if the deadline had
already passed, which fails with an EINVAL error during the call to
futex. The code in MessageQueue::ReceiveMessage now checks for negative
timeouts before calling TimedWait.

Test: m test-art-host-gtest-art_runtime_tests
Bug: 174652565
Change-Id: I107af849e2a01727719f7662a7685d989b42d176

1 files changed, 172 insertions, 0 deletions

diff --git a/runtime/base/message_queue.h b/runtime/base/message_queue.h
new file mode 100644
index 0000000000..b0f1460a16
--- /dev/null
+++ b/runtime/base/message_queue.h
@@ -0,0 +1,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_