Merge "adb: sysdeps: add support for joining threads."

adb/Android.mk

@@ -58,6 +58,7 @@ LIBADB_SRC_FILES := \
 LIBADB_TEST_SRCS := \
     adb_io_test.cpp \
     adb_utils_test.cpp \
+    sysdeps_test.cpp \
     transport_test.cpp \
 
 LIBADB_CFLAGS := \

adb/sysdeps.h

@@ -30,6 +30,7 @@
 #include <vector>
 
 // Include this before open/unlink are defined as macros below.
+#include <android-base/errors.h>
 #include <android-base/utf8.h>
 
 /*
@@ -114,13 +115,57 @@ static __inline__ void  adb_mutex_unlock( adb_mutex_t*  lock )
     LeaveCriticalSection( lock );
 }
 
-typedef  void*  (*adb_thread_func_t)(void*  arg);
+typedef void* (*adb_thread_func_t)(void* arg);
+typedef HANDLE adb_thread_t;
 
-typedef  void (*win_thread_func_t)(void*  arg);
+struct win_thread_args {
+    adb_thread_func_t func;
+    void* arg;
+};
 
-static __inline__ bool adb_thread_create(adb_thread_func_t func, void* arg) {
-    uintptr_t tid = _beginthread((win_thread_func_t)func, 0, arg);
-    return (tid != static_cast<uintptr_t>(-1L));
+static unsigned __stdcall win_thread_wrapper(void* args) {
+    win_thread_args thread_args = *static_cast<win_thread_args*>(args);
+    delete static_cast<win_thread_args*>(args);
+    void* result = thread_args.func(thread_args.arg);
+    return reinterpret_cast<unsigned>(result);
+}
+
+static __inline__ bool adb_thread_create(adb_thread_func_t func, void* arg,
+                                         adb_thread_t* thread = nullptr) {
+    win_thread_args* args = new win_thread_args{.func = func, .arg = arg};
+    uintptr_t handle = _beginthreadex(nullptr, 0, win_thread_wrapper, args, 0, nullptr);
+    if (handle != static_cast<uintptr_t>(0)) {
+        if (thread) {
+            *thread = reinterpret_cast<HANDLE>(handle);
+        } else {
+            CloseHandle(thread);
+        }
+        return true;
+    }
+    return false;
+}
+
+static __inline__ bool adb_thread_join(adb_thread_t thread) {
+    switch (WaitForSingleObject(thread, INFINITE)) {
+        case WAIT_OBJECT_0:
+            CloseHandle(thread);
+            return true;
+
+        case WAIT_FAILED:
+            fprintf(stderr, "adb_thread_join failed: %s\n",
+                    android::base::SystemErrorCodeToString(GetLastError()).c_str());
+            break;
+
+        default:
+            abort();
+    }
+
+    return false;
+}
+
+static __inline__ bool adb_thread_detach(adb_thread_t thread) {
+    CloseHandle(thread);
+    return true;
 }
 
 static __inline__ int adb_thread_setname(const std::string& name) {
@@ -658,14 +703,32 @@ static __inline__ int  adb_socket_accept(int  serverfd, struct sockaddr*  addr,
 
 typedef void*  (*adb_thread_func_t)( void*  arg );
 
-static __inline__ bool adb_thread_create(adb_thread_func_t start, void* arg) {
+typedef pthread_t adb_thread_t;
+
+static __inline__ bool adb_thread_create(adb_thread_func_t start, void* arg,
+                                         adb_thread_t* thread = nullptr) {
+    pthread_t temp;
     pthread_attr_t attr;
     pthread_attr_init(&attr);
-    pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED);
+    pthread_attr_setdetachstate(&attr, thread ? PTHREAD_CREATE_JOINABLE : PTHREAD_CREATE_DETACHED);
+    errno = pthread_create(&temp, &attr, start, arg);
+    if (errno == 0) {
+        if (thread) {
+            *thread = temp;
+        }
+        return true;
+    }
+    return false;
+}
 
-    pthread_t thread;
-    errno = pthread_create(&thread, &attr, start, arg);
-    return (errno == 0);
+static __inline__ bool adb_thread_join(adb_thread_t thread) {
+    errno = pthread_join(thread, nullptr);
+    return errno == 0;
+}
+
+static __inline__ bool adb_thread_detach(adb_thread_t thread) {
+    errno = pthread_detach(thread);
+    return errno == 0;
 }
 
 static __inline__ int adb_thread_setname(const std::string& name) {

adb/sysdeps_test.cpp

@@ -0,0 +1,59 @@
+/*
+ * Copyright (C) 2065 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 <gtest/gtest.h>
+#include <unistd.h>
+#include <atomic>
+
+#include "sysdeps.h"
+
+static void* increment_atomic_int(void* c) {
+    sleep(1);
+    reinterpret_cast<std::atomic<int>*>(c)->fetch_add(1);
+    return nullptr;
+}
+
+TEST(sysdeps_thread, smoke) {
+    std::atomic<int> counter(0);
+
+    for (int i = 0; i < 100; ++i) {
+        ASSERT_TRUE(adb_thread_create(increment_atomic_int, &counter));
+    }
+
+    sleep(2);
+    ASSERT_EQ(100, counter.load());
+}
+
+TEST(sysdeps_thread, join) {
+    std::atomic<int> counter(0);
+    std::vector<adb_thread_t> threads(500);
+    for (size_t i = 0; i < threads.size(); ++i) {
+        ASSERT_TRUE(adb_thread_create(increment_atomic_int, &counter, &threads[i]));
+    }
+
+    int current = counter.load();
+    ASSERT_GE(current, 0);
+    // Make sure that adb_thread_create actually creates threads, and doesn't do something silly
+    // like synchronously run the function passed in. The sleep in increment_atomic_int should be
+    // enough to keep this from being flakey.
+    ASSERT_LT(current, 500);
+
+    for (const auto& thread : threads) {
+        ASSERT_TRUE(adb_thread_join(thread));
+    }
+
+    ASSERT_EQ(500, counter.load());
+}