external/bsd/libevent/dist/evbuffer.c

   1 /*      $NetBSD$        */
   2 /*
   3  * Copyright (c) 2002-2004 Niels Provos <provos@citi.umich.edu>
   4  * All rights reserved.
   5  *
   6  * Redistribution and use in source and binary forms, with or without
   7  * modification, are permitted provided that the following conditions
   8  * are met:
   9  * 1. Redistributions of source code must retain the above copyright
  10  *    notice, this list of conditions and the following disclaimer.
  11  * 2. Redistributions in binary form must reproduce the above copyright
  12  *    notice, this list of conditions and the following disclaimer in the
  13  *    documentation and/or other materials provided with the distribution.
  14  * 3. The name of the author may not be used to endorse or promote products
  15  *    derived from this software without specific prior written permission.
  16  *
  17  * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR
  18  * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
  19  * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
  20  * IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT,
  21  * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
  22  * NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
  23  * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
  24  * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  25  * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
  26  * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  27  */
  28
  29 #include <sys/types.h>
  30
  31 #ifdef HAVE_CONFIG_H
  32 #include "config.h"
  33 #endif
  34
  35 #ifdef HAVE_SYS_TIME_H
  36 #include <sys/time.h>
  37 #endif
  38
  39 #include <errno.h>
  40 #include <stdio.h>
  41 #include <stdlib.h>
  42 #include <string.h>
  43 #ifdef HAVE_STDARG_H
  44 #include <stdarg.h>
  45 #endif
  46
  47 #ifdef WIN32
  48 #include <winsock2.h>
  49 #endif
  50
  51 #include "evutil.h"
  52 #include "event.h"
  53
  54 /* prototypes */
  55
  56 void bufferevent_read_pressure_cb(struct evbuffer *, size_t, size_t, void *);
  57
  58 static int
  59 bufferevent_add(struct event *ev, int timeout)
  60 {
  61         struct timeval tv, *ptv = NULL;
  62
  63         if (timeout) {
  64                 evutil_timerclear(&tv);
  65                 tv.tv_sec = timeout;
  66                 ptv = &tv;
  67         }
  68
  69         return (event_add(ev, ptv));
  70 }
  71
  72 /*
  73  * This callback is executed when the size of the input buffer changes.
  74  * We use it to apply back pressure on the reading side.
  75  */
  76
  77 void
  78 bufferevent_read_pressure_cb(struct evbuffer *buf, size_t old, size_t now,
  79     void *arg) {
  80         struct bufferevent *bufev = arg;
  81         /*
  82          * If we are below the watermark then reschedule reading if it's
  83          * still enabled.
  84          */
  85         if (bufev->wm_read.high == 0 || now < bufev->wm_read.high) {
  86                 evbuffer_setcb(buf, NULL, NULL);
  87
  88                 if (bufev->enabled & EV_READ)
  89                         bufferevent_add(&bufev->ev_read, bufev->timeout_read);
  90         }
  91 }
  92
  93 static void
  94 bufferevent_readcb(int fd, short event, void *arg)
  95 {
  96         struct bufferevent *bufev = arg;
  97         int res = 0;
  98         short what = EVBUFFER_READ;
  99         size_t len;
 100         int howmuch = -1;
 101
 102         if (event == EV_TIMEOUT) {
 103                 what |= EVBUFFER_TIMEOUT;
 104                 goto error;
 105         }
 106
 107         /*
 108          * If we have a high watermark configured then we don't want to
 109          * read more data than would make us reach the watermark.
 110          */
 111         if (bufev->wm_read.high != 0) {
 112                 howmuch = bufev->wm_read.high - EVBUFFER_LENGTH(bufev->input);
 113                 /* we might have lowered the watermark, stop reading */
 114                 if (howmuch <= 0) {
 115                         struct evbuffer *buf = bufev->input;
 116                         event_del(&bufev->ev_read);
 117                         evbuffer_setcb(buf,
 118                             bufferevent_read_pressure_cb, bufev);
 119                         return;
 120                 }
 121         }
 122
 123         res = evbuffer_read(bufev->input, fd, howmuch);
 124         if (res == -1) {
 125                 if (errno == EAGAIN || errno == EINTR)
 126                         goto reschedule;
 127                 /* error case */
 128                 what |= EVBUFFER_ERROR;
 129         } else if (res == 0) {
 130                 /* eof case */
 131                 what |= EVBUFFER_EOF;
 132         }
 133
 134         if (res <= 0)
 135                 goto error;
 136
 137         bufferevent_add(&bufev->ev_read, bufev->timeout_read);
 138
 139         /* See if this callbacks meets the water marks */
 140         len = EVBUFFER_LENGTH(bufev->input);
 141         if (bufev->wm_read.low != 0 && len < bufev->wm_read.low)
 142                 return;
 143         if (bufev->wm_read.high != 0 && len >= bufev->wm_read.high) {
 144                 struct evbuffer *buf = bufev->input;
 145                 event_del(&bufev->ev_read);
 146
 147                 /* Now schedule a callback for us when the buffer changes */
 148                 evbuffer_setcb(buf, bufferevent_read_pressure_cb, bufev);
 149         }
 150
 151         /* Invoke the user callback - must always be called last */
 152         if (bufev->readcb != NULL)
 153                 (*bufev->readcb)(bufev, bufev->cbarg);
 154         return;
 155
 156  reschedule:
 157         bufferevent_add(&bufev->ev_read, bufev->timeout_read);
 158         return;
 159
 160  error:
 161         (*bufev->errorcb)(bufev, what, bufev->cbarg);
 162 }
 163
 164 static void
 165 bufferevent_writecb(int fd, short event, void *arg)
 166 {
 167         struct bufferevent *bufev = arg;
 168         int res = 0;
 169         short what = EVBUFFER_WRITE;
 170
 171         if (event == EV_TIMEOUT) {
 172                 what |= EVBUFFER_TIMEOUT;
 173                 goto error;
 174         }
 175
 176         if (EVBUFFER_LENGTH(bufev->output)) {
 177             res = evbuffer_write(bufev->output, fd);
 178             if (res == -1) {
 179 #ifndef WIN32
 180 /*todo. evbuffer uses WriteFile when WIN32 is set. WIN32 system calls do not
 181  *set errno. thus this error checking is not portable*/
 182                     if (errno == EAGAIN ||
 183                         errno == EINTR ||
 184                         errno == EINPROGRESS)
 185                             goto reschedule;
 186                     /* error case */
 187                     what |= EVBUFFER_ERROR;
 188
 189 #else
 190                                 goto reschedule;
 191 #endif
 192
 193             } else if (res == 0) {
 194                     /* eof case */
 195                     what |= EVBUFFER_EOF;
 196             }
 197             if (res <= 0)
 198                     goto error;
 199         }
 200
 201         if (EVBUFFER_LENGTH(bufev->output) != 0)
 202                 bufferevent_add(&bufev->ev_write, bufev->timeout_write);
 203
 204         /*
 205          * Invoke the user callback if our buffer is drained or below the
 206          * low watermark.
 207          */
 208         if (bufev->writecb != NULL &&
 209             EVBUFFER_LENGTH(bufev->output) <= bufev->wm_write.low)
 210                 (*bufev->writecb)(bufev, bufev->cbarg);
 211
 212         return;
 213
 214  reschedule:
 215         if (EVBUFFER_LENGTH(bufev->output) != 0)
 216                 bufferevent_add(&bufev->ev_write, bufev->timeout_write);
 217         return;
 218
 219  error:
 220         (*bufev->errorcb)(bufev, what, bufev->cbarg);
 221 }
 222
 223 /*
 224  * Create a new buffered event object.
 225  *
 226  * The read callback is invoked whenever we read new data.
 227  * The write callback is invoked whenever the output buffer is drained.
 228  * The error callback is invoked on a write/read error or on EOF.
 229  *
 230  * Both read and write callbacks maybe NULL.  The error callback is not
 231  * allowed to be NULL and have to be provided always.
 232  */
 233
 234 struct bufferevent *
 235 bufferevent_new(int fd, evbuffercb readcb, evbuffercb writecb,
 236     everrorcb errorcb, void *cbarg)
 237 {
 238         struct bufferevent *bufev;
 239
 240         if ((bufev = calloc(1, sizeof(struct bufferevent))) == NULL)
 241                 return (NULL);
 242
 243         if ((bufev->input = evbuffer_new()) == NULL) {
 244                 free(bufev);
 245                 return (NULL);
 246         }
 247
 248         if ((bufev->output = evbuffer_new()) == NULL) {
 249                 evbuffer_free(bufev->input);
 250                 free(bufev);
 251                 return (NULL);
 252         }
 253
 254         event_set(&bufev->ev_read, fd, EV_READ, bufferevent_readcb, bufev);
 255         event_set(&bufev->ev_write, fd, EV_WRITE, bufferevent_writecb, bufev);
 256
 257         bufferevent_setcb(bufev, readcb, writecb, errorcb, cbarg);
 258
 259         /*
 260          * Set to EV_WRITE so that using bufferevent_write is going to
 261          * trigger a callback.  Reading needs to be explicitly enabled
 262          * because otherwise no data will be available.
 263          */
 264         bufev->enabled = EV_WRITE;
 265
 266         return (bufev);
 267 }
 268
 269 void
 270 bufferevent_setcb(struct bufferevent *bufev,
 271     evbuffercb readcb, evbuffercb writecb, everrorcb errorcb, void *cbarg)
 272 {
 273         bufev->readcb = readcb;
 274         bufev->writecb = writecb;
 275         bufev->errorcb = errorcb;
 276
 277         bufev->cbarg = cbarg;
 278 }
 279
 280 void
 281 bufferevent_setfd(struct bufferevent *bufev, int fd)
 282 {
 283         event_del(&bufev->ev_read);
 284         event_del(&bufev->ev_write);
 285
 286         event_set(&bufev->ev_read, fd, EV_READ, bufferevent_readcb, bufev);
 287         event_set(&bufev->ev_write, fd, EV_WRITE, bufferevent_writecb, bufev);
 288         if (bufev->ev_base != NULL) {
 289                 event_base_set(bufev->ev_base, &bufev->ev_read);
 290                 event_base_set(bufev->ev_base, &bufev->ev_write);
 291         }
 292
 293         /* might have to manually trigger event registration */
 294 }
 295
 296 int
 297 bufferevent_priority_set(struct bufferevent *bufev, int priority)
 298 {
 299         if (event_priority_set(&bufev->ev_read, priority) == -1)
 300                 return (-1);
 301         if (event_priority_set(&bufev->ev_write, priority) == -1)
 302                 return (-1);
 303
 304         return (0);
 305 }
 306
 307 /* Closing the file descriptor is the responsibility of the caller */
 308
 309 void
 310 bufferevent_free(struct bufferevent *bufev)
 311 {
 312         event_del(&bufev->ev_read);
 313         event_del(&bufev->ev_write);
 314
 315         evbuffer_free(bufev->input);
 316         evbuffer_free(bufev->output);
 317
 318         free(bufev);
 319 }
 320
 321 /*
 322  * Returns 0 on success;
 323  *        -1 on failure.
 324  */
 325
 326 int
 327 bufferevent_write(struct bufferevent *bufev, const void *data, size_t size)
 328 {
 329         int res;
 330
 331         res = evbuffer_add(bufev->output, data, size);
 332
 333         if (res == -1)
 334                 return (res);
 335
 336         /* If everything is okay, we need to schedule a write */
 337         if (size > 0 && (bufev->enabled & EV_WRITE))
 338                 bufferevent_add(&bufev->ev_write, bufev->timeout_write);
 339
 340         return (res);
 341 }
 342
 343 int
 344 bufferevent_write_buffer(struct bufferevent *bufev, struct evbuffer *buf)
 345 {
 346         int res;
 347
 348         res = bufferevent_write(bufev, buf->buffer, buf->off);
 349         if (res != -1)
 350                 evbuffer_drain(buf, buf->off);
 351
 352         return (res);
 353 }
 354
 355 size_t
 356 bufferevent_read(struct bufferevent *bufev, void *data, size_t size)
 357 {
 358         struct evbuffer *buf = bufev->input;
 359
 360         if (buf->off < size)
 361                 size = buf->off;
 362
 363         /* Copy the available data to the user buffer */
 364         memcpy(data, buf->buffer, size);
 365
 366         if (size)
 367                 evbuffer_drain(buf, size);
 368
 369         return (size);
 370 }
 371
 372 int
 373 bufferevent_enable(struct bufferevent *bufev, short event)
 374 {
 375         if (event & EV_READ) {
 376                 if (bufferevent_add(&bufev->ev_read, bufev->timeout_read) == -1)
 377                         return (-1);
 378         }
 379         if (event & EV_WRITE) {
 380                 if (bufferevent_add(&bufev->ev_write, bufev->timeout_write) == -1)
 381                         return (-1);
 382         }
 383
 384         bufev->enabled |= event;
 385         return (0);
 386 }
 387
 388 int
 389 bufferevent_disable(struct bufferevent *bufev, short event)
 390 {
 391         if (event & EV_READ) {
 392                 if (event_del(&bufev->ev_read) == -1)
 393                         return (-1);
 394         }
 395         if (event & EV_WRITE) {
 396                 if (event_del(&bufev->ev_write) == -1)
 397                         return (-1);
 398         }
 399
 400         bufev->enabled &= ~event;
 401         return (0);
 402 }
 403
 404 /*
 405  * Sets the read and write timeout for a buffered event.
 406  */
 407
 408 void
 409 bufferevent_settimeout(struct bufferevent *bufev,
 410     int timeout_read, int timeout_write) {
 411         bufev->timeout_read = timeout_read;
 412         bufev->timeout_write = timeout_write;
 413
 414         if (event_pending(&bufev->ev_read, EV_READ, NULL))
 415                 bufferevent_add(&bufev->ev_read, timeout_read);
 416         if (event_pending(&bufev->ev_write, EV_WRITE, NULL))
 417                 bufferevent_add(&bufev->ev_write, timeout_write);
 418 }
 419
 420 /*
 421  * Sets the water marks
 422  */
 423
 424 void
 425 bufferevent_setwatermark(struct bufferevent *bufev, short events,
 426     size_t lowmark, size_t highmark)
 427 {
 428         if (events & EV_READ) {
 429                 bufev->wm_read.low = lowmark;
 430                 bufev->wm_read.high = highmark;
 431         }
 432
 433         if (events & EV_WRITE) {
 434                 bufev->wm_write.low = lowmark;
 435                 bufev->wm_write.high = highmark;
 436         }
 437
 438         /* If the watermarks changed then see if we should call read again */
 439         bufferevent_read_pressure_cb(bufev->input,
 440             0, EVBUFFER_LENGTH(bufev->input), bufev);
 441 }
 442
 443 int
 444 bufferevent_base_set(struct event_base *base, struct bufferevent *bufev)
 445 {
 446         int res;
 447
 448         bufev->ev_base = base;
 449
 450         res = event_base_set(base, &bufev->ev_read);
 451         if (res == -1)
 452                 return (res);
 453
 454         res = event_base_set(base, &bufev->ev_write);
 455         return (res);
 456 }