justin/jsix
1
1
Fork 0
mirror of https://github.com/justinian/jsix.git synced 2025-12-10 00:14:32 -08:00
Code Issues Packages Projects Releases Wiki Activity

[kernel] Re-design thread blocking

Browse Source 
In preparation for the new mailbox IPC model, blocking threads needed an
overhaul. The `wait_on_*` and `wake_on_*` methods are gone, and the
`block()` and `wake()` calls on threads now pass a value between the
waker and the blocked thread.

As part of this change, the concept of signals on the base kobject class
was removed, along with the queue of blocked threads waiting on any
given object. Signals are now exclusively the domain of the event object
type, and the new wait_queue utility class helps manage waiting threads
when an object does actually need this functionality. In some cases (eg,
logger) an event object is used instead of the lower-level wait_queue.

Since this change has a lot of ramifications, this large commit includes
the following additional changes:

- The j6_object_wait, j6_object_wait_many, and j6_thread_pause syscalls
  have been removed.
- The j6_event_clear syscall has been removed - events are "cleared" by
  reading them now. A new j6_event_wait syscall has been added to read
  events.
- The generic close() method on kobject has been removed.
- The on_no_handles() method on kobject now deletes the object by
  default, and needs to be overridden by classes that should not be.
- The j6_system_bind_irq syscall now takes an event handle, as well as a
  signal that the IRQ should set on the event. IRQs will cause a waiting
  thread to be woken with the appropriate bit set.
- Threads waking due to timeout is simplified to just having a
  wake_timeout() accessor that returns a timestamp.
- The new wait_queue uses util::deque, which caused the disovery of two
  bugs in the deque implementation: empty deques could still have a
  single array allocated and thus return true for empty(), and new
  arrays getting allocated were not being zeroed first.
- Exposed a new erase() method on util::map that takes a node pointer
  instead of a key, skipping lookup.
This commit is contained in:
Justin C. Miller
2022-02-21 19:16:20 -08:00
parent f93d80b8d2
commit f7ae2e2220
40 changed files with 419 additions and 628 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
src/kernel/objects/channel.cpp
View File

@@ -15,7 +15,7 @@ channel::channel() :
m_len(0),
m_data(g_kernel_buffers.get_section()),
m_buffer(reinterpret_cast<uint8_t*>(m_data), buffer_bytes),
kobject(kobject::type::channel, j6_signal_channel_can_send)
kobject(kobject::type::channel)
{
}
@@ -41,10 +41,10 @@ channel::enqueue(const util::buffer &data)
m_buffer.commit(len);
if (len)
assert_signal(j6_signal_channel_can_recv);
m_can_recv = true;
if (m_buffer.free_space() == 0)
deassert_signal(j6_signal_channel_can_send);
m_can_send = false;
return len;
}
@@ -64,10 +64,10 @@ channel::dequeue(util::buffer buffer)
m_buffer.consume(len);
if (len)
assert_signal(j6_signal_channel_can_send);
m_can_send = true;
if (m_buffer.size() == 0)
deassert_signal(j6_signal_channel_can_recv);
m_can_recv = false;
return len;
}
@@ -76,16 +76,8 @@ void
channel::close()
{
util::scoped_lock lock {m_close_lock};
kobject::close();
g_kernel_buffers.return_section(m_data);
}
void
channel::on_no_handles()
{
kobject::on_no_handles();
delete this;
m_closed = true;
}
} // namespace obj

14
src/kernel/objects/channel.h
View File

@@ -3,7 +3,6 @@
/// Definition of channel objects and related functions
#include <j6/caps.h>
#include <j6/signals.h>
#include <util/bip_buffer.h>
#include <util/counted.h>
#include <util/spinlock.h>
@@ -26,10 +25,10 @@ public:
static constexpr kobject::type type = kobject::type::channel;
/// Check if the channel has space for a message to be sent
inline bool can_send() const { return check_signal(j6_signal_channel_can_send); }
inline bool can_send() const { return m_can_send; }
/// Check if the channel has a message wiating already
inline bool can_receive() const { return check_signal(j6_signal_channel_can_recv); }
inline bool can_receive() const { return m_can_recv; }
/// Put a message into the channel
/// \arg data Buffer of data to write
@@ -43,14 +42,17 @@ public:
/// Mark this channel as closed, all future calls to enqueue or
/// dequeue messages will fail with j6_status_closed.
virtual void close() override;
void close();
protected:
virtual void on_no_handles() override;
/// Check if this channel has been closed.
inline bool closed() const { return m_closed; }
private:
size_t m_len;
uintptr_t m_data;
bool m_closed;
bool m_can_send;
bool m_can_recv;
util::bip_buffer m_buffer;
util::spinlock m_close_lock;
};

48
src/kernel/objects/event.cpp Normal file
View File

@@ -0,0 +1,48 @@
#include "objects/event.h"
#include "objects/thread.h"
namespace obj {
event::event() :
kobject {type::event},
m_signals {0}
{}
event::~event() {}
void
event::signal(uint64_t s)
{
uint64_t after = __atomic_or_fetch(&m_signals, s, __ATOMIC_SEQ_CST);
if (after) wake_observer();
}
uint64_t
event::wait()
{
// If events are pending, grab those and return immediately
uint64_t value = read();
if (value)
return value;
// Wait for event::signal() to wake us with a value
thread &current = thread::current();
m_queue.add_thread(&current);
return current.block();
}
void
event::wake_observer()
{
util::scoped_lock lock {m_queue.get_lock()};
thread *t = m_queue.get_next_unlocked();
if (!t) return;
uint64_t value = read();
if (value) {
m_queue.pop_next_unlocked();
t->wake(value);
}
}
} // namespace obj

37
src/kernel/objects/event.h
View File

@@ -2,21 +2,48 @@
/// \file event.h
/// Definition of event kobject types
#include <stdint.h>
#include <j6/caps.h>
#include "objects/kobject.h"
#include "wait_queue.h"
namespace obj {
class thread;
class event :
public kobject
{
public:
/// Capabilities on a newly constructed event handle
constexpr static j6_cap_t creation_caps = 0;
event() :
kobject(type::event) {}
constexpr static j6_cap_t creation_caps = j6_cap_event_all;
static constexpr kobject::type type = kobject::type::event;
event();
~event();
/// Signal the given event(s).
/// \arg s The events to signal, as a bitset
void signal(uint64_t s);
/// Wait for events to be signalled. If events have already
/// been signaled since the last call to wait, return them
/// immediately.
/// \returns The events which have been signalled, as a bitset
uint64_t wait();
private:
/// Read and consume the current signaled events
inline uint64_t read() {
return __atomic_exchange_n(&m_signals, 0, __ATOMIC_SEQ_CST);
}
/// Wake an observer to handle an incoming event
void wake_observer();
uint64_t m_signals;
wait_queue m_queue;
};
} // namespace obj

83
src/kernel/objects/kobject.cpp
View File

@@ -1,5 +1,4 @@
#include <j6/errors.h>
#include <j6/signals.h>
#include <j6/types.h>
#include "assert.h"
@@ -10,17 +9,12 @@ namespace obj {
static j6_koid_t next_koids [static_cast<size_t>(kobject::type::max)] = { 0 };
kobject::kobject(type t, j6_signal_t signals) :
kobject::kobject(type t) :
m_koid(koid_generate(t)),
m_signals(signals),
m_handle_count(0)
{}
kobject::~kobject()
{
for (auto *t : m_blocked_threads)
t->wake_on_result(this, j6_status_destroyed);
}
kobject::~kobject() {}
j6_koid_t
kobject::koid_generate(type t)
@@ -37,81 +31,10 @@ kobject::koid_type(j6_koid_t koid)
return static_cast<type>((koid >> 48) & 0xffffull);
}
j6_signal_t
kobject::assert_signal(j6_signal_t s)
{
j6_signal_t old =
__atomic_fetch_or(&m_signals, s, __ATOMIC_SEQ_CST);
notify_signal_observers();
return old;
}
j6_signal_t
kobject::deassert_signal(j6_signal_t s)
{
return __atomic_fetch_and(&m_signals, ~s, __ATOMIC_SEQ_CST);
}
void
kobject::compact_blocked_threads()
{
// Clean up what we can of the list
while (!m_blocked_threads.empty() && !m_blocked_threads.first())
m_blocked_threads.pop_front();
while (!m_blocked_threads.empty() && !m_blocked_threads.last())
m_blocked_threads.pop_back();
}
void
kobject::notify_signal_observers()
{
for (auto &entry : m_blocked_threads) {
if (entry == nullptr) continue;
if (entry->wake_on_signals(this, m_signals))
entry = nullptr;
}
compact_blocked_threads();
}
void
kobject::notify_object_observers(size_t count)
{
if (!count) return;
for (auto &entry : m_blocked_threads) {
if (entry == nullptr) continue;
if (entry->wake_on_object(this)) {
entry = nullptr;
if (--count) break;
}
}
compact_blocked_threads();
}
void
kobject::remove_blocked_thread(thread *t)
{
// Can't really remove from a deque, so just
// null out removed entries.
for (auto &entry : m_blocked_threads) {
if (entry == t) {
entry = nullptr;
break;
}
}
compact_blocked_threads();
}
void
kobject::close()
{
assert_signal(j6_signal_closed);
}
void
kobject::on_no_handles()
{
assert_signal(j6_signal_no_handles);
delete this;
}
} // namespace obj

55
src/kernel/objects/kobject.h
View File

@@ -3,9 +3,7 @@
/// Definition of base type for user-interactable kernel objects
#include <j6/errors.h>
#include <j6/signals.h>
#include <j6/types.h>
#include <util/deque.h>
namespace obj {
@@ -25,7 +23,7 @@ public:
max
};
kobject(type t, j6_signal_t signals = 0ull);
kobject(type t);
virtual ~kobject();
/// Generate a new koid for a given type
@@ -44,23 +42,6 @@ public:
/// Get this object's koid
inline j6_koid_t koid() const { return m_koid; }
/// Set the given signals active on this object
/// \arg s The set of signals to assert
/// \returns The previous state of the signals
j6_signal_t assert_signal(j6_signal_t s);
/// Clear the given signals on this object
/// \arg s The set of signals to deassert
/// \returns The previous state of the signals
j6_signal_t deassert_signal(j6_signal_t s);
/// Check if the given signals are set on this object
/// \arg s The set of signals to check
inline bool check_signal(j6_signal_t s) const { return (m_signals & s) == s; }
/// Get the current object signal state
inline j6_signal_t signals() const { return m_signals; }
/// Increment the handle refcount
inline void handle_retain() { ++m_handle_count; }
@@ -69,45 +50,21 @@ public:
if (--m_handle_count == 0) on_no_handles();
}
/// Add the given thread to the list of threads waiting on this object.
inline void add_blocked_thread(thread *t) { m_blocked_threads.push_back(t); }
/// Remove the given thread from the list of threads waiting on this object.
void remove_blocked_thread(thread *t);
/// Perform any cleanup actions necessary to mark this object closed
virtual void close();
/// Check if this object has been closed
inline bool closed() const { return check_signal(j6_signal_closed); }
protected:
/// Interface for subclasses to handle when all handles are closed. Subclasses
/// should either call the base version, or assert j6_signal_no_handles.
/// Interface for subclasses to handle when all handles are closed.
/// Default implementation deletes the object.
virtual void on_no_handles();
/// Get the current number of handles to this object
inline uint16_t handle_count() const { return m_handle_count; }
private:
kobject() = delete;
kobject(const kobject &other) = delete;
kobject(const kobject &&other) = delete;
/// Notifiy observers of this object
/// \arg result The result to pass to the observers
void notify_signal_observers();
/// Compact the blocked threads deque
void compact_blocked_threads();
j6_koid_t m_koid;
j6_signal_t m_signals;
uint16_t m_handle_count;
protected:
/// Notify threads waiting on this object
/// \arg count Number of observers to wake, or -1ull for all
void notify_object_observers(size_t count = -1ull);
util::deque<thread*, 8> m_blocked_threads;
};
} // namespace obj

1
src/kernel/objects/process.cpp
View File

@@ -53,7 +53,6 @@ process::exit(int32_t code)
// TODO: make this thread-safe
m_state = state::exited;
m_return_code = code;
close();
for (auto *thread : m_threads) {
thread->exit(code);

5
src/kernel/objects/process.h
View File

@@ -103,6 +103,11 @@ public:
/// \returns The kernel process object
static process * create_kernel_process(page_table *pml4);
protected:
/// Don't delete a process on no handles, the scheduler takes
/// care of that.
virtual void on_no_handles() override {}
private:
// This constructor is called by create_kernel_process
process(page_table *kpml4);

135
src/kernel/objects/thread.cpp
View File

@@ -1,4 +1,3 @@
#include <j6/signals.h>
#include <util/pointers.h>
#include "cpu.h"
@@ -15,15 +14,12 @@ extern obj::vm_area_guarded &g_kernel_stacks;
namespace obj {
static constexpr j6_signal_t thread_default_signals = 0;
thread::thread(process &parent, uint8_t pri, uintptr_t rsp0) :
kobject(kobject::type::thread, thread_default_signals),
m_parent(parent),
m_state(state::none),
m_wait_type(wait_type::none),
m_wait_data(0),
m_wait_obj(0)
kobject {kobject::type::thread},
m_parent {parent},
m_state {state::none},
m_wake_value {0},
m_wake_timeout {0}
{
parent.space().initialize_tcb(m_tcb);
m_tcb.priority = pri;
@@ -45,129 +41,20 @@ thread::~thread()
thread & thread::current() { return *current_cpu().thread; }
j6_status_t
thread::wait_on_signals(j6_signal_t signals)
{
util::scoped_lock lock {m_wait_lock};
m_wait_type = wait_type::signal;
m_wait_data = signals;
lock.release();
block();
return m_wait_result;
}
j6_status_t
thread::wait_on_time(uint64_t t)
{
util::scoped_lock lock {m_wait_lock};
m_wait_type = wait_type::time;
m_wait_time = t;
lock.release();
block();
return m_wait_result;
}
j6_status_t
thread::wait_on_object(void *o, uint64_t t)
{
util::scoped_lock lock {m_wait_lock};
m_wait_type = wait_type::object;
m_wait_data = reinterpret_cast<uint64_t>(o);
if (t) {
m_wait_type |= wait_type::time;
m_wait_time = t;
}
lock.release();
block();
return m_wait_result;
}
bool
thread::wake_on_signals(kobject *obj, j6_signal_t signals)
{
util::scoped_lock lock {m_wait_lock};
if (!(m_wait_type & wait_type::signal) ||
(signals & m_wait_data) == 0)
return false;
m_wait_type = wait_type::none;
m_wait_result = j6_status_ok;
m_wait_data = signals;
m_wait_obj = obj->koid();
set_state(state::ready);
return true;
}
bool
thread::wake_on_time(uint64_t now)
{
util::scoped_lock lock {m_wait_lock};
if (!(m_wait_type & wait_type::time) ||
now < m_wait_time)
return false;
if (!(m_wait_type & ~wait_type::none))
m_wait_result = j6_status_ok;
else
m_wait_result = j6_err_timed_out;
m_wait_type = wait_type::none;
m_wait_data = now;
m_wait_obj = 0;
set_state(state::ready);
return true;
}
bool
thread::wake_on_object(void *o, uint64_t id)
{
util::scoped_lock lock {m_wait_lock};
if (!(m_wait_type & wait_type::object) ||
reinterpret_cast<uint64_t>(o) != m_wait_data)
return false;
m_wait_type = wait_type::none;
m_wait_result = j6_status_ok;
m_wait_obj = id;
set_state(state::ready);
return true;
}
void
thread::wake_on_result(kobject *obj, j6_status_t result)
{
util::scoped_lock lock {m_wait_lock};
m_wait_type = wait_type::none;
m_wait_result = result;
m_wait_data = 0;
m_wait_obj = obj->koid();
set_state(state::ready);
}
void
uint64_t
thread::block()
{
clear_state(state::ready);
if (current_cpu().thread == this)
scheduler::get().schedule();
return m_wake_value;
}
void
thread::wake()
thread::wake(uint64_t value)
{
m_wake_value = value;
m_wake_timeout = 0;
set_state(state::ready);
}
@@ -175,8 +62,8 @@ void
thread::exit(int32_t code)
{
m_return_code = code;
m_wake_timeout = 0;
set_state(state::exited);
close();
block();
}

87
src/kernel/objects/thread.h
View File

@@ -29,14 +29,14 @@ struct TCB
obj::thread* thread;
uint8_t priority;
// note: 3 bytes padding
// TODO: move state into TCB?
uintptr_t kernel_stack;
uint32_t time_left;
uint64_t last_ran;
uintptr_t kernel_stack;
};
using tcb_list = util::linked_list<TCB>;
@@ -90,6 +90,10 @@ public:
/// \returns True if the thread has constant priority.
inline bool constant() const { return has_state(state::constant); }
/// Get the `exited` state of the thread.
/// \returns True if the thread has exited.
inline bool exited() const { return has_state(state::exited); }
/// Get the thread priority.
inline uint8_t priority() const { return m_tcb.priority; }
@@ -97,58 +101,21 @@ public:
/// \arg p The new thread priority
inline void set_priority(uint8_t p) { if (!constant()) m_tcb.priority = p; }
/// Block the thread, waiting an object's signals.
/// \arg signals Mask of signals to wait for
/// \returns j6_status_ok on success
j6_status_t wait_on_signals(j6_signal_t signals);
/// Block this thread, waiting for a value
/// \returns The value passed to wake()
uint64_t block();
/// Block the thread, waiting for a given clock value
/// \arg t Clock value to wait for
/// \returns j6_status_ok on success
j6_status_t wait_on_time(uint64_t t);
/// Wake this thread, giving it a value
/// \arg value The value that block() should return
void wake(uint64_t value = 0);
/// Block the thread, waiting on the given object
/// \arg o The object that should wake this thread
/// \arg t The timeout clock value to wait for
/// \returns j6_status_ok on success
j6_status_t wait_on_object(void *o, uint64_t t = 0);
/// Set a timeout to unblock this thread
/// \arg time The clock time at which to wake. 0 for no timeout.
inline void set_wake_timeout(uint64_t time) { m_wake_timeout = time; }
/// Wake the thread if it is waiting on signals.
/// \arg obj Object that changed signals
/// \arg signals Signal state of the object
/// \returns True if this action unblocked the thread
bool wake_on_signals(kobject *obj, j6_signal_t signals);
/// Wake the thread if it is waiting on the clock.
/// \arg now Current clock value
/// \returns True if this action unblocked the thread
bool wake_on_time(uint64_t now);
/// Wake the thread if it is waiting on the given object.
/// \arg o Object trying to wake the thread
/// \arg id Id of the object trying to wake the thread
/// \returns True if this action unblocked the thread
bool wake_on_object(void *o, uint64_t id = 0);
/// Wake the thread with a given result code.
/// \arg obj Object that changed signals
/// \arg result Result code to return to the thread
void wake_on_result(kobject *obj, j6_status_t result);
/// Get the result status code from the last blocking operation
j6_status_t get_wait_result() const { return m_wait_result; }
/// Get the current blocking opreation's wait data
uint64_t get_wait_data() const { return m_wait_data; }
/// Get the current blocking operation's wait ojbect (as a handle)
j6_koid_t get_wait_object() const { return m_wait_obj; }
/// Primitive thread blocking, instead of using wait framework
void block();
/// Primitive thread unblocking, instead of using wait framework
void wake();
/// Get the timeout at which to unblock this thread
/// \returns The clock time at which to wake. 0 for no timeout.
inline uint64_t wake_timeout() const { return m_wake_timeout; }
inline bool has_state(state s) const {
return static_cast<uint8_t>(m_state) & static_cast<uint8_t>(s);
@@ -189,6 +156,11 @@ public:
/// \arg rsp The existing stack for the idle thread
static thread * create_idle_thread(process &kernel, uint8_t pri, uintptr_t rsp);
protected:
/// Don't delete a thread on no handles, the scheduler takes
/// care of that.
virtual void on_no_handles() override {}
private:
thread() = delete;
thread(const thread &other) = delete;
@@ -210,16 +182,13 @@ private:
thread *m_creator;
state m_state;
wait_type m_wait_type;
// There should be 1 byte of padding here
// There should be 3 bytes of padding here
int32_t m_return_code;
uint64_t m_wait_data;
uint64_t m_wait_time;
j6_status_t m_wait_result;
uint64_t m_wait_obj;
util::spinlock m_wait_lock;
uint64_t m_wake_value;
uint64_t m_wake_timeout;
j6_handle_t m_self_handle;
};

4
src/kernel/objects/vm_area.cpp
View File

@@ -35,15 +35,13 @@ void
vm_area::remove_from(vm_space *space)
{
m_spaces.remove_swap(space);
if (!m_spaces.count() &&
check_signal(j6_signal_no_handles))
if (!m_spaces.count() && !handle_count())
delete this;
}
void
vm_area::on_no_handles()
{
kobject::on_no_handles();
if (!m_spaces.count())
delete this;
}

4
src/kernel/objects/vm_area.h
View File

@@ -6,7 +6,6 @@
#include <stdint.h>
#include <j6/caps.h>
#include <j6/signals.h>
#include <util/vector.h>
#include <util/enum_bitfields.h>
@@ -76,7 +75,10 @@ public:
virtual bool get_page(uintptr_t offset, uintptr_t &phys) = 0;
protected:
/// A VMA is not deleted until both no handles remain AND it's not
/// mapped by any VM space.
virtual void on_no_handles() override;
bool can_resize(size_t size);
size_t m_size;