[kernel] Make channels stream based

Multiple changes regarding channels. Mainly channels are now stream
based and can handle partial reads or writes. Channels now use the
kernel buffers area with the related buffer_cache. Added a fake stdout
stream channel and kernel task to read its contents to the screen in
preparation for handing channels as stdin/stdout to processes.

src/kernel/main.cpp

@@ -17,6 +17,7 @@
 #include "kernel_args.h"
 #include "kernel_memory.h"
 #include "log.h"
+#include "objects/channel.h"
 #include "objects/event.h"
 #include "objects/handle.h"
 #include "page_manager.h"
@@ -49,28 +50,6 @@ void memory_initialize_post_ctors(kernel::args::header *kargs);
 
 using namespace kernel;
 
-/*
-class test_observer :
-	public kobject::observer
-{
-public:
-	test_observer(const char *name) : m_name(name) {}
-
-	virtual bool on_signals_changed(
-			kobject *obj,
-			j6_signal_t s,
-			j6_signal_t ds,
-			j6_status_t result)
-	{
-		log::info(logs::objs, "  %s: Signals %016lx changed, object %p, result %016lx",
-				m_name, ds, obj, result);
-		return false;
-	}
-
-	const char *m_name;
-};
-*/
-
 void
 init_console()
 {
@@ -84,6 +63,40 @@ init_console()
 	logger_init();
 }
 
+channel *std_out = nullptr;
+
+void
+stdout_task()
+{
+	uint8_t buffer[257];
+	auto *ent = reinterpret_cast<log::logger::entry *>(buffer);
+	auto *cons = console::get();
+
+	log::info(logs::task, "Starting kernel stdout task");
+
+	scheduler &s = scheduler::get();
+	thread *th = thread::from_tcb(s.current());
+
+	while (true) {
+		j6_signal_t current = std_out->signals();
+		if (!(current & j6_signal_channel_can_recv)) {
+			th->wait_on_signals(std_out, j6_signal_channel_can_recv);
+			s.schedule();
+		}
+
+		size_t n = 256;
+		j6_status_t status = std_out->dequeue(&n, buffer);
+		if (status != j6_status_ok) {
+			log::warn(logs::task, "Kernel stdout error: %x", status);
+			return;
+		}
+
+		buffer[n] = 0;
+		cons->puts(reinterpret_cast<const char *>(buffer));
+	}
+}
+
+
 void
 kernel_main(args::header *header)
 {
@@ -190,22 +203,14 @@ kernel_main(args::header *header)
 		}
 	}
 
-	/*
-	log::info(logs::objs, "Testing object system:");
-	test_observer obs1("event");
-	test_observer obs2("no handles");
-	{
-		event e;
-
-		e.register_signal_observer(&obs1, j6_signal_user0);
-		e.register_signal_observer(&obs2, j6_signal_no_handles);
-
-		e.assert_signal(j6_signal_user0);
-
-		handle h(1, 0, &e);
-	}
-	*/
+	std_out = new channel;
 
 	sched->create_kernel_task(logger_task, scheduler::max_priority-1, true);
+	sched->create_kernel_task(stdout_task, scheduler::max_priority-1, true);
+
+	const char stdout_message[] = "Hello on the fake stdout channel\n";
+	size_t message_size = sizeof(stdout_message);
+	std_out->enqueue(&message_size, reinterpret_cast<const void*>(stdout_message));
+
 	sched->start();
 }

src/kernel/memory_bootstrap.cpp

@@ -16,6 +16,7 @@ using memory::frame_size;
 using memory::heap_start;
 using memory::kernel_max_heap;
 using memory::kernel_offset;
+using memory::heap_start;
 using memory::page_offset;
 using memory::pml4e_kernel;
 using memory::pml4e_offset;
@@ -23,7 +24,7 @@ using memory::table_entries;
 
 using namespace kernel;
 
-kutil::vm_space g_kernel_space {kernel_offset, (page_offset-kernel_offset)};
+kutil::vm_space g_kernel_space {kernel_offset, (heap_start-kernel_offset)};
 
 
 // These objects are initialized _before_ global constructors are called,

src/kernel/objects/channel.cpp

@@ -1,40 +1,49 @@
 #include "kutil/assert.h"
 
+#include "buffer_cache.h"
+#include "kernel_memory.h"
 #include "objects/channel.h"
 
+using memory::frame_size;
+using memory::kernel_buffer_pages;
+static constexpr size_t buffer_bytes = kernel_buffer_pages * frame_size;
+
 channel::channel() :
 	m_len(0),
-	m_capacity(0),
-	m_data(nullptr),
+	m_data(g_kbuffer_cache.get_buffer()),
+	m_buffer(reinterpret_cast<uint8_t*>(m_data), buffer_bytes),
 	kobject(kobject::type::channel, j6_signal_channel_can_send)
 {
 }
 
 channel::~channel()
 {
-	kutil::kfree(m_data);
+	if (!closed()) close();
 }
 
 j6_status_t
-channel::enqueue(size_t len, void *data)
+channel::enqueue(size_t *len, const void *data)
 {
 	// TODO: Make this thread safe!
 	if (closed())
 		return j6_status_closed;
 
-	if (!can_send())
+	if (!len || !*len)
+		return j6_err_invalid_arg;
+
+	if (m_buffer.free_space() == 0)
 		return j6_err_not_ready;
 
-	if (m_capacity < len) {
-		kutil::kfree(m_data);
-		m_data = kutil::kalloc(len);
-		m_capacity = len;
-		kassert(m_data, "Failed to allocate memory to copy channel message");
-	}
+	void *buffer = nullptr;
+	size_t avail = m_buffer.reserve(*len, &buffer);
+	*len = *len > avail ? avail : *len;
+
+	kutil::memcpy(buffer, data, *len);
+	m_buffer.commit(*len);
 
-	m_len = len;
-	kutil::memcpy(m_data, data, len);
 	assert_signal(j6_signal_channel_can_recv);
+	if (m_buffer.free_space() == 0)
+		deassert_signal(j6_signal_channel_can_send);
 
 	return j6_status_ok;
 }
@@ -46,22 +55,33 @@ channel::dequeue(size_t *len, void *data)
 	if (closed())
 		return j6_status_closed;
 
-	if (!can_receive())
-		return j6_err_not_ready;
-
-	if (!len)
+	if (!len || !*len)
 		return j6_err_invalid_arg;
 
-	if (*len < m_len)
-		return j6_err_insufficient;
+	if (m_buffer.size() == 0)
+		return j6_err_not_ready;
+
+	void *buffer = nullptr;
+	size_t avail = m_buffer.get_block(&buffer);
+	*len = *len > avail ? avail : *len;
+
+	kutil::memcpy(data, buffer, *len);
+	m_buffer.consume(*len);
 
-	kutil::memcpy(data, m_data, m_len);
-	*len = m_len;
 	assert_signal(j6_signal_channel_can_send);
+	if (m_buffer.size() == 0)
+		deassert_signal(j6_signal_channel_can_recv);
 
 	return j6_status_ok;
 }
 
+void
+channel::close()
+{
+	g_kbuffer_cache.return_buffer(m_data);
+	assert_signal(j6_signal_channel_closed);
+}
+
 void
 channel::on_no_handles()
 {

src/kernel/objects/channel.h

@@ -3,6 +3,7 @@
 /// Definition of channel objects and related functions
 
 #include "j6/signals.h"
+#include "kutil/bip_buffer.h"
 #include "objects/kobject.h"
 
 /// Channels are bi-directional means of sending messages
@@ -20,10 +21,10 @@ public:
 	inline bool can_receive() const { return check_signal(j6_signal_channel_can_recv); }
 
 	/// Put a message into the channel
-	/// \arg len  Length of data, in bytes
+	/// \arg len  [in] Bytes in data buffer [out] number of bytes written
 	/// \arg data Pointer to the message data
 	/// \returns  j6_status_ok on success
-	j6_status_t enqueue(size_t len, void *data);
+	j6_status_t enqueue(size_t *len, const void *data);
 
 	/// Get a message from the channel, copied into a provided buffer
 	/// \arg len  On input, the size of the provided buffer. On output,
@@ -34,7 +35,7 @@ public:
 
 	/// Mark this channel as closed, all future calls to enqueue or
 	/// dequeue messages will fail with j6_status_closed.
-	inline void close() { assert_signal(j6_signal_channel_closed); }
+	void close();
 
 	/// Check if this channel has been closed
 	inline bool closed() { return check_signal(j6_signal_channel_closed); }
@@ -44,6 +45,6 @@ protected:
 
 private:
 	size_t m_len;
-	size_t m_capacity;
-	void *m_data;
+	uintptr_t m_data;
+	kutil::bip_buffer m_buffer;
 };

src/kernel/syscalls/channel.cpp

@@ -42,7 +42,7 @@ channel_close(j6_handle_t handle)
 }
 
 j6_status_t
-channel_send(j6_handle_t handle, size_t len, void *data)
+channel_send(j6_handle_t handle, size_t *len, void *data)
 {
 	scheduler &s = scheduler::get();
 	TCB *tcb = s.current();