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