Kernel: Semaphores and Threads can now be blocked with timeout

kernel/include/kernel/Scheduler.h

@@ -19,9 +19,9 @@ namespace Kernel
 		void reschedule();
 		void reschedule_if_idling();
 
-		void set_current_thread_sleeping(uint64_t);
+		void set_current_thread_sleeping(uint64_t wake_time);
 
-		void block_current_thread(Semaphore*);
+		void block_current_thread(Semaphore*, uint64_t wake_time);
 		void unblock_threads(Semaphore*);
 		// Makes sleeping or blocked thread with tid active.
 		void unblock_thread(pid_t tid);
@@ -36,6 +36,8 @@ namespace Kernel
 	private:
 		Scheduler() = default;
 
+		void set_current_thread_sleeping_impl(uint64_t wake_time);
+
 		void wake_threads();
 		[[nodiscard]] bool save_current_thread();
 		void remove_and_advance_current_thread();
@@ -50,18 +52,14 @@ namespace Kernel
 				: thread(thread)
 			{}
 
-			Thread* thread;
-			union
-			{
-				uint64_t	wake_time;
-				Semaphore*	semaphore;
-			};
+			Thread*		thread;
+			uint64_t	wake_time;
+			Semaphore*	semaphore;
 		};
 
 		Thread* m_idle_thread { nullptr };
 		BAN::LinkedList<SchedulerThread> m_active_threads;
 		BAN::LinkedList<SchedulerThread> m_sleeping_threads;
-		BAN::LinkedList<SchedulerThread> m_blocking_threads;
 
 		BAN::LinkedList<SchedulerThread>::iterator m_current_thread;
 

kernel/include/kernel/Semaphore.h

@@ -6,7 +6,9 @@ namespace Kernel
 	class Semaphore
 	{
 	public:
-		void block();
+		void block_indefinite();
+		void block_with_timeout(uint64_t timeout_ms);
+		void block_with_wake_time(uint64_t wake_time_ms);
 		void unblock();
 	};
 

kernel/include/kernel/Thread.h

@@ -47,8 +47,10 @@ namespace Kernel
 		void handle_signal(int signal = 0);
 		bool add_signal(int signal);
 
-		// blocks semaphore and returns either on unblock, eintr or spuriously
-		BAN::ErrorOr<void> block_or_eintr(Semaphore&);
+		// blocks semaphore and returns either on unblock, eintr, spuriously or after timeout
+		BAN::ErrorOr<void> block_or_eintr_indefinite(Semaphore& semaphore);
+		BAN::ErrorOr<void> block_or_eintr_or_timeout(Semaphore& semaphore, uint64_t timeout_ms, bool etimedout);
+		BAN::ErrorOr<void> block_or_eintr_or_waketime(Semaphore& semaphore, uint64_t wake_time_ms, bool etimedout);
 
 		void set_return_rsp(uintptr_t& rsp) { m_return_rsp = &rsp; }
 		void set_return_rip(uintptr_t& rip) { m_return_rip = &rip; }

kernel/kernel/FS/DevFS/FileSystem.cpp

@@ -64,9 +64,8 @@ namespace Kernel
 				{
 					while (!s_instance->m_should_sync)
 					{
-						s_instance->m_device_lock.unlock();
-						s_instance->m_sync_semaphore.block();
-						s_instance->m_device_lock.lock();
+						LockFreeGuard _(s_instance->m_device_lock);
+						s_instance->m_sync_semaphore.block_indefinite();
 					}
 
 					for (auto& device : s_instance->m_devices)
@@ -105,7 +104,7 @@ namespace Kernel
 			m_sync_semaphore.unblock();
 		}
 		if (should_block)
-			m_sync_done.block();
+			m_sync_done.block_indefinite();
 	}
 
 	void DevFileSystem::add_device(BAN::RefPtr<Device> device)

kernel/kernel/FS/Pipe.cpp

@@ -1,5 +1,6 @@
 #include <kernel/FS/Pipe.h>
 #include <kernel/LockGuard.h>
+#include <kernel/Thread.h>
 #include <kernel/Timer/Timer.h>
 
 namespace Kernel
@@ -46,9 +47,8 @@ namespace Kernel
 		{
 			if (m_writing_count == 0)
 				return 0;
-			m_lock.unlock();
-			m_semaphore.block();
-			m_lock.lock();
+			LockFreeGuard lock_free(m_lock);
+			TRY(Thread::current().block_or_eintr_indefinite(m_semaphore));
 		}
 
 		size_t to_copy = BAN::Math::min<size_t>(buffer.size(), m_buffer.size());

kernel/kernel/Input/PS2/Keyboard.cpp

@@ -195,7 +195,7 @@ namespace Kernel::Input
 		while (true)
 		{
 			if (m_event_queue.empty())
-				TRY(Thread::current().block_or_eintr(m_semaphore));
+				TRY(Thread::current().block_or_eintr_indefinite(m_semaphore));
 
 			CriticalScope _;
 			if (m_event_queue.empty())

kernel/kernel/Input/PS2/Mouse.cpp

@@ -179,7 +179,7 @@ namespace Kernel::Input
 		while (true)
 		{
 			if (m_event_queue.empty())
-				TRY(Thread::current().block_or_eintr(m_semaphore));
+				TRY(Thread::current().block_or_eintr_indefinite(m_semaphore));
 
 			CriticalScope _;
 			if (m_event_queue.empty())

kernel/kernel/Networking/ARPTable.cpp

@@ -155,7 +155,7 @@ namespace Kernel
 
 			if (!pending.has_value())
 			{
-				m_pending_semaphore.block();
+				m_pending_semaphore.block_indefinite();
 				continue;
 			}
 

kernel/kernel/Networking/IPv4Layer.cpp

@@ -239,7 +239,7 @@ namespace Kernel
 
 			if (!pending.has_value())
 			{
-				m_pending_semaphore.block();
+				m_pending_semaphore.block_indefinite();
 				continue;
 			}
 

kernel/kernel/Networking/UDPSocket.cpp

@@ -65,7 +65,7 @@ namespace Kernel
 	BAN::ErrorOr<size_t> UDPSocket::read_packet(BAN::ByteSpan buffer, sockaddr_in* sender_addr)
 	{
 		while (m_packets.empty())
-			TRY(Thread::current().block_or_eintr(m_packet_semaphore));
+			TRY(Thread::current().block_or_eintr_indefinite(m_packet_semaphore));
 
 		LockGuard _(m_packet_lock);
 		if (m_packets.empty())

kernel/kernel/Networking/UNIX/Socket.cpp

@@ -66,7 +66,7 @@ namespace Kernel
 			return BAN::Error::from_errno(EINVAL);
 
 		while (connection_info.pending_connections.empty())
-			TRY(Thread::current().block_or_eintr(connection_info.pending_semaphore));
+			TRY(Thread::current().block_or_eintr_indefinite(connection_info.pending_semaphore));
 
 		BAN::RefPtr<UnixDomainSocket> pending;
 
@@ -158,7 +158,7 @@ namespace Kernel
 					break;
 				}
 			}
-			TRY(Thread::current().block_or_eintr(target_info.pending_semaphore));
+			TRY(Thread::current().block_or_eintr_indefinite(target_info.pending_semaphore));
 		}
 
 		while (!connection_info.connection_done)
@@ -234,7 +234,7 @@ namespace Kernel
 		while (m_packet_sizes.full() || m_packet_size_total + packet.size() > s_packet_buffer_size)
 		{
 			LockFreeGuard _(m_lock);
-			TRY(Thread::current().block_or_eintr(m_packet_semaphore));
+			TRY(Thread::current().block_or_eintr_indefinite(m_packet_semaphore));
 		}
 
 		uint8_t* packet_buffer = reinterpret_cast<uint8_t*>(m_packet_buffer->vaddr() + m_packet_size_total);
@@ -321,7 +321,7 @@ namespace Kernel
 		while (m_packet_size_total == 0)
 		{
 			LockFreeGuard _(m_lock);
-			TRY(Thread::current().block_or_eintr(m_packet_semaphore));
+			TRY(Thread::current().block_or_eintr_indefinite(m_packet_semaphore));
 		}
 
 		uint8_t* packet_buffer = reinterpret_cast<uint8_t*>(m_packet_buffer->vaddr());

kernel/kernel/Process.cpp

@@ -568,7 +568,7 @@ namespace Kernel
 			return BAN::Error::from_errno(ECHILD);
 
 		while (!target->m_exit_status.exited)
-			TRY(Thread::current().block_or_eintr(target->m_exit_status.semaphore));
+			TRY(Thread::current().block_or_eintr_indefinite(target->m_exit_status.semaphore));
 
 		int exit_status = target->m_exit_status.exit_code;
 		target->m_exit_status.waiting--;

kernel/kernel/Scheduler.cpp

@@ -278,12 +278,9 @@ namespace Kernel
 		ASSERT_NOT_REACHED();
 	}
 
-	void Scheduler::set_current_thread_sleeping(uint64_t wake_time)
+	void Scheduler::set_current_thread_sleeping_impl(uint64_t wake_time)
 	{
-		VERIFY_STI();
-		DISABLE_INTERRUPTS();
-
-		ASSERT(m_current_thread);
+		VERIFY_CLI();
 
 		if (save_current_thread())
 		{
@@ -310,42 +307,37 @@ namespace Kernel
 		ASSERT_NOT_REACHED();
 	}
 
-	void Scheduler::block_current_thread(Semaphore* semaphore)
+	void Scheduler::set_current_thread_sleeping(uint64_t wake_time)
 	{
 		VERIFY_STI();
 		DISABLE_INTERRUPTS();
 
 		ASSERT(m_current_thread);
 
-		if (save_current_thread())
-		{
-			ENABLE_INTERRUPTS();
-			return;
-		}
+		m_current_thread->semaphore = nullptr;
+		set_current_thread_sleeping_impl(wake_time);
+	}
+
+	void Scheduler::block_current_thread(Semaphore* semaphore, uint64_t wake_time)
+	{
+		VERIFY_STI();
+		DISABLE_INTERRUPTS();
+
+		ASSERT(m_current_thread);
 
 		m_current_thread->semaphore = semaphore;
-		m_active_threads.move_element_to_other_linked_list(
-			m_blocking_threads,
-			m_blocking_threads.end(),
-			m_current_thread
-		);
-
-		m_current_thread = {};
-		advance_current_thread();
-
-		execute_current_thread();
-		ASSERT_NOT_REACHED();
+		set_current_thread_sleeping_impl(wake_time);
 	}
 
 	void Scheduler::unblock_threads(Semaphore* semaphore)
 	{
 		CriticalScope critical;
 
-		for (auto it = m_blocking_threads.begin(); it != m_blocking_threads.end();)
+		for (auto it = m_sleeping_threads.begin(); it != m_sleeping_threads.end();)
 		{
 			if (it->semaphore == semaphore)
 			{
-				it = m_blocking_threads.move_element_to_other_linked_list(
+				it = m_sleeping_threads.move_element_to_other_linked_list(
 					m_active_threads,
 					m_active_threads.end(),
 					it
@@ -362,19 +354,6 @@ namespace Kernel
 	{
 		CriticalScope _;
 
-		for (auto it = m_blocking_threads.begin(); it != m_blocking_threads.end(); it++)
-		{
-			if (it->thread->tid() == tid)
-			{
-				m_blocking_threads.move_element_to_other_linked_list(
-					m_active_threads,
-					m_active_threads.end(),
-					it
-				);
-				return;
-			}
-		}
-
 		for (auto it = m_sleeping_threads.begin(); it != m_sleeping_threads.end(); it++)
 		{
 			if (it->thread->tid() == tid)

kernel/kernel/Semaphore.cpp

@@ -1,12 +1,23 @@
 #include <kernel/Scheduler.h>
 #include <kernel/Semaphore.h>
+#include <kernel/Timer/Timer.h>
 
 namespace Kernel
 {
 
-	void Semaphore::block()
+	void Semaphore::block_indefinite()
 	{
-		Scheduler::get().block_current_thread(this);
+		Scheduler::get().block_current_thread(this, ~(uint64_t)0);
+	}
+
+	void Semaphore::block_with_timeout(uint64_t timeout_ms)
+	{
+		Scheduler::get().block_current_thread(this, SystemTimer::get().ms_since_boot() + timeout_ms);
+	}
+
+	void Semaphore::block_with_wake_time(uint64_t wake_time)
+	{
+		Scheduler::get().block_current_thread(this, wake_time);
 	}
 
 	void Semaphore::unblock()

kernel/kernel/Storage/NVMe/Queue.cpp

@@ -67,13 +67,12 @@ namespace Kernel
 
 		while (SystemTimer::get().ms_since_boot() < start_time + s_nvme_command_timeout_ms)
 		{
-			if (!m_done)
+			if (m_done)
 			{
-				m_semaphore.block();
-				continue;
+				m_done = false;
+				return m_status;
 			}
-			m_done = false;
-			return m_status;
+			m_semaphore.block_with_wake_time(start_time + s_nvme_command_timeout_ms);
 		}
 
 		return 0xFFFF;

kernel/kernel/Terminal/TTY.cpp

@@ -92,7 +92,7 @@ namespace Kernel
 				while (true)
 				{
 					while (!TTY::current()->m_tty_ctrl.receive_input)
-						TTY::current()->m_tty_ctrl.semaphore.block();
+						TTY::current()->m_tty_ctrl.semaphore.block_indefinite();
 
 					Input::KeyEvent event;
 					size_t read = MUST(inode->read(0, BAN::ByteSpan::from(event)));
@@ -323,7 +323,7 @@ namespace Kernel
 			uint32_t depth = m_lock.lock_depth();
 			for (uint32_t i = 0; i < depth; i++)
 				m_lock.unlock();
-			auto eintr = Thread::current().block_or_eintr(m_output.semaphore);
+			auto eintr = Thread::current().block_or_eintr_indefinite(m_output.semaphore);
 			for (uint32_t i = 0; i < depth; i++)
 				m_lock.lock();
 			if (eintr.is_error())

kernel/kernel/Thread.cpp

@@ -8,6 +8,7 @@
 #include <kernel/Process.h>
 #include <kernel/Scheduler.h>
 #include <kernel/Thread.h>
+#include <kernel/Timer/Timer.h>
 
 namespace Kernel
 {
@@ -30,7 +31,6 @@ namespace Kernel
 	void Thread::terminate()
 	{
 		CriticalScope _;
-		ASSERT(this == &Thread::current());
 		m_state = Thread::State::Terminated;
 		if (this == &Thread::current())
 			Scheduler::get().execute_current_thread();
@@ -343,16 +343,34 @@ namespace Kernel
 		return false;
 	}
 
-	BAN::ErrorOr<void> Thread::block_or_eintr(Semaphore& semaphore)
+	BAN::ErrorOr<void> Thread::block_or_eintr_indefinite(Semaphore& semaphore)
 	{
 		if (is_interrupted_by_signal())
 			return BAN::Error::from_errno(EINTR);
-		semaphore.block();
+		semaphore.block_indefinite();
 		if (is_interrupted_by_signal())
 			return BAN::Error::from_errno(EINTR);
 		return {};
 	}
 
+	BAN::ErrorOr<void> Thread::block_or_eintr_or_timeout(Semaphore& semaphore, uint64_t timeout_ms, bool etimedout)
+	{
+		uint64_t wake_time_ms = SystemTimer::get().ms_since_boot() + timeout_ms;
+		return block_or_eintr_or_waketime(semaphore, wake_time_ms, etimedout);
+	}
+
+	BAN::ErrorOr<void> Thread::block_or_eintr_or_waketime(Semaphore& semaphore, uint64_t wake_time_ms, bool etimedout)
+	{
+		if (is_interrupted_by_signal())
+			return BAN::Error::from_errno(EINTR);
+		semaphore.block_with_wake_time(wake_time_ms);
+		if (is_interrupted_by_signal())
+			return BAN::Error::from_errno(EINTR);
+		if (etimedout && SystemTimer::get().ms_since_boot() >= wake_time_ms)
+			return BAN::Error::from_errno(ETIMEDOUT);
+		return {};
+	}
+
 	void Thread::validate_stack() const
 	{
 		if (stack_base() <= m_rsp && m_rsp <= stack_base() + stack_size())