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 "osi_thread"
20
21#include <assert.h>
22#include <errno.h>
23#include <pthread.h>
24#include <string.h>
25#include <sys/prctl.h>
26#include <sys/types.h>
27#include <utils/Log.h>
28
29#include "fixed_queue.h"
30#include "reactor.h"
31#include "semaphore.h"
32#include "thread.h"
33
34struct thread_t {
35  pthread_t pthread;
36  pid_t tid;
37  char name[THREAD_NAME_MAX + 1];
38  reactor_t *reactor;
39  fixed_queue_t *work_queue;
40};
41
42struct start_arg {
43  thread_t *thread;
44  semaphore_t *start_sem;
45  int error;
46};
47
48typedef struct {
49  thread_fn func;
50  void *context;
51} work_item_t;
52
53static void *run_thread(void *start_arg);
54static void work_queue_read_cb(void *context);
55
56static const size_t WORK_QUEUE_CAPACITY = 128;
57
58thread_t *thread_new(const char *name) {
59  assert(name != NULL);
60
61  // Start is on the stack, but we use a semaphore, so it's safe
62  thread_t *ret = calloc(1, sizeof(thread_t));
63  if (!ret)
64    goto error;
65
66  ret->reactor = reactor_new();
67  if (!ret->reactor)
68    goto error;
69
70  ret->work_queue = fixed_queue_new(WORK_QUEUE_CAPACITY);
71  if (!ret->work_queue)
72    goto error;
73
74  struct start_arg start;
75  start.start_sem = semaphore_new(0);
76  if (!start.start_sem)
77    goto error;
78
79  strncpy(ret->name, name, THREAD_NAME_MAX);
80  start.thread = ret;
81  start.error = 0;
82  pthread_create(&ret->pthread, NULL, run_thread, &start);
83  semaphore_wait(start.start_sem);
84  semaphore_free(start.start_sem);
85  if (start.error)
86    goto error;
87  return ret;
88
89error:;
90  if (ret) {
91    fixed_queue_free(ret->work_queue, free);
92    reactor_free(ret->reactor);
93  }
94  free(ret);
95  return NULL;
96}
97
98void thread_free(thread_t *thread) {
99  if (!thread)
100    return;
101
102  thread_stop(thread);
103  pthread_join(thread->pthread, NULL);
104  fixed_queue_free(thread->work_queue, free);
105  reactor_free(thread->reactor);
106  free(thread);
107}
108
109bool thread_post(thread_t *thread, thread_fn func, void *context) {
110  assert(thread != NULL);
111  assert(func != NULL);
112
113  // TODO(sharvil): if the current thread == |thread| and we've run out
114  // of queue space, we should abort this operation, otherwise we'll
115  // deadlock.
116
117  // Queue item is freed either when the queue itself is destroyed
118  // or when the item is removed from the queue for dispatch.
119  work_item_t *item = (work_item_t *)malloc(sizeof(work_item_t));
120  if (!item) {
121    ALOGE("%s unable to allocate memory: %s", __func__, strerror(errno));
122    return false;
123  }
124  item->func = func;
125  item->context = context;
126  fixed_queue_enqueue(thread->work_queue, item);
127  return true;
128}
129
130void thread_stop(thread_t *thread) {
131  assert(thread != NULL);
132  reactor_stop(thread->reactor);
133}
134
135const char *thread_name(const thread_t *thread) {
136  assert(thread != NULL);
137  return thread->name;
138}
139
140static void *run_thread(void *start_arg) {
141  assert(start_arg != NULL);
142
143  struct start_arg *start = start_arg;
144  thread_t *thread = start->thread;
145
146  assert(thread != NULL);
147
148  if (prctl(PR_SET_NAME, (unsigned long)thread->name) == -1) {
149    ALOGE("%s unable to set thread name: %s", __func__, strerror(errno));
150    start->error = errno;
151    semaphore_post(start->start_sem);
152    return NULL;
153  }
154  thread->tid = gettid();
155
156  semaphore_post(start->start_sem);
157
158  reactor_object_t work_queue_object;
159  work_queue_object.context = thread->work_queue;
160  work_queue_object.fd = fixed_queue_get_dequeue_fd(thread->work_queue);
161  work_queue_object.interest = REACTOR_INTEREST_READ;
162  work_queue_object.read_ready = work_queue_read_cb;
163
164  reactor_register(thread->reactor, &work_queue_object);
165  reactor_start(thread->reactor);
166
167  // Make sure we dispatch all queued work items before exiting the thread.
168  // This allows a caller to safely tear down by enqueuing a teardown
169  // work item and then joining the thread.
170  size_t count = 0;
171  work_item_t *item = fixed_queue_try_dequeue(thread->work_queue);
172  while (item && count <= WORK_QUEUE_CAPACITY) {
173    item->func(item->context);
174    free(item);
175    item = fixed_queue_try_dequeue(thread->work_queue);
176    ++count;
177  }
178
179  if (count > WORK_QUEUE_CAPACITY)
180    ALOGD("%s growing event queue on shutdown.", __func__);
181
182  return NULL;
183}
184
185static void work_queue_read_cb(void *context) {
186  assert(context != NULL);
187
188  fixed_queue_t *queue = (fixed_queue_t *)context;
189  work_item_t *item = fixed_queue_dequeue(queue);
190  item->func(item->context);
191  free(item);
192}
193