kernel: add channel/port ipc mechanism

init/main.c

@@ -33,7 +33,7 @@ static void hang(void)
 
 	while (1) {
 #if 0
-		printk("[cpu %u, task %u, thread %u]: tick",
+		printk("[cpu %u, task %ld, thread %u]: tick",
 		       this_cpu(),
 		       self->t_id,
 		       thread->tr_id);
@@ -74,6 +74,9 @@ void kernel_init(uintptr_t arg)
 {
 	ml_init(arg);
 
+	port_type_init();
+	channel_type_init();
+
 	struct boot_module bsp_image = {0};
 	bsp_get_location(&bsp_image);
 

kernel/channel.c

@@ -0,0 +1,240 @@
+#include <kernel/channel.h>
+#include <kernel/msg.h>
+#include <kernel/util.h>
+#include <kernel/vm-region.h>
+
+#define CHANNEL_CAST(p) OBJECT_C_CAST(struct channel, c_base, &channel_type, p)
+
+static struct object_type channel_type = {
+	.ob_name = "channel",
+	.ob_size = sizeof(struct channel),
+	.ob_header_offset = offsetof(struct channel, c_base),
+};
+
+BTREE_DEFINE_SIMPLE_GET(struct kmsg, msgid_t, msg_node, msg_id, get_msg_with_id)
+
+kern_status_t channel_type_init(void)
+{
+	return object_type_register(&channel_type);
+}
+
+extern struct channel *channel_create(void)
+{
+	struct object *channel_object = object_create(&channel_type);
+	if (!channel_object) {
+		return NULL;
+	}
+
+	struct channel *channel = CHANNEL_CAST(channel_object);
+
+	return channel;
+}
+
+static bool try_enqueue(struct btree *tree, struct kmsg *msg)
+{
+	if (!tree->b_root) {
+		tree->b_root = &msg->msg_node;
+		btree_insert_fixup(tree, &msg->msg_node);
+		return true;
+	}
+
+	struct btree_node *cur = tree->b_root;
+	while (1) {
+		struct kmsg *cur_node
+			= BTREE_CONTAINER(struct kmsg, msg_node, cur);
+		struct btree_node *next = NULL;
+
+		if (msg->msg_id > cur_node->msg_id) {
+			next = btree_right(cur);
+
+			if (!next) {
+				btree_put_right(cur, &msg->msg_node);
+				break;
+			}
+		} else if (msg->msg_id < cur_node->msg_id) {
+			next = btree_left(cur);
+
+			if (!next) {
+				btree_put_left(cur, &msg->msg_node);
+				break;
+			}
+		} else {
+			return false;
+		}
+
+		cur = next;
+	}
+
+	btree_insert_fixup(tree, &msg->msg_node);
+	return true;
+}
+
+static void kmsg_reply_error(struct kmsg *msg, kern_status_t status)
+{
+	msg->msg_status = KMSG_REPLY_SENT;
+	msg->msg_status = status;
+	thread_awaken(msg->msg_sender_thread);
+}
+
+static struct kmsg *get_next_msg(struct channel *channel)
+{
+	unsigned long flags;
+	struct btree_node *cur = btree_first(&channel->c_msg);
+	while (cur) {
+		struct kmsg *msg = BTREE_CONTAINER(struct kmsg, msg_node, cur);
+		spin_lock_irqsave(&msg->msg_lock, &flags);
+		if (msg->msg_status == KMSG_WAIT_RECEIVE) {
+			msg->msg_status = KMSG_WAIT_REPLY;
+			return msg;
+		}
+
+		spin_unlock_irqrestore(&msg->msg_lock, flags);
+		cur = btree_next(cur);
+	}
+
+	return NULL;
+}
+
+extern kern_status_t channel_enqueue_msg(
+	struct channel *channel,
+	struct kmsg *msg)
+{
+	fill_random(&msg->msg_id, sizeof msg->msg_id);
+	while (!try_enqueue(&channel->c_msg, msg)) {
+		msg->msg_id++;
+	}
+
+	wakeup_one(&channel->c_wq);
+
+	return KERN_OK;
+}
+
+extern kern_status_t channel_recv_msg(
+	struct channel *channel,
+	struct msg *out_msg,
+	msgid_t *out_id,
+	unsigned long *irq_flags)
+{
+	struct wait_item waiter;
+	struct thread *self = current_thread();
+	struct kmsg *msg = NULL;
+
+	wait_item_init(&waiter, self);
+	for (;;) {
+		thread_wait_begin(&waiter, &channel->c_wq);
+		msg = get_next_msg(channel);
+		if (msg) {
+			break;
+		}
+
+		object_unlock_irqrestore(&channel->c_base, *irq_flags);
+		schedule(SCHED_NORMAL);
+		object_lock_irqsave(&channel->c_base, irq_flags);
+	}
+	thread_wait_end(&waiter, &channel->c_wq);
+
+	/* msg is now set to the next message to process */
+
+	struct task *sender = msg->msg_sender_thread->tr_parent;
+	struct task *receiver = self->tr_parent;
+
+	kern_status_t status = vm_region_memmove_v(
+		receiver->t_address_space,
+		0,
+		out_msg->msg_data,
+		out_msg->msg_data_count,
+		sender->t_address_space,
+		0,
+		msg->msg_req->msg_data,
+		msg->msg_req->msg_data_count,
+		VM_REGION_COPY_ALL);
+	if (status != KERN_OK) {
+		kmsg_reply_error(msg, status);
+		return status;
+	}
+
+	status = handle_list_transfer(
+		receiver->t_handles,
+		out_msg->msg_handles,
+		out_msg->msg_handles_count,
+		sender->t_handles,
+		msg->msg_req->msg_handles,
+		msg->msg_req->msg_handles_count);
+	if (status != KERN_OK) {
+		kmsg_reply_error(msg, status);
+		return status;
+	}
+
+	kmsg_reply_error(msg, KERN_OK);
+	return KERN_OK;
+}
+
+extern kern_status_t channel_reply_msg(
+	struct channel *channel,
+	msgid_t id,
+	const struct msg *resp,
+	unsigned long *irq_flags)
+{
+	struct kmsg *msg = get_msg_with_id(&channel->c_msg, id);
+	if (!msg || msg->msg_status != KMSG_WAIT_REPLY) {
+		return KERN_INVALID_ARGUMENT;
+	}
+
+	struct thread *self = current_thread();
+	struct task *sender = msg->msg_sender_thread->tr_parent;
+	struct task *receiver = self->tr_parent;
+
+	kern_status_t status = vm_region_memmove_v(
+		receiver->t_address_space,
+		0,
+		msg->msg_resp->msg_data,
+		msg->msg_resp->msg_data_count,
+		sender->t_address_space,
+		0,
+		resp->msg_data,
+		resp->msg_data_count,
+		VM_REGION_COPY_ALL);
+	if (status != KERN_OK) {
+		kmsg_reply_error(msg, status);
+		return status;
+	}
+
+	status = handle_list_transfer(
+		receiver->t_handles,
+		msg->msg_resp->msg_handles,
+		msg->msg_resp->msg_handles_count,
+		sender->t_handles,
+		resp->msg_handles,
+		resp->msg_handles_count);
+	if (status != KERN_OK) {
+		kmsg_reply_error(msg, status);
+		return status;
+	}
+
+	msg->msg_status = KERN_OK;
+	msg->msg_status = KMSG_REPLY_SENT;
+
+	return KERN_UNIMPLEMENTED;
+}
+
+extern kern_status_t channel_read_msg(
+	struct channel *channel,
+	msgid_t msg,
+	size_t offset,
+	void *buf,
+	size_t len,
+	size_t *nr_read)
+{
+	return KERN_UNIMPLEMENTED;
+}
+
+extern kern_status_t channel_write_msg(
+	struct channel *channel,
+	msgid_t msg,
+	size_t offset,
+	const void *buf,
+	size_t len,
+	size_t *nr_written)
+{
+	return KERN_UNIMPLEMENTED;
+}

kernel/port.c

@@ -0,0 +1,78 @@
+#include <kernel/channel.h>
+#include <kernel/port.h>
+#include <kernel/util.h>
+
+#define PORT_CAST(p) OBJECT_C_CAST(struct port, p_base, &port_type, p)
+
+static struct object_type port_type = {
+	.ob_name = "port",
+	.ob_size = sizeof(struct port),
+	.ob_header_offset = offsetof(struct port, p_base),
+};
+
+kern_status_t port_type_init(void)
+{
+	return object_type_register(&port_type);
+}
+
+struct port *port_cast(struct object *obj)
+{
+	return PORT_CAST(obj);
+}
+
+static void wait_for_reply(struct port *port)
+{
+}
+
+struct port *port_create(void)
+{
+	struct object *port_object = object_create(&port_type);
+	if (!port_object) {
+		return NULL;
+	}
+
+	struct port *port = PORT_CAST(port_object);
+
+	port->p_status = PORT_OFFLINE;
+
+	return port;
+}
+
+kern_status_t port_connect(struct port *port, struct channel *remote)
+{
+	if (port->p_status != PORT_OFFLINE) {
+		return KERN_BAD_STATE;
+	}
+
+	port->p_remote = remote;
+	port->p_status = PORT_READY;
+	return KERN_OK;
+}
+
+kern_status_t port_send_msg(
+	struct port *port,
+	const struct msg *req,
+	struct msg *resp)
+{
+	if (port->p_status != PORT_READY) {
+		return KERN_BAD_STATE;
+	}
+
+	struct thread *self = current_thread();
+	struct kmsg *msg = &self->tr_msg;
+	msg->msg_sender_thread = self;
+	msg->msg_sender_port = port;
+	msg->msg_req = req;
+	msg->msg_resp = resp;
+
+	unsigned long flags;
+	channel_lock_irqsave(port->p_remote, &flags);
+	channel_enqueue_msg(port->p_remote, msg);
+	channel_unlock_irqrestore(port->p_remote, flags);
+
+	port->p_status = PORT_SEND_BLOCKED;
+
+	wait_for_reply(port);
+
+	return msg->msg_result;
+}