adb: make fdevent_run_on_main_thread's fd nonblocking.

If we get a ton of fdevent_run_on_main_thread calls while running one
of the handlers, the socket might become full, which will result in a
deadlock in fdevent_run_on_main_thread when a write to the fd blocks
with the mutex taken. Resolve this by making the fd nonblocking, which
is safe because we always write after appending to the list, and read
before emptying the list, which guarantees that if the byte we write is
consumed, the std::function we appended will be run.

Bug: http://b/74616284
Test: adb_test
Test: python test_device.py
Change-Id: I29319bda2ad7b5a5cdcd91d1d0ddf39f7ab7d115

adb/fdevent.cpp

@@ -402,6 +402,10 @@ static void fdevent_run_setup() {
             PLOG(FATAL) << "failed to create run queue notify socketpair";
         }
 
+        if (!set_file_block_mode(s[0], false) || !set_file_block_mode(s[1], false)) {
+            PLOG(FATAL) << "failed to make run queue notify socket nonblocking";
+        }
+
         run_queue_notify_fd.reset(s[0]);
         fdevent* fde = fdevent_create(s[1], fdevent_run_func, nullptr);
         CHECK(fde != nullptr);
@@ -418,7 +422,12 @@ void fdevent_run_on_main_thread(std::function<void()> fn) {
     // run_queue_notify_fd could still be -1 if we're called before fdevent has finished setting up.
     // In that case, rely on the setup code to flush the queue without a notification being needed.
     if (run_queue_notify_fd != -1) {
-        if (adb_write(run_queue_notify_fd.get(), "", 1) != 1) {
+        int rc = adb_write(run_queue_notify_fd.get(), "", 1);
+
+        // It's possible that we get EAGAIN here, if lots of notifications came in while handling.
+        if (rc == 0) {
+            PLOG(FATAL) << "run queue notify fd was closed?";
+        } else if (rc == -1 && errno != EAGAIN) {
             PLOG(FATAL) << "failed to write to run queue notify fd";
         }
     }

adb/fdevent_test.cpp

@@ -180,7 +180,13 @@ TEST_F(FdeventTest, run_on_main_thread) {
     PrepareThread();
     std::thread thread(fdevent_loop);
 
-    for (int i = 0; i < 100; ++i) {
+    // Block the main thread for a long time while we queue our callbacks.
+    fdevent_run_on_main_thread([]() {
+        check_main_thread();
+        std::this_thread::sleep_for(std::chrono::seconds(1));
+    });
+
+    for (int i = 0; i < 1000000; ++i) {
         fdevent_run_on_main_thread([i, &vec]() {
             check_main_thread();
             vec.push_back(i);
@@ -189,8 +195,8 @@ TEST_F(FdeventTest, run_on_main_thread) {
 
     TerminateThread(thread);
 
-    ASSERT_EQ(100u, vec.size());
+    ASSERT_EQ(1000000u, vec.size());
-    for (int i = 0; i < 100; ++i) {
+    for (int i = 0; i < 1000000; ++i) {
         ASSERT_EQ(i, vec[i]);
     }
 }