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 
thread_new_sized(const char * name,size_t work_queue_capacity)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 
thread_new(const char * name)105 thread_t *thread_new(const char *name) {
106   return thread_new_sized(name, DEFAULT_WORK_QUEUE_CAPACITY);
107 }
108 
thread_free(thread_t * thread)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 
thread_join(thread_t * thread)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 
thread_post(thread_t * thread,thread_fn func,void * context)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 
thread_stop(thread_t * thread)152 void thread_stop(thread_t *thread) {
153   assert(thread != NULL);
154   reactor_stop(thread->reactor);
155 }
156 
thread_is_self(const thread_t * thread)157 bool thread_is_self(const thread_t *thread) {
158   assert(thread != NULL);
159   return !!pthread_equal(pthread_self(), thread->pthread);
160 }
161 
thread_get_reactor(const thread_t * thread)162 reactor_t *thread_get_reactor(const thread_t *thread) {
163   assert(thread != NULL);
164   return thread->reactor;
165 }
166 
thread_name(const thread_t * thread)167 const char *thread_name(const thread_t *thread) {
168   assert(thread != NULL);
169   return thread->name;
170 }
171 
run_thread(void * start_arg)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 
work_queue_read_cb(void * context)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 }
223