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
42struct 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
51struct start_arg {
52  thread_t* thread;
53  semaphore_t* start_sem;
54  int error;
55};
56
57typedef struct {
58  thread_fn func;
59  void* context;
60} work_item_t;
61
62static void* run_thread(void* start_arg);
63static void work_queue_read_cb(void* context);
64
65static const size_t DEFAULT_WORK_QUEUE_CAPACITY = 128;
66
67thread_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
95error:;
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
104thread_t* thread_new(const char* name) {
105  return thread_new_sized(name, DEFAULT_WORK_QUEUE_CAPACITY);
106}
107
108void 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
119void 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
126bool 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
143void thread_stop(thread_t* thread) {
144  CHECK(thread != NULL);
145  reactor_stop(thread->reactor);
146}
147
148bool 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
162bool 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
179bool thread_is_self(const thread_t* thread) {
180  CHECK(thread != NULL);
181  return !!pthread_equal(pthread_self(), thread->pthread);
182}
183
184reactor_t* thread_get_reactor(const thread_t* thread) {
185  CHECK(thread != NULL);
186  return thread->reactor;
187}
188
189const char* thread_name(const thread_t* thread) {
190  CHECK(thread != NULL);
191  return thread->name;
192}
193
194static 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
246static 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