libqb/lib/loop_poll.c

/*
 * Copyright (C) 2010 Red Hat, Inc.
 *
 * Author: Angus Salkeld <asalkeld@redhat.com>
 *
 * This file is part of libqb.
 *
 * libqb is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Lesser General Public License as published by
 * the Free Software Foundation, either version 2.1 of the License, or
 * (at your option) any later version.
 *
 * libqb is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU Lesser General Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public License
 * along with libqb.  If not, see <http://www.gnu.org/licenses/>.
 */
#include "os_base.h"

/* due to MinGW/splint emitting "< Location unknown >: Previous use of" */
#if defined(HAVE_SYS_RESOURCE_H) && !defined(S_SPLINT_S)
#include <sys/resource.h>
#endif

#include <signal.h>

#if defined(__DARWIN_NSIG)
#define QB_MAX_NUM_SIGNALS __DARWIN_NSIG
#else
  #if defined(NSIG)
  #define QB_MAX_NUM_SIGNALS NSIG - 1
  #else
  #define QB_MAX_NUM_SIGNALS 31
  #endif
#endif

#include "loop_poll_int.h"

/*
 * Define this to log slow (>10ms) jobs.
 */
#undef DEBUG_DISPATCH_TIME

/* logs, std(in|out|err), pipe */
#define POLL_FDS_USED_MISC 50

#ifdef HAVE_EPOLL
#define USE_EPOLL 1
#else
 #ifdef HAVE_KQUEUE
 #define USE_KQUEUE 1
 #else
 #define USE_POLL 1
 #endif /* HAVE_KQUEUE */
#endif /* HAVE_EPOLL */

static int32_t _qb_signal_add_to_jobs_(struct qb_loop *l,
				       struct qb_poll_entry *pe);

static void
_poll_entry_check_generate_(struct qb_poll_entry *pe)
{
	int32_t i;

	for (i = 0; i < 200; i++) {
		pe->check = random();

		if (pe->check != 0 && pe->check != UINT32_MAX) {
			break;
		}
	}
}

static void
_poll_entry_mark_deleted_(struct qb_poll_entry *pe)
{
	pe->ufd.fd = -1;
	pe->state = QB_POLL_ENTRY_DELETED;
	pe->check = 0;
}

static void
_poll_entry_empty_(struct qb_poll_entry *pe)
{
	memset(pe, 0, sizeof(struct qb_poll_entry));
	pe->ufd.fd = -1;
}

static void
_poll_dispatch_and_take_back_(struct qb_loop_item *item,
			      enum qb_loop_priority p)
{
	struct qb_poll_entry *pe = (struct qb_poll_entry *)item;
	int32_t res;
#ifdef DEBUG_DISPATCH_TIME
	uint64_t start;
	uint64_t stop;
	int32_t log_warn = QB_FALSE;

	start = qb_util_nano_current_get();
#endif /* DEBUG_DISPATCH_TIME */

	assert(pe->state == QB_POLL_ENTRY_JOBLIST);
	assert(pe->item.type == QB_LOOP_FD);

	res = pe->poll_dispatch_fn(pe->ufd.fd,
				   pe->ufd.revents,
				   pe->item.user_data);
	if (res < 0) {
		_poll_entry_mark_deleted_(pe);
	} else if (pe->state != QB_POLL_ENTRY_DELETED) {
		pe->state = QB_POLL_ENTRY_ACTIVE;
		pe->ufd.revents = 0;
	}
#ifdef DEBUG_DISPATCH_TIME
	if (pe->state == QB_POLL_ENTRY_ACTIVE) {
		pe->runs++;
		if ((pe->runs % 50) == 0) {
			log_warn = QB_TRUE;
		}
		stop = qb_util_nano_current_get();
		if ((stop - start) > (10 * QB_TIME_NS_IN_MSEC)) {
			log_warn = QB_TRUE;
		}

		if (log_warn && pe->item.type == QB_LOOP_FD) {
			qb_util_log(LOG_INFO,
				    "[fd:%d] dispatch:%p runs:%d duration:%d ms",
				    pe->ufd.fd, pe->poll_dispatch_fn,
				    pe->runs,
				    (int32_t) ((stop -
						start) / QB_TIME_NS_IN_MSEC));
		}
	}
#endif /* DEBUG_DISPATCH_TIME */
}

void
qb_poll_fds_usage_check_(struct qb_poll_source *s)
{
	struct rlimit lim;
	static int32_t socks_limit = 0;
	int32_t send_event = QB_FALSE;
	int32_t socks_used = 0;
	int32_t socks_avail = 0;
	struct qb_poll_entry *pe;
	int32_t i;

	if (socks_limit == 0) {
		if (getrlimit(RLIMIT_NOFILE, &lim) == -1) {
			qb_util_perror(LOG_WARNING, "getrlimit");
			return;
		}
		socks_limit = lim.rlim_cur;
		socks_limit -= POLL_FDS_USED_MISC;
		if (socks_limit < 0) {
			socks_limit = 0;
		}
	}

	for (i = 0; i < s->poll_entry_count; i++) {
		assert(qb_array_index(s->poll_entries, i, (void **)&pe) == 0);
		if ((pe->state == QB_POLL_ENTRY_ACTIVE ||
		     pe->state == QB_POLL_ENTRY_JOBLIST) && pe->ufd.fd != -1) {
			socks_used++;
		}
		if (pe->state == QB_POLL_ENTRY_DELETED) {
			_poll_entry_empty_(pe);
		}
	}

	socks_avail = socks_limit - socks_used;
	if (socks_avail < 0) {
		socks_avail = 0;
	}
	send_event = QB_FALSE;
	if (s->not_enough_fds) {
		if (socks_avail > 2) {
			s->not_enough_fds = QB_FALSE;
			send_event = QB_TRUE;
		}
	} else {
		if (socks_avail <= 1) {
			s->not_enough_fds = QB_TRUE;
			send_event = QB_TRUE;
		}
	}
	if (send_event && s->low_fds_event_fn) {
		s->low_fds_event_fn(s->not_enough_fds, socks_avail);
	}
}


struct qb_loop_source *
qb_loop_poll_create(struct qb_loop *l)
{
	struct qb_poll_source *s = malloc(sizeof(struct qb_poll_source));
	if (s == NULL) {
		return NULL;
	}
	s->s.l = l;
	s->s.dispatch_and_take_back = _poll_dispatch_and_take_back_;

	s->poll_entries = qb_array_create_2(16, sizeof(struct qb_poll_entry), 16);
	s->poll_entry_count = 0;
	s->low_fds_event_fn = NULL;
	s->not_enough_fds = QB_FALSE;

#ifdef USE_EPOLL
	(void)qb_epoll_init(s);
#endif
#ifdef USE_KQUEUE
	(void)qb_kqueue_init(s);
#endif
#ifdef USE_POLL
	(void)qb_poll_init(s);
#endif /* USE_POLL */

	return (struct qb_loop_source *)s;
}

void
qb_loop_poll_destroy(struct qb_loop *l)
{
	struct qb_poll_source *s = (struct qb_poll_source *)l->fd_source;
	qb_array_free(s->poll_entries);

	s->driver.fini(s);

	free(s);
}

int32_t
qb_loop_poll_low_fds_event_set(struct qb_loop *l,
			       qb_loop_poll_low_fds_event_fn fn)
{
	struct qb_poll_source *s = (struct qb_poll_source *)l->fd_source;
	s->low_fds_event_fn = fn;

	return 0;
}

static int32_t
_get_empty_array_position_(struct qb_poll_source *s)
{
	int32_t found = QB_FALSE;
	uint32_t install_pos;
	int32_t res = 0;
	struct qb_poll_entry *pe;

	for (install_pos = 0;
	     install_pos < s->poll_entry_count; install_pos++) {
		assert(qb_array_index
		       (s->poll_entries, install_pos, (void **)&pe) == 0);
		if (pe->state == QB_POLL_ENTRY_EMPTY) {
			found = QB_TRUE;
			break;
		}
	}

	if (found == QB_FALSE) {
#ifdef USE_POLL
		struct pollfd *ufds;
		int32_t new_size = (s->poll_entry_count + 1) * sizeof(struct pollfd);
		ufds = realloc(s->ufds, new_size);
		if (ufds == NULL) {
			return -ENOMEM;
		}
		s->ufds = ufds;
#endif /* USE_POLL */
		/*
		 * Grow pollfd list
		 */
		res = qb_array_grow(s->poll_entries, s->poll_entry_count + 1);
		if (res != 0) {
			return res;
		}

		s->poll_entry_count += 1;
		install_pos = s->poll_entry_count - 1;
	}
	return install_pos;
}

static int32_t
_poll_add_(struct qb_loop *l,
	   enum qb_loop_priority p,
	   int32_t fd, int32_t events, void *data, struct qb_poll_entry **pe_pt)
{
	struct qb_poll_entry *pe;
	uint32_t install_pos;
	int32_t res = 0;
	struct qb_poll_source *s;

	if (l == NULL) {
		return -EINVAL;
	}

	s = (struct qb_poll_source *)l->fd_source;

	install_pos = _get_empty_array_position_(s);

	assert(qb_array_index(s->poll_entries, install_pos, (void **)&pe) == 0);
	pe->state = QB_POLL_ENTRY_ACTIVE;
	pe->install_pos = install_pos;
	_poll_entry_check_generate_(pe);
	pe->ufd.fd = fd;
	pe->ufd.events = events;
	pe->ufd.revents = 0;
	pe->item.user_data = data;
	pe->item.source = (struct qb_loop_source *)l->fd_source;
	pe->p = p;
	pe->runs = 0;
	res = s->driver.add(s, pe, fd, events);
	if (res == 0) {
		*pe_pt = pe;
		return 0;
	} else {
		pe->state = QB_POLL_ENTRY_EMPTY;
		return res;
	}
}

static int32_t
_qb_poll_add_to_jobs_(struct qb_loop *l, struct qb_poll_entry *pe)
{
	assert(pe->item.type == QB_LOOP_FD);
	qb_loop_level_item_add(&l->level[pe->p], &pe->item);
	pe->state = QB_POLL_ENTRY_JOBLIST;
	return 1;
}

int32_t
qb_loop_poll_add(struct qb_loop * lp,
		 enum qb_loop_priority p,
		 int32_t fd,
		 int32_t events,
		 void *data, qb_loop_poll_dispatch_fn dispatch_fn)
{
	struct qb_poll_entry *pe = NULL;
	int32_t size;
	int32_t new_size;
	int32_t res;
	struct qb_loop *l = lp;

	if (l == NULL) {
		l = qb_loop_default_get();
	}

	size = ((struct qb_poll_source *)l->fd_source)->poll_entry_count;
	res = _poll_add_(l, p, fd, events, data, &pe);
	if (res != 0) {
		qb_util_perror(LOG_ERR,
			       "couldn't add poll entryfor FD %d", fd);
		return res;
	}
	new_size = ((struct qb_poll_source *)l->fd_source)->poll_entry_count;

	pe->poll_dispatch_fn = dispatch_fn;
	pe->item.type = QB_LOOP_FD;
	pe->add_to_jobs = _qb_poll_add_to_jobs_;

	if (new_size > size) {
		qb_util_log(LOG_TRACE,
			    "grown poll array to %d for FD %d", new_size, fd);
	}

	return res;
}

int32_t
qb_loop_poll_mod(struct qb_loop * lp,
		 enum qb_loop_priority p,
		 int32_t fd,
		 int32_t events,
		 void *data, qb_loop_poll_dispatch_fn dispatch_fn)
{
	uint32_t i;
	int32_t res = 0;
	struct qb_poll_entry *pe;
	struct qb_poll_source *s;
	struct qb_loop *l = lp;

	if (l == NULL) {
		l = qb_loop_default_get();
	}
	s = (struct qb_poll_source *)l->fd_source;

	/*
	 * Find file descriptor to modify events and dispatch function
	 */
	for (i = 0; i < s->poll_entry_count; i++) {
		assert(qb_array_index(s->poll_entries, i, (void **)&pe) == 0);
		if (pe->ufd.fd != fd) {
			continue;
		}
		if (pe->state == QB_POLL_ENTRY_DELETED || pe->check == 0) {
			qb_util_log(LOG_ERR,
				    "poll_mod : can't modify entry already deleted");
			return -EBADF;
		}
		pe->poll_dispatch_fn = dispatch_fn;
		pe->item.user_data = data;
		pe->p = p;
		if (pe->ufd.events != events) {
			res = s->driver.mod(s, pe, fd, events);
			pe->ufd.events = events;
		}
		return res;
	}

	return -EBADF;
}

int32_t
qb_loop_poll_del(struct qb_loop * lp, int32_t fd)
{
	int32_t i;
	int32_t res = 0;
	struct qb_poll_entry *pe;
	struct qb_poll_source *s;
	struct qb_loop *l = lp;

	if (l == NULL) {
		l = qb_loop_default_get();
	}
	s = (struct qb_poll_source *)l->fd_source;
	for (i = 0; i < s->poll_entry_count; i++) {
		assert(qb_array_index(s->poll_entries, i, (void **)&pe) == 0);
		if (pe->ufd.fd != fd || pe->item.type != QB_LOOP_FD) {
			continue;
		}
		if (pe->state == QB_POLL_ENTRY_DELETED ||
		    pe->state == QB_POLL_ENTRY_EMPTY) {
			return 0;
		}
		if (pe->state == QB_POLL_ENTRY_JOBLIST) {
			qb_loop_level_item_del(&l->level[pe->p], &pe->item);
		}
		res = s->driver.del(s, pe, fd, i);
		_poll_entry_mark_deleted_(pe);
		return res;
	}

	return -EBADF;
}

static int32_t pipe_fds[2] = { -1, -1 };

struct qb_signal_source {
	struct qb_loop_source s;
	struct qb_list_head sig_head;
	sigset_t signal_superset;
};

struct qb_loop_sig {
	struct qb_loop_item item;
	int32_t signal;
	enum qb_loop_priority p;
	qb_loop_signal_dispatch_fn dispatch_fn;
	struct qb_loop_sig *cloned_from;
};

static void
_handle_real_signal_(int signal_num, siginfo_t * si, void *context)
{
	int32_t sig = signal_num;
	int32_t res = 0;

	if (pipe_fds[1] > 0) {
try_again:
		res = write(pipe_fds[1], &sig, sizeof(int32_t));
		if (res == -1 && errno == EAGAIN) {
			goto try_again;
		}
	}
}

static void
_signal_dispatch_and_take_back_(struct qb_loop_item *item,
				enum qb_loop_priority p)
{
	struct qb_loop_sig *sig = (struct qb_loop_sig *)item;
	int32_t res;

	res = sig->dispatch_fn(sig->signal, sig->item.user_data);
	if (res != 0) {
		(void)qb_loop_signal_del(sig->cloned_from->item.source->l,
					 sig->cloned_from);
	}
	free(sig);
}

struct qb_loop_source *
qb_loop_signals_create(struct qb_loop *l)
{
	int32_t res = 0;
	struct qb_poll_entry *pe;
	struct qb_signal_source *s = calloc(1, sizeof(struct qb_signal_source));

	if (s == NULL) {
		return NULL;
	}
	s->s.l = l;
	s->s.dispatch_and_take_back = _signal_dispatch_and_take_back_;
	s->s.poll = NULL;
	qb_list_init(&s->sig_head);
	sigemptyset(&s->signal_superset);

	if (pipe_fds[0] < 0) {
		res = pipe(pipe_fds);
		if (res == -1) {
			res = -errno;
			qb_util_perror(LOG_ERR, "Can't light pipe");
			goto error_exit;
		}
		(void)qb_sys_fd_nonblock_cloexec_set(pipe_fds[0]);
		(void)qb_sys_fd_nonblock_cloexec_set(pipe_fds[1]);

		res = _poll_add_(l, QB_LOOP_HIGH,
				 pipe_fds[0], POLLIN, NULL, &pe);
		if (res == 0) {
			pe->poll_dispatch_fn = NULL;
			pe->item.type = QB_LOOP_SIG;
			pe->add_to_jobs = _qb_signal_add_to_jobs_;
		} else {
			qb_util_perror(LOG_ERR, "Can't smoke pipe");
			goto error_exit;
		}
	}

	return (struct qb_loop_source *)s;

error_exit:
	errno = -res;
	free(s);
	if (pipe_fds[0] >= 0) {
		close(pipe_fds[0]);
	}
	if (pipe_fds[1] >= 0) {
		close(pipe_fds[1]);
	}
	return NULL;
}

void
qb_loop_signals_destroy(struct qb_loop *l)
{
	struct qb_signal_source *s =
	    (struct qb_signal_source *)l->signal_source;
	struct qb_list_head *list;
	struct qb_list_head *n;
	struct qb_loop_item *item;

	close(pipe_fds[0]);
	pipe_fds[0] = -1;
	close(pipe_fds[1]);
	pipe_fds[1] = -1;

	qb_list_for_each_safe(list, n, &s->sig_head) {
		item = qb_list_entry(list, struct qb_loop_item, list);
		qb_list_del(&item->list);
		free(item);
	}

	free(l->signal_source);
}

static int32_t
_qb_signal_add_to_jobs_(struct qb_loop *l, struct qb_poll_entry *pe)
{
	struct qb_signal_source *s =
	    (struct qb_signal_source *)l->signal_source;
	struct qb_list_head *list;
	struct qb_loop_sig *sig;
	struct qb_loop_item *item;
	struct qb_loop_sig *new_sig_job;
	int32_t the_signal;
	ssize_t res;
	int32_t jobs_added = 0;

	res = read(pipe_fds[0], &the_signal, sizeof(int32_t));
	if (res != sizeof(int32_t)) {
		qb_util_perror(LOG_WARNING, "failed to read pipe");
		return 0;
	}
	pe->ufd.revents = 0;

	qb_list_for_each(list, &s->sig_head) {
		item = qb_list_entry(list, struct qb_loop_item, list);
		sig = (struct qb_loop_sig *)item;
		if (sig->signal == the_signal) {
			new_sig_job = calloc(1, sizeof(struct qb_loop_sig));
			if (new_sig_job == NULL) {
				return jobs_added;
			}
			memcpy(new_sig_job, sig, sizeof(struct qb_loop_sig));

			qb_util_log(LOG_TRACE,
				    "adding signal [%d] to job queue %p",
				    the_signal, sig);

			new_sig_job->cloned_from = sig;
			qb_loop_level_item_add(&l->level[sig->p],
					       &new_sig_job->item);
			jobs_added++;
		}
	}
	return jobs_added;
}

static void
_adjust_sigactions_(struct qb_signal_source *s)
{
	struct qb_loop_sig *sig;
	struct qb_loop_item *item;
	struct sigaction sa;
	int32_t i;
	int32_t needed;

	sa.sa_flags = SA_SIGINFO;
	sa.sa_sigaction = _handle_real_signal_;
	sigemptyset(&s->signal_superset);
	sigemptyset(&sa.sa_mask);

	/* re-set to default */
	for (i = 0; i < QB_MAX_NUM_SIGNALS; i++) {
		needed = QB_FALSE;
		qb_list_for_each_entry(item, &s->sig_head, list) {
			sig = (struct qb_loop_sig *)item;
			if (i == sig->signal) {
				needed = QB_TRUE;
				break;
			}
		}
		if (needed) {
			sigaddset(&s->signal_superset, i);
			sigaction(i, &sa, NULL);
		}
	}
}

int32_t
qb_loop_signal_add(qb_loop_t * lp,
		   enum qb_loop_priority p,
		   int32_t the_sig,
		   void *data,
		   qb_loop_signal_dispatch_fn dispatch_fn,
		   qb_loop_signal_handle * handle)
{
	struct qb_loop_sig *sig;
	struct qb_signal_source *s;
	struct qb_loop *l = lp;

	if (l == NULL) {
		l = qb_loop_default_get();
	}
	if (l == NULL || dispatch_fn == NULL) {
		return -EINVAL;
	}
	if (p < QB_LOOP_LOW || p > QB_LOOP_HIGH) {
		return -EINVAL;
	}
	s = (struct qb_signal_source *)l->signal_source;
	sig = calloc(1, sizeof(struct qb_loop_sig));
	if (sig == NULL) {
		return -errno;
	}

	sig->dispatch_fn = dispatch_fn;
	sig->p = p;
	sig->signal = the_sig;
	sig->item.user_data = data;
	sig->item.source = l->signal_source;
	sig->item.type = QB_LOOP_SIG;

	qb_list_init(&sig->item.list);
	qb_list_add_tail(&sig->item.list, &s->sig_head);

	if (sigismember(&s->signal_superset, the_sig) != 1) {
		_adjust_sigactions_(s);
	}
	if (handle) {
		*handle = sig;
	}

	return 0;
}

int32_t
qb_loop_signal_mod(qb_loop_t * lp,
		   enum qb_loop_priority p,
		   int32_t the_sig,
		   void *data,
		   qb_loop_signal_dispatch_fn dispatch_fn,
		   qb_loop_signal_handle handle)
{
	struct qb_signal_source *s;
	struct qb_loop_sig *sig = (struct qb_loop_sig *)handle;
	struct qb_loop *l = lp;

	if (l == NULL) {
		l = qb_loop_default_get();
	}
	if (l == NULL || dispatch_fn == NULL || handle == NULL) {
		return -EINVAL;
	}
	if (p < QB_LOOP_LOW || p > QB_LOOP_HIGH) {
		return -EINVAL;
	}
	s = (struct qb_signal_source *)l->signal_source;

	sig->item.user_data = data;
	sig->item.type = QB_LOOP_SIG;
	sig->dispatch_fn = dispatch_fn;
	sig->p = p;

	if (sig->signal != the_sig) {
		(void)signal(sig->signal, SIG_DFL);
		sig->signal = the_sig;
		_adjust_sigactions_(s);
	}

	return 0;
}

int32_t
qb_loop_signal_del(qb_loop_t * lp, qb_loop_signal_handle handle)
{
	struct qb_signal_source *s;
	struct qb_loop_sig *sig = (struct qb_loop_sig *)handle;
	struct qb_loop_sig *sig_clone;
	struct qb_loop *l = lp;
	struct qb_loop_item *item;

	if (l == NULL) {
		l = qb_loop_default_get();
	}
	if (l == NULL || handle == NULL) {
		return -EINVAL;
	}
	s = (struct qb_signal_source *)l->signal_source;

	qb_list_for_each_entry(item, &l->level[sig->p].wait_head, list) {
		if (item->type != QB_LOOP_SIG) {
			continue;
		}
		sig_clone = (struct qb_loop_sig *)item;
		if (sig_clone->cloned_from == sig) {
			qb_util_log(LOG_TRACE, "deleting sig in WAITLIST");
			qb_list_del(&sig_clone->item.list);
			free(sig_clone);
			break;
		}
	}

	qb_list_for_each_entry(item, &l->level[sig->p].job_head, list) {
		if (item->type != QB_LOOP_SIG) {
			continue;
		}
		sig_clone = (struct qb_loop_sig *)item;
		if (sig_clone->cloned_from == sig) {
			qb_loop_level_item_del(&l->level[sig->p], item);
			qb_util_log(LOG_TRACE, "deleting sig in JOBLIST");
			break;
		}
	}

	qb_list_del(&sig->item.list);
	(void)signal(sig->signal, SIG_DFL);
	free(sig);
	_adjust_sigactions_(s);
	return 0;
}