osi/src/reactor.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_reactor"
  20
  21 #include <assert.h>
  22 #include <errno.h>
  23 #include <pthread.h>
  24 #include <stdlib.h>
  25 #include <string.h>
  26 #include <sys/epoll.h>
  27 #include <sys/eventfd.h>
  28
  29 #include "osi/include/allocator.h"
  30 #include "osi/include/list.h"
  31 #include "osi/include/log.h"
  32 #include "osi/include/reactor.h"
  33
  34 #if !defined(EFD_SEMAPHORE)
  35 #  define EFD_SEMAPHORE (1 << 0)
  36 #endif
  37
  38 struct reactor_t {
  39   int epoll_fd;
  40   int event_fd;
  41   pthread_mutex_t list_lock;  // protects invalidation_list.
  42   list_t *invalidation_list;  // reactor objects that have been unregistered.
  43   pthread_t run_thread;       // the pthread on which reactor_run is executing.
  44   bool is_running;            // indicates whether |run_thread| is valid.
  45   bool object_removed;
  46 };
  47
  48 struct reactor_object_t {
  49   int fd;                              // the file descriptor to monitor for events.
  50   void *context;                       // a context that's passed back to the *_ready functions.
  51   reactor_t *reactor;                  // the reactor instance this object is registered with.
  52   pthread_mutex_t lock;                // protects the lifetime of this object and all variables.
  53
  54   void (*read_ready)(void *context);   // function to call when the file descriptor becomes readable.
  55   void (*write_ready)(void *context);  // function to call when the file descriptor becomes writeable.
  56 };
  57
  58 static reactor_status_t run_reactor(reactor_t *reactor, int iterations);
  59
  60 static const size_t MAX_EVENTS = 64;
  61 static const eventfd_t EVENT_REACTOR_STOP = 1;
  62
  63 reactor_t *reactor_new(void) {
  64   reactor_t *ret = (reactor_t *)osi_calloc(sizeof(reactor_t));
  65   if (!ret)
  66     return NULL;
  67
  68   ret->epoll_fd = INVALID_FD;
  69   ret->event_fd = INVALID_FD;
  70
  71   ret->epoll_fd = epoll_create(MAX_EVENTS);
  72   if (ret->epoll_fd == INVALID_FD) {
  73     LOG_ERROR("%s unable to create epoll instance: %s", __func__, strerror(errno));
  74     goto error;
  75   }
  76
  77   ret->event_fd = eventfd(0, 0);
  78   if (ret->event_fd == INVALID_FD) {
  79     LOG_ERROR("%s unable to create eventfd: %s", __func__, strerror(errno));
  80     goto error;
  81   }
  82
  83   pthread_mutex_init(&ret->list_lock, NULL);
  84   ret->invalidation_list = list_new(NULL);
  85   if (!ret->invalidation_list) {
  86     LOG_ERROR("%s unable to allocate object invalidation list.", __func__);
  87     goto error;
  88   }
  89
  90   struct epoll_event event;
  91   memset(&event, 0, sizeof(event));
  92   event.events = EPOLLIN;
  93   event.data.ptr = NULL;
  94   if (epoll_ctl(ret->epoll_fd, EPOLL_CTL_ADD, ret->event_fd, &event) == -1) {
  95     LOG_ERROR("%s unable to register eventfd with epoll set: %s", __func__, strerror(errno));
  96     goto error;
  97   }
  98
  99   return ret;
 100
 101 error:;
 102   reactor_free(ret);
 103   return NULL;
 104 }
 105
 106 void reactor_free(reactor_t *reactor) {
 107   if (!reactor)
 108     return;
 109
 110   list_free(reactor->invalidation_list);
 111   close(reactor->event_fd);
 112   close(reactor->epoll_fd);
 113   osi_free(reactor);
 114 }
 115
 116 reactor_status_t reactor_start(reactor_t *reactor) {
 117   assert(reactor != NULL);
 118   return run_reactor(reactor, 0);
 119 }
 120
 121 reactor_status_t reactor_run_once(reactor_t *reactor) {
 122   assert(reactor != NULL);
 123   return run_reactor(reactor, 1);
 124 }
 125
 126 void reactor_stop(reactor_t *reactor) {
 127   assert(reactor != NULL);
 128
 129   eventfd_write(reactor->event_fd, EVENT_REACTOR_STOP);
 130 }
 131
 132 reactor_object_t *reactor_register(reactor_t *reactor,
 133     int fd, void *context,
 134     void (*read_ready)(void *context),
 135     void (*write_ready)(void *context)) {
 136   assert(reactor != NULL);
 137   assert(fd != INVALID_FD);
 138
 139   reactor_object_t *object = (reactor_object_t *)osi_calloc(sizeof(reactor_object_t));
 140   if (!object) {
 141     LOG_ERROR("%s unable to allocate reactor object: %s", __func__, strerror(errno));
 142     return NULL;
 143   }
 144
 145   object->reactor = reactor;
 146   object->fd = fd;
 147   object->context = context;
 148   object->read_ready = read_ready;
 149   object->write_ready = write_ready;
 150   pthread_mutex_init(&object->lock, NULL);
 151
 152   struct epoll_event event;
 153   memset(&event, 0, sizeof(event));
 154   if (read_ready)
 155     event.events |= (EPOLLIN | EPOLLRDHUP);
 156   if (write_ready)
 157     event.events |= EPOLLOUT;
 158   event.data.ptr = object;
 159
 160   if (epoll_ctl(reactor->epoll_fd, EPOLL_CTL_ADD, fd, &event) == -1) {
 161     LOG_ERROR("%s unable to register fd %d to epoll set: %s", __func__, fd, strerror(errno));
 162     pthread_mutex_destroy(&object->lock);
 163     osi_free(object);
 164     return NULL;
 165   }
 166
 167   return object;
 168 }
 169
 170 bool reactor_change_registration(reactor_object_t *object,
 171     void (*read_ready)(void *context),
 172     void (*write_ready)(void *context)) {
 173   assert(object != NULL);
 174
 175   struct epoll_event event;
 176   memset(&event, 0, sizeof(event));
 177   if (read_ready)
 178     event.events |= (EPOLLIN | EPOLLRDHUP);
 179   if (write_ready)
 180     event.events |= EPOLLOUT;
 181   event.data.ptr = object;
 182
 183   if (epoll_ctl(object->reactor->epoll_fd, EPOLL_CTL_MOD, object->fd, &event) == -1) {
 184     LOG_ERROR("%s unable to modify interest set for fd %d: %s", __func__, object->fd, strerror(errno));
 185     return false;
 186   }
 187
 188   pthread_mutex_lock(&object->lock);
 189   object->read_ready = read_ready;
 190   object->write_ready = write_ready;
 191   pthread_mutex_unlock(&object->lock);
 192
 193   return true;
 194 }
 195
 196 void reactor_unregister(reactor_object_t *obj) {
 197   assert(obj != NULL);
 198
 199   reactor_t *reactor = obj->reactor;
 200
 201   if (epoll_ctl(reactor->epoll_fd, EPOLL_CTL_DEL, obj->fd, NULL) == -1)
 202     LOG_ERROR("%s unable to unregister fd %d from epoll set: %s", __func__, obj->fd, strerror(errno));
 203
 204   if (reactor->is_running && pthread_equal(pthread_self(), reactor->run_thread)) {
 205     reactor->object_removed = true;
 206     return;
 207   }
 208
 209   pthread_mutex_lock(&reactor->list_lock);
 210   list_append(reactor->invalidation_list, obj);
 211   pthread_mutex_unlock(&reactor->list_lock);
 212
 213   // Taking the object lock here makes sure a callback for |obj| isn't
 214   // currently executing. The reactor thread must then either be before
 215   // the callbacks or after. If after, we know that the object won't be
 216   // referenced because it has been taken out of the epoll set. If before,
 217   // it won't be referenced because the reactor thread will check the
 218   // invalidation_list and find it in there. So by taking this lock, we
 219   // are waiting until the reactor thread drops all references to |obj|.
 220   // One the wait completes, we can unlock and destroy |obj| safely.
 221   pthread_mutex_lock(&obj->lock);
 222   pthread_mutex_unlock(&obj->lock);
 223   pthread_mutex_destroy(&obj->lock);
 224   osi_free(obj);
 225 }
 226
 227 // Runs the reactor loop for a maximum of |iterations|.
 228 // 0 |iterations| means loop forever.
 229 // |reactor| may not be NULL.
 230 static reactor_status_t run_reactor(reactor_t *reactor, int iterations) {
 231   assert(reactor != NULL);
 232
 233   reactor->run_thread = pthread_self();
 234   reactor->is_running = true;
 235
 236   struct epoll_event events[MAX_EVENTS];
 237   for (int i = 0; iterations == 0 || i < iterations; ++i) {
 238     pthread_mutex_lock(&reactor->list_lock);
 239     list_clear(reactor->invalidation_list);
 240     pthread_mutex_unlock(&reactor->list_lock);
 241
 242     int ret;
 243     do {
 244       ret = epoll_wait(reactor->epoll_fd, events, MAX_EVENTS, -1);
 245     } while (ret == -1 && errno == EINTR);
 246
 247     if (ret == -1) {
 248       LOG_ERROR("%s error in epoll_wait: %s", __func__, strerror(errno));
 249       reactor->is_running = false;
 250       return REACTOR_STATUS_ERROR;
 251     }
 252
 253     for (int j = 0; j < ret; ++j) {
 254       // The event file descriptor is the only one that registers with
 255       // a NULL data pointer. We use the NULL to identify it and break
 256       // out of the reactor loop.
 257       if (events[j].data.ptr == NULL) {
 258         eventfd_t value;
 259         eventfd_read(reactor->event_fd, &value);
 260         reactor->is_running = false;
 261         return REACTOR_STATUS_STOP;
 262       }
 263
 264       reactor_object_t *object = (reactor_object_t *)events[j].data.ptr;
 265
 266       pthread_mutex_lock(&reactor->list_lock);
 267       if (list_contains(reactor->invalidation_list, object)) {
 268         pthread_mutex_unlock(&reactor->list_lock);
 269         continue;
 270       }
 271
 272       // Downgrade the list lock to an object lock.
 273       pthread_mutex_lock(&object->lock);
 274       pthread_mutex_unlock(&reactor->list_lock);
 275
 276       reactor->object_removed = false;
 277       if (events[j].events & (EPOLLIN | EPOLLHUP | EPOLLRDHUP | EPOLLERR) && object->read_ready)
 278         object->read_ready(object->context);
 279       if (!reactor->object_removed && events[j].events & EPOLLOUT && object->write_ready)
 280         object->write_ready(object->context);
 281       pthread_mutex_unlock(&object->lock);
 282
 283       if (reactor->object_removed) {
 284         pthread_mutex_destroy(&object->lock);
 285         osi_free(object);
 286       }
 287     }
 288   }
 289
 290   reactor->is_running = false;
 291   return REACTOR_STATUS_DONE;
 292 }