Add unit tests for local socket.

Add has_write_error flag in asocket, so it will not wait on local_socket_closing_list
to write pending packets in local_socket_close(). Although it doesn't fix any problem,
it helps to make the code more stable.
Add a missing put_apacket() in error handling.
Add a check when adding local socket in local_socket_closing_list.

Bug: 23314034

Change-Id: I75b07ba8ee59b7f277fba2fb919db63065b291be

9 files changed, 485 insertions, 133 deletions

diff --git a/Android.mk b/Android.mk
index 6d68bec..5786d18 100644
--- a/Android.mk
+++ b/Android.mk
@@ -74,9 +74,11 @@ LIBADB_windows_SRC_FILES := \
 
 LIBADB_TEST_linux_SRCS := \
     fdevent_test.cpp \
+    socket_test.cpp \
 
 LIBADB_TEST_darwin_SRCS := \
     fdevent_test.cpp \
+    socket_test.cpp \
 
 LIBADB_TEST_windows_SRCS := \
     sysdeps_win32_test.cpp \
diff --git a/adb.cpp b/adb.cpp
index 1249822..20094e4 100644
--- a/adb.cpp
+++ b/adb.cpp
@@ -162,6 +162,9 @@ std::string get_trace_setting() {
 // adbd's comes from the system property persist.adb.trace_mask.
 static void setup_trace_mask() {
     const std::string trace_setting = get_trace_setting();
+    if (trace_setting.empty()) {
+        return;
+    }
 
     std::unordered_map<std::string, int> trace_flags = {
         {"1", 0},
@@ -184,7 +187,7 @@ static void setup_trace_mask() {
     for (const auto& elem : elements) {
         const auto& flag = trace_flags.find(elem);
         if (flag == trace_flags.end()) {
-            D("Unknown trace flag: %s", flag->first.c_str());
+            D("Unknown trace flag: %s", elem.c_str());
             continue;
         }
 
diff --git a/adb.h b/adb.h
index 4922040..037c010 100644
--- a/adb.h
+++ b/adb.h
@@ -26,6 +26,7 @@
 
 #include "adb_trace.h"
 #include "fdevent.h"
+#include "socket.h"
 
 constexpr size_t MAX_PAYLOAD_V1 = 4 * 1024;
 constexpr size_t MAX_PAYLOAD_V2 = 256 * 1024;
@@ -74,80 +75,6 @@ struct apacket
     unsigned char data[MAX_PAYLOAD];
 };
 
-/* An asocket represents one half of a connection between a local and
-** remote entity.  A local asocket is bound to a file descriptor.  A
-** remote asocket is bound to the protocol engine.
-*/
-struct asocket {
-        /* chain pointers for the local/remote list of
-        ** asockets that this asocket lives in
-        */
-    asocket *next;
-    asocket *prev;
-
-        /* the unique identifier for this asocket
-        */
-    unsigned id;
-
-        /* flag: set when the socket's peer has closed
-        ** but packets are still queued for delivery
-        */
-    int    closing;
-
-        /* flag: quit adbd when both ends close the
-        ** local service socket
-        */
-    int    exit_on_close;
-
-        /* the asocket we are connected to
-        */
-
-    asocket *peer;
-
-        /* For local asockets, the fde is used to bind
-        ** us to our fd event system.  For remote asockets
-        ** these fields are not used.
-        */
-    fdevent fde;
-    int fd;
-
-        /* queue of apackets waiting to be written
-        */
-    apacket *pkt_first;
-    apacket *pkt_last;
-
-        /* enqueue is called by our peer when it has data
-        ** for us.  It should return 0 if we can accept more
-        ** data or 1 if not.  If we return 1, we must call
-        ** peer->ready() when we once again are ready to
-        ** receive data.
-        */
-    int (*enqueue)(asocket *s, apacket *pkt);
-
-        /* ready is called by the peer when it is ready for
-        ** us to send data via enqueue again
-        */
-    void (*ready)(asocket *s);
-
-        /* shutdown is called by the peer before it goes away.
-        ** the socket should not do any further calls on its peer.
-        ** Always followed by a call to close. Optional, i.e. can be NULL.
-        */
-    void (*shutdown)(asocket *s);
-
-        /* close is called by the peer when it has gone away.
-        ** we are not allowed to make any further calls on the
-        ** peer once our close method is called.
-        */
-    void (*close)(asocket *s);
-
-        /* A socket is bound to atransport */
-    atransport *transport;
-
-    size_t get_max_payload() const;
-};
-
-
 /* the adisconnect structure is used to record a callback that
 ** will be called whenever a transport is disconnected (e.g. by the user)
 ** this should be used to cleanup objects that depend on the
@@ -215,18 +142,7 @@ struct alistener
 
 void print_packet(const char *label, apacket *p);
 
-asocket *find_local_socket(unsigned local_id, unsigned remote_id);
-void install_local_socket(asocket *s);
-void remove_socket(asocket *s);
-void close_all_sockets(atransport *t);
-
-asocket *create_local_socket(int fd);
-asocket *create_local_service_socket(const char* destination,
-                                     const atransport* transport);
 
-asocket *create_remote_socket(unsigned id, atransport *t);
-void connect_to_remote(asocket *s, const char *destination);
-void connect_to_smartsocket(asocket *s);
 
 void fatal(const char *fmt, ...) __attribute__((noreturn));
 void fatal_errno(const char *fmt, ...) __attribute__((noreturn));
diff --git a/fdevent.cpp b/fdevent.cpp
index dc03807..666a15f 100644
--- a/fdevent.cpp
+++ b/fdevent.cpp
@@ -119,10 +119,10 @@ void fdevent_install(fdevent* fde, int fd, fd_func func, void* arg) {
     fde->func = func;
     fde->arg = arg;
     if (fcntl(fd, F_SETFL, O_NONBLOCK) != 0) {
-      // Here is not proper to handle the error. If it fails here, some error is
-      // likely to be detected by poll(), then we can let the callback function
-      // to handle it.
-      LOG(ERROR) << "failed to fcntl(" << fd << ") to be nonblock";
+        // Here is not proper to handle the error. If it fails here, some error is
+        // likely to be detected by poll(), then we can let the callback function
+        // to handle it.
+        LOG(ERROR) << "failed to fcntl(" << fd << ") to be nonblock";
     }
     auto pair = g_poll_node_map.emplace(fde->fd, PollNode(fde));
     CHECK(pair.second) << "install existing fd " << fd;
@@ -215,10 +215,13 @@ static void fdevent_process() {
     D("poll(), pollfds = %s", dump_pollfds(pollfds).c_str());
     int ret = TEMP_FAILURE_RETRY(poll(&pollfds[0], pollfds.size(), -1));
     if (ret == -1) {
-      PLOG(ERROR) << "poll(), ret = " << ret;
-      return;
+        PLOG(ERROR) << "poll(), ret = " << ret;
+        return;
     }
     for (auto& pollfd : pollfds) {
+        if (pollfd.revents != 0) {
+            D("for fd %d, revents = %x", pollfd.fd, pollfd.revents);
+        }
         unsigned events = 0;
         if (pollfd.revents & POLLIN) {
             events |= FDE_READ;
@@ -337,3 +340,12 @@ void fdevent_loop()
         }
     }
 }
+
+size_t fdevent_installed_count() {
+    return g_poll_node_map.size();
+}
+
+void fdevent_reset() {
+    g_poll_node_map.clear();
+    g_pending_list.clear();
+}
diff --git a/fdevent.h b/fdevent.h
index ca1494c..657fde5 100644
--- a/fdevent.h
+++ b/fdevent.h
@@ -17,6 +17,7 @@
 #ifndef __FDEVENT_H
 #define __FDEVENT_H
 
+#include <stddef.h>
 #include <stdint.h>  /* for int64_t */
 
 /* events that may be observed */
@@ -27,10 +28,22 @@
 /* features that may be set (via the events set/add/del interface) */
 #define FDE_DONT_CLOSE        0x0080
 
-struct fdevent;
-
 typedef void (*fd_func)(int fd, unsigned events, void *userdata);
 
+struct fdevent {
+    fdevent *next;
+    fdevent *prev;
+
+    int fd;
+    int force_eof;
+
+    uint16_t state;
+    uint16_t events;
+
+    fd_func func;
+    void *arg;
+};
+
 /* Allocate and initialize a new fdevent object
  * Note: use FD_TIMER as 'fd' to create a fd-less object
  * (used to implement timers).
@@ -63,18 +76,9 @@ void fdevent_set_timeout(fdevent *fde, int64_t  timeout_ms);
 */
 void fdevent_loop();
 
-struct fdevent {
-    fdevent *next;
-    fdevent *prev;
-
-    int fd;
-    int force_eof;
-
-    uint16_t state;
-    uint16_t events;
-
-    fd_func func;
-    void *arg;
-};
+// For debugging only.
+size_t fdevent_installed_count();
+// For debugging only.
+void fdevent_reset();
 
 #endif
diff --git a/fdevent_test.cpp b/fdevent_test.cpp
index 33034d8..7457712 100644
--- a/fdevent_test.cpp
+++ b/fdevent_test.cpp
@@ -28,25 +28,6 @@
 
 #include "adb_io.h"
 
-class SignalHandlerRegister {
-  public:
-    SignalHandlerRegister(const std::vector<int>& signums, void (*handler)(int)) {
-        for (auto& sig : signums) {
-            sig_t old_handler = signal(sig, handler);
-            saved_signal_handlers_.push_back(std::make_pair(sig, old_handler));
-        }
-    }
-
-    ~SignalHandlerRegister() {
-        for (auto& pair : saved_signal_handlers_) {
-            signal(pair.first, pair.second);
-        }
-    }
-
-  private:
-    std::vector<std::pair<int, sig_t>> saved_signal_handlers_;
-};
-
 class FdHandler {
   public:
     FdHandler(int read_fd, int write_fd) : read_fd_(read_fd), write_fd_(write_fd) {
@@ -95,6 +76,19 @@ static void signal_handler(int) {
     pthread_exit(nullptr);
 }
 
+class FdeventTest : public ::testing::Test {
+  protected:
+    static void SetUpTestCase() {
+        ASSERT_NE(SIG_ERR, signal(SIGUSR1, signal_handler));
+        ASSERT_NE(SIG_ERR, signal(SIGPIPE, SIG_IGN));
+    }
+
+    virtual void SetUp() {
+        fdevent_reset();
+        ASSERT_EQ(0u, fdevent_installed_count());
+    }
+};
+
 struct ThreadArg {
     int first_read_fd;
     int last_write_fd;
@@ -102,8 +96,6 @@ struct ThreadArg {
 };
 
 static void FdEventThreadFunc(ThreadArg* arg) {
-    SignalHandlerRegister signal_handler_register({SIGUSR1}, signal_handler);
-
     std::vector<int> read_fds;
     std::vector<int> write_fds;
 
@@ -124,7 +116,7 @@ static void FdEventThreadFunc(ThreadArg* arg) {
     fdevent_loop();
 }
 
-TEST(fdevent, smoke) {
+TEST_F(FdeventTest, smoke) {
     const size_t PIPE_COUNT = 10;
     const size_t MESSAGE_LOOP_COUNT = 100;
     const std::string MESSAGE = "fdevent_test";
@@ -154,6 +146,8 @@ TEST(fdevent, smoke) {
 
     ASSERT_EQ(0, pthread_kill(thread, SIGUSR1));
     ASSERT_EQ(0, pthread_join(thread, nullptr));
+    ASSERT_EQ(0, close(writer));
+    ASSERT_EQ(0, close(reader));
 }
 
 struct InvalidFdArg {
@@ -171,7 +165,7 @@ static void InvalidFdEventCallback(int fd, unsigned events, void* userdata) {
     }
 }
 
-void InvalidFdThreadFunc(void*) {
+static void InvalidFdThreadFunc(void*) {
     const int INVALID_READ_FD = std::numeric_limits<int>::max() - 1;
     size_t happened_event_count = 0;
     InvalidFdArg read_arg;
@@ -189,7 +183,7 @@ void InvalidFdThreadFunc(void*) {
     fdevent_loop();
 }
 
-TEST(fdevent, invalid_fd) {
+TEST_F(FdeventTest, invalid_fd) {
     pthread_t thread;
     ASSERT_EQ(0, pthread_create(&thread, nullptr,
                                 reinterpret_cast<void* (*)(void*)>(InvalidFdThreadFunc),
diff --git a/socket.h b/socket.h
new file mode 100644
index 0000000..4083036
--- /dev/null
+++ b/socket.h
@@ -0,0 +1,117 @@
+/*
+ * Copyright (C) 2015 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 __ADB_SOCKET_H
+#define __ADB_SOCKET_H
+
+#include <stddef.h>
+
+#include "fdevent.h"
+
+struct apacket;
+class atransport;
+
+/* An asocket represents one half of a connection between a local and
+** remote entity.  A local asocket is bound to a file descriptor.  A
+** remote asocket is bound to the protocol engine.
+*/
+struct asocket {
+        /* chain pointers for the local/remote list of
+        ** asockets that this asocket lives in
+        */
+    asocket *next;
+    asocket *prev;
+
+        /* the unique identifier for this asocket
+        */
+    unsigned id;
+
+        /* flag: set when the socket's peer has closed
+        ** but packets are still queued for delivery
+        */
+    int    closing;
+
+    // flag: set when the socket failed to write, so the socket will not wait to
+    // write packets and close directly.
+    bool has_write_error;
+
+        /* flag: quit adbd when both ends close the
+        ** local service socket
+        */
+    int    exit_on_close;
+
+        /* the asocket we are connected to
+        */
+
+    asocket *peer;
+
+        /* For local asockets, the fde is used to bind
+        ** us to our fd event system.  For remote asockets
+        ** these fields are not used.
+        */
+    fdevent fde;
+    int fd;
+
+        /* queue of apackets waiting to be written
+        */
+    apacket *pkt_first;
+    apacket *pkt_last;
+
+        /* enqueue is called by our peer when it has data
+        ** for us.  It should return 0 if we can accept more
+        ** data or 1 if not.  If we return 1, we must call
+        ** peer->ready() when we once again are ready to
+        ** receive data.
+        */
+    int (*enqueue)(asocket *s, apacket *pkt);
+
+        /* ready is called by the peer when it is ready for
+        ** us to send data via enqueue again
+        */
+    void (*ready)(asocket *s);
+
+        /* shutdown is called by the peer before it goes away.
+        ** the socket should not do any further calls on its peer.
+        ** Always followed by a call to close. Optional, i.e. can be NULL.
+        */
+    void (*shutdown)(asocket *s);
+
+        /* close is called by the peer when it has gone away.
+        ** we are not allowed to make any further calls on the
+        ** peer once our close method is called.
+        */
+    void (*close)(asocket *s);
+
+        /* A socket is bound to atransport */
+    atransport *transport;
+
+    size_t get_max_payload() const;
+};
+
+asocket *find_local_socket(unsigned local_id, unsigned remote_id);
+void install_local_socket(asocket *s);
+void remove_socket(asocket *s);
+void close_all_sockets(atransport *t);
+
+asocket *create_local_socket(int fd);
+asocket *create_local_service_socket(const char* destination,
+                                     const atransport* transport);
+
+asocket *create_remote_socket(unsigned id, atransport *t);
+void connect_to_remote(asocket *s, const char *destination);
+void connect_to_smartsocket(asocket *s);
+
+#endif  // __ADB_SOCKET_H
diff --git a/socket_test.cpp b/socket_test.cpp
new file mode 100644
index 0000000..8f395d1
--- /dev/null
+++ b/socket_test.cpp
@@ -0,0 +1,300 @@
+/*
+ * Copyright (C) 2015 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.
+ */
+
+#include "fdevent.h"
+
+#include <gtest/gtest.h>
+
+#include <limits>
+#include <queue>
+#include <string>
+#include <vector>
+
+#include <pthread.h>
+#include <signal.h>
+#include <unistd.h>
+
+#include "adb.h"
+#include "adb_io.h"
+#include "socket.h"
+#include "sysdeps.h"
+
+static void signal_handler(int) {
+    ASSERT_EQ(1u, fdevent_installed_count());
+    pthread_exit(nullptr);
+}
+
+// On host, register a dummy socket, so fdevet_loop() will not abort when previously
+// registered local sockets are all closed. On device, fdevent_subproc_setup() installs
+// one fdevent which can be considered as dummy socket.
+static void InstallDummySocket() {
+#if ADB_HOST
+    int dummy_fds[2];
+    ASSERT_EQ(0, pipe(dummy_fds));
+    asocket* dummy_socket = create_local_socket(dummy_fds[0]);
+    ASSERT_TRUE(dummy_socket != nullptr);
+    dummy_socket->ready(dummy_socket);
+#endif
+}
+
+struct ThreadArg {
+    int first_read_fd;
+    int last_write_fd;
+    size_t middle_pipe_count;
+};
+
+static void FdEventThreadFunc(ThreadArg* arg) {
+    std::vector<int> read_fds;
+    std::vector<int> write_fds;
+
+    read_fds.push_back(arg->first_read_fd);
+    for (size_t i = 0; i < arg->middle_pipe_count; ++i) {
+        int fds[2];
+        ASSERT_EQ(0, adb_socketpair(fds));
+        read_fds.push_back(fds[0]);
+        write_fds.push_back(fds[1]);
+    }
+    write_fds.push_back(arg->last_write_fd);
+
+    for (size_t i = 0; i < read_fds.size(); ++i) {
+        asocket* reader = create_local_socket(read_fds[i]);
+        ASSERT_TRUE(reader != nullptr);
+        asocket* writer = create_local_socket(write_fds[i]);
+        ASSERT_TRUE(writer != nullptr);
+        reader->peer = writer;
+        writer->peer = reader;
+        reader->ready(reader);
+    }
+
+    InstallDummySocket();
+    fdevent_loop();
+}
+
+class LocalSocketTest : public ::testing::Test {
+  protected:
+    static void SetUpTestCase() {
+        ASSERT_NE(SIG_ERR, signal(SIGUSR1, signal_handler));
+        ASSERT_NE(SIG_ERR, signal(SIGPIPE, SIG_IGN));
+    }
+
+    virtual void SetUp() {
+        fdevent_reset();
+        ASSERT_EQ(0u, fdevent_installed_count());
+    }
+};
+
+TEST_F(LocalSocketTest, smoke) {
+    const size_t PIPE_COUNT = 100;
+    const size_t MESSAGE_LOOP_COUNT = 100;
+    const std::string MESSAGE = "socket_test";
+    int fd_pair1[2];
+    int fd_pair2[2];
+    ASSERT_EQ(0, adb_socketpair(fd_pair1));
+    ASSERT_EQ(0, adb_socketpair(fd_pair2));
+    pthread_t thread;
+    ThreadArg thread_arg;
+    thread_arg.first_read_fd = fd_pair1[0];
+    thread_arg.last_write_fd = fd_pair2[1];
+    thread_arg.middle_pipe_count = PIPE_COUNT;
+    int writer = fd_pair1[1];
+    int reader = fd_pair2[0];
+
+    ASSERT_EQ(0, pthread_create(&thread, nullptr,
+                                reinterpret_cast<void* (*)(void*)>(FdEventThreadFunc),
+                                &thread_arg));
+
+    usleep(1000);
+    for (size_t i = 0; i < MESSAGE_LOOP_COUNT; ++i) {
+        std::string read_buffer = MESSAGE;
+        std::string write_buffer(MESSAGE.size(), 'a');
+        ASSERT_TRUE(WriteFdExactly(writer, read_buffer.c_str(), read_buffer.size()));
+        ASSERT_TRUE(ReadFdExactly(reader, &write_buffer[0], write_buffer.size()));
+        ASSERT_EQ(read_buffer, write_buffer);
+    }
+    ASSERT_EQ(0, adb_close(writer));
+    ASSERT_EQ(0, adb_close(reader));
+    // Wait until the local sockets are closed.
+    sleep(1);
+
+    ASSERT_EQ(0, pthread_kill(thread, SIGUSR1));
+    ASSERT_EQ(0, pthread_join(thread, nullptr));
+}
+
+struct CloseWithPacketArg {
+    int socket_fd;
+    size_t bytes_written;
+    int cause_close_fd;
+};
+
+static void CloseWithPacketThreadFunc(CloseWithPacketArg* arg) {
+    asocket* s = create_local_socket(arg->socket_fd);
+    ASSERT_TRUE(s != nullptr);
+    arg->bytes_written = 0;
+    while (true) {
+        apacket* p = get_apacket();
+        p->len = sizeof(p->data);
+        arg->bytes_written += p->len;
+        int ret = s->enqueue(s, p);
+        if (ret == 1) {
+            // The writer has one packet waiting to send.
+            break;
+        }
+    }
+
+    asocket* cause_close_s = create_local_socket(arg->cause_close_fd);
+    ASSERT_TRUE(cause_close_s != nullptr);
+    cause_close_s->peer = s;
+    s->peer = cause_close_s;
+    cause_close_s->ready(cause_close_s);
+
+    InstallDummySocket();
+    fdevent_loop();
+}
+
+// This test checks if we can close local socket in the following situation:
+// The socket is closing but having some packets, so it is not closed. Then
+// some write error happens in the socket's file handler, e.g., the file
+// handler is closed.
+TEST_F(LocalSocketTest, close_with_packet) {
+    int socket_fd[2];
+    ASSERT_EQ(0, adb_socketpair(socket_fd));
+    int cause_close_fd[2];
+    ASSERT_EQ(0, adb_socketpair(cause_close_fd));
+    CloseWithPacketArg arg;
+    arg.socket_fd = socket_fd[1];
+    arg.cause_close_fd = cause_close_fd[1];
+    pthread_t thread;
+    ASSERT_EQ(0, pthread_create(&thread, nullptr,
+                                reinterpret_cast<void* (*)(void*)>(CloseWithPacketThreadFunc),
+                                &arg));
+    // Wait until the fdevent_loop() starts.
+    sleep(1);
+    ASSERT_EQ(0, adb_close(cause_close_fd[0]));
+    sleep(1);
+    ASSERT_EQ(2u, fdevent_installed_count());
+    ASSERT_EQ(0, adb_close(socket_fd[0]));
+    // Wait until the socket is closed.
+    sleep(1);
+
+    ASSERT_EQ(0, pthread_kill(thread, SIGUSR1));
+    ASSERT_EQ(0, pthread_join(thread, nullptr));
+}
+
+#undef shutdown
+
+// This test checks if we can read packets from a closing local socket.
+// The socket's file handler may be non readable if the other side has
+// called shutdown(SHUT_WR). But we should always write packets
+// successfully to the other side.
+TEST_F(LocalSocketTest, half_close_with_packet) {
+    int socket_fd[2];
+    ASSERT_EQ(0, adb_socketpair(socket_fd));
+    int cause_close_fd[2];
+    ASSERT_EQ(0, adb_socketpair(cause_close_fd));
+    CloseWithPacketArg arg;
+    arg.socket_fd = socket_fd[1];
+    arg.cause_close_fd = cause_close_fd[1];
+
+    pthread_t thread;
+    ASSERT_EQ(0, pthread_create(&thread, nullptr,
+                                reinterpret_cast<void* (*)(void*)>(CloseWithPacketThreadFunc),
+                                &arg));
+    // Wait until the fdevent_loop() starts.
+    sleep(1);
+    ASSERT_EQ(0, adb_close(cause_close_fd[0]));
+    sleep(1);
+    ASSERT_EQ(2u, fdevent_installed_count());
+    ASSERT_EQ(0, shutdown(socket_fd[0], SHUT_WR));
+
+    // Verify if we can read successfully.
+    std::vector<char> buf(arg.bytes_written);
+    ASSERT_EQ(true, ReadFdExactly(socket_fd[0], buf.data(), buf.size()));
+    ASSERT_EQ(0, adb_close(socket_fd[0]));
+
+    // Wait until the socket is closed.
+    sleep(1);
+
+    ASSERT_EQ(0, pthread_kill(thread, SIGUSR1));
+    ASSERT_EQ(0, pthread_join(thread, nullptr));
+}
+
+// This test checks if we can close local socket in the following situation:
+// The socket is not closed and has some packets. When it fails to write to
+// the socket's file handler because the other end is closed, we check if the
+// socket is closed.
+TEST_F(LocalSocketTest, write_error_when_having_packets) {
+    int socket_fd[2];
+    ASSERT_EQ(0, adb_socketpair(socket_fd));
+    int cause_close_fd[2];
+    ASSERT_EQ(0, adb_socketpair(cause_close_fd));
+    CloseWithPacketArg arg;
+    arg.socket_fd = socket_fd[1];
+    arg.cause_close_fd = cause_close_fd[1];
+
+    pthread_t thread;
+    ASSERT_EQ(0, pthread_create(&thread, nullptr,
+                                reinterpret_cast<void* (*)(void*)>(CloseWithPacketThreadFunc),
+                                &arg));
+    // Wait until the fdevent_loop() starts.
+    sleep(1);
+    ASSERT_EQ(3u, fdevent_installed_count());
+    ASSERT_EQ(0, adb_close(socket_fd[0]));
+
+    // Wait until the socket is closed.
+    sleep(1);
+
+    ASSERT_EQ(0, pthread_kill(thread, SIGUSR1));
+    ASSERT_EQ(0, pthread_join(thread, nullptr));
+}
+
+struct CloseNoEventsArg {
+    int socket_fd;
+};
+
+static void CloseNoEventsThreadFunc(CloseNoEventsArg* arg) {
+    asocket* s = create_local_socket(arg->socket_fd);
+    ASSERT_TRUE(s != nullptr);
+
+    InstallDummySocket();
+    fdevent_loop();
+}
+
+// This test checks when a local socket doesn't enable FDE_READ/FDE_WRITE/FDE_ERROR, it
+// can still be closed when some error happens on its file handler.
+// This test successes on linux but fails on mac because of different implementation of
+// poll(). I think the function tested here is useful to make adb server more stable on
+// linux.
+TEST_F(LocalSocketTest, close_with_no_events_installed) {
+    int socket_fd[2];
+    ASSERT_EQ(0, adb_socketpair(socket_fd));
+
+    CloseNoEventsArg arg;
+    arg.socket_fd = socket_fd[1];
+    pthread_t thread;
+    ASSERT_EQ(0, pthread_create(&thread, nullptr,
+                                reinterpret_cast<void* (*)(void*)>(CloseNoEventsThreadFunc),
+                                &arg));
+    // Wait until the fdevent_loop() starts.
+    sleep(1);
+    ASSERT_EQ(2u, fdevent_installed_count());
+    ASSERT_EQ(0, adb_close(socket_fd[0]));
+
+    // Wait until the socket is closed.
+    sleep(1);
+
+    ASSERT_EQ(0, pthread_kill(thread, SIGUSR1));
+    ASSERT_EQ(0, pthread_join(thread, nullptr));
+}
diff --git a/sockets.cpp b/sockets.cpp
index 104ad6b..bd33d79 100644
--- a/sockets.cpp
+++ b/sockets.cpp
@@ -157,6 +157,8 @@ static int local_socket_enqueue(asocket *s, apacket *p)
         }
         if((r == 0) || (errno != EAGAIN)) {
             D( "LS(%d): not ready, errno=%d: %s", s->id, errno, strerror(errno) );
+            put_apacket(p);
+            s->has_write_error = true;
             s->close(s);
             return 1; /* not ready (error) */
         } else {
@@ -252,7 +254,7 @@ static void local_socket_close_locked(asocket *s)
         /* If we are already closing, or if there are no
         ** pending packets, destroy immediately
         */
-    if (s->closing || s->pkt_first == NULL) {
+    if (s->closing || s->has_write_error || s->pkt_first == NULL) {
         int   id = s->id;
         local_socket_destroy(s);
         D("LS(%d): closed", id);
@@ -267,6 +269,7 @@ static void local_socket_close_locked(asocket *s)
     remove_socket(s);
     D("LS(%d): put on socket_closing_list fd=%d", s->id, s->fd);
     insert_local_socket(s, &local_socket_closing_list);
+    CHECK_EQ(FDE_WRITE, s->fde.state & FDE_WRITE);
 }
 
 static void local_socket_event_func(int fd, unsigned ev, void* _s)
@@ -296,6 +299,7 @@ static void local_socket_event_func(int fd, unsigned ev, void* _s)
                 }
 
                 D(" closing after write because r=%d and errno is %d", r, errno);
+                s->has_write_error = true;
                 s->close(s);
                 return;
             }
@@ -392,6 +396,7 @@ static void local_socket_event_func(int fd, unsigned ev, void* _s)
             D(" closing because is_eof=%d r=%d s->fde.force_eof=%d",
               is_eof, r, s->fde.force_eof);
             s->close(s);
+            return;
         }
     }
 
@@ -401,7 +406,6 @@ static void local_socket_event_func(int fd, unsigned ev, void* _s)
             ** bytes of readable data.
             */
         D("LS(%d): FDE_ERROR (fd=%d)", s->id, s->fd);
-
         return;
     }
 }