Kernel: Rewrite the whole scheduler and re-architecture SMP handling
Change Semaphore -> ThreadBlocker
This was not a semaphore, I just named it one because I didn't know
what semaphore was. I have meant to change this sooner, but it was in
no way urgent :D
Implement SMP events. Processors can now be sent SMP events through
IPIs. SMP events can be sent either to a single processor or broadcasted
to every processor.
PageTable::{map_page,map_range,unmap_page,unmap_range}() now send SMP
event to invalidate TLB caches for the changed pages.
Scheduler no longer uses a global run queue. Each processor has its own
scheduler that keeps track of the load on the processor. Once every
second schedulers do load balancing. Schedulers have no access to other
processors' schedulers, they just see approximate loads. If scheduler
decides that it has too much load, it will send a thread to another
processor through a SMP event.
Schedulers are currently run using the timer interrupt on BSB. This
should be not the case, and each processor should use its LAPIC timer
for interrupts. There is no reason to broadcast SMP event to all
processors when BSB gets timer interrupt.
Old scheduler only achieved 20% idle load on qemu. That was probably a
very inefficient implementation. This new scheduler seems to average
around 1% idle load. This is much closer to what I would expect. On my
own laptop idle load seems to be only around 0.5% on each processor.
This commit is contained in:
+640
-185
@@ -1,260 +1,715 @@
|
||||
#include <kernel/Arch.h>
|
||||
#include <kernel/Attributes.h>
|
||||
#include <kernel/GDT.h>
|
||||
#include <BAN/Optional.h>
|
||||
#include <BAN/Sort.h>
|
||||
#include <kernel/InterruptController.h>
|
||||
#include <kernel/Process.h>
|
||||
#include <kernel/Scheduler.h>
|
||||
#include <kernel/Thread.h>
|
||||
#include <kernel/Timer/Timer.h>
|
||||
|
||||
#define SCHEDULER_VERIFY_STACK 1
|
||||
#define DEBUG_SCHEDULER 0
|
||||
#define SCHEDULER_ASSERT 1
|
||||
|
||||
#if SCHEDULER_ASSERT == 0
|
||||
#undef ASSERT
|
||||
#define ASSERT(...)
|
||||
#endif
|
||||
|
||||
namespace Kernel
|
||||
{
|
||||
|
||||
static Scheduler* s_instance = nullptr;
|
||||
static constexpr uint64_t s_reschedule_interval_ns = 10'000'000;
|
||||
static constexpr uint64_t s_load_balance_interval_ns = 1'000'000'000;
|
||||
|
||||
static BAN::Atomic<uint8_t> s_schedulers_initialized { 0 };
|
||||
|
||||
struct ProcessorInfo
|
||||
{
|
||||
uint64_t idle_time_ns { s_load_balance_interval_ns };
|
||||
uint32_t max_load_threads { 0 };
|
||||
};
|
||||
|
||||
static SpinLock s_processor_info_time_lock;
|
||||
static BAN::Array<ProcessorInfo, 0xFF> s_processor_infos;
|
||||
|
||||
|
||||
static BAN::Atomic<size_t> s_next_processor_index { 0 };
|
||||
|
||||
|
||||
void SchedulerQueue::add_thread_to_back(Node* node)
|
||||
{
|
||||
ASSERT(Processor::get_interrupt_state() == InterruptState::Disabled);
|
||||
node->next = nullptr;
|
||||
node->prev = m_tail;
|
||||
(m_tail ? m_tail->next : m_head) = node;
|
||||
m_tail = node;
|
||||
}
|
||||
|
||||
void SchedulerQueue::add_thread_with_wake_time(Node* node)
|
||||
{
|
||||
ASSERT(Processor::get_interrupt_state() == InterruptState::Disabled);
|
||||
|
||||
if (m_tail == nullptr || node->wake_time_ns >= m_tail->wake_time_ns)
|
||||
return add_thread_to_back(node);
|
||||
|
||||
Node* next = m_head;
|
||||
Node* prev = nullptr;
|
||||
while (next && node->wake_time_ns > next->wake_time_ns)
|
||||
{
|
||||
prev = next;
|
||||
next = next->next;
|
||||
}
|
||||
|
||||
node->next = next;
|
||||
node->prev = prev;
|
||||
(next ? next->prev : m_tail) = node;
|
||||
(prev ? prev->next : m_head) = node;
|
||||
}
|
||||
|
||||
template<typename F>
|
||||
SchedulerQueue::Node* SchedulerQueue::remove_with_condition(F callback)
|
||||
{
|
||||
ASSERT(Processor::get_interrupt_state() == InterruptState::Disabled);
|
||||
|
||||
for (Node* node = m_head; node; node = node->next)
|
||||
{
|
||||
if (!callback(node))
|
||||
continue;
|
||||
remove_node(node);
|
||||
return node;
|
||||
}
|
||||
|
||||
return nullptr;
|
||||
}
|
||||
|
||||
void SchedulerQueue::remove_node(Node* node)
|
||||
{
|
||||
(node->prev ? node->prev->next : m_head) = node->next;
|
||||
(node->next ? node->next->prev : m_tail) = node->prev;
|
||||
node->prev = nullptr;
|
||||
node->next = nullptr;
|
||||
}
|
||||
|
||||
SchedulerQueue::Node* SchedulerQueue::front()
|
||||
{
|
||||
ASSERT(Processor::get_interrupt_state() == InterruptState::Disabled);
|
||||
ASSERT(!empty());
|
||||
return m_head;
|
||||
}
|
||||
|
||||
SchedulerQueue::Node* SchedulerQueue::pop_front()
|
||||
{
|
||||
ASSERT(Processor::get_interrupt_state() == InterruptState::Disabled);
|
||||
if (empty())
|
||||
return nullptr;
|
||||
Node* result = m_head;
|
||||
m_head = m_head->next;
|
||||
(m_head ? m_head->prev : m_tail) = nullptr;
|
||||
result->next = nullptr;
|
||||
return result;
|
||||
}
|
||||
|
||||
BAN::ErrorOr<Scheduler*> Scheduler::create()
|
||||
{
|
||||
auto* scheduler = new Scheduler();
|
||||
if (scheduler == nullptr)
|
||||
return BAN::Error::from_errno(ENOMEM);
|
||||
return scheduler;
|
||||
}
|
||||
|
||||
BAN::ErrorOr<void> Scheduler::initialize()
|
||||
{
|
||||
ASSERT(s_instance == nullptr);
|
||||
s_instance = new Scheduler();
|
||||
ASSERT(s_instance);
|
||||
Processor::allocate_idle_thread();
|
||||
m_idle_thread = TRY(Thread::create_kernel([](void*) { asm volatile("1: hlt; jmp 1b"); }, nullptr, nullptr));
|
||||
ASSERT(m_idle_thread);
|
||||
|
||||
size_t processor_index = 0;
|
||||
for (; processor_index < Processor::count(); processor_index++)
|
||||
if (Processor::id_from_index(processor_index) == Processor::current_id())
|
||||
break;
|
||||
ASSERT(processor_index < Processor::count());
|
||||
|
||||
// each CPU does load balance at different times. This calulates the offset to other CPUs
|
||||
m_last_load_balance_ns = s_load_balance_interval_ns * processor_index / Processor::count();
|
||||
m_idle_ns = -m_last_load_balance_ns;
|
||||
|
||||
s_schedulers_initialized++;
|
||||
while (s_schedulers_initialized < Processor::count())
|
||||
__builtin_ia32_pause();
|
||||
|
||||
return {};
|
||||
}
|
||||
|
||||
Scheduler& Scheduler::get()
|
||||
{
|
||||
ASSERT(s_instance);
|
||||
return *s_instance;
|
||||
}
|
||||
|
||||
void Scheduler::start()
|
||||
void Scheduler::add_current_to_most_loaded(SchedulerQueue* target_queue)
|
||||
{
|
||||
ASSERT(Processor::get_interrupt_state() == InterruptState::Disabled);
|
||||
ASSERT(!m_active_threads.empty());
|
||||
|
||||
// broadcast ipi (yield) for each processor
|
||||
InterruptController::get().broadcast_ipi();
|
||||
yield();
|
||||
|
||||
ASSERT_NOT_REACHED();
|
||||
}
|
||||
|
||||
Thread& Scheduler::current_thread()
|
||||
{
|
||||
auto* current = Processor::get_current_thread();
|
||||
return current ? *current->thread : *Processor::idle_thread();
|
||||
}
|
||||
|
||||
pid_t Scheduler::current_tid()
|
||||
{
|
||||
if (s_instance == nullptr)
|
||||
return 0;
|
||||
return Scheduler::get().current_thread().tid();
|
||||
}
|
||||
|
||||
void Scheduler::setup_next_thread()
|
||||
{
|
||||
ASSERT(m_lock.current_processor_has_lock());
|
||||
|
||||
if (auto* current = Processor::get_current_thread())
|
||||
bool has_current = false;
|
||||
for (auto& info : m_most_loaded_threads)
|
||||
{
|
||||
auto* thread = current->thread;
|
||||
|
||||
if (thread->state() == Thread::State::Terminated)
|
||||
if (info.node == m_current)
|
||||
{
|
||||
PageTable::kernel().load();
|
||||
delete thread;
|
||||
delete current;
|
||||
}
|
||||
else
|
||||
{
|
||||
// thread->state() can be NotStarted when calling exec or cleaning up process
|
||||
if (thread->state() != Thread::State::NotStarted)
|
||||
{
|
||||
thread->interrupt_stack() = Processor::get_interrupt_stack();
|
||||
thread->interrupt_registers() = Processor::get_interrupt_registers();
|
||||
}
|
||||
|
||||
if (current->should_block)
|
||||
{
|
||||
current->should_block = false;
|
||||
m_blocking_threads.add_with_wake_time(current);
|
||||
}
|
||||
else
|
||||
{
|
||||
m_active_threads.push_back(current);
|
||||
}
|
||||
info.queue = target_queue;
|
||||
has_current = true;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
SchedulerQueue::Node* node = nullptr;
|
||||
while (!m_active_threads.empty())
|
||||
if (!has_current)
|
||||
{
|
||||
node = m_active_threads.pop_front();
|
||||
if (node->thread->state() != Thread::State::Terminated)
|
||||
size_t index = 0;
|
||||
for (; index < m_most_loaded_threads.size() - 1; index++)
|
||||
if (m_most_loaded_threads[index].node == nullptr)
|
||||
break;
|
||||
m_most_loaded_threads[index].queue = target_queue;
|
||||
m_most_loaded_threads[index].node = m_current;
|
||||
}
|
||||
|
||||
BAN::sort::sort(m_most_loaded_threads.begin(), m_most_loaded_threads.end(),
|
||||
[](const ThreadInfo& a, const ThreadInfo& b) -> bool
|
||||
{
|
||||
if (a.node == nullptr || b.node == nullptr)
|
||||
return a.node;
|
||||
return a.node->time_used_ns > b.node->time_used_ns;
|
||||
}
|
||||
);
|
||||
}
|
||||
|
||||
void Scheduler::update_most_loaded_node_queue(SchedulerQueue::Node* node, SchedulerQueue* target_queue)
|
||||
{
|
||||
ASSERT(Processor::get_interrupt_state() == InterruptState::Disabled);
|
||||
|
||||
for (auto& info : m_most_loaded_threads)
|
||||
{
|
||||
if (info.node == node)
|
||||
{
|
||||
info.queue = target_queue;
|
||||
break;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
void Scheduler::remove_node_from_most_loaded(SchedulerQueue::Node* node)
|
||||
{
|
||||
ASSERT(Processor::get_interrupt_state() == InterruptState::Disabled);
|
||||
|
||||
size_t i = 0;
|
||||
for (; i < m_most_loaded_threads.size(); i++)
|
||||
if (m_most_loaded_threads[i].node == node)
|
||||
break;
|
||||
|
||||
PageTable::kernel().load();
|
||||
delete node->thread;
|
||||
delete node;
|
||||
node = nullptr;
|
||||
for (; i < m_most_loaded_threads.size() - 1; i++)
|
||||
m_most_loaded_threads[i] = m_most_loaded_threads[i + 1];
|
||||
|
||||
m_most_loaded_threads.back().node = nullptr;
|
||||
m_most_loaded_threads.back().queue = nullptr;
|
||||
}
|
||||
|
||||
void Scheduler::reschedule(InterruptStack* interrupt_stack, InterruptRegisters* interrupt_registers)
|
||||
{
|
||||
ASSERT(Processor::get_interrupt_state() == InterruptState::Disabled);
|
||||
|
||||
// If there are no other threads in run queue, reschedule can be no-op :)
|
||||
if (m_run_queue.empty() && !m_current_will_block && current_thread().state() == Thread::State::Executing)
|
||||
return;
|
||||
|
||||
if (m_current == nullptr)
|
||||
m_idle_ns += SystemTimer::get().ns_since_boot() - m_idle_start_ns;
|
||||
else
|
||||
{
|
||||
switch (m_current->thread->state())
|
||||
{
|
||||
case Thread::State::Terminated:
|
||||
remove_node_from_most_loaded(m_current);
|
||||
PageTable::kernel().load();
|
||||
delete m_current->thread;
|
||||
delete m_current;
|
||||
m_thread_count--;
|
||||
break;
|
||||
case Thread::State::Executing:
|
||||
{
|
||||
const uint64_t current_ns = SystemTimer::get().ns_since_boot();
|
||||
m_current->thread->interrupt_stack() = *interrupt_stack;
|
||||
m_current->thread->interrupt_registers() = *interrupt_registers;
|
||||
m_current->time_used_ns += current_ns - m_current->last_start_ns;
|
||||
add_current_to_most_loaded(m_current_will_block ? &m_block_queue : &m_run_queue);
|
||||
if (!m_current_will_block)
|
||||
m_run_queue.add_thread_to_back(m_current);
|
||||
else
|
||||
{
|
||||
m_current_will_block = false;
|
||||
m_block_queue.add_thread_with_wake_time(m_current);
|
||||
}
|
||||
break;
|
||||
}
|
||||
case Thread::State::NotStarted:
|
||||
ASSERT(!m_current_will_block);
|
||||
m_current->time_used_ns = 0;
|
||||
remove_node_from_most_loaded(m_current);
|
||||
m_run_queue.add_thread_to_back(m_current);
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
Processor::set_current_thread(node);
|
||||
|
||||
auto* thread = node ? node->thread : Processor::idle_thread();
|
||||
|
||||
if (thread->has_process())
|
||||
thread->process().page_table().load();
|
||||
else
|
||||
while ((m_current = m_run_queue.pop_front()))
|
||||
{
|
||||
if (m_current->thread->state() != Thread::State::Terminated)
|
||||
break;
|
||||
remove_node_from_most_loaded(m_current);
|
||||
PageTable::kernel().load();
|
||||
delete m_current->thread;
|
||||
delete m_current;
|
||||
m_thread_count--;
|
||||
}
|
||||
|
||||
if (m_current == nullptr)
|
||||
{
|
||||
PageTable::kernel().load();
|
||||
*interrupt_stack = m_idle_thread->interrupt_stack();
|
||||
*interrupt_registers = m_idle_thread->interrupt_registers();
|
||||
m_idle_thread->m_state = Thread::State::Executing;
|
||||
m_idle_start_ns = SystemTimer::get().ns_since_boot();
|
||||
return;
|
||||
}
|
||||
|
||||
update_most_loaded_node_queue(m_current, nullptr);
|
||||
|
||||
auto* thread = m_current->thread;
|
||||
|
||||
auto& page_table = thread->has_process() ? thread->process().page_table() : PageTable::kernel();
|
||||
page_table.load();
|
||||
|
||||
if (thread->state() == Thread::State::NotStarted)
|
||||
thread->m_state = Thread::State::Executing;
|
||||
|
||||
Processor::gdt().set_tss_stack(thread->kernel_stack_top());
|
||||
Processor::get_interrupt_stack() = thread->interrupt_stack();
|
||||
Processor::get_interrupt_registers() = thread->interrupt_registers();
|
||||
*interrupt_stack = thread->interrupt_stack();
|
||||
*interrupt_registers = thread->interrupt_registers();
|
||||
|
||||
m_current->last_start_ns = SystemTimer::get().ns_since_boot();
|
||||
}
|
||||
|
||||
void Scheduler::timer_reschedule()
|
||||
void Scheduler::reschedule_if_idle()
|
||||
{
|
||||
ASSERT(Processor::get_interrupt_state() == InterruptState::Disabled);
|
||||
if (!m_current && !m_run_queue.empty())
|
||||
Processor::yield();
|
||||
}
|
||||
|
||||
void Scheduler::preempt()
|
||||
{
|
||||
ASSERT(Processor::get_interrupt_state() == InterruptState::Disabled);
|
||||
|
||||
if (Processor::is_smp_enabled())
|
||||
do_load_balancing();
|
||||
|
||||
{
|
||||
SpinLockGuard _(m_lock);
|
||||
m_blocking_threads.remove_with_wake_time(m_active_threads, SystemTimer::get().ms_since_boot());
|
||||
const uint64_t current_ns = SystemTimer::get().ns_since_boot();
|
||||
while (!m_block_queue.empty() && current_ns >= m_block_queue.front()->wake_time_ns)
|
||||
{
|
||||
auto* node = m_block_queue.pop_front();
|
||||
update_most_loaded_node_queue(node, &m_run_queue);
|
||||
m_run_queue.add_thread_to_back(node);
|
||||
}
|
||||
}
|
||||
|
||||
// Broadcast IPI to all other processors for them
|
||||
// to perform reschedule
|
||||
InterruptController::get().broadcast_ipi();
|
||||
yield();
|
||||
}
|
||||
|
||||
void Scheduler::yield()
|
||||
{
|
||||
auto state = Processor::get_interrupt_state();
|
||||
Processor::set_interrupt_state(InterruptState::Disabled);
|
||||
|
||||
ASSERT(!m_lock.current_processor_has_lock());
|
||||
|
||||
#if ARCH(x86_64)
|
||||
asm volatile(
|
||||
"movq %%rsp, %%rcx;"
|
||||
"movq %[load_sp], %%rsp;"
|
||||
"int %[yield];"
|
||||
"movq %%rcx, %%rsp;"
|
||||
// NOTE: This is offset by 2 pointers since interrupt without PL change
|
||||
// does not push SP and SS. This allows accessing "whole" interrupt stack.
|
||||
:: [load_sp]"r"(Processor::current_stack_top() - 2 * sizeof(uintptr_t)),
|
||||
[yield]"i"(IRQ_VECTOR_BASE + IRQ_YIELD)
|
||||
: "memory", "rcx"
|
||||
);
|
||||
#elif ARCH(i686)
|
||||
asm volatile(
|
||||
"movl %%esp, %%ecx;"
|
||||
"movl %[load_sp], %%esp;"
|
||||
"int %[yield];"
|
||||
"movl %%ecx, %%esp;"
|
||||
// NOTE: This is offset by 2 pointers since interrupt without PL change
|
||||
// does not push SP and SS. This allows accessing "whole" interrupt stack.
|
||||
:: [load_sp]"r"(Processor::current_stack_top() - 2 * sizeof(uintptr_t)),
|
||||
[yield]"i"(IRQ_VECTOR_BASE + IRQ_YIELD)
|
||||
: "memory", "ecx"
|
||||
);
|
||||
#else
|
||||
#error
|
||||
#endif
|
||||
|
||||
Processor::set_interrupt_state(state);
|
||||
}
|
||||
|
||||
void Scheduler::irq_reschedule()
|
||||
{
|
||||
SpinLockGuard _(m_lock);
|
||||
setup_next_thread();
|
||||
}
|
||||
|
||||
void Scheduler::reschedule_if_idling()
|
||||
{
|
||||
{
|
||||
SpinLockGuard _(m_lock);
|
||||
if (Processor::get_current_thread())
|
||||
return;
|
||||
if (m_active_threads.empty())
|
||||
return;
|
||||
const uint64_t current_ns = SystemTimer::get().ns_since_boot();
|
||||
if (current_ns >= m_last_reschedule_ns + s_reschedule_interval_ns)
|
||||
{
|
||||
m_last_reschedule_ns = current_ns;
|
||||
Processor::yield();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
void Scheduler::timer_interrupt()
|
||||
{
|
||||
ASSERT(Processor::get_interrupt_state() == InterruptState::Disabled);
|
||||
|
||||
// FIXME: all processors should LAPIC for their preemption
|
||||
if (Processor::is_smp_enabled())
|
||||
{
|
||||
ASSERT(Processor::current_is_bsb());
|
||||
Processor::broadcast_smp_message({
|
||||
.type = Processor::SMPMessage::Type::SchedulerPreemption,
|
||||
.scheduler_preemption = 0 // dummy value
|
||||
});
|
||||
}
|
||||
|
||||
yield();
|
||||
preempt();
|
||||
}
|
||||
|
||||
void Scheduler::handle_unblock_request(const UnblockRequest& request)
|
||||
{
|
||||
ASSERT(Processor::get_interrupt_state() == InterruptState::Disabled);
|
||||
|
||||
switch (request.type)
|
||||
{
|
||||
case UnblockRequest::Type::ThreadBlocker:
|
||||
do_unblock(request.blocker);
|
||||
break;
|
||||
case UnblockRequest::Type::ThreadID:
|
||||
do_unblock(request.tid);
|
||||
break;
|
||||
default:
|
||||
ASSERT_NOT_REACHED();
|
||||
}
|
||||
}
|
||||
|
||||
void Scheduler::handle_new_thread_request(const NewThreadRequest& reqeuest)
|
||||
{
|
||||
ASSERT(Processor::get_interrupt_state() == InterruptState::Disabled);
|
||||
|
||||
if (reqeuest.blocked)
|
||||
m_block_queue.add_thread_with_wake_time(reqeuest.node);
|
||||
else
|
||||
m_run_queue.add_thread_to_back(reqeuest.node);
|
||||
}
|
||||
|
||||
bool Scheduler::do_unblock(ThreadBlocker* blocker)
|
||||
{
|
||||
ASSERT(Processor::get_interrupt_state() == InterruptState::Disabled);
|
||||
|
||||
// FIXME: This could _easily_ be O(1)
|
||||
|
||||
bool did_unblock = false;
|
||||
|
||||
if (m_current && m_current->blocker == blocker && m_current_will_block)
|
||||
{
|
||||
m_current_will_block = false;
|
||||
did_unblock = true;
|
||||
}
|
||||
|
||||
SchedulerQueue::Node* match;
|
||||
while ((match = m_block_queue.remove_with_condition([blocker](const auto* node) { return node->blocker == blocker; })))
|
||||
{
|
||||
dprintln_if(DEBUG_SCHEDULER, "CPU {}: unblock blocker {} (tid {})", Processor::current_id(), blocker, match->thread->tid());
|
||||
update_most_loaded_node_queue(match, &m_run_queue);
|
||||
m_run_queue.add_thread_to_back(match);
|
||||
did_unblock = true;
|
||||
}
|
||||
|
||||
return did_unblock;
|
||||
}
|
||||
|
||||
bool Scheduler::do_unblock(pid_t tid)
|
||||
{
|
||||
ASSERT(Processor::get_interrupt_state() == InterruptState::Disabled);
|
||||
|
||||
// FIXME: This could _easily_ be O(1)
|
||||
|
||||
if (m_current && m_current->thread->tid() == tid && m_current_will_block)
|
||||
{
|
||||
m_current_will_block = false;
|
||||
return true;
|
||||
}
|
||||
|
||||
auto* match = m_block_queue.remove_with_condition([tid](const auto* node) { return node->thread->tid() == tid; });
|
||||
if (match == nullptr)
|
||||
return false;
|
||||
|
||||
dprintln_if(DEBUG_SCHEDULER, "CPU {}: unblock tid {}", Processor::current_id(), tid);
|
||||
update_most_loaded_node_queue(match, &m_run_queue);
|
||||
m_run_queue.add_thread_to_back(match);
|
||||
return true;
|
||||
}
|
||||
|
||||
ProcessorID Scheduler::find_least_loaded_processor() const
|
||||
{
|
||||
ProcessorID least_loaded_id = Processor::current_id();
|
||||
uint64_t most_idle_ns = m_idle_ns;
|
||||
uint32_t least_max_load_threads = static_cast<uint32_t>(-1);
|
||||
for (uint8_t i = 0; i < Processor::count(); i++)
|
||||
{
|
||||
auto processor_id = Processor::id_from_index(i);
|
||||
if (processor_id == Processor::current_id())
|
||||
continue;
|
||||
const auto& info = s_processor_infos[i];
|
||||
if (info.idle_time_ns < most_idle_ns || info.max_load_threads > least_max_load_threads)
|
||||
continue;
|
||||
least_loaded_id = processor_id;
|
||||
most_idle_ns = info.idle_time_ns;
|
||||
least_max_load_threads = info.max_load_threads;
|
||||
}
|
||||
return least_loaded_id;
|
||||
}
|
||||
|
||||
void Scheduler::do_load_balancing()
|
||||
{
|
||||
ASSERT(Processor::get_interrupt_state() == InterruptState::Disabled);
|
||||
|
||||
const uint64_t current_ns = SystemTimer::get().ns_since_boot();
|
||||
if (current_ns < m_last_load_balance_ns + s_load_balance_interval_ns)
|
||||
return;
|
||||
|
||||
if (m_current == nullptr)
|
||||
{
|
||||
m_idle_ns += current_ns - m_idle_start_ns;
|
||||
m_idle_start_ns = current_ns;
|
||||
}
|
||||
else
|
||||
{
|
||||
m_current->time_used_ns += current_ns - m_current->last_start_ns;
|
||||
m_current->last_start_ns = current_ns;
|
||||
add_current_to_most_loaded(nullptr);
|
||||
}
|
||||
|
||||
if constexpr(DEBUG_SCHEDULER)
|
||||
{
|
||||
const uint64_t duration_ns = current_ns - m_last_load_balance_ns;
|
||||
const uint64_t processing_ns = duration_ns - m_idle_ns;
|
||||
|
||||
{
|
||||
const uint64_t load_percent_x1000 = BAN::Math::div_round_up<uint64_t>(processing_ns * 100'000, duration_ns);
|
||||
dprintln("CPU {}: { 2}.{3}% ({} threads)", Processor::current_id(), load_percent_x1000 / 1000, load_percent_x1000 % 1000, m_thread_count);
|
||||
}
|
||||
|
||||
if (m_current)
|
||||
{
|
||||
const char* name = "unknown";
|
||||
if (m_current->thread->has_process() && m_current->thread->process().is_userspace() && m_current->thread->process().userspace_info().argv)
|
||||
name = m_current->thread->process().userspace_info().argv[0];
|
||||
const uint64_t load_percent_x1000 = BAN::Math::div_round_up<uint64_t>(m_current->time_used_ns * 100'000, processing_ns);
|
||||
dprintln(" tid { 2}: { 3}.{3}% <{}> current", m_current->thread->tid(), load_percent_x1000 / 1000, load_percent_x1000 % 1000, name);
|
||||
}
|
||||
m_run_queue.remove_with_condition(
|
||||
[&](SchedulerQueue::Node* node)
|
||||
{
|
||||
const uint64_t load_percent_x1000 = BAN::Math::div_round_up<uint64_t>(node->time_used_ns * 100'000, processing_ns);
|
||||
dprintln(" tid { 2}: { 3}.{3}% active", node->thread->tid(), load_percent_x1000 / 1000, load_percent_x1000 % 1000);
|
||||
return false;
|
||||
}
|
||||
);
|
||||
m_block_queue.remove_with_condition(
|
||||
[&](SchedulerQueue::Node* node)
|
||||
{
|
||||
const uint64_t load_percent_x1000 = BAN::Math::div_round_up<uint64_t>(node->time_used_ns * 100'000, processing_ns);
|
||||
dprintln(" tid { 2}: { 3}.{3}% blocked", node->thread->tid(), load_percent_x1000 / 1000, load_percent_x1000 % 1000);
|
||||
return false;
|
||||
}
|
||||
);
|
||||
}
|
||||
|
||||
if (!s_processor_info_time_lock.try_lock_interrupts_disabled())
|
||||
{
|
||||
dprintln_if(DEBUG_SCHEDULER, "Load balancing cannot keep up");
|
||||
return;
|
||||
}
|
||||
|
||||
if (m_idle_ns == 0 && m_should_calculate_max_load_threads)
|
||||
{
|
||||
const auto& most_loaded_thread = m_most_loaded_threads.front();
|
||||
if (most_loaded_thread.node == nullptr || most_loaded_thread.node->time_used_ns == 0)
|
||||
s_processor_infos[Processor::current_id().as_u32()].max_load_threads = 0;
|
||||
else
|
||||
{
|
||||
const uint64_t duration_ns = current_ns - m_last_load_balance_ns;
|
||||
const uint64_t max_thread_load_x1000 = 1000 * m_most_loaded_threads.front().node->time_used_ns / duration_ns;
|
||||
const uint64_t max_load_thread_count = ((2000 / max_thread_load_x1000) + 1) / 2;
|
||||
s_processor_infos[Processor::current_id().as_u32()].max_load_threads = max_load_thread_count;
|
||||
}
|
||||
}
|
||||
|
||||
constexpr auto absolute_difference_u64 = [](uint64_t a, uint64_t b) { return (a < b) ? (b - a) : (a - b); };
|
||||
|
||||
for (size_t i = 1; i < m_most_loaded_threads.size(); i++)
|
||||
{
|
||||
auto& thread_info = m_most_loaded_threads[i];
|
||||
if (thread_info.node == nullptr)
|
||||
break;
|
||||
if (thread_info.node == m_current || thread_info.queue == nullptr)
|
||||
continue;
|
||||
|
||||
auto least_loaded_id = find_least_loaded_processor();
|
||||
if (least_loaded_id == Processor::current_id())
|
||||
break;
|
||||
|
||||
auto& most_idle_info = s_processor_infos[least_loaded_id.as_u32()];
|
||||
auto& my_info = s_processor_infos[Processor::current_id().as_u32()];
|
||||
|
||||
if (m_idle_ns == 0)
|
||||
{
|
||||
if (my_info.max_load_threads == 0)
|
||||
break;
|
||||
|
||||
if (most_idle_info.idle_time_ns == 0)
|
||||
{
|
||||
if (most_idle_info.max_load_threads + 1 > my_info.max_load_threads - 1)
|
||||
break;
|
||||
|
||||
my_info.max_load_threads -= 1;
|
||||
most_idle_info.max_load_threads += 1;
|
||||
|
||||
dprintln_if(DEBUG_SCHEDULER, "CPU {}: sending tid {} to CPU {} (max load)", Processor::current_id(), thread_info.node->thread->tid(), least_loaded_id);
|
||||
}
|
||||
else
|
||||
{
|
||||
my_info.max_load_threads -= 1;
|
||||
most_idle_info.idle_time_ns = 0;
|
||||
most_idle_info.max_load_threads = 1;
|
||||
|
||||
dprintln_if(DEBUG_SCHEDULER, "CPU {}: sending tid {} to CPU {}", Processor::current_id(), thread_info.node->thread->tid(), least_loaded_id);
|
||||
}
|
||||
}
|
||||
else
|
||||
{
|
||||
const uint64_t my_current_proc_ns = s_load_balance_interval_ns - BAN::Math::min(s_load_balance_interval_ns, m_idle_ns);
|
||||
const uint64_t other_current_proc_ns = s_load_balance_interval_ns - BAN::Math::min(s_load_balance_interval_ns, most_idle_info.idle_time_ns);
|
||||
const uint64_t current_proc_diff_ns = absolute_difference_u64(my_current_proc_ns, other_current_proc_ns);
|
||||
|
||||
const uint64_t my_new_proc_ns = my_current_proc_ns - BAN::Math::min(thread_info.node->time_used_ns, my_current_proc_ns);
|
||||
const uint64_t other_new_proc_ns = other_current_proc_ns + thread_info.node->time_used_ns;
|
||||
const uint64_t new_proc_diff_ns = absolute_difference_u64(my_new_proc_ns, other_new_proc_ns);
|
||||
|
||||
// require 10% decrease between CPU loads to do send thread to other CPU
|
||||
if (new_proc_diff_ns >= current_proc_diff_ns || (100 * (current_proc_diff_ns - new_proc_diff_ns) / current_proc_diff_ns) < 10)
|
||||
continue;
|
||||
|
||||
most_idle_info.idle_time_ns -= BAN::Math::min(thread_info.node->time_used_ns, most_idle_info.idle_time_ns);
|
||||
m_idle_ns += thread_info.node->time_used_ns;
|
||||
|
||||
dprintln_if(DEBUG_SCHEDULER, "CPU {}: sending tid {} to CPU {}", Processor::current_id(), thread_info.node->thread->tid(), least_loaded_id);
|
||||
}
|
||||
|
||||
thread_info.node->time_used_ns = 0;
|
||||
|
||||
{
|
||||
auto& my_queue = (thread_info.queue == &m_run_queue) ? m_run_queue : m_block_queue;
|
||||
my_queue.remove_node(thread_info.node);
|
||||
m_thread_count--;
|
||||
}
|
||||
|
||||
Processor::send_smp_message(least_loaded_id, {
|
||||
.type = Processor::SMPMessage::Type::NewThread,
|
||||
.new_thread = {
|
||||
.node = thread_info.node,
|
||||
.blocked = thread_info.queue == &m_block_queue
|
||||
}
|
||||
});
|
||||
|
||||
thread_info.node = nullptr;
|
||||
thread_info.queue = nullptr;
|
||||
|
||||
if (m_idle_ns == 0)
|
||||
break;
|
||||
}
|
||||
|
||||
s_processor_infos[Processor::current_id().as_u32()].idle_time_ns = m_idle_ns;
|
||||
s_processor_info_time_lock.unlock(InterruptState::Disabled);
|
||||
|
||||
if (m_current)
|
||||
m_current->time_used_ns = 0;
|
||||
for (auto& thread_info : m_most_loaded_threads)
|
||||
thread_info = {};
|
||||
m_run_queue .remove_with_condition([&](SchedulerQueue::Node* node) { node->time_used_ns = 0; return false; });
|
||||
m_block_queue.remove_with_condition([&](SchedulerQueue::Node* node) { node->time_used_ns = 0; return false; });
|
||||
m_idle_ns = 0;
|
||||
|
||||
m_should_calculate_max_load_threads = true;
|
||||
|
||||
m_last_load_balance_ns += s_load_balance_interval_ns;
|
||||
}
|
||||
|
||||
BAN::ErrorOr<void> Scheduler::add_thread(Thread* thread)
|
||||
{
|
||||
auto* node = new SchedulerQueue::Node(thread);
|
||||
if (node == nullptr)
|
||||
auto* new_node = new SchedulerQueue::Node(thread);
|
||||
if (new_node == nullptr)
|
||||
return BAN::Error::from_errno(ENOMEM);
|
||||
SpinLockGuard _(m_lock);
|
||||
m_active_threads.push_back(node);
|
||||
|
||||
const size_t processor_index = s_next_processor_index++ % Processor::count();
|
||||
const auto processor_id = Processor::id_from_index(processor_index);
|
||||
|
||||
if (processor_id == Processor::current_id())
|
||||
{
|
||||
auto state = Processor::get_interrupt_state();
|
||||
Processor::set_interrupt_state(InterruptState::Disabled);
|
||||
m_run_queue.add_thread_to_back(new_node);
|
||||
m_thread_count++;
|
||||
Processor::set_interrupt_state(state);
|
||||
}
|
||||
else
|
||||
{
|
||||
Processor::send_smp_message(processor_id, {
|
||||
.type = Processor::SMPMessage::Type::NewThread,
|
||||
.new_thread = {
|
||||
.node = new_node,
|
||||
.blocked = false
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
return {};
|
||||
}
|
||||
|
||||
void Scheduler::terminate_thread(Thread* thread)
|
||||
void Scheduler::block_current_thread(ThreadBlocker* blocker, uint64_t wake_time_ns)
|
||||
{
|
||||
auto state = m_lock.lock();
|
||||
auto state = Processor::get_interrupt_state();
|
||||
Processor::set_interrupt_state(InterruptState::Disabled);
|
||||
|
||||
ASSERT(thread->state() == Thread::State::Executing);
|
||||
thread->m_state = Thread::State::Terminated;
|
||||
thread->interrupt_stack().sp = Processor::current_stack_top();
|
||||
m_current->blocker = blocker;
|
||||
m_current->wake_time_ns = wake_time_ns;
|
||||
m_current_will_block = true;
|
||||
Processor::yield();
|
||||
|
||||
m_lock.unlock(InterruptState::Disabled);
|
||||
Processor::set_interrupt_state(state);
|
||||
}
|
||||
|
||||
// actual deletion will be done while rescheduling
|
||||
void Scheduler::unblock_threads(ThreadBlocker* blocker)
|
||||
{
|
||||
auto state = Processor::get_interrupt_state();
|
||||
Processor::set_interrupt_state(InterruptState::Disabled);
|
||||
|
||||
if (¤t_thread() == thread)
|
||||
do_unblock(blocker);
|
||||
|
||||
Processor::broadcast_smp_message({
|
||||
.type = Processor::SMPMessage::Type::UnblockThread,
|
||||
.unblock_thread = {
|
||||
.type = UnblockRequest::Type::ThreadBlocker,
|
||||
.blocker = blocker
|
||||
}
|
||||
});
|
||||
|
||||
Processor::set_interrupt_state(state);
|
||||
}
|
||||
|
||||
void Scheduler::unblock_thread(pid_t tid)
|
||||
{
|
||||
auto state = Processor::get_interrupt_state();
|
||||
Processor::set_interrupt_state(InterruptState::Disabled);
|
||||
|
||||
if (!do_unblock(tid))
|
||||
{
|
||||
yield();
|
||||
ASSERT_NOT_REACHED();
|
||||
Processor::broadcast_smp_message({
|
||||
.type = Processor::SMPMessage::Type::UnblockThread,
|
||||
.unblock_thread = {
|
||||
.type = UnblockRequest::Type::ThreadID,
|
||||
.tid = tid
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
Processor::set_interrupt_state(state);
|
||||
}
|
||||
|
||||
void Scheduler::set_current_thread_sleeping_impl(Semaphore* semaphore, uint64_t wake_time)
|
||||
Thread& Scheduler::current_thread()
|
||||
{
|
||||
auto state = m_lock.lock();
|
||||
|
||||
auto* current = Processor::get_current_thread();
|
||||
current->semaphore = semaphore;
|
||||
current->wake_time = wake_time;
|
||||
current->should_block = true;
|
||||
|
||||
m_lock.unlock(InterruptState::Disabled);
|
||||
|
||||
yield();
|
||||
|
||||
Processor::set_interrupt_state(state);
|
||||
if (m_current)
|
||||
return *m_current->thread;
|
||||
return *m_idle_thread;
|
||||
}
|
||||
|
||||
void Scheduler::set_current_thread_sleeping(uint64_t wake_time)
|
||||
Thread& Scheduler::idle_thread()
|
||||
{
|
||||
set_current_thread_sleeping_impl(nullptr, wake_time);
|
||||
return *m_idle_thread;
|
||||
}
|
||||
|
||||
void Scheduler::block_current_thread(Semaphore* semaphore, uint64_t wake_time)
|
||||
pid_t Scheduler::current_tid() const
|
||||
{
|
||||
set_current_thread_sleeping_impl(semaphore, wake_time);
|
||||
return m_current ? m_current->thread->tid() : 0;
|
||||
}
|
||||
|
||||
void Scheduler::unblock_threads(Semaphore* semaphore)
|
||||
bool Scheduler::is_idle() const
|
||||
{
|
||||
SpinLockGuard _(m_lock);
|
||||
m_blocking_threads.remove_with_condition(m_active_threads, [&](auto* node) { return node->semaphore == semaphore; });
|
||||
}
|
||||
|
||||
void Scheduler::unblock_thread(pid_t tid)
|
||||
{
|
||||
SpinLockGuard _(m_lock);
|
||||
m_blocking_threads.remove_with_condition(m_active_threads, [&](auto* node) { return node->thread->tid() == tid; });
|
||||
return m_current == nullptr;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user