From 45d9d0d24760b5c1ffcad6a81c6aee8fc1a0d2b1 Mon Sep 17 00:00:00 2001
From: Zach Johnson <zachoverflow@google.com>
Date: Mon, 8 Feb 2021 11:34:16 -0800
Subject: [PATCH] rusty-gd: sequence immediate message loop posts

this ensures order of operations is preserved

Bug: 171749953
Tag: #gd-refactor
Test: gd/cert/run --rhost DirectHciTest
Change-Id: I6368fd2e07e95b505ac67c3e67343ba2a488b842
---
 gd/rust/shim/src/message_loop_thread.rs | 39 +++++++++++++++++++++++----------
 1 file changed, 27 insertions(+), 12 deletions(-)

diff --git a/gd/rust/shim/src/message_loop_thread.rs b/gd/rust/shim/src/message_loop_thread.rs
index 377479527..df32e6d12 100644
--- a/gd/rust/shim/src/message_loop_thread.rs
+++ b/gd/rust/shim/src/message_loop_thread.rs
@@ -5,6 +5,7 @@ use std::convert::TryInto;
 use std::sync::Arc;
 use std::time::Duration;
 use tokio::runtime::Runtime;
+use tokio::sync::mpsc::{unbounded_channel, UnboundedSender};
 
 #[cxx::bridge(namespace = bluetooth::shim::rust)]
 mod ffi {
@@ -32,12 +33,21 @@ unsafe impl Send for ffi::OnceClosure {}
 
 pub struct MessageLoopThread {
     rt: Arc<Runtime>,
+    tx: UnboundedSender<cxx::UniquePtr<ffi::OnceClosure>>,
 }
 
 pub fn main_message_loop_thread_create() -> Box<MessageLoopThread> {
     assert!(init_flags::gd_rust_is_enabled());
 
-    Box::new(MessageLoopThread { rt: crate::stack::RUNTIME.clone() })
+    let rt = crate::stack::RUNTIME.clone();
+    let (tx, mut rx) = unbounded_channel::<cxx::UniquePtr<ffi::OnceClosure>>();
+    rt.spawn(async move {
+        while let Some(c) = rx.recv().await {
+            c.Run();
+        }
+    });
+
+    Box::new(MessageLoopThread { rt, tx })
 }
 
 pub fn main_message_loop_thread_start(thread: &mut MessageLoopThread) -> i32 {
@@ -52,15 +62,20 @@ pub fn main_message_loop_thread_do_delayed(
     delay_ms: i64,
 ) {
     assert!(init_flags::gd_rust_is_enabled());
-
-    thread.rt.spawn(async move {
-        // NOTE: tokio's sleep can't wake up the system...
-        // but hey, neither could the message loop from libchrome.
-        //
-        // ...and this way we don't use timerfds arbitrarily.
-        //
-        // #yolo
-        tokio::time::sleep(Duration::from_millis(delay_ms.try_into().unwrap_or(0))).await;
-        closure.Run();
-    });
+    if delay_ms == 0 {
+        if thread.tx.send(closure).is_err() {
+            log::error!("could not post task - shutting down?");
+        }
+    } else {
+        thread.rt.spawn(async move {
+            // NOTE: tokio's sleep can't wake up the system...
+            // but hey, neither could the message loop from libchrome.
+            //
+            // ...and this way we don't use timerfds arbitrarily.
+            //
+            // #yolo
+            tokio::time::sleep(Duration::from_millis(delay_ms.try_into().unwrap_or(0))).await;
+            closure.Run();
+        });
+    }
 }
-- 
2.11.0