adb: allow reentrant calls to fdevent_run_on_main_thread.

Previously, reentrant calls to fdevent_run_on_main_thread would
deadlock.

Test: adb_test on host
Change-Id: I0783be0558dcaf61ddbe76d13ac6917fc2de0be0

adb/fdevent.cpp

@@ -26,6 +26,7 @@
 #include <unistd.h>
 
 #include <atomic>
+#include <deque>
 #include <functional>
 #include <list>
 #include <mutex>
@@ -81,7 +82,7 @@ static unsigned long main_thread_id;
 
 static auto& run_queue_notify_fd = *new unique_fd();
 static auto& run_queue_mutex = *new std::mutex();
-static auto& run_queue GUARDED_BY(run_queue_mutex) = *new std::vector<std::function<void()>>();
+static auto& run_queue GUARDED_BY(run_queue_mutex) = *new std::deque<std::function<void()>>();
 
 void check_main_thread() {
     if (main_thread_valid) {
@@ -359,11 +360,21 @@ static void fdevent_subproc_setup() {
 }
 #endif // !ADB_HOST
 
-static void fdevent_run_flush() REQUIRES(run_queue_mutex) {
-    for (auto& f : run_queue) {
-        f();
+static void fdevent_run_flush() EXCLUDES(run_queue_mutex) {
+    // We need to be careful around reentrancy here, since a function we call can queue up another
+    // function.
+    while (true) {
+        std::function<void()> fn;
+        {
+            std::lock_guard<std::mutex> lock(run_queue_mutex);
+            if (run_queue.empty()) {
+                break;
+            }
+            fn = run_queue.front();
+            run_queue.pop_front();
+        }
+        fn();
     }
-    run_queue.clear();
 }
 
 static void fdevent_run_func(int fd, unsigned ev, void* /* userdata */) {
@@ -377,22 +388,23 @@ static void fdevent_run_func(int fd, unsigned ev, void* /* userdata */) {
         PLOG(FATAL) << "failed to empty run queue notify fd";
     }
 
-    std::lock_guard<std::mutex> lock(run_queue_mutex);
     fdevent_run_flush();
 }
 
 static void fdevent_run_setup() {
-    std::lock_guard<std::mutex> lock(run_queue_mutex);
-    CHECK(run_queue_notify_fd.get() == -1);
-    int s[2];
-    if (adb_socketpair(s) != 0) {
-        PLOG(FATAL) << "failed to create run queue notify socketpair";
-    }
+    {
+        std::lock_guard<std::mutex> lock(run_queue_mutex);
+        CHECK(run_queue_notify_fd.get() == -1);
+        int s[2];
+        if (adb_socketpair(s) != 0) {
+            PLOG(FATAL) << "failed to create run queue notify socketpair";
+        }
 
-    run_queue_notify_fd.reset(s[0]);
-    fdevent* fde = fdevent_create(s[1], fdevent_run_func, nullptr);
-    CHECK(fde != nullptr);
-    fdevent_add(fde, FDE_READ);
+        run_queue_notify_fd.reset(s[0]);
+        fdevent* fde = fdevent_create(s[1], fdevent_run_func, nullptr);
+        CHECK(fde != nullptr);
+        fdevent_add(fde, FDE_READ);
+    }
 
     fdevent_run_flush();
 }

adb/fdevent_test.cpp

@@ -194,3 +194,31 @@ TEST_F(FdeventTest, run_on_main_thread) {
         ASSERT_EQ(i, vec[i]);
     }
 }
+
+static std::function<void()> make_appender(std::vector<int>* vec, int value) {
+    return [vec, value]() {
+        check_main_thread();
+        if (value == 100) {
+            return;
+        }
+
+        vec->push_back(value);
+        fdevent_run_on_main_thread(make_appender(vec, value + 1));
+    };
+}
+
+TEST_F(FdeventTest, run_on_main_thread_reentrant) {
+    std::vector<int> vec;
+
+    PrepareThread();
+    std::thread thread(fdevent_loop);
+
+    fdevent_run_on_main_thread(make_appender(&vec, 0));
+
+    TerminateThread(thread);
+
+    ASSERT_EQ(100u, vec.size());
+    for (int i = 0; i < 100; ++i) {
+        ASSERT_EQ(i, vec[i]);
+    }
+}