src/utils-upoll.c

   1 /*
   2  * Copyright 2016 IoT.bzh
   3  * Author: José Bollo <jose.bollo@iot.bzh>
   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 #include <sys/epoll.h>
  19 #include <pthread.h>
  20 #include <stdlib.h>
  21 #include <errno.h>
  22 #include <unistd.h>
  23 #include <assert.h>
  24
  25 #include "utils-upoll.h"
  26
  27
  28 struct upollfd;
  29
  30 struct upoll
  31 {
  32         struct upollfd *fd;
  33         void (*read)(void *);
  34         void (*write)(void *);
  35         void (*hangup)(void *);
  36         void *closure;
  37         struct upoll *next;
  38 };
  39
  40 struct upollfd
  41 {
  42         int fd;
  43         uint32_t events;
  44         struct upollfd *next;
  45         struct upoll *head;
  46 };
  47
  48 static int pollfd = 0;
  49 static struct upollfd *head = NULL;
  50 static pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
  51
  52 static int update(struct upollfd *ufd)
  53 {
  54         int rc;
  55         struct upoll *u;
  56         struct epoll_event e;
  57         uint32_t events;
  58         struct upollfd **prv;
  59
  60         events = 0;
  61         pthread_mutex_lock(&mutex);
  62         u = ufd->head;
  63         if (u == NULL) {
  64                 /* no more watchers */
  65                 prv = &head;
  66                 while(*prv) {
  67                         if (*prv == ufd) {
  68                                 *prv = ufd->next;
  69                                 break;
  70                         }
  71                         prv = &(*prv)->next;
  72                 }
  73                 pthread_mutex_unlock(&mutex);
  74                 epoll_ctl(pollfd, EPOLL_CTL_DEL, ufd->fd, NULL);
  75                 free(ufd);
  76                 return 0;
  77         }
  78         /* compute the events for the watchers */
  79         while (u != NULL) {
  80                 if (u->read != NULL)
  81                         events |= EPOLLIN;
  82                 if (u->write != NULL)
  83                         events |= EPOLLOUT;
  84                 u = u->next;
  85         }
  86         pthread_mutex_unlock(&mutex);
  87         if (ufd->events == events)
  88                 return 0;
  89         e.events = events;
  90         e.data.ptr = ufd;
  91         rc = epoll_ctl(pollfd, EPOLL_CTL_MOD, ufd->fd, &e);
  92         if (rc == 0)
  93                 ufd->events = events;
  94         return rc;
  95 }
  96
  97 static struct upollfd *get_fd(int fd)
  98 {
  99         struct epoll_event e;
 100         struct upollfd *result;
 101         int rc;
 102
 103         /* opens the epoll stream */
 104         if (pollfd == 0) {
 105                 pollfd = epoll_create1(EPOLL_CLOEXEC);
 106                 if (pollfd == 0) {
 107                         pollfd = dup(0);
 108                         close(0);
 109                 }
 110                 if (pollfd < 0) {
 111                         pollfd = 0;
 112                         return NULL;
 113                 }
 114         }
 115
 116         /* search */
 117         result = head;
 118         while (result != NULL) {
 119                 if (result->fd == fd)
 120                         return result;
 121                 result = result->next;
 122         }
 123
 124         /* allocates */
 125         result = calloc(1, sizeof *result);
 126         if (result == NULL)
 127                 return NULL;
 128
 129         /* init */
 130         result->fd = fd;
 131         pthread_mutex_lock(&mutex);
 132         result->next = head;
 133         head = result;
 134         pthread_mutex_unlock(&mutex);
 135
 136         /* records */
 137         e.events = 0;
 138         e.data.ptr = result;
 139         rc = epoll_ctl(pollfd, EPOLL_CTL_ADD, fd, &e);
 140         if (rc == 0)
 141                 return result;
 142
 143         /* revert on error */
 144         rc = errno;
 145         update(result);
 146         errno = rc;
 147         return NULL;
 148 }
 149
 150 int upoll_is_valid(struct upoll *upoll)
 151 {
 152         struct upollfd *itfd = head;
 153         struct upoll *it;
 154         while (itfd != NULL) {
 155                 it = itfd->head;
 156                 while (it != NULL) {
 157                         if (it == upoll)
 158                                 return 1;
 159                         it = it->next;
 160                 }
 161                 itfd = itfd->next;
 162         }
 163         return 0;
 164 }
 165
 166 struct upoll *upoll_open(int fd, void *closure)
 167 {
 168         struct upollfd *ufd;
 169         struct upoll *result;
 170
 171         /* allocates */
 172         result = calloc(1, sizeof *result);
 173         if (result == NULL)
 174                 return NULL;
 175
 176         /* get for fd */
 177         ufd = get_fd(fd);
 178         if (ufd == NULL) {
 179                 free(result);
 180                 return NULL;
 181         }
 182
 183         /* init */
 184         result->fd = ufd;
 185         result->closure = closure;
 186         pthread_mutex_lock(&mutex);
 187         result->next = ufd->head;
 188         ufd->head = result;
 189         pthread_mutex_unlock(&mutex);
 190         return result;
 191 }
 192
 193 int upoll_on_readable(struct upoll *upoll, void (*process)(void *))
 194 {
 195         assert(pollfd != 0);
 196         assert(upoll_is_valid(upoll));
 197
 198         upoll->read = process;
 199         return update(upoll->fd);
 200 }
 201
 202 int upoll_on_writable(struct upoll *upoll, void (*process)(void *))
 203 {
 204         assert(pollfd != 0);
 205         assert(upoll_is_valid(upoll));
 206
 207         upoll->write = process;
 208         return update(upoll->fd);
 209 }
 210
 211 void upoll_on_hangup(struct upoll *upoll, void (*process)(void *))
 212 {
 213         assert(pollfd != 0);
 214         assert(upoll_is_valid(upoll));
 215
 216         upoll->hangup = process;
 217 }
 218
 219 void upoll_close(struct upoll *upoll)
 220 {
 221         struct upoll **it;
 222         struct upollfd *ufd;
 223
 224         assert(pollfd != 0);
 225         assert(upoll_is_valid(upoll));
 226
 227         ufd = upoll->fd;
 228         pthread_mutex_lock(&mutex);
 229         it = &ufd->head;
 230         while (*it != upoll)
 231                 it = &(*it)->next;
 232         *it = upoll->next;
 233         pthread_mutex_unlock(&mutex);
 234         free(upoll);
 235         update(ufd);
 236 }
 237
 238 int upoll_wait(int timeout)
 239 {
 240         int rc;
 241         struct epoll_event e;
 242         struct upollfd *ufd;
 243         struct upoll *u;
 244
 245         if (pollfd == 0) {
 246                 errno = ECANCELED;
 247                 return -1;
 248         }
 249
 250         do {
 251                 rc = epoll_wait(pollfd, &e, 1, timeout);
 252         } while (rc < 0 && errno == EINTR);
 253         if (rc == 1) {
 254                 ufd = e.data.ptr;
 255                 u = ufd->head;
 256                 while (u != NULL) {
 257                         if ((e.events & EPOLLIN) && u->read) {
 258                                 u->read(u->closure);
 259                                 break;
 260                         }
 261                         if ((e.events & EPOLLOUT) && u->write) {
 262                                 u->write(u->closure);
 263                                 break;
 264                         }
 265                         if ((e.events & EPOLLHUP) && u->hangup) {
 266                                 u->hangup(u->closure);
 267                                 break;
 268                         }
 269                         u = u->next;
 270                 }
 271         }
 272         return rc < 0 ? rc : 0;
 273 }
 274