osi/src/thread.c

   1 /******************************************************************************
   2  *
   3  *  Copyright (C) 2014 Google, Inc.
   4  *
   5  *  Licensed under the Apache License, Version 2.0 (the "License");
   6  *  you may not use this file except in compliance with the License.
   7  *  You may obtain a copy of the License at:
   8  *
   9  *  http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  *  Unless required by applicable law or agreed to in writing, software
  12  *  distributed under the License is distributed on an "AS IS" BASIS,
  13  *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14  *  See the License for the specific language governing permissions and
  15  *  limitations under the License.
  16  *
  17  ******************************************************************************/
  18
  19 #define LOG_TAG "bt_osi_thread"
  20
  21 #include <assert.h>
  22 #include <errno.h>
  23 #include <malloc.h>
  24 #include <pthread.h>
  25 #include <string.h>
  26 #include <sys/prctl.h>
  27 #include <sys/types.h>
  28
  29 #include "osi/include/allocator.h"
  30 #include "osi/include/compat.h"
  31 #include "osi/include/fixed_queue.h"
  32 #include "osi/include/log.h"
  33 #include "osi/include/reactor.h"
  34 #include "osi/include/semaphore.h"
  35 #include "osi/include/thread.h"
  36
  37 struct thread_t {
  38   bool is_joined;
  39   pthread_t pthread;
  40   pid_t tid;
  41   char name[THREAD_NAME_MAX + 1];
  42   reactor_t *reactor;
  43   fixed_queue_t *work_queue;
  44 };
  45
  46 struct start_arg {
  47   thread_t *thread;
  48   semaphore_t *start_sem;
  49   int error;
  50 };
  51
  52 typedef struct {
  53   thread_fn func;
  54   void *context;
  55 } work_item_t;
  56
  57 static void *run_thread(void *start_arg);
  58 static void work_queue_read_cb(void *context);
  59
  60 static const size_t DEFAULT_WORK_QUEUE_CAPACITY = 128;
  61
  62 thread_t *thread_new_sized(const char *name, size_t work_queue_capacity) {
  63   assert(name != NULL);
  64   assert(work_queue_capacity != 0);
  65
  66   thread_t *ret = osi_calloc(sizeof(thread_t));
  67   if (!ret)
  68     goto error;
  69
  70   ret->reactor = reactor_new();
  71   if (!ret->reactor)
  72     goto error;
  73
  74   ret->work_queue = fixed_queue_new(work_queue_capacity);
  75   if (!ret->work_queue)
  76     goto error;
  77
  78   // Start is on the stack, but we use a semaphore, so it's safe
  79   struct start_arg start;
  80   start.start_sem = semaphore_new(0);
  81   if (!start.start_sem)
  82     goto error;
  83
  84   strncpy(ret->name, name, THREAD_NAME_MAX);
  85   start.thread = ret;
  86   start.error = 0;
  87   pthread_create(&ret->pthread, NULL, run_thread, &start);
  88   semaphore_wait(start.start_sem);
  89   semaphore_free(start.start_sem);
  90
  91   if (start.error)
  92     goto error;
  93
  94   return ret;
  95
  96 error:;
  97   if (ret) {
  98     fixed_queue_free(ret->work_queue, osi_free);
  99     reactor_free(ret->reactor);
 100   }
 101   osi_free(ret);
 102   return NULL;
 103 }
 104
 105 thread_t *thread_new(const char *name) {
 106   return thread_new_sized(name, DEFAULT_WORK_QUEUE_CAPACITY);
 107 }
 108
 109 void thread_free(thread_t *thread) {
 110   if (!thread)
 111     return;
 112
 113   thread_stop(thread);
 114   thread_join(thread);
 115
 116   fixed_queue_free(thread->work_queue, osi_free);
 117   reactor_free(thread->reactor);
 118   osi_free(thread);
 119 }
 120
 121 void thread_join(thread_t *thread) {
 122   assert(thread != NULL);
 123
 124   // TODO(zachoverflow): use a compare and swap when ready
 125   if (!thread->is_joined) {
 126     thread->is_joined = true;
 127     pthread_join(thread->pthread, NULL);
 128   }
 129 }
 130
 131 bool thread_post(thread_t *thread, thread_fn func, void *context) {
 132   assert(thread != NULL);
 133   assert(func != NULL);
 134
 135   // TODO(sharvil): if the current thread == |thread| and we've run out
 136   // of queue space, we should abort this operation, otherwise we'll
 137   // deadlock.
 138
 139   // Queue item is freed either when the queue itself is destroyed
 140   // or when the item is removed from the queue for dispatch.
 141   work_item_t *item = (work_item_t *)osi_malloc(sizeof(work_item_t));
 142   if (!item) {
 143     LOG_ERROR("%s unable to allocate memory: %s", __func__, strerror(errno));
 144     return false;
 145   }
 146   item->func = func;
 147   item->context = context;
 148   fixed_queue_enqueue(thread->work_queue, item);
 149   return true;
 150 }
 151
 152 void thread_stop(thread_t *thread) {
 153   assert(thread != NULL);
 154   reactor_stop(thread->reactor);
 155 }
 156
 157 bool thread_is_self(const thread_t *thread) {
 158   assert(thread != NULL);
 159   return !!pthread_equal(pthread_self(), thread->pthread);
 160 }
 161
 162 reactor_t *thread_get_reactor(const thread_t *thread) {
 163   assert(thread != NULL);
 164   return thread->reactor;
 165 }
 166
 167 const char *thread_name(const thread_t *thread) {
 168   assert(thread != NULL);
 169   return thread->name;
 170 }
 171
 172 static void *run_thread(void *start_arg) {
 173   assert(start_arg != NULL);
 174
 175   struct start_arg *start = start_arg;
 176   thread_t *thread = start->thread;
 177
 178   assert(thread != NULL);
 179
 180   if (prctl(PR_SET_NAME, (unsigned long)thread->name) == -1) {
 181     LOG_ERROR("%s unable to set thread name: %s", __func__, strerror(errno));
 182     start->error = errno;
 183     semaphore_post(start->start_sem);
 184     return NULL;
 185   }
 186   thread->tid = gettid();
 187
 188   semaphore_post(start->start_sem);
 189
 190   int fd = fixed_queue_get_dequeue_fd(thread->work_queue);
 191   void *context = thread->work_queue;
 192
 193   reactor_object_t *work_queue_object = reactor_register(thread->reactor, fd, context, work_queue_read_cb, NULL);
 194   reactor_start(thread->reactor);
 195   reactor_unregister(work_queue_object);
 196
 197   // Make sure we dispatch all queued work items before exiting the thread.
 198   // This allows a caller to safely tear down by enqueuing a teardown
 199   // work item and then joining the thread.
 200   size_t count = 0;
 201   work_item_t *item = fixed_queue_try_dequeue(thread->work_queue);
 202   while (item && count <= fixed_queue_capacity(thread->work_queue)) {
 203     item->func(item->context);
 204     osi_free(item);
 205     item = fixed_queue_try_dequeue(thread->work_queue);
 206     ++count;
 207   }
 208
 209   if (count > fixed_queue_capacity(thread->work_queue))
 210     LOG_DEBUG("%s growing event queue on shutdown.", __func__);
 211
 212   return NULL;
 213 }
 214
 215 static void work_queue_read_cb(void *context) {
 216   assert(context != NULL);
 217
 218   fixed_queue_t *queue = (fixed_queue_t *)context;
 219   work_item_t *item = fixed_queue_dequeue(queue);
 220   item->func(item->context);
 221   osi_free(item);
 222 }