122347Spst/*
222347Spst * Copyright (c) 2002-2007 Niels Provos <provos@citi.umich.edu>
322347Spst * Copyright (c) 2007-2012 Niels Provos, Nick Mathewson
429964Sache *
592906Smarkm * Redistribution and use in source and binary forms, with or without
622347Spst * modification, are permitted provided that the following conditions
722347Spst * are met:
822347Spst * 1. Redistributions of source code must retain the above copyright
922347Spst *    notice, this list of conditions and the following disclaimer.
1022347Spst * 2. Redistributions in binary form must reproduce the above copyright
1122347Spst *    notice, this list of conditions and the following disclaimer in the
1222347Spst *    documentation and/or other materials provided with the distribution.
1322347Spst * 3. The name of the author may not be used to endorse or promote products
1422347Spst *    derived from this software without specific prior written permission.
1522347Spst *
1622347Spst * THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``AS IS'' AND ANY EXPRESS OR
1722347Spst * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
1829964Sache * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
1922347Spst * IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY DIRECT, INDIRECT,
2022347Spst * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT
2122347Spst * NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
2222347Spst * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
2322347Spst * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
2422347Spst * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
2522347Spst * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
2622347Spst */
2722347Spst
2822347Spst#include "event2/event-config.h"
2922347Spst#include "evconfig-private.h"
3022347Spst
3122347Spst#include <sys/types.h>
3222347Spst
3322347Spst#ifdef EVENT__HAVE_SYS_TIME_H
3422347Spst#include <sys/time.h>
3522347Spst#endif
3622347Spst
3722347Spst#include <errno.h>
3822347Spst#include <stdio.h>
3922347Spst#include <stdlib.h>
4022347Spst#include <string.h>
4122347Spst#ifdef EVENT__HAVE_STDARG_H
4222347Spst#include <stdarg.h>
4322347Spst#endif
4422347Spst
4522347Spst#ifdef _WIN32
4629964Sache#include <winsock2.h>
4729964Sache#endif
4829964Sache#include <errno.h>
4929964Sache
5022347Spst#include "event2/util.h"
5122347Spst#include "event2/buffer.h"
5222347Spst#include "event2/buffer_compat.h"
5322347Spst#include "event2/bufferevent.h"
5422347Spst#include "event2/bufferevent_struct.h"
5522347Spst#include "event2/bufferevent_compat.h"
5622347Spst#include "event2/event.h"
5722347Spst#include "event-internal.h"
5822347Spst#include "log-internal.h"
5922347Spst#include "mm-internal.h"
6022347Spst#include "bufferevent-internal.h"
6122347Spst#include "evbuffer-internal.h"
6222347Spst#include "util-internal.h"
6322347Spst
6422347Spststatic void bufferevent_cancel_all_(struct bufferevent *bev);
6522347Spststatic void bufferevent_finalize_cb_(struct event_callback *evcb, void *arg_);
6622347Spst
6722347Spstvoid
6822347Spstbufferevent_suspend_read_(struct bufferevent *bufev, bufferevent_suspend_flags what)
6922347Spst{
7022347Spst	struct bufferevent_private *bufev_private =
7122347Spst	    EVUTIL_UPCAST(bufev, struct bufferevent_private, bev);
7222347Spst	BEV_LOCK(bufev);
7322347Spst	if (!bufev_private->read_suspended)
7422347Spst		bufev->be_ops->disable(bufev, EV_READ);
7522347Spst	bufev_private->read_suspended |= what;
7622347Spst	BEV_UNLOCK(bufev);
7722347Spst}
7822347Spst
7922347Spstvoid
8022347Spstbufferevent_unsuspend_read_(struct bufferevent *bufev, bufferevent_suspend_flags what)
8122347Spst{
8222347Spst	struct bufferevent_private *bufev_private =
8322347Spst	    EVUTIL_UPCAST(bufev, struct bufferevent_private, bev);
8422347Spst	BEV_LOCK(bufev);
8522347Spst	bufev_private->read_suspended &= ~what;
8622347Spst	if (!bufev_private->read_suspended && (bufev->enabled & EV_READ))
8722347Spst		bufev->be_ops->enable(bufev, EV_READ);
8822347Spst	BEV_UNLOCK(bufev);
8922347Spst}
9022347Spst
9122347Spstvoid
9222347Spstbufferevent_suspend_write_(struct bufferevent *bufev, bufferevent_suspend_flags what)
9322347Spst{
9422347Spst	struct bufferevent_private *bufev_private =
9522347Spst	    EVUTIL_UPCAST(bufev, struct bufferevent_private, bev);
9622347Spst	BEV_LOCK(bufev);
9722347Spst	if (!bufev_private->write_suspended)
9822347Spst		bufev->be_ops->disable(bufev, EV_WRITE);
9922347Spst	bufev_private->write_suspended |= what;
10022347Spst	BEV_UNLOCK(bufev);
10122347Spst}
10222347Spst
10322347Spstvoid
10422347Spstbufferevent_unsuspend_write_(struct bufferevent *bufev, bufferevent_suspend_flags what)
10522347Spst{
10622347Spst	struct bufferevent_private *bufev_private =
10722347Spst	    EVUTIL_UPCAST(bufev, struct bufferevent_private, bev);
10822347Spst	BEV_LOCK(bufev);
10922347Spst	bufev_private->write_suspended &= ~what;
11022347Spst	if (!bufev_private->write_suspended && (bufev->enabled & EV_WRITE))
11122347Spst		bufev->be_ops->enable(bufev, EV_WRITE);
11222347Spst	BEV_UNLOCK(bufev);
11322347Spst}
11422347Spst
11522347Spst
11622347Spst/* Callback to implement watermarks on the input buffer.  Only enabled
11722347Spst * if the watermark is set. */
11822347Spststatic void
11922347Spstbufferevent_inbuf_wm_cb(struct evbuffer *buf,
12022347Spst    const struct evbuffer_cb_info *cbinfo,
12122347Spst    void *arg)
12222347Spst{
12322347Spst	struct bufferevent *bufev = arg;
12422347Spst	size_t size;
12522347Spst
12622347Spst	size = evbuffer_get_length(buf);
12722347Spst
12822347Spst	if (size >= bufev->wm_read.high)
12922347Spst		bufferevent_wm_suspend_read(bufev);
13022347Spst	else
13122347Spst		bufferevent_wm_unsuspend_read(bufev);
13222347Spst}
13322347Spst
13422347Spststatic void
13522347Spstbufferevent_run_deferred_callbacks_locked(struct event_callback *cb, void *arg)
13622347Spst{
13722347Spst	struct bufferevent_private *bufev_private = arg;
13822347Spst	struct bufferevent *bufev = &bufev_private->bev;
13922347Spst
14022347Spst	BEV_LOCK(bufev);
14122347Spst	if ((bufev_private->eventcb_pending & BEV_EVENT_CONNECTED) &&
14222347Spst	    bufev->errorcb) {
14322347Spst		/* The "connected" happened before any reads or writes, so
14422347Spst		   send it first. */
14522347Spst		bufev_private->eventcb_pending &= ~BEV_EVENT_CONNECTED;
14622347Spst		bufev->errorcb(bufev, BEV_EVENT_CONNECTED, bufev->cbarg);
14722347Spst	}
14822347Spst	if (bufev_private->readcb_pending && bufev->readcb) {
14922347Spst		bufev_private->readcb_pending = 0;
15022347Spst		bufev->readcb(bufev, bufev->cbarg);
15122347Spst	}
15222347Spst	if (bufev_private->writecb_pending && bufev->writecb) {
15322347Spst		bufev_private->writecb_pending = 0;
15422347Spst		bufev->writecb(bufev, bufev->cbarg);
15522347Spst	}
15622347Spst	if (bufev_private->eventcb_pending && bufev->errorcb) {
15722347Spst		short what = bufev_private->eventcb_pending;
15822347Spst		int err = bufev_private->errno_pending;
15922347Spst		bufev_private->eventcb_pending = 0;
16022347Spst		bufev_private->errno_pending = 0;
16122347Spst		EVUTIL_SET_SOCKET_ERROR(err);
16222347Spst		bufev->errorcb(bufev, what, bufev->cbarg);
16322347Spst	}
16422347Spst	bufferevent_decref_and_unlock_(bufev);
16522347Spst}
16622347Spst
16722347Spststatic void
16822347Spstbufferevent_run_deferred_callbacks_unlocked(struct event_callback *cb, void *arg)
16922347Spst{
17022347Spst	struct bufferevent_private *bufev_private = arg;
171	struct bufferevent *bufev = &bufev_private->bev;
172
173	BEV_LOCK(bufev);
174#define UNLOCKED(stmt) \
175	do { BEV_UNLOCK(bufev); stmt; BEV_LOCK(bufev); } while(0)
176
177	if ((bufev_private->eventcb_pending & BEV_EVENT_CONNECTED) &&
178	    bufev->errorcb) {
179		/* The "connected" happened before any reads or writes, so
180		   send it first. */
181		bufferevent_event_cb errorcb = bufev->errorcb;
182		void *cbarg = bufev->cbarg;
183		bufev_private->eventcb_pending &= ~BEV_EVENT_CONNECTED;
184		UNLOCKED(errorcb(bufev, BEV_EVENT_CONNECTED, cbarg));
185	}
186	if (bufev_private->readcb_pending && bufev->readcb) {
187		bufferevent_data_cb readcb = bufev->readcb;
188		void *cbarg = bufev->cbarg;
189		bufev_private->readcb_pending = 0;
190		UNLOCKED(readcb(bufev, cbarg));
191	}
192	if (bufev_private->writecb_pending && bufev->writecb) {
193		bufferevent_data_cb writecb = bufev->writecb;
194		void *cbarg = bufev->cbarg;
195		bufev_private->writecb_pending = 0;
196		UNLOCKED(writecb(bufev, cbarg));
197	}
198	if (bufev_private->eventcb_pending && bufev->errorcb) {
199		bufferevent_event_cb errorcb = bufev->errorcb;
200		void *cbarg = bufev->cbarg;
201		short what = bufev_private->eventcb_pending;
202		int err = bufev_private->errno_pending;
203		bufev_private->eventcb_pending = 0;
204		bufev_private->errno_pending = 0;
205		EVUTIL_SET_SOCKET_ERROR(err);
206		UNLOCKED(errorcb(bufev,what,cbarg));
207	}
208	bufferevent_decref_and_unlock_(bufev);
209#undef UNLOCKED
210}
211
212#define SCHEDULE_DEFERRED(bevp)						\
213	do {								\
214		if (event_deferred_cb_schedule_(			\
215			    (bevp)->bev.ev_base,			\
216			&(bevp)->deferred))				\
217			bufferevent_incref_(&(bevp)->bev);		\
218	} while (0)
219
220
221void
222bufferevent_run_readcb_(struct bufferevent *bufev, int options)
223{
224	/* Requires that we hold the lock and a reference */
225	struct bufferevent_private *p =
226	    EVUTIL_UPCAST(bufev, struct bufferevent_private, bev);
227	if (bufev->readcb == NULL)
228		return;
229	if ((p->options|options) & BEV_OPT_DEFER_CALLBACKS) {
230		p->readcb_pending = 1;
231		SCHEDULE_DEFERRED(p);
232	} else {
233		bufev->readcb(bufev, bufev->cbarg);
234	}
235}
236
237void
238bufferevent_run_writecb_(struct bufferevent *bufev, int options)
239{
240	/* Requires that we hold the lock and a reference */
241	struct bufferevent_private *p =
242	    EVUTIL_UPCAST(bufev, struct bufferevent_private, bev);
243	if (bufev->writecb == NULL)
244		return;
245	if ((p->options|options) & BEV_OPT_DEFER_CALLBACKS) {
246		p->writecb_pending = 1;
247		SCHEDULE_DEFERRED(p);
248	} else {
249		bufev->writecb(bufev, bufev->cbarg);
250	}
251}
252
253#define BEV_TRIG_ALL_OPTS (			\
254		BEV_TRIG_IGNORE_WATERMARKS|	\
255		BEV_TRIG_DEFER_CALLBACKS	\
256	)
257
258void
259bufferevent_trigger(struct bufferevent *bufev, short iotype, int options)
260{
261	bufferevent_incref_and_lock_(bufev);
262	bufferevent_trigger_nolock_(bufev, iotype, options&BEV_TRIG_ALL_OPTS);
263	bufferevent_decref_and_unlock_(bufev);
264}
265
266void
267bufferevent_run_eventcb_(struct bufferevent *bufev, short what, int options)
268{
269	/* Requires that we hold the lock and a reference */
270	struct bufferevent_private *p =
271	    EVUTIL_UPCAST(bufev, struct bufferevent_private, bev);
272	if (bufev->errorcb == NULL)
273		return;
274	if ((p->options|options) & BEV_OPT_DEFER_CALLBACKS) {
275		p->eventcb_pending |= what;
276		p->errno_pending = EVUTIL_SOCKET_ERROR();
277		SCHEDULE_DEFERRED(p);
278	} else {
279		bufev->errorcb(bufev, what, bufev->cbarg);
280	}
281}
282
283void
284bufferevent_trigger_event(struct bufferevent *bufev, short what, int options)
285{
286	bufferevent_incref_and_lock_(bufev);
287	bufferevent_run_eventcb_(bufev, what, options&BEV_TRIG_ALL_OPTS);
288	bufferevent_decref_and_unlock_(bufev);
289}
290
291int
292bufferevent_init_common_(struct bufferevent_private *bufev_private,
293    struct event_base *base,
294    const struct bufferevent_ops *ops,
295    enum bufferevent_options options)
296{
297	struct bufferevent *bufev = &bufev_private->bev;
298
299	if (!bufev->input) {
300		if ((bufev->input = evbuffer_new()) == NULL)
301			return -1;
302	}
303
304	if (!bufev->output) {
305		if ((bufev->output = evbuffer_new()) == NULL) {
306			evbuffer_free(bufev->input);
307			return -1;
308		}
309	}
310
311	bufev_private->refcnt = 1;
312	bufev->ev_base = base;
313
314	/* Disable timeouts. */
315	evutil_timerclear(&bufev->timeout_read);
316	evutil_timerclear(&bufev->timeout_write);
317
318	bufev->be_ops = ops;
319
320	bufferevent_ratelim_init_(bufev_private);
321
322	/*
323	 * Set to EV_WRITE so that using bufferevent_write is going to
324	 * trigger a callback.  Reading needs to be explicitly enabled
325	 * because otherwise no data will be available.
326	 */
327	bufev->enabled = EV_WRITE;
328
329#ifndef EVENT__DISABLE_THREAD_SUPPORT
330	if (options & BEV_OPT_THREADSAFE) {
331		if (bufferevent_enable_locking_(bufev, NULL) < 0) {
332			/* cleanup */
333			evbuffer_free(bufev->input);
334			evbuffer_free(bufev->output);
335			bufev->input = NULL;
336			bufev->output = NULL;
337			return -1;
338		}
339	}
340#endif
341	if ((options & (BEV_OPT_DEFER_CALLBACKS|BEV_OPT_UNLOCK_CALLBACKS))
342	    == BEV_OPT_UNLOCK_CALLBACKS) {
343		event_warnx("UNLOCK_CALLBACKS requires DEFER_CALLBACKS");
344		return -1;
345	}
346	if (options & BEV_OPT_UNLOCK_CALLBACKS)
347		event_deferred_cb_init_(
348		    &bufev_private->deferred,
349		    event_base_get_npriorities(base) / 2,
350		    bufferevent_run_deferred_callbacks_unlocked,
351		    bufev_private);
352	else
353		event_deferred_cb_init_(
354		    &bufev_private->deferred,
355		    event_base_get_npriorities(base) / 2,
356		    bufferevent_run_deferred_callbacks_locked,
357		    bufev_private);
358
359	bufev_private->options = options;
360
361	evbuffer_set_parent_(bufev->input, bufev);
362	evbuffer_set_parent_(bufev->output, bufev);
363
364	return 0;
365}
366
367void
368bufferevent_setcb(struct bufferevent *bufev,
369    bufferevent_data_cb readcb, bufferevent_data_cb writecb,
370    bufferevent_event_cb eventcb, void *cbarg)
371{
372	BEV_LOCK(bufev);
373
374	bufev->readcb = readcb;
375	bufev->writecb = writecb;
376	bufev->errorcb = eventcb;
377
378	bufev->cbarg = cbarg;
379	BEV_UNLOCK(bufev);
380}
381
382void
383bufferevent_getcb(struct bufferevent *bufev,
384    bufferevent_data_cb *readcb_ptr,
385    bufferevent_data_cb *writecb_ptr,
386    bufferevent_event_cb *eventcb_ptr,
387    void **cbarg_ptr)
388{
389	BEV_LOCK(bufev);
390	if (readcb_ptr)
391		*readcb_ptr = bufev->readcb;
392	if (writecb_ptr)
393		*writecb_ptr = bufev->writecb;
394	if (eventcb_ptr)
395		*eventcb_ptr = bufev->errorcb;
396	if (cbarg_ptr)
397		*cbarg_ptr = bufev->cbarg;
398
399	BEV_UNLOCK(bufev);
400}
401
402struct evbuffer *
403bufferevent_get_input(struct bufferevent *bufev)
404{
405	return bufev->input;
406}
407
408struct evbuffer *
409bufferevent_get_output(struct bufferevent *bufev)
410{
411	return bufev->output;
412}
413
414struct event_base *
415bufferevent_get_base(struct bufferevent *bufev)
416{
417	return bufev->ev_base;
418}
419
420int
421bufferevent_get_priority(const struct bufferevent *bufev)
422{
423	if (event_initialized(&bufev->ev_read)) {
424		return event_get_priority(&bufev->ev_read);
425	} else {
426		return event_base_get_npriorities(bufev->ev_base) / 2;
427	}
428}
429
430int
431bufferevent_write(struct bufferevent *bufev, const void *data, size_t size)
432{
433	if (evbuffer_add(bufev->output, data, size) == -1)
434		return (-1);
435
436	return 0;
437}
438
439int
440bufferevent_write_buffer(struct bufferevent *bufev, struct evbuffer *buf)
441{
442	if (evbuffer_add_buffer(bufev->output, buf) == -1)
443		return (-1);
444
445	return 0;
446}
447
448size_t
449bufferevent_read(struct bufferevent *bufev, void *data, size_t size)
450{
451	return (evbuffer_remove(bufev->input, data, size));
452}
453
454int
455bufferevent_read_buffer(struct bufferevent *bufev, struct evbuffer *buf)
456{
457	return (evbuffer_add_buffer(buf, bufev->input));
458}
459
460int
461bufferevent_enable(struct bufferevent *bufev, short event)
462{
463	struct bufferevent_private *bufev_private =
464	    EVUTIL_UPCAST(bufev, struct bufferevent_private, bev);
465	short impl_events = event;
466	int r = 0;
467
468	bufferevent_incref_and_lock_(bufev);
469	if (bufev_private->read_suspended)
470		impl_events &= ~EV_READ;
471	if (bufev_private->write_suspended)
472		impl_events &= ~EV_WRITE;
473
474	bufev->enabled |= event;
475
476	if (impl_events && bufev->be_ops->enable(bufev, impl_events) < 0)
477		r = -1;
478
479	bufferevent_decref_and_unlock_(bufev);
480	return r;
481}
482
483int
484bufferevent_set_timeouts(struct bufferevent *bufev,
485			 const struct timeval *tv_read,
486			 const struct timeval *tv_write)
487{
488	int r = 0;
489	BEV_LOCK(bufev);
490	if (tv_read) {
491		bufev->timeout_read = *tv_read;
492	} else {
493		evutil_timerclear(&bufev->timeout_read);
494	}
495	if (tv_write) {
496		bufev->timeout_write = *tv_write;
497	} else {
498		evutil_timerclear(&bufev->timeout_write);
499	}
500
501	if (bufev->be_ops->adj_timeouts)
502		r = bufev->be_ops->adj_timeouts(bufev);
503	BEV_UNLOCK(bufev);
504
505	return r;
506}
507
508
509/* Obsolete; use bufferevent_set_timeouts */
510void
511bufferevent_settimeout(struct bufferevent *bufev,
512		       int timeout_read, int timeout_write)
513{
514	struct timeval tv_read, tv_write;
515	struct timeval *ptv_read = NULL, *ptv_write = NULL;
516
517	memset(&tv_read, 0, sizeof(tv_read));
518	memset(&tv_write, 0, sizeof(tv_write));
519
520	if (timeout_read) {
521		tv_read.tv_sec = timeout_read;
522		ptv_read = &tv_read;
523	}
524	if (timeout_write) {
525		tv_write.tv_sec = timeout_write;
526		ptv_write = &tv_write;
527	}
528
529	bufferevent_set_timeouts(bufev, ptv_read, ptv_write);
530}
531
532
533int
534bufferevent_disable_hard_(struct bufferevent *bufev, short event)
535{
536	int r = 0;
537	struct bufferevent_private *bufev_private =
538	    EVUTIL_UPCAST(bufev, struct bufferevent_private, bev);
539
540	BEV_LOCK(bufev);
541	bufev->enabled &= ~event;
542
543	bufev_private->connecting = 0;
544	if (bufev->be_ops->disable(bufev, event) < 0)
545		r = -1;
546
547	BEV_UNLOCK(bufev);
548	return r;
549}
550
551int
552bufferevent_disable(struct bufferevent *bufev, short event)
553{
554	int r = 0;
555
556	BEV_LOCK(bufev);
557	bufev->enabled &= ~event;
558
559	if (bufev->be_ops->disable(bufev, event) < 0)
560		r = -1;
561
562	BEV_UNLOCK(bufev);
563	return r;
564}
565
566/*
567 * Sets the water marks
568 */
569
570void
571bufferevent_setwatermark(struct bufferevent *bufev, short events,
572    size_t lowmark, size_t highmark)
573{
574	struct bufferevent_private *bufev_private =
575	    EVUTIL_UPCAST(bufev, struct bufferevent_private, bev);
576
577	BEV_LOCK(bufev);
578	if (events & EV_WRITE) {
579		bufev->wm_write.low = lowmark;
580		bufev->wm_write.high = highmark;
581	}
582
583	if (events & EV_READ) {
584		bufev->wm_read.low = lowmark;
585		bufev->wm_read.high = highmark;
586
587		if (highmark) {
588			/* There is now a new high-water mark for read.
589			   enable the callback if needed, and see if we should
590			   suspend/bufferevent_wm_unsuspend. */
591
592			if (bufev_private->read_watermarks_cb == NULL) {
593				bufev_private->read_watermarks_cb =
594				    evbuffer_add_cb(bufev->input,
595						    bufferevent_inbuf_wm_cb,
596						    bufev);
597			}
598			evbuffer_cb_set_flags(bufev->input,
599				      bufev_private->read_watermarks_cb,
600				      EVBUFFER_CB_ENABLED|EVBUFFER_CB_NODEFER);
601
602			if (evbuffer_get_length(bufev->input) >= highmark)
603				bufferevent_wm_suspend_read(bufev);
604			else if (evbuffer_get_length(bufev->input) < highmark)
605				bufferevent_wm_unsuspend_read(bufev);
606		} else {
607			/* There is now no high-water mark for read. */
608			if (bufev_private->read_watermarks_cb)
609				evbuffer_cb_clear_flags(bufev->input,
610				    bufev_private->read_watermarks_cb,
611				    EVBUFFER_CB_ENABLED);
612			bufferevent_wm_unsuspend_read(bufev);
613		}
614	}
615	BEV_UNLOCK(bufev);
616}
617
618int
619bufferevent_getwatermark(struct bufferevent *bufev, short events,
620    size_t *lowmark, size_t *highmark)
621{
622	if (events == EV_WRITE) {
623		BEV_LOCK(bufev);
624		if (lowmark)
625			*lowmark = bufev->wm_write.low;
626		if (highmark)
627			*highmark = bufev->wm_write.high;
628		BEV_UNLOCK(bufev);
629		return 0;
630	}
631
632	if (events == EV_READ) {
633		BEV_LOCK(bufev);
634		if (lowmark)
635			*lowmark = bufev->wm_read.low;
636		if (highmark)
637			*highmark = bufev->wm_read.high;
638		BEV_UNLOCK(bufev);
639		return 0;
640	}
641	return -1;
642}
643
644int
645bufferevent_flush(struct bufferevent *bufev,
646    short iotype,
647    enum bufferevent_flush_mode mode)
648{
649	int r = -1;
650	BEV_LOCK(bufev);
651	if (bufev->be_ops->flush)
652		r = bufev->be_ops->flush(bufev, iotype, mode);
653	BEV_UNLOCK(bufev);
654	return r;
655}
656
657void
658bufferevent_incref_and_lock_(struct bufferevent *bufev)
659{
660	struct bufferevent_private *bufev_private =
661	    BEV_UPCAST(bufev);
662	BEV_LOCK(bufev);
663	++bufev_private->refcnt;
664}
665
666#if 0
667static void
668bufferevent_transfer_lock_ownership_(struct bufferevent *donor,
669    struct bufferevent *recipient)
670{
671	struct bufferevent_private *d = BEV_UPCAST(donor);
672	struct bufferevent_private *r = BEV_UPCAST(recipient);
673	if (d->lock != r->lock)
674		return;
675	if (r->own_lock)
676		return;
677	if (d->own_lock) {
678		d->own_lock = 0;
679		r->own_lock = 1;
680	}
681}
682#endif
683
684int
685bufferevent_decref_and_unlock_(struct bufferevent *bufev)
686{
687	struct bufferevent_private *bufev_private =
688	    EVUTIL_UPCAST(bufev, struct bufferevent_private, bev);
689	int n_cbs = 0;
690#define MAX_CBS 16
691	struct event_callback *cbs[MAX_CBS];
692
693	EVUTIL_ASSERT(bufev_private->refcnt > 0);
694
695	if (--bufev_private->refcnt) {
696		BEV_UNLOCK(bufev);
697		return 0;
698	}
699
700	if (bufev->be_ops->unlink)
701		bufev->be_ops->unlink(bufev);
702
703	/* Okay, we're out of references. Let's finalize this once all the
704	 * callbacks are done running. */
705	cbs[0] = &bufev->ev_read.ev_evcallback;
706	cbs[1] = &bufev->ev_write.ev_evcallback;
707	cbs[2] = &bufev_private->deferred;
708	n_cbs = 3;
709	if (bufev_private->rate_limiting) {
710		struct event *e = &bufev_private->rate_limiting->refill_bucket_event;
711		if (event_initialized(e))
712			cbs[n_cbs++] = &e->ev_evcallback;
713	}
714	n_cbs += evbuffer_get_callbacks_(bufev->input, cbs+n_cbs, MAX_CBS-n_cbs);
715	n_cbs += evbuffer_get_callbacks_(bufev->output, cbs+n_cbs, MAX_CBS-n_cbs);
716
717	event_callback_finalize_many_(bufev->ev_base, n_cbs, cbs,
718	    bufferevent_finalize_cb_);
719
720#undef MAX_CBS
721	BEV_UNLOCK(bufev);
722
723	return 1;
724}
725
726static void
727bufferevent_finalize_cb_(struct event_callback *evcb, void *arg_)
728{
729	struct bufferevent *bufev = arg_;
730	struct bufferevent *underlying;
731	struct bufferevent_private *bufev_private =
732	    EVUTIL_UPCAST(bufev, struct bufferevent_private, bev);
733
734	BEV_LOCK(bufev);
735	underlying = bufferevent_get_underlying(bufev);
736
737	/* Clean up the shared info */
738	if (bufev->be_ops->destruct)
739		bufev->be_ops->destruct(bufev);
740
741	/* XXX what happens if refcnt for these buffers is > 1?
742	 * The buffers can share a lock with this bufferevent object,
743	 * but the lock might be destroyed below. */
744	/* evbuffer will free the callbacks */
745	evbuffer_free(bufev->input);
746	evbuffer_free(bufev->output);
747
748	if (bufev_private->rate_limiting) {
749		if (bufev_private->rate_limiting->group)
750			bufferevent_remove_from_rate_limit_group_internal_(bufev,0);
751		mm_free(bufev_private->rate_limiting);
752		bufev_private->rate_limiting = NULL;
753	}
754
755
756	BEV_UNLOCK(bufev);
757
758	if (bufev_private->own_lock)
759		EVTHREAD_FREE_LOCK(bufev_private->lock,
760		    EVTHREAD_LOCKTYPE_RECURSIVE);
761
762	/* Free the actual allocated memory. */
763	mm_free(((char*)bufev) - bufev->be_ops->mem_offset);
764
765	/* Release the reference to underlying now that we no longer need the
766	 * reference to it.  We wait this long mainly in case our lock is
767	 * shared with underlying.
768	 *
769	 * The 'destruct' function will also drop a reference to underlying
770	 * if BEV_OPT_CLOSE_ON_FREE is set.
771	 *
772	 * XXX Should we/can we just refcount evbuffer/bufferevent locks?
773	 * It would probably save us some headaches.
774	 */
775	if (underlying)
776		bufferevent_decref_(underlying);
777}
778
779int
780bufferevent_decref_(struct bufferevent *bufev)
781{
782	BEV_LOCK(bufev);
783	return bufferevent_decref_and_unlock_(bufev);
784}
785
786void
787bufferevent_free(struct bufferevent *bufev)
788{
789	BEV_LOCK(bufev);
790	bufferevent_setcb(bufev, NULL, NULL, NULL, NULL);
791	bufferevent_cancel_all_(bufev);
792	bufferevent_decref_and_unlock_(bufev);
793}
794
795void
796bufferevent_incref_(struct bufferevent *bufev)
797{
798	struct bufferevent_private *bufev_private =
799	    EVUTIL_UPCAST(bufev, struct bufferevent_private, bev);
800
801	BEV_LOCK(bufev);
802	++bufev_private->refcnt;
803	BEV_UNLOCK(bufev);
804}
805
806int
807bufferevent_enable_locking_(struct bufferevent *bufev, void *lock)
808{
809#ifdef EVENT__DISABLE_THREAD_SUPPORT
810	return -1;
811#else
812	struct bufferevent *underlying;
813
814	if (BEV_UPCAST(bufev)->lock)
815		return -1;
816	underlying = bufferevent_get_underlying(bufev);
817
818	if (!lock && underlying && BEV_UPCAST(underlying)->lock) {
819		lock = BEV_UPCAST(underlying)->lock;
820		BEV_UPCAST(bufev)->lock = lock;
821		BEV_UPCAST(bufev)->own_lock = 0;
822	} else if (!lock) {
823		EVTHREAD_ALLOC_LOCK(lock, EVTHREAD_LOCKTYPE_RECURSIVE);
824		if (!lock)
825			return -1;
826		BEV_UPCAST(bufev)->lock = lock;
827		BEV_UPCAST(bufev)->own_lock = 1;
828	} else {
829		BEV_UPCAST(bufev)->lock = lock;
830		BEV_UPCAST(bufev)->own_lock = 0;
831	}
832	evbuffer_enable_locking(bufev->input, lock);
833	evbuffer_enable_locking(bufev->output, lock);
834
835	if (underlying && !BEV_UPCAST(underlying)->lock)
836		bufferevent_enable_locking_(underlying, lock);
837
838	return 0;
839#endif
840}
841
842int
843bufferevent_setfd(struct bufferevent *bev, evutil_socket_t fd)
844{
845	union bufferevent_ctrl_data d;
846	int res = -1;
847	d.fd = fd;
848	BEV_LOCK(bev);
849	if (bev->be_ops->ctrl)
850		res = bev->be_ops->ctrl(bev, BEV_CTRL_SET_FD, &d);
851	BEV_UNLOCK(bev);
852	return res;
853}
854
855evutil_socket_t
856bufferevent_getfd(struct bufferevent *bev)
857{
858	union bufferevent_ctrl_data d;
859	int res = -1;
860	d.fd = -1;
861	BEV_LOCK(bev);
862	if (bev->be_ops->ctrl)
863		res = bev->be_ops->ctrl(bev, BEV_CTRL_GET_FD, &d);
864	BEV_UNLOCK(bev);
865	return (res<0) ? -1 : d.fd;
866}
867
868enum bufferevent_options
869bufferevent_get_options_(struct bufferevent *bev)
870{
871	struct bufferevent_private *bev_p =
872	    EVUTIL_UPCAST(bev, struct bufferevent_private, bev);
873	enum bufferevent_options options;
874
875	BEV_LOCK(bev);
876	options = bev_p->options;
877	BEV_UNLOCK(bev);
878	return options;
879}
880
881
882static void
883bufferevent_cancel_all_(struct bufferevent *bev)
884{
885	union bufferevent_ctrl_data d;
886	memset(&d, 0, sizeof(d));
887	BEV_LOCK(bev);
888	if (bev->be_ops->ctrl)
889		bev->be_ops->ctrl(bev, BEV_CTRL_CANCEL_ALL, &d);
890	BEV_UNLOCK(bev);
891}
892
893short
894bufferevent_get_enabled(struct bufferevent *bufev)
895{
896	short r;
897	BEV_LOCK(bufev);
898	r = bufev->enabled;
899	BEV_UNLOCK(bufev);
900	return r;
901}
902
903struct bufferevent *
904bufferevent_get_underlying(struct bufferevent *bev)
905{
906	union bufferevent_ctrl_data d;
907	int res = -1;
908	d.ptr = NULL;
909	BEV_LOCK(bev);
910	if (bev->be_ops->ctrl)
911		res = bev->be_ops->ctrl(bev, BEV_CTRL_GET_UNDERLYING, &d);
912	BEV_UNLOCK(bev);
913	return (res<0) ? NULL : d.ptr;
914}
915
916static void
917bufferevent_generic_read_timeout_cb(evutil_socket_t fd, short event, void *ctx)
918{
919	struct bufferevent *bev = ctx;
920	bufferevent_incref_and_lock_(bev);
921	bufferevent_disable(bev, EV_READ);
922	bufferevent_run_eventcb_(bev, BEV_EVENT_TIMEOUT|BEV_EVENT_READING, 0);
923	bufferevent_decref_and_unlock_(bev);
924}
925static void
926bufferevent_generic_write_timeout_cb(evutil_socket_t fd, short event, void *ctx)
927{
928	struct bufferevent *bev = ctx;
929	bufferevent_incref_and_lock_(bev);
930	bufferevent_disable(bev, EV_WRITE);
931	bufferevent_run_eventcb_(bev, BEV_EVENT_TIMEOUT|BEV_EVENT_WRITING, 0);
932	bufferevent_decref_and_unlock_(bev);
933}
934
935void
936bufferevent_init_generic_timeout_cbs_(struct bufferevent *bev)
937{
938	event_assign(&bev->ev_read, bev->ev_base, -1, EV_FINALIZE,
939	    bufferevent_generic_read_timeout_cb, bev);
940	event_assign(&bev->ev_write, bev->ev_base, -1, EV_FINALIZE,
941	    bufferevent_generic_write_timeout_cb, bev);
942}
943
944int
945bufferevent_generic_adj_timeouts_(struct bufferevent *bev)
946{
947	const short enabled = bev->enabled;
948	struct bufferevent_private *bev_p =
949	    EVUTIL_UPCAST(bev, struct bufferevent_private, bev);
950	int r1=0, r2=0;
951	if ((enabled & EV_READ) && !bev_p->read_suspended &&
952	    evutil_timerisset(&bev->timeout_read))
953		r1 = event_add(&bev->ev_read, &bev->timeout_read);
954	else
955		r1 = event_del(&bev->ev_read);
956
957	if ((enabled & EV_WRITE) && !bev_p->write_suspended &&
958	    evutil_timerisset(&bev->timeout_write) &&
959	    evbuffer_get_length(bev->output))
960		r2 = event_add(&bev->ev_write, &bev->timeout_write);
961	else
962		r2 = event_del(&bev->ev_write);
963	if (r1 < 0 || r2 < 0)
964		return -1;
965	return 0;
966}
967
968int
969bufferevent_add_event_(struct event *ev, const struct timeval *tv)
970{
971	if (tv->tv_sec == 0 && tv->tv_usec == 0)
972		return event_add(ev, NULL);
973	else
974		return event_add(ev, tv);
975}
976
977/* For use by user programs only; internally, we should be calling
978   either bufferevent_incref_and_lock_(), or BEV_LOCK. */
979void
980bufferevent_lock(struct bufferevent *bev)
981{
982	bufferevent_incref_and_lock_(bev);
983}
984
985void
986bufferevent_unlock(struct bufferevent *bev)
987{
988	bufferevent_decref_and_unlock_(bev);
989}
990