Home | History | Annotate | Download | only in src
      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 "osi/include/thread.h"
     22 
     23 #include <atomic>
     24 
     25 #include <base/logging.h>
     26 #include <errno.h>
     27 #include <malloc.h>
     28 #include <pthread.h>
     29 #include <string.h>
     30 #include <sys/prctl.h>
     31 #include <sys/resource.h>
     32 #include <sys/types.h>
     33 #include <unistd.h>
     34 
     35 #include "osi/include/allocator.h"
     36 #include "osi/include/compat.h"
     37 #include "osi/include/fixed_queue.h"
     38 #include "osi/include/log.h"
     39 #include "osi/include/reactor.h"
     40 #include "osi/include/semaphore.h"
     41 
     42 struct thread_t {
     43   std::atomic_bool is_joined{false};
     44   pthread_t pthread;
     45   pid_t tid;
     46   char name[THREAD_NAME_MAX + 1];
     47   reactor_t* reactor;
     48   fixed_queue_t* work_queue;
     49 };
     50 
     51 struct start_arg {
     52   thread_t* thread;
     53   semaphore_t* start_sem;
     54   int error;
     55 };
     56 
     57 typedef struct {
     58   thread_fn func;
     59   void* context;
     60 } work_item_t;
     61 
     62 static void* run_thread(void* start_arg);
     63 static void work_queue_read_cb(void* context);
     64 
     65 static const size_t DEFAULT_WORK_QUEUE_CAPACITY = 128;
     66 
     67 thread_t* thread_new_sized(const char* name, size_t work_queue_capacity) {
     68   CHECK(name != NULL);
     69   CHECK(work_queue_capacity != 0);
     70 
     71   thread_t* ret = static_cast<thread_t*>(osi_calloc(sizeof(thread_t)));
     72 
     73   ret->reactor = reactor_new();
     74   if (!ret->reactor) goto error;
     75 
     76   ret->work_queue = fixed_queue_new(work_queue_capacity);
     77   if (!ret->work_queue) goto error;
     78 
     79   // Start is on the stack, but we use a semaphore, so it's safe
     80   struct start_arg start;
     81   start.start_sem = semaphore_new(0);
     82   if (!start.start_sem) 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) goto error;
     92 
     93   return ret;
     94 
     95 error:;
     96   if (ret) {
     97     fixed_queue_free(ret->work_queue, osi_free);
     98     reactor_free(ret->reactor);
     99   }
    100   osi_free(ret);
    101   return NULL;
    102 }
    103 
    104 thread_t* thread_new(const char* name) {
    105   return thread_new_sized(name, DEFAULT_WORK_QUEUE_CAPACITY);
    106 }
    107 
    108 void thread_free(thread_t* thread) {
    109   if (!thread) return;
    110 
    111   thread_stop(thread);
    112   thread_join(thread);
    113 
    114   fixed_queue_free(thread->work_queue, osi_free);
    115   reactor_free(thread->reactor);
    116   osi_free(thread);
    117 }
    118 
    119 void thread_join(thread_t* thread) {
    120   CHECK(thread != NULL);
    121 
    122   if (!std::atomic_exchange(&thread->is_joined, true))
    123     pthread_join(thread->pthread, NULL);
    124 }
    125 
    126 bool thread_post(thread_t* thread, thread_fn func, void* context) {
    127   CHECK(thread != NULL);
    128   CHECK(func != NULL);
    129 
    130   // TODO(sharvil): if the current thread == |thread| and we've run out
    131   // of queue space, we should abort this operation, otherwise we'll
    132   // deadlock.
    133 
    134   // Queue item is freed either when the queue itself is destroyed
    135   // or when the item is removed from the queue for dispatch.
    136   work_item_t* item = (work_item_t*)osi_malloc(sizeof(work_item_t));
    137   item->func = func;
    138   item->context = context;
    139   fixed_queue_enqueue(thread->work_queue, item);
    140   return true;
    141 }
    142 
    143 void thread_stop(thread_t* thread) {
    144   CHECK(thread != NULL);
    145   reactor_stop(thread->reactor);
    146 }
    147 
    148 bool thread_set_priority(thread_t* thread, int priority) {
    149   if (!thread) return false;
    150 
    151   const int rc = setpriority(PRIO_PROCESS, thread->tid, priority);
    152   if (rc < 0) {
    153     LOG_ERROR(LOG_TAG,
    154               "%s unable to set thread priority %d for tid %d, error %d",
    155               __func__, priority, thread->tid, rc);
    156     return false;
    157   }
    158 
    159   return true;
    160 }
    161 
    162 bool thread_set_rt_priority(thread_t* thread, int priority) {
    163   if (!thread) return false;
    164 
    165   struct sched_param rt_params;
    166   rt_params.sched_priority = priority;
    167 
    168   const int rc = sched_setscheduler(thread->tid, SCHED_FIFO, &rt_params);
    169   if (rc != 0) {
    170     LOG_ERROR(LOG_TAG,
    171               "%s unable to set SCHED_FIFO priority %d for tid %d, error %s",
    172               __func__, priority, thread->tid, strerror(errno));
    173     return false;
    174   }
    175 
    176   return true;
    177 }
    178 
    179 bool thread_is_self(const thread_t* thread) {
    180   CHECK(thread != NULL);
    181   return !!pthread_equal(pthread_self(), thread->pthread);
    182 }
    183 
    184 reactor_t* thread_get_reactor(const thread_t* thread) {
    185   CHECK(thread != NULL);
    186   return thread->reactor;
    187 }
    188 
    189 const char* thread_name(const thread_t* thread) {
    190   CHECK(thread != NULL);
    191   return thread->name;
    192 }
    193 
    194 static void* run_thread(void* start_arg) {
    195   CHECK(start_arg != NULL);
    196 
    197   struct start_arg* start = static_cast<struct start_arg*>(start_arg);
    198   thread_t* thread = start->thread;
    199 
    200   CHECK(thread != NULL);
    201 
    202   if (prctl(PR_SET_NAME, (unsigned long)thread->name) == -1) {
    203     LOG_ERROR(LOG_TAG, "%s unable to set thread name: %s", __func__,
    204               strerror(errno));
    205     start->error = errno;
    206     semaphore_post(start->start_sem);
    207     return NULL;
    208   }
    209   thread->tid = gettid();
    210 
    211   LOG_INFO(LOG_TAG, "%s: thread id %d, thread name %s started", __func__,
    212            thread->tid, thread->name);
    213 
    214   semaphore_post(start->start_sem);
    215 
    216   int fd = fixed_queue_get_dequeue_fd(thread->work_queue);
    217   void* context = thread->work_queue;
    218 
    219   reactor_object_t* work_queue_object =
    220       reactor_register(thread->reactor, fd, context, work_queue_read_cb, NULL);
    221   reactor_start(thread->reactor);
    222   reactor_unregister(work_queue_object);
    223 
    224   // Make sure we dispatch all queued work items before exiting the thread.
    225   // This allows a caller to safely tear down by enqueuing a teardown
    226   // work item and then joining the thread.
    227   size_t count = 0;
    228   work_item_t* item =
    229       static_cast<work_item_t*>(fixed_queue_try_dequeue(thread->work_queue));
    230   while (item && count <= fixed_queue_capacity(thread->work_queue)) {
    231     item->func(item->context);
    232     osi_free(item);
    233     item =
    234         static_cast<work_item_t*>(fixed_queue_try_dequeue(thread->work_queue));
    235     ++count;
    236   }
    237 
    238   if (count > fixed_queue_capacity(thread->work_queue))
    239     LOG_DEBUG(LOG_TAG, "%s growing event queue on shutdown.", __func__);
    240 
    241   LOG_WARN(LOG_TAG, "%s: thread id %d, thread name %s exited", __func__,
    242            thread->tid, thread->name);
    243   return NULL;
    244 }
    245 
    246 static void work_queue_read_cb(void* context) {
    247   CHECK(context != NULL);
    248 
    249   fixed_queue_t* queue = (fixed_queue_t*)context;
    250   work_item_t* item = static_cast<work_item_t*>(fixed_queue_dequeue(queue));
    251   item->func(item->context);
    252   osi_free(item);
    253 }
    254