evbuffer.c 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455
  1. /*
  2. * Copyright (c) 2002-2004 Niels Provos <provos@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. #include <sys/types.h>
  28. #ifdef HAVE_CONFIG_H
  29. #include "config.h"
  30. #endif
  31. #ifdef HAVE_SYS_TIME_H
  32. #include <sys/time.h>
  33. #endif
  34. #include <errno.h>
  35. #include <stdio.h>
  36. #include <stdlib.h>
  37. #include <string.h>
  38. #ifdef HAVE_STDARG_H
  39. #include <stdarg.h>
  40. #endif
  41. #ifdef WIN32
  42. #include <winsock2.h>
  43. #endif
  44. #include "evutil.h"
  45. #include "event.h"
  46. /* prototypes */
  47. void bufferevent_read_pressure_cb(struct evbuffer *, size_t, size_t, void *);
  48. static int
  49. bufferevent_add(struct event *ev, int timeout)
  50. {
  51. struct timeval tv, *ptv = NULL;
  52. if (timeout) {
  53. evutil_timerclear(&tv);
  54. tv.tv_sec = timeout;
  55. ptv = &tv;
  56. }
  57. return (event_add(ev, ptv));
  58. }
  59. /*
  60. * This callback is executed when the size of the input buffer changes.
  61. * We use it to apply back pressure on the reading side.
  62. */
  63. void
  64. bufferevent_read_pressure_cb(struct evbuffer *buf, size_t old, size_t now,
  65. void *arg) {
  66. struct bufferevent *bufev = arg;
  67. /*
  68. * If we are below the watermark then reschedule reading if it's
  69. * still enabled.
  70. */
  71. if (bufev->wm_read.high == 0 || now < bufev->wm_read.high) {
  72. evbuffer_setcb(buf, NULL, NULL);
  73. if (bufev->enabled & EV_READ)
  74. bufferevent_add(&bufev->ev_read, bufev->timeout_read);
  75. }
  76. }
  77. static void
  78. bufferevent_readcb(int fd, short event, void *arg)
  79. {
  80. struct bufferevent *bufev = arg;
  81. int res = 0;
  82. short what = EVBUFFER_READ;
  83. size_t len;
  84. int howmuch = -1;
  85. if (event == EV_TIMEOUT) {
  86. what |= EVBUFFER_TIMEOUT;
  87. goto error;
  88. }
  89. /*
  90. * If we have a high watermark configured then we don't want to
  91. * read more data than would make us reach the watermark.
  92. */
  93. if (bufev->wm_read.high != 0) {
  94. howmuch = bufev->wm_read.high - EVBUFFER_LENGTH(bufev->input);
  95. /* we might have lowered the watermark, stop reading */
  96. if (howmuch <= 0) {
  97. struct evbuffer *buf = bufev->input;
  98. event_del(&bufev->ev_read);
  99. evbuffer_setcb(buf,
  100. bufferevent_read_pressure_cb, bufev);
  101. return;
  102. }
  103. }
  104. res = evbuffer_read(bufev->input, fd, howmuch);
  105. if (res == -1) {
  106. if (errno == EAGAIN || errno == EINTR)
  107. goto reschedule;
  108. /* error case */
  109. what |= EVBUFFER_ERROR;
  110. } else if (res == 0) {
  111. /* eof case */
  112. what |= EVBUFFER_EOF;
  113. }
  114. if (res <= 0)
  115. goto error;
  116. bufferevent_add(&bufev->ev_read, bufev->timeout_read);
  117. /* See if this callbacks meets the water marks */
  118. len = EVBUFFER_LENGTH(bufev->input);
  119. if (bufev->wm_read.low != 0 && len < bufev->wm_read.low)
  120. return;
  121. if (bufev->wm_read.high != 0 && len >= bufev->wm_read.high) {
  122. struct evbuffer *buf = bufev->input;
  123. event_del(&bufev->ev_read);
  124. /* Now schedule a callback for us when the buffer changes */
  125. evbuffer_setcb(buf, bufferevent_read_pressure_cb, bufev);
  126. }
  127. /* Invoke the user callback - must always be called last */
  128. if (bufev->readcb != NULL)
  129. (*bufev->readcb)(bufev, bufev->cbarg);
  130. return;
  131. reschedule:
  132. bufferevent_add(&bufev->ev_read, bufev->timeout_read);
  133. return;
  134. error:
  135. (*bufev->errorcb)(bufev, what, bufev->cbarg);
  136. }
  137. static void
  138. bufferevent_writecb(int fd, short event, void *arg)
  139. {
  140. struct bufferevent *bufev = arg;
  141. int res = 0;
  142. short what = EVBUFFER_WRITE;
  143. if (event == EV_TIMEOUT) {
  144. what |= EVBUFFER_TIMEOUT;
  145. goto error;
  146. }
  147. if (EVBUFFER_LENGTH(bufev->output)) {
  148. res = evbuffer_write(bufev->output, fd);
  149. if (res == -1) {
  150. #ifndef WIN32
  151. /*todo. evbuffer uses WriteFile when WIN32 is set. WIN32 system calls do not
  152. *set errno. thus this error checking is not portable*/
  153. if (errno == EAGAIN ||
  154. errno == EINTR ||
  155. errno == EINPROGRESS)
  156. goto reschedule;
  157. /* error case */
  158. what |= EVBUFFER_ERROR;
  159. #else
  160. goto reschedule;
  161. #endif
  162. } else if (res == 0) {
  163. /* eof case */
  164. what |= EVBUFFER_EOF;
  165. }
  166. if (res <= 0)
  167. goto error;
  168. }
  169. if (EVBUFFER_LENGTH(bufev->output) != 0)
  170. bufferevent_add(&bufev->ev_write, bufev->timeout_write);
  171. /*
  172. * Invoke the user callback if our buffer is drained or below the
  173. * low watermark.
  174. */
  175. if (bufev->writecb != NULL &&
  176. EVBUFFER_LENGTH(bufev->output) <= bufev->wm_write.low)
  177. (*bufev->writecb)(bufev, bufev->cbarg);
  178. return;
  179. reschedule:
  180. if (EVBUFFER_LENGTH(bufev->output) != 0)
  181. bufferevent_add(&bufev->ev_write, bufev->timeout_write);
  182. return;
  183. error:
  184. (*bufev->errorcb)(bufev, what, bufev->cbarg);
  185. }
  186. /*
  187. * Create a new buffered event object.
  188. *
  189. * The read callback is invoked whenever we read new data.
  190. * The write callback is invoked whenever the output buffer is drained.
  191. * The error callback is invoked on a write/read error or on EOF.
  192. *
  193. * Both read and write callbacks maybe NULL. The error callback is not
  194. * allowed to be NULL and have to be provided always.
  195. */
  196. struct bufferevent *
  197. bufferevent_new(int fd, evbuffercb readcb, evbuffercb writecb,
  198. everrorcb errorcb, void *cbarg)
  199. {
  200. struct bufferevent *bufev;
  201. if ((bufev = calloc(1, sizeof(struct bufferevent))) == NULL)
  202. return (NULL);
  203. if ((bufev->input = evbuffer_new()) == NULL) {
  204. free(bufev);
  205. return (NULL);
  206. }
  207. if ((bufev->output = evbuffer_new()) == NULL) {
  208. evbuffer_free(bufev->input);
  209. free(bufev);
  210. return (NULL);
  211. }
  212. event_set(&bufev->ev_read, fd, EV_READ, bufferevent_readcb, bufev);
  213. event_set(&bufev->ev_write, fd, EV_WRITE, bufferevent_writecb, bufev);
  214. bufferevent_setcb(bufev, readcb, writecb, errorcb, cbarg);
  215. /*
  216. * Set to EV_WRITE so that using bufferevent_write is going to
  217. * trigger a callback. Reading needs to be explicitly enabled
  218. * because otherwise no data will be available.
  219. */
  220. bufev->enabled = EV_WRITE;
  221. return (bufev);
  222. }
  223. void
  224. bufferevent_setcb(struct bufferevent *bufev,
  225. evbuffercb readcb, evbuffercb writecb, everrorcb errorcb, void *cbarg)
  226. {
  227. bufev->readcb = readcb;
  228. bufev->writecb = writecb;
  229. bufev->errorcb = errorcb;
  230. bufev->cbarg = cbarg;
  231. }
  232. void
  233. bufferevent_setfd(struct bufferevent *bufev, int fd)
  234. {
  235. event_del(&bufev->ev_read);
  236. event_del(&bufev->ev_write);
  237. event_set(&bufev->ev_read, fd, EV_READ, bufferevent_readcb, bufev);
  238. event_set(&bufev->ev_write, fd, EV_WRITE, bufferevent_writecb, bufev);
  239. if (bufev->ev_base != NULL) {
  240. event_base_set(bufev->ev_base, &bufev->ev_read);
  241. event_base_set(bufev->ev_base, &bufev->ev_write);
  242. }
  243. /* might have to manually trigger event registration */
  244. }
  245. int
  246. bufferevent_priority_set(struct bufferevent *bufev, int priority)
  247. {
  248. if (event_priority_set(&bufev->ev_read, priority) == -1)
  249. return (-1);
  250. if (event_priority_set(&bufev->ev_write, priority) == -1)
  251. return (-1);
  252. return (0);
  253. }
  254. /* Closing the file descriptor is the responsibility of the caller */
  255. void
  256. bufferevent_free(struct bufferevent *bufev)
  257. {
  258. event_del(&bufev->ev_read);
  259. event_del(&bufev->ev_write);
  260. evbuffer_free(bufev->input);
  261. evbuffer_free(bufev->output);
  262. free(bufev);
  263. }
  264. /*
  265. * Returns 0 on success;
  266. * -1 on failure.
  267. */
  268. int
  269. bufferevent_write(struct bufferevent *bufev, const void *data, size_t size)
  270. {
  271. int res;
  272. res = evbuffer_add(bufev->output, data, size);
  273. if (res == -1)
  274. return (res);
  275. /* If everything is okay, we need to schedule a write */
  276. if (size > 0 && (bufev->enabled & EV_WRITE))
  277. bufferevent_add(&bufev->ev_write, bufev->timeout_write);
  278. return (res);
  279. }
  280. int
  281. bufferevent_write_buffer(struct bufferevent *bufev, struct evbuffer *buf)
  282. {
  283. int res;
  284. res = bufferevent_write(bufev, buf->buffer, buf->off);
  285. if (res != -1)
  286. evbuffer_drain(buf, buf->off);
  287. return (res);
  288. }
  289. size_t
  290. bufferevent_read(struct bufferevent *bufev, void *data, size_t size)
  291. {
  292. struct evbuffer *buf = bufev->input;
  293. if (buf->off < size)
  294. size = buf->off;
  295. /* Copy the available data to the user buffer */
  296. memcpy(data, buf->buffer, size);
  297. if (size)
  298. evbuffer_drain(buf, size);
  299. return (size);
  300. }
  301. int
  302. bufferevent_enable(struct bufferevent *bufev, short event)
  303. {
  304. if (event & EV_READ) {
  305. if (bufferevent_add(&bufev->ev_read, bufev->timeout_read) == -1)
  306. return (-1);
  307. }
  308. if (event & EV_WRITE) {
  309. if (bufferevent_add(&bufev->ev_write, bufev->timeout_write) == -1)
  310. return (-1);
  311. }
  312. bufev->enabled |= event;
  313. return (0);
  314. }
  315. int
  316. bufferevent_disable(struct bufferevent *bufev, short event)
  317. {
  318. if (event & EV_READ) {
  319. if (event_del(&bufev->ev_read) == -1)
  320. return (-1);
  321. }
  322. if (event & EV_WRITE) {
  323. if (event_del(&bufev->ev_write) == -1)
  324. return (-1);
  325. }
  326. bufev->enabled &= ~event;
  327. return (0);
  328. }
  329. /*
  330. * Sets the read and write timeout for a buffered event.
  331. */
  332. void
  333. bufferevent_settimeout(struct bufferevent *bufev,
  334. int timeout_read, int timeout_write) {
  335. bufev->timeout_read = timeout_read;
  336. bufev->timeout_write = timeout_write;
  337. if (event_pending(&bufev->ev_read, EV_READ, NULL))
  338. bufferevent_add(&bufev->ev_read, timeout_read);
  339. if (event_pending(&bufev->ev_write, EV_WRITE, NULL))
  340. bufferevent_add(&bufev->ev_write, timeout_write);
  341. }
  342. /*
  343. * Sets the water marks
  344. */
  345. void
  346. bufferevent_setwatermark(struct bufferevent *bufev, short events,
  347. size_t lowmark, size_t highmark)
  348. {
  349. if (events & EV_READ) {
  350. bufev->wm_read.low = lowmark;
  351. bufev->wm_read.high = highmark;
  352. }
  353. if (events & EV_WRITE) {
  354. bufev->wm_write.low = lowmark;
  355. bufev->wm_write.high = highmark;
  356. }
  357. /* If the watermarks changed then see if we should call read again */
  358. bufferevent_read_pressure_cb(bufev->input,
  359. 0, EVBUFFER_LENGTH(bufev->input), bufev);
  360. }
  361. int
  362. bufferevent_base_set(struct event_base *base, struct bufferevent *bufev)
  363. {
  364. int res;
  365. bufev->ev_base = base;
  366. res = event_base_set(base, &bufev->ev_read);
  367. if (res == -1)
  368. return (res);
  369. res = event_base_set(base, &bufev->ev_write);
  370. return (res);
  371. }