Kernel: Rewrite pipes

Pipes have now a fixed size buffer and pipe clone and close is working
again.

kernel/include/kernel/FS/Pipe.h

@@ -1,5 +1,6 @@
 #pragma once
 
+#include <BAN/Array.h>
 #include <kernel/FS/Inode.h>
 #include <kernel/ThreadBlocker.h>
 
@@ -33,7 +34,7 @@ namespace Kernel
 		virtual BAN::ErrorOr<size_t> read_impl(off_t, BAN::ByteSpan) override;
 		virtual BAN::ErrorOr<size_t> write_impl(off_t, BAN::ConstByteSpan) override;
 
-		virtual bool can_read_impl() const override { return !m_buffer.empty(); }
+		virtual bool can_read_impl() const override { return m_buffer_size == 0; }
 		virtual bool can_write_impl() const override { return true; }
 		virtual bool has_error_impl() const override { return false; }
 
@@ -46,10 +47,13 @@ namespace Kernel
 		timespec m_atime {};
 		timespec m_mtime {};
 		timespec m_ctime {};
-		BAN::Vector<uint8_t> m_buffer;
 		ThreadBlocker m_thread_blocker;
 
-		uint32_t m_writing_count { 1 };
+		BAN::Array<uint8_t, PAGE_SIZE> m_buffer;
+		BAN::Atomic<size_t> m_buffer_size { 0 };
+		size_t m_buffer_tail { 0 };
+
+		BAN::Atomic<uint32_t> m_writing_count { 1 };
 	};
 
 }

kernel/kernel/FS/Pipe.cpp

@@ -3,6 +3,8 @@
 #include <kernel/Thread.h>
 #include <kernel/Timer/Timer.h>
 
+#include <kernel/Process.h>
+
 namespace Kernel
 {
 
@@ -26,24 +28,23 @@ namespace Kernel
 
 	void Pipe::clone_writing()
 	{
-		LockGuard _(m_mutex);
+		[[maybe_unused]] auto old_writing_count = m_writing_count.fetch_add(1);
-		ASSERT(m_writing_count > 0);
+		ASSERT(old_writing_count > 0);
-		m_writing_count++;
 	}
 
 	void Pipe::close_writing()
 	{
-		LockGuard _(m_mutex);
+		auto old_writing_count = m_writing_count.fetch_sub(1);
-		ASSERT(m_writing_count > 0);
+		ASSERT(old_writing_count > 0);
-		m_writing_count--;
+		if (old_writing_count == 1)
-		if (m_writing_count == 0)
 			m_thread_blocker.unblock();
 	}
 
 	BAN::ErrorOr<size_t> Pipe::read_impl(off_t, BAN::ByteSpan buffer)
 	{
 		LockGuard _(m_mutex);
-		while (m_buffer.empty())
+
+		while (m_buffer_size == 0)
 		{
 			if (m_writing_count == 0)
 				return 0;
@@ -51,11 +52,20 @@ namespace Kernel
 			TRY(Thread::current().block_or_eintr_indefinite(m_thread_blocker));
 		}
 
-		size_t to_copy = BAN::Math::min<size_t>(buffer.size(), m_buffer.size());
+		const size_t to_copy = BAN::Math::min<size_t>(buffer.size(), m_buffer_size);
-		memcpy(buffer.data(), m_buffer.data(), to_copy);
 
-		memmove(m_buffer.data(), m_buffer.data() + to_copy, m_buffer.size() - to_copy);
+		if (m_buffer_tail + to_copy <= m_buffer.size())
-		MUST(m_buffer.resize(m_buffer.size() - to_copy));
+			memcpy(buffer.data(), m_buffer.data() + m_buffer_tail, to_copy);
+		else
+		{
+			const size_t before_wrap = m_buffer.size() - m_buffer_tail;
+			const size_t after_wrap = to_copy - before_wrap;
+			memcpy(buffer.data(), m_buffer.data() + m_buffer_tail, before_wrap);
+			memcpy(buffer.data() + before_wrap, m_buffer.data(), after_wrap);
+		}
+
+		m_buffer_tail = (m_buffer_tail + to_copy) % m_buffer.size();
+		m_buffer_size -= to_copy;
 
 		m_atime = SystemTimer::get().real_time();
 
@@ -68,10 +78,29 @@ namespace Kernel
 	{
 		LockGuard _(m_mutex);
 
-		size_t old_size = m_buffer.size();
+		if (buffer.size() > m_buffer.size())
+			buffer = buffer.slice(0, m_buffer.size());
 
-		TRY(m_buffer.resize(old_size + buffer.size()));
+		while (m_buffer.size() - m_buffer_size < buffer.size())
-		memcpy(m_buffer.data() + old_size, buffer.data(), buffer.size());
+		{
+			LockFreeGuard lock_free(m_mutex);
+			TRY(Thread::current().block_or_eintr_indefinite(m_thread_blocker));
+		}
+
+		const size_t to_copy = buffer.size();
+		const size_t buffer_head = (m_buffer_tail + m_buffer_size) % m_buffer.size();
+
+		if (buffer_head + to_copy <= m_buffer.size())
+			memcpy(m_buffer.data() + buffer_head, buffer.data(), to_copy);
+		else
+		{
+			const size_t before_wrap = m_buffer.size() - buffer_head;
+			const size_t after_wrap = to_copy - before_wrap;
+			memcpy(m_buffer.data() + buffer_head, buffer.data(), before_wrap);
+			memcpy(m_buffer.data(), buffer.data() + before_wrap, after_wrap);
+		}
+
+		m_buffer_size += to_copy;
 
 		timespec current_time = SystemTimer::get().real_time();
 		m_mtime = current_time;
@@ -79,7 +108,7 @@ namespace Kernel
 
 		m_thread_blocker.unblock();
 
-		return buffer.size();
+		return to_copy;
 	}
 
 }

kernel/kernel/OpenFileDescriptorSet.cpp

@@ -48,8 +48,11 @@ namespace Kernel
 
 			m_open_files[fd] = result.release_value();
 
-			if (m_open_files[fd]->flags & O_WRONLY && m_open_files[fd]->inode->is_pipe())
+			if (m_open_files[fd]->path == "<pipe wr>"_sv)
-				((Pipe*)m_open_files[fd]->inode.ptr())->clone_writing();
+			{
+				ASSERT(m_open_files[fd]->inode->is_pipe());
+				static_cast<Pipe*>(m_open_files[fd]->inode.ptr())->clone_writing();
+			}
 		}
 
 		return {};
@@ -59,7 +62,7 @@ namespace Kernel
 	{
 		ASSERT(file.inode);
 
-		if (flags & ~(O_ACCMODE | O_NOFOLLOW | O_APPEND | O_TRUNC | O_CLOEXEC | O_TTY_INIT | O_DIRECTORY | O_CREAT | O_EXCL | O_NONBLOCK))
+		if (flags & ~(O_ACCMODE | O_NOFOLLOW | O_APPEND | O_TRUNC | O_CLOEXEC | O_TTY_INIT | O_NOCTTY | O_DIRECTORY | O_CREAT | O_EXCL | O_NONBLOCK))
 			return BAN::Error::from_errno(ENOTSUP);
 
 		if ((flags & O_ACCMODE) != O_RDWR && __builtin_popcount(flags & O_ACCMODE) != 1)
@@ -139,8 +142,8 @@ namespace Kernel
 		TRY(get_free_fd_pair(fds));
 
 		auto pipe = TRY(Pipe::create(m_credentials));
-		m_open_files[fds[0]] = TRY(BAN::RefPtr<OpenFileDescription>::create(pipe, ""_sv, 0, O_RDONLY));
+		m_open_files[fds[0]] = TRY(BAN::RefPtr<OpenFileDescription>::create(pipe, "<pipe rd>"_sv, 0, O_RDONLY));
-		m_open_files[fds[1]] = TRY(BAN::RefPtr<OpenFileDescription>::create(pipe, ""_sv, 0, O_WRONLY));
+		m_open_files[fds[1]] = TRY(BAN::RefPtr<OpenFileDescription>::create(pipe, "<pipe wr>"_sv, 0, O_WRONLY));
 
 		return {};
 	}
@@ -152,8 +155,11 @@ namespace Kernel
 		int result = TRY(get_free_fd());
 		m_open_files[result] = m_open_files[fildes];
 
-		if (m_open_files[result]->flags & O_WRONLY && m_open_files[result]->inode->is_pipe())
+		if (m_open_files[result]->path == "<pipe wr>"_sv)
-			((Pipe*)m_open_files[result]->inode.ptr())->clone_writing();
+		{
+			ASSERT(m_open_files[result]->inode->is_pipe());
+			static_cast<Pipe*>(m_open_files[result]->inode.ptr())->clone_writing();
+		}
 
 		return result;
 	}
@@ -172,8 +178,11 @@ namespace Kernel
 		m_open_files[fildes2] = m_open_files[fildes];
 		m_open_files[fildes2]->flags &= ~O_CLOEXEC;
 
-		if (m_open_files[fildes]->flags & O_WRONLY && m_open_files[fildes]->inode->is_pipe())
+		if (m_open_files[fildes2]->path == "<pipe wr>"_sv)
-			((Pipe*)m_open_files[fildes]->inode.ptr())->clone_writing();
+		{
+			ASSERT(m_open_files[fildes2]->inode->is_pipe());
+			static_cast<Pipe*>(m_open_files[fildes2]->inode.ptr())->clone_writing();
+		}
 
 		return fildes;
 	}
@@ -303,8 +312,11 @@ namespace Kernel
 	{
 		TRY(validate_fd(fd));
 
-		if (m_open_files[fd]->flags & O_WRONLY && m_open_files[fd]->inode->is_pipe())
+		if (m_open_files[fd]->path == "<pipe wr>"_sv)
-			((Pipe*)m_open_files[fd]->inode.ptr())->close_writing();
+		{
+			ASSERT(m_open_files[fd]->inode->is_pipe());
+			static_cast<Pipe*>(m_open_files[fd]->inode.ptr())->close_writing();
+		}
 
 		m_open_files[fd].clear();