libei/src/util-sources.c

/*
 * Copyright © 2020 Red Hat, Inc.
 *
 * Permission is hereby granted, free of charge, to any person obtaining a
 * copy of this software and associated documentation files (the "Software"),
 * to deal in the Software without restriction, including without limitation
 * the rights to use, copy, modify, merge, publish, distribute, sublicense,
 * and/or sell copies of the Software, and to permit persons to whom the
 * Software is furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice (including the next
 * paragraph) shall be included in all copies or substantial portions of the
 * Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.  IN NO EVENT SHALL
 * THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
 * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
 * DEALINGS IN THE SOFTWARE.
 */

#include "config.h"

#include <assert.h>
#include <errno.h>
#include <sys/epoll.h>

#include "util-object.h"
#include "util-io.h"
#include "util-list.h"
#include "util-sources.h"

struct sink {
	struct object object;
	int epollfd;
	struct list sources;
	struct list sources_removed;
};

enum source_close_behavior {
	SOURCE_CLOSE_FD_ON_REMOVE = 1, /* default */
	SOURCE_CLOSE_FD_ON_DESTROY,
	SOURCE_CLOSE_FD_NEVER,
};

struct source {
	struct object object;
	struct sink *sink;
	struct list link; /* sink.sources or sink.sources_removed */
	source_dispatch_t dispatch;
	void *user_data;
	enum source_close_behavior close_behavior;
	int fd;
};

OBJECT_IMPLEMENT_REF(source);
OBJECT_IMPLEMENT_UNREF(source);
OBJECT_IMPLEMENT_GETTER(source, fd, int);
OBJECT_IMPLEMENT_GETTER(source, user_data, void*);
OBJECT_IMPLEMENT_SETTER(source, user_data, void*);

/**
 * Remove the source, closing the fd. The source is tagged as removed and
 * will be removed whenever sink_dispatch() finishes (or is called next).
 */
void
source_remove(struct source *source)
{
	if (source->fd != -1) {
		epoll_ctl(source->sink->epollfd, EPOLL_CTL_DEL, source->fd, NULL);
		if (source->close_behavior == SOURCE_CLOSE_FD_ON_REMOVE)
			source->fd = xclose(source->fd);
	}
	/* Note: epollfd was the owner of the source, new owner is the removed
	   list */
	list_remove(&source->link);
	list_append(&source->sink->sources_removed, &source->link);
}

/* Ignore, use source_unref() */
static void
source_destroy(struct source *source)
{
	if (source->fd != -1) {
		source_remove(source);
		if (source->close_behavior == SOURCE_CLOSE_FD_ON_DESTROY)
			source->fd = xclose(source->fd);
	}
}

OBJECT_IMPLEMENT_CREATE(source);

static inline struct source *
source_add(struct sink *sink, int sourcefd,
	   source_dispatch_t dispatch,
	   void *user_data,
	   enum source_close_behavior close_behavior)
{
	struct source *source = source_create(NULL);

	source->dispatch = dispatch;
	source->user_data = user_data;
	source->fd = sourcefd;
	source->sink = sink;
	source->close_behavior = close_behavior;

	struct epoll_event e = {
		.events = EPOLLIN,
		.data.ptr = source, /* epollfd is the owner of this source */
	};

	if (epoll_ctl(sink->epollfd, EPOLL_CTL_ADD, sourcefd, &e) < 0) {
		free(source);
		return NULL;
	}

	list_append(&sink->sources, &source->link);

	return source;
}

struct source *
source_add_autoclose(struct sink *sink, int sourcefd,
		     source_dispatch_t dispatch,
		     void *user_data)
{
	return source_add(sink, sourcefd, dispatch, user_data, SOURCE_CLOSE_FD_ON_REMOVE);
}

static void
sink_destroy(struct sink *sink)
{
	struct source *s, *tmp;
	list_for_each_safe(s, tmp, &sink->sources, link) {
		source_remove(s);
	}
	list_for_each_safe(s, tmp, &sink->sources_removed, link) {
		source_unref(s);
	}
	xclose(sink->epollfd);
}

OBJECT_IMPLEMENT_UNREF(sink);
OBJECT_IMPLEMENT_CREATE(sink);

int
sink_get_fd(struct sink *sink)
{
	assert(sink);
	return sink->epollfd;
}

struct sink *
sink_new(void)
{
	int fd = epoll_create1(EPOLL_CLOEXEC);
	if (fd < 0)
		return NULL;

	struct sink *sink = sink_create(NULL);

	sink->epollfd = fd;
	list_init(&sink->sources);
	list_init(&sink->sources_removed);

	return sink;
}

int
sink_dispatch(struct sink *sink)
{
	struct epoll_event ep[32];
	int count = epoll_wait(sink->epollfd, ep, sizeof(ep)/sizeof(ep[0]), 0);
	if (count < 0)
		return -errno;

	for (int i = 0; i < count; ++i) {
		struct source *source = ep[i].data.ptr;
		if (source->fd == -1)
			continue;

		source->dispatch(source, source->user_data);
	}

	struct source *s, *tmp;
	list_for_each_safe(s, tmp, &sink->sources_removed, link) {
		list_remove(&s->link);
		source_unref(s);
	}

	return 0;
}