eventqueue.c

// SPDX-License-Identifier: GPL-2.0-only
#include <linux/module.h>
#include <linux/miscdevice.h>
#include <linux/fs.h>
#include <linux/mutex.h>
#include <linux/poll.h>
#include <linux/ktime.h>
#include <linux/uaccess.h>
#include "include/event_ring.h"

static DEFINE_MUTEX(queue_lock);
static DECLARE_WAIT_QUEUE_HEAD(readers);
static DECLARE_WAIT_QUEUE_HEAD(writers);
static struct event_ring queue;
/* Atomic readiness is a hint; the queue is always rechecked under queue_lock. */
static atomic_t depth = ATOMIC_INIT(0);
static unsigned long long next_sequence;

static ssize_t event_read(struct file *file, char __user *buf, size_t len,
			  loff_t *offset)
{
	int ret;
	(void)offset;
	if (!len)
		return 0;
	if (len < sizeof(struct event_record))
		return -EINVAL;
	for (;;) {
		if (mutex_lock_interruptible(&queue_lock))
			return -ERESTARTSYS;
		if (queue.count)
			break;
		mutex_unlock(&queue_lock);
		if (file->f_flags & O_NONBLOCK)
			return -EAGAIN;
		ret = wait_event_interruptible(readers, atomic_read(&depth) > 0);
		if (ret)
			return ret;
	}
	/* Keep the record queued on EFAULT. A mutex permits sleeping user copies. */
	if (copy_to_user(buf, ring_peek(&queue), sizeof(struct event_record))) {
		mutex_unlock(&queue_lock);
		return -EFAULT;
	}
	ring_pop(&queue);
	atomic_set(&depth, queue.count);
	mutex_unlock(&queue_lock);
	wake_up_interruptible(&writers);
	return sizeof(struct event_record);
}

static ssize_t event_write(struct file *file, const char __user *buf, size_t len,
			   loff_t *offset)
{
	struct event_record event;
	int ret;
	(void)offset;
	if (!len)
		return 0;
	if (len != sizeof(event.value))
		return -EINVAL;
	if (copy_from_user(&event.value, buf, sizeof(event.value)))
		return -EFAULT;
	for (;;) {
		if (mutex_lock_interruptible(&queue_lock))
			return -ERESTARTSYS;
		if (queue.count < EVENT_CAPACITY)
			break;
		mutex_unlock(&queue_lock);
		if (file->f_flags & O_NONBLOCK)
			return -EAGAIN;
		ret = wait_event_interruptible(writers,
				atomic_read(&depth) < EVENT_CAPACITY);
		if (ret)
			return ret;
	}
	event.sequence = next_sequence++;
	event.monotonic_ns = ktime_get_ns();
	ring_push(&queue, event);
	atomic_set(&depth, queue.count);
	mutex_unlock(&queue_lock);
	wake_up_interruptible(&readers);
	return sizeof(event.value);
}

static __poll_t event_poll(struct file *file, poll_table *wait)
{
	__poll_t ready = 0;
	poll_wait(file, &readers, wait);
	poll_wait(file, &writers, wait);
	mutex_lock(&queue_lock);
	if (queue.count)
		ready |= EPOLLIN | EPOLLRDNORM;
	if (queue.count < EVENT_CAPACITY)
		ready |= EPOLLOUT | EPOLLWRNORM;
	mutex_unlock(&queue_lock);
	return ready;
}

static const struct file_operations event_ops = {
	.owner = THIS_MODULE,
	.read = event_read,
	.write = event_write,
	.poll = event_poll,
	.llseek = no_llseek,
};
static struct miscdevice event_device = {
	.minor = MISC_DYNAMIC_MINOR,
	.name = "telecom_eventq",
	.fops = &event_ops,
	.mode = 0600,
};
static int __init event_init(void)
{
	BUILD_BUG_ON(sizeof(struct event_record) != 24);
	return misc_register(&event_device);
}
static void __exit event_exit(void)
{
	misc_deregister(&event_device);
}
module_init(event_init);
module_exit(event_exit);
MODULE_LICENSE("GPL");
MODULE_DESCRIPTION("Bounded event queue lab: blocking I/O, poll and backpressure");