Merge pull request #1191 from nicolasnoble/gpr_thd_join

Adding joinable threads, and gpr_thd_join.

3 files changed, 128 insertions, 26 deletions

diff --git a/src/core/support/thd.c b/src/core/support/thd.c
new file mode 100644
index 0000000000..ec308f3119
--- /dev/null
+++ b/src/core/support/thd.c
@@ -0,0 +1,66 @@
+/*
+ *
+ * Copyright 2015, Google Inc.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are
+ * met:
+ *
+ *     * Redistributions of source code must retain the above copyright
+ * notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above
+ * copyright notice, this list of conditions and the following disclaimer
+ * in the documentation and/or other materials provided with the
+ * distribution.
+ *     * Neither the name of Google Inc. nor the names of its
+ * contributors may be used to endorse or promote products derived from
+ * this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ *
+ */
+
+/* Posix implementation for gpr threads. */
+
+#include <memory.h>
+
+#include <grpc/support/thd.h>
+
+enum {
+  GPR_THD_JOINABLE = 1
+};
+
+gpr_thd_options gpr_thd_options_default(void) {
+  gpr_thd_options options;
+  memset(&options, 0, sizeof(options));
+  return options;
+}
+
+void gpr_thd_options_set_detached(gpr_thd_options *options) {
+  options->flags &= ~GPR_THD_JOINABLE;
+}
+
+void gpr_thd_options_set_joinable(gpr_thd_options *options) {
+  options->flags |= GPR_THD_JOINABLE;
+}
+
+int gpr_thd_options_is_detached(const gpr_thd_options *options) {
+  if (!options) return 1;
+  return (options->flags & GPR_THD_JOINABLE) == 0;
+}
+
+int gpr_thd_options_is_joinable(const gpr_thd_options *options) {
+  if (!options) return 0;
+  return (options->flags & GPR_THD_JOINABLE) == GPR_THD_JOINABLE;
+}
diff --git a/src/core/support/thd_posix.c b/src/core/support/thd_posix.c
index f50ea58335..7bf527201d 100644
--- a/src/core/support/thd_posix.c
+++ b/src/core/support/thd_posix.c
@@ -68,7 +68,11 @@ int gpr_thd_new(gpr_thd_id *t, void (*thd_body)(void *arg), void *arg,
   a->arg = arg;
 
   GPR_ASSERT(pthread_attr_init(&attr) == 0);
-  GPR_ASSERT(pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED) == 0);
+  if (gpr_thd_options_is_detached(options)) {
+    GPR_ASSERT(pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_DETACHED) == 0);
+  } else {
+    GPR_ASSERT(pthread_attr_setdetachstate(&attr, PTHREAD_CREATE_JOINABLE) == 0);
+  }
   thread_started = (pthread_create(&p, &attr, &thread_body, a) == 0);
   GPR_ASSERT(pthread_attr_destroy(&attr) == 0);
   if (!thread_started) {
@@ -78,14 +82,12 @@ int gpr_thd_new(gpr_thd_id *t, void (*thd_body)(void *arg), void *arg,
   return thread_started;
 }
 
-gpr_thd_options gpr_thd_options_default(void) {
-  gpr_thd_options options;
-  memset(&options, 0, sizeof(options));
-  return options;
-}
-
 gpr_thd_id gpr_thd_currentid(void) {
   return (gpr_thd_id)pthread_self();
 }
 
+void gpr_thd_join(gpr_thd_id t) {
+  pthread_join(t, NULL);
+}
+
 #endif /* GPR_POSIX_SYNC */
diff --git a/src/core/support/thd_win32.c b/src/core/support/thd_win32.c
index 347cad57e3..f92fb64a5c 100644
--- a/src/core/support/thd_win32.c
+++ b/src/core/support/thd_win32.c
@@ -31,7 +31,7 @@
  *
  */
 
-/* Posix implementation for gpr threads. */
+/* Windows implementation for gpr threads. */
 
 #include <grpc/support/port_platform.h>
 
@@ -40,47 +40,81 @@
 #include <windows.h>
 #include <string.h>
 #include <grpc/support/alloc.h>
+#include <grpc/support/log.h>
 #include <grpc/support/thd.h>
 
-struct thd_arg {
+#if defined(_MSC_VER)
+#define thread_local __declspec(thread)
+#elif defined(__GNUC__)
+#define thread_local __thread
+#else
+#error "Unknown compiler - please file a bug report"
+#endif
+
+struct thd_info {
   void (*body)(void *arg); /* body of a thread */
   void *arg;               /* argument to a thread */
+  HANDLE join_event;       /* if joinable, the join event */
+  int joinable;            /* true if not detached */
 };
 
+static thread_local struct thd_info *g_thd_info;
+
+/* Destroys a thread info */
+static destroy_thread(struct thd_info *t) {
+  if (t->joinable) CloseHandle(t->join_event);
+  gpr_free(t);
+}
+
 /* Body of every thread started via gpr_thd_new. */
 static DWORD WINAPI thread_body(void *v) {
-  struct thd_arg a = *(struct thd_arg *)v;
-  gpr_free(v);
-  (*a.body)(a.arg);
+  g_thd_info = (struct thd_info *)v;
+  g_thd_info->body(g_thd_info->arg);
+  if (g_thd_info->joinable) {
+    BOOL ret = SetEvent(g_thd_info->join_event);
+    GPR_ASSERT(ret);
+  } else {
+    destroy_thread(g_thd_info);
+  }
   return 0;
 }
 
 int gpr_thd_new(gpr_thd_id *t, void (*thd_body)(void *arg), void *arg,
                 const gpr_thd_options *options) {
   HANDLE handle;
-  DWORD thread_id;
-  struct thd_arg *a = gpr_malloc(sizeof(*a));
-  a->body = thd_body;
-  a->arg = arg;
+  struct thd_info *info = gpr_malloc(sizeof(*info));
+  info->body = thd_body;
+  info->arg = arg;
   *t = 0;
-  handle = CreateThread(NULL, 64 * 1024, thread_body, a, 0, &thread_id);
+  if (gpr_thd_options_is_joinable(options)) {
+    info->joinable = 1;
+    info->join_event = CreateEvent(NULL, FALSE, FALSE, NULL);
+    if (info->join_event == NULL) {
+      gpr_free(info);
+      return 0;
+    }
+  } else {
+    info->joinable = 0;
+  }
+  handle = CreateThread(NULL, 64 * 1024, thread_body, info, 0, NULL);
   if (handle == NULL) {
-    gpr_free(a);
+    destroy_thread(info);
   } else {
-    CloseHandle(handle); /* threads are "detached" */
+    *t = (gpr_thd_id)info;
+    CloseHandle(handle);
   }
-  *t = (gpr_thd_id)thread_id;
   return handle != NULL;
 }
 
-gpr_thd_options gpr_thd_options_default(void) {
-  gpr_thd_options options;
-  memset(&options, 0, sizeof(options));
-  return options;
+gpr_thd_id gpr_thd_currentid(void) {
+  return (gpr_thd_id)g_thd_info;
 }
 
-gpr_thd_id gpr_thd_currentid(void) {
-  return (gpr_thd_id)GetCurrentThreadId();
+void gpr_thd_join(gpr_thd_id t) {
+  struct thd_info *info = (struct thd_info *)t;
+  DWORD ret = WaitForSingleObject(info->join_event, INFINITE);
+  GPR_ASSERT(ret == WAIT_OBJECT_0);
+  destroy_thread(info);
 }
 
 #endif /* GPR_WIN32 */