kernel: implement a generic object signalling system

kernel/channel.c

@@ -5,6 +5,7 @@
 #include <kernel/thread.h>
 #include <kernel/util.h>
 #include <kernel/vm-region.h>
+#include <mango/signal.h>
 
 #define CHANNEL_CAST(p) OBJECT_C_CAST(struct channel, c_base, &channel_type, p)
 
@@ -100,6 +101,7 @@ static struct msg *get_next_msg(
 		if (msg->msg_status == KMSG_WAIT_RECEIVE) {
 			msg->msg_status = KMSG_WAIT_REPLY;
 			msg->msg_sender_port->p_status = PORT_REPLY_BLOCKED;
+			channel->c_msg_waiting--;
 			return msg;
 		}
 
@@ -119,7 +121,8 @@ extern kern_status_t channel_enqueue_msg(
 		msg->msg_id++;
 	}
 
-	wakeup_one(&channel->c_wq);
+	channel->c_msg_waiting++;
+	object_assert_signal(&channel->c_base, CHANNEL_SIGNAL_MSG_RECEIVED);
 
 	return KERN_OK;
 }
@@ -129,11 +132,21 @@ extern kern_status_t channel_recv_msg(
 	kern_msg_t *out_msg,
 	unsigned long *irq_flags)
 {
-	struct wait_item waiter;
 	struct thread *self = current_thread();
 	struct msg *msg = NULL;
 	unsigned long msg_lock_flags;
 
+	msg = get_next_msg(channel, &msg_lock_flags);
+	if (!msg) {
+		return KERN_NO_ENTRY;
+	}
+
+	if (channel->c_msg_waiting == 0) {
+		object_clear_signal(
+			&channel->c_base,
+			CHANNEL_SIGNAL_MSG_RECEIVED);
+	}
+#if 0
 	wait_item_init(&waiter, self);
 	for (;;) {
 		thread_wait_begin(&waiter, &channel->c_wq);
@@ -147,6 +160,7 @@ extern kern_status_t channel_recv_msg(
 		object_lock_irqsave(&channel->c_base, irq_flags);
 	}
 	thread_wait_end(&waiter, &channel->c_wq);
+#endif
 
 	/* msg is now set to the next message to process */
 

kernel/object.c

@@ -226,3 +226,35 @@ struct object *object_header(void *p)
 
 	return obj;
 }
+
+void object_assert_signal(struct object *obj, uint32_t signals)
+{
+	obj->ob_signals |= signals;
+	wakeup_queue(&obj->ob_wq);
+}
+
+void object_clear_signal(struct object *obj, uint32_t signals)
+{
+	obj->ob_signals &= ~signals;
+}
+
+void object_wait_signal(
+	struct object *obj,
+	uint32_t signals,
+	unsigned long *irq_flags)
+{
+	struct thread *self = current_thread();
+	struct wait_item waiter;
+	wait_item_init(&waiter, self);
+	for (;;) {
+		thread_wait_begin(&waiter, &obj->ob_wq);
+		if (obj->ob_signals & signals) {
+			break;
+		}
+
+		object_unlock_irqrestore(obj, *irq_flags);
+		schedule(SCHED_NORMAL);
+		object_lock_irqsave(obj, irq_flags);
+	}
+	thread_wait_end(&waiter, &obj->ob_wq);
+}