diff --git a/sched/include/xsched/sched/policy/cfs.h b/sched/include/xsched/sched/policy/cfs.h index 5ec20e3a..bd2d22f3 100644 --- a/sched/include/xsched/sched/policy/cfs.h +++ b/sched/include/xsched/sched/policy/cfs.h @@ -26,8 +26,9 @@ class CompletelyFairSchedulerPolicy : public Policy { virtual void RecvHint(std::shared_ptr hint) override; private: - std::unordered_map cfs_infos_; - std::chrono::microseconds time_slice_{1000}; // set timeslice to 1ms + std::unordered_map cfs_infos_; + std::unordered_map> pending_hints_; + std::chrono::microseconds time_slice_{1000}; }; } // namespace xsched::sched diff --git a/sched/include/xsched/sched/policy/kedf.h b/sched/include/xsched/sched/policy/kedf.h index 6b8e2e16..1abcaf30 100644 --- a/sched/include/xsched/sched/policy/kedf.h +++ b/sched/include/xsched/sched/policy/kedf.h @@ -9,9 +9,9 @@ namespace xsched::sched { -struct DeadlineEntry +struct ProcessDeadlineEntry { - XQueueHandle xqueue; + PID pid; std::chrono::system_clock::time_point deadline; }; @@ -25,6 +25,8 @@ class KEarliestDeadlineFirstPolicy : public Policy virtual void RecvHint(std::shared_ptr hint) override; private: + void SwitchProcess(PID pid, const Status &status); + size_t k_ = 1; std::unordered_map deadlines_; }; diff --git a/sched/include/xsched/sched/policy/lax.h b/sched/include/xsched/sched/policy/lax.h index 44683613..cabde2ba 100644 --- a/sched/include/xsched/sched/policy/lax.h +++ b/sched/include/xsched/sched/policy/lax.h @@ -25,7 +25,10 @@ class LaxityPolicy : public Policy Priority lax_prio; Priority crit_prio; }; - std::unordered_map laxity_infos_; + std::unordered_map laxity_infos_; + std::unordered_map handle_to_pid_; + + void SwitchProcess(PID pid, const Status &status); }; } // namespace xsched::sched diff --git a/sched/include/xsched/sched/policy/mlfq.h b/sched/include/xsched/sched/policy/mlfq.h index 6110bb77..2f987339 100644 --- a/sched/include/xsched/sched/policy/mlfq.h +++ b/sched/include/xsched/sched/policy/mlfq.h @@ -9,21 +9,20 @@ namespace xsched { namespace sched { struct MLFQNode { - int priority = 0; // Current priority level (0 is highest) - bool is_running = false; // Whether it's currently running - bool was_ready_last_tick = false; // Edge detection for idle/ready + int priority = 0; + bool is_running = false; + bool was_ready_last_tick = false; using TimePoint = std::chrono::time_point; - TimePoint i_a; // (1) Time became idle - TimePoint p_a; // (2) Time of last priority update - TimePoint q_a; // (3) Time of most recent request + TimePoint i_a; + TimePoint p_a; + TimePoint q_a; - // For pending time tracking and time slice tracking std::chrono::microseconds accumulated_pending_time{0}; - TimePoint last_pending_start; // When it entered pending state + TimePoint last_pending_start; std::chrono::microseconds time_slice_used{0}; - TimePoint last_resume_time; // When it started running + TimePoint last_resume_time; }; class MultiLevelFeedbackQueuePolicy : public Policy { @@ -35,16 +34,13 @@ class MultiLevelFeedbackQueuePolicy : public Policy { void RecvHint(std::shared_ptr hint) override; private: - std::map mlfq_infos_; + std::map mlfq_infos_; - // Configuration - const int max_priority_ = 3; // Levels 0, 1, 2, 3 - const std::chrono::microseconds recovery_threshold_{100000}; // 100ms - const std::chrono::microseconds default_tick_{5000}; // 5ms scheduling tick + const int max_priority_ = 3; + const std::chrono::microseconds recovery_threshold_{100000}; + const std::chrono::microseconds default_tick_{5000}; - // Get time slice based on priority level std::chrono::microseconds get_time_slice(int prio) const { - // e.g., prio 0: 10ms, prio 1: 20ms, prio 2: 40ms, prio 3: 80ms return std::chrono::microseconds(10000 * (1 << prio)); } }; diff --git a/sched/include/xsched/sched/policy/up.h b/sched/include/xsched/sched/policy/up.h index 1fa9349c..3cf16ab2 100644 --- a/sched/include/xsched/sched/policy/up.h +++ b/sched/include/xsched/sched/policy/up.h @@ -22,12 +22,14 @@ class UtilizationPartitionPolicy : public Policy private: std::chrono::microseconds GetBudget(Utilization util); + bool ProcessReady(PID pid, const Status &status); void SwitchToAny(const Status &status); - void SwitchTo(XQueueHandle handle, Utilization util, const Status &status); + void SwitchProcess(PID pid, Utilization util, const Status &status); - XQueueHandle cur_running_ = 0; + PID cur_running_ = 0; std::chrono::system_clock::time_point cur_end_; - std::map utils_; + std::map utils_; + std::map handle_to_pid_; std::chrono::microseconds timeslice_ = std::chrono::microseconds(TIMESLICE_DEFAULT); }; diff --git a/sched/src/policy/cfs.cpp b/sched/src/policy/cfs.cpp index 3dd2cd1a..9c38804b 100644 --- a/sched/src/policy/cfs.cpp +++ b/sched/src/policy/cfs.cpp @@ -1,6 +1,7 @@ #include #include #include +#include #include "xsched/utils/xassert.h" #include "xsched/sched/policy/cfs.h" @@ -12,30 +13,35 @@ void CompletelyFairSchedulerPolicy::Sched(const Status &status) auto now = std::chrono::system_clock::now(); bool has_ready_tasks = false; - // update vruntime for running queues for (auto &st : status.xqueue_status) { + PID pid = st.second->pid; XQueueHandle handle = st.second->handle; - auto it = cfs_infos_.find(handle); + + auto pit = pending_hints_.find(handle); + if (pit != pending_hints_.end()) { + cfs_infos_[pid].priority = pit->second.first; + cfs_infos_[pid].weight = pit->second.second; + pending_hints_.erase(pit); + } + + auto it = cfs_infos_.find(pid); if (it != cfs_infos_.end() && it->second.is_running) { auto delta_us = std::chrono::duration_cast(now - it->second.last_resume_time).count(); - // Update virtual time: real time * (base weight / current task weight) it->second.vruntime += delta_us * (1024.0 / it->second.weight); } } - // Find the ready task with the minimum vruntime on each physical GPU - std::map min_vruntime_handles; + std::map min_vruntime_pids; std::map min_vruntimes; - // find the current min_vruntime for each device among existing tasks for (auto &st : status.xqueue_status) { if (!st.second->ready) continue; has_ready_tasks = true; XDevice device = st.second->device; - XQueueHandle handle = st.second->handle; + PID pid = st.second->pid; - auto it = cfs_infos_.find(handle); + auto it = cfs_infos_.find(pid); if (it != cfs_infos_.end()) { double current_vruntime = it->second.vruntime; if (min_vruntimes.find(device) == min_vruntimes.end() || current_vruntime < min_vruntimes[device]) { @@ -44,62 +50,66 @@ void CompletelyFairSchedulerPolicy::Sched(const Status &status) } } - // handle initialization for new tasks and select the final task to run for (auto &st : status.xqueue_status) { if (!st.second->ready) continue; XDevice device = st.second->device; - XQueueHandle handle = st.second->handle; + PID pid = st.second->pid; - // If this is the first time seeing this queue, initialize its CFS info - if (cfs_infos_.find(handle) == cfs_infos_.end()) { - cfs_infos_[handle] = CFSNode(); - cfs_infos_[handle].last_resume_time = now; - // Inherit the minimum vruntime of the current device to prevent new tasks from starving old tasks + if (cfs_infos_.find(pid) == cfs_infos_.end()) { + cfs_infos_[pid] = CFSNode(); + cfs_infos_[pid].last_resume_time = now; if (min_vruntimes.find(device) != min_vruntimes.end()) { - cfs_infos_[handle].vruntime = min_vruntimes[device]; + cfs_infos_[pid].vruntime = min_vruntimes[device]; } else { - cfs_infos_[handle].vruntime = 0.0; + cfs_infos_[pid].vruntime = 0.0; } - // Ensure min_vruntimes contains the vruntime of the new task (mainly for cases where the device has no old tasks) if (min_vruntimes.find(device) == min_vruntimes.end()) { - min_vruntimes[device] = cfs_infos_[handle].vruntime; + min_vruntimes[device] = cfs_infos_[pid].vruntime; } } - double current_vruntime = cfs_infos_[handle].vruntime; + double current_vruntime = cfs_infos_[pid].vruntime; - // Find the final minimum value and corresponding handle on this device - if (min_vruntime_handles.find(device) == min_vruntime_handles.end() || current_vruntime < cfs_infos_[min_vruntime_handles[device]].vruntime) { - min_vruntime_handles[device] = handle; + if (min_vruntime_pids.find(device) == min_vruntime_pids.end() || + current_vruntime < cfs_infos_[min_vruntime_pids[device]].vruntime) { + min_vruntime_pids[device] = pid; } } - // Resume the task with the minimum vruntime, Suspend all others for (auto &st : status.xqueue_status) { - XDevice device = st.second->device; - XQueueHandle handle = st.second->handle; - + PID pid = st.second->pid; if (!st.second->ready) { - cfs_infos_[handle].is_running = false; + cfs_infos_[pid].is_running = false; continue; } + } - if (min_vruntime_handles[device] == handle) { - if (!cfs_infos_[handle].is_running) { - this->Resume(handle); - cfs_infos_[handle].is_running = true; - cfs_infos_[handle].last_resume_time = now; - } else { - cfs_infos_[handle].last_resume_time = now; - } + std::set running_pids; + for (const auto &pair : min_vruntime_pids) { + PID best_pid = pair.second; + auto &node = cfs_infos_[best_pid]; + if (!node.is_running) { + node.is_running = true; + node.last_resume_time = now; } else { - this->Suspend(handle); - cfs_infos_[handle].is_running = false; + node.last_resume_time = now; + } + running_pids.insert(best_pid); + } + + for (auto &st : status.xqueue_status) { + PID pid = st.second->pid; + if (!st.second->ready) continue; + + if (running_pids.count(pid)) { + this->Resume(st.first); + } else { + this->Suspend(st.first); + cfs_infos_[pid].is_running = false; } } - // force a new scheduling round after the time slice if (has_ready_tasks) { this->AddTimer(now + time_slice_); } @@ -111,14 +121,10 @@ void CompletelyFairSchedulerPolicy::RecvHint(std::shared_ptr hint) auto h = std::dynamic_pointer_cast(hint); if (h == nullptr) return; - XQueueHandle handle = h->Handle(); Priority prio = h->Prio(); - - // Calculate weight: assuming base is 1024. For each priority increase, weight increases by 20% double weight = 1024.0 * std::pow(1.2, prio); - cfs_infos_[handle].priority = prio; - cfs_infos_[handle].weight = weight; + pending_hints_[h->Handle()] = {prio, weight}; - XINFO("CFS: set priority %d (weight %.2f) for XQueue 0x" FMT_64X, prio, weight, handle); + XINFO("CFS: set priority %d (weight %.2f) for XQueue 0x" FMT_64X, prio, weight, h->Handle()); } diff --git a/sched/src/policy/kedf.cpp b/sched/src/policy/kedf.cpp index c5db98e1..a6e07fd1 100644 --- a/sched/src/policy/kedf.cpp +++ b/sched/src/policy/kedf.cpp @@ -9,44 +9,69 @@ using namespace xsched::sched; void KEarliestDeadlineFirstPolicy::Sched(const Status &status) { - std::vector ddls; - ddls.reserve(status.xqueue_status.size()); + std::vector ddls; + ddls.reserve(status.process_status.size()); - // calculate the deadline of each xqueue - for (auto &status : status.xqueue_status) { - XQueueHandle handle = status.second->handle; + for (auto &process : status.process_status) { + PID pid = process.first; auto ddl = (std::chrono::system_clock::time_point::max)(); - if (!status.second->ready) { - ddls.emplace_back(DeadlineEntry{.xqueue=handle,.deadline=ddl}); - continue; + + bool any_ready = false; + for (auto handle : process.second->running_xqueues) { + auto xq_it = status.xqueue_status.find(handle); + if (xq_it == status.xqueue_status.end()) continue; + if (!xq_it->second->ready) continue; + any_ready = true; + auto d_it = deadlines_.find(handle); + if (d_it == deadlines_.end()) continue; + auto xq_ddl = xq_it->second->ready_time + std::chrono::microseconds(d_it->second); + if (xq_ddl < ddl) ddl = xq_ddl; + } + for (auto handle : process.second->suspended_xqueues) { + auto xq_it = status.xqueue_status.find(handle); + if (xq_it == status.xqueue_status.end()) continue; + if (!xq_it->second->ready) continue; + any_ready = true; + auto d_it = deadlines_.find(handle); + if (d_it == deadlines_.end()) continue; + auto xq_ddl = xq_it->second->ready_time + std::chrono::microseconds(d_it->second); + if (xq_ddl < ddl) ddl = xq_ddl; } - auto it = deadlines_.find(handle); - if (it == deadlines_.end()) { - ddls.emplace_back(DeadlineEntry{.xqueue=handle,.deadline=ddl}); + if (!any_ready) { + ddls.emplace_back(ProcessDeadlineEntry{.pid=pid,.deadline=(std::chrono::system_clock::time_point::max)()}); continue; } - ddl = status.second->ready_time + std::chrono::microseconds(it->second); - ddls.emplace_back(DeadlineEntry{.xqueue=handle,.deadline=ddl}); + ddls.emplace_back(ProcessDeadlineEntry{.pid=pid,.deadline=ddl}); } - // sort the xqueues by deadline, from the earliest to the latest - std::sort(ddls.begin(), ddls.end(), [](const DeadlineEntry &a, const DeadlineEntry &b) { + std::sort(ddls.begin(), ddls.end(), [](const ProcessDeadlineEntry &a, const ProcessDeadlineEntry &b) { return a.deadline < b.deadline; }); - // resume the first k_ xqueues for (size_t i = 0; i < k_ && i < ddls.size(); ++i) { - this->Resume(ddls[i].xqueue); + SwitchProcess(ddls[i].pid, status); } - - // suspend all other xqueues for (size_t i = k_; i < ddls.size(); ++i) { - this->Suspend(ddls[i].xqueue); + const auto it = status.process_status.find(ddls[i].pid); + if (it == status.process_status.end()) continue; + std::vector running; + for (auto xq : it->second->running_xqueues) running.push_back(xq); + for (auto xq : running) this->Suspend(xq); } } +void KEarliestDeadlineFirstPolicy::SwitchProcess(PID pid, const Status &status) +{ + const auto it = status.process_status.find(pid); + if (it == status.process_status.end()) return; + + std::vector suspended; + for (auto xq : it->second->suspended_xqueues) suspended.push_back(xq); + for (auto xq : suspended) this->Resume(xq); +} + void KEarliestDeadlineFirstPolicy::RecvHint(std::shared_ptr hint) { switch (hint->Type()) diff --git a/sched/src/policy/lax.cpp b/sched/src/policy/lax.cpp index fe47036c..3ab29752 100644 --- a/sched/src/policy/lax.cpp +++ b/sched/src/policy/lax.cpp @@ -1,4 +1,5 @@ #include +#include #include "xsched/utils/xassert.h" #include "xsched/sched/policy/lax.h" @@ -9,22 +10,46 @@ void LaxityPolicy::Sched(const Status &status) { auto current = std::chrono::system_clock::now(); - // find the task with the highest priority and the earliest laxity + for (const auto &xq : status.xqueue_status) { + handle_to_pid_[xq.first] = xq.second->pid; + } + Priority highest_prio = PRIORITY_MIN; bool has_laxity = false; auto earliest_laxity = (std::chrono::system_clock::time_point::max)(); - for (auto &status : status.xqueue_status) { - if (!status.second->ready) continue; - XQueueHandle handle = status.second->handle; - auto it = laxity_infos_.find(handle); + for (const auto &process : status.process_status) { + PID pid = process.first; + + auto earliest_ready = (std::chrono::system_clock::time_point::max)(); + bool any_ready = false; + for (auto handle : process.second->running_xqueues) { + auto it = status.xqueue_status.find(handle); + if (it == status.xqueue_status.end()) continue; + if (it->second->ready) { + any_ready = true; + if (it->second->ready_time < earliest_ready) + earliest_ready = it->second->ready_time; + } + } + for (auto handle : process.second->suspended_xqueues) { + auto it = status.xqueue_status.find(handle); + if (it == status.xqueue_status.end()) continue; + if (it->second->ready) { + any_ready = true; + if (it->second->ready_time < earliest_ready) + earliest_ready = it->second->ready_time; + } + } + if (!any_ready) continue; + + auto it = laxity_infos_.find(pid); if (it == laxity_infos_.end()) { - // no laxity info set, use PRIO_DEFAULT if (PRIORITY_DEFAULT > highest_prio) highest_prio = PRIORITY_DEFAULT; continue; } - auto laxity = status.second->ready_time + std::chrono::microseconds(it->second.lax); + auto laxity = earliest_ready + std::chrono::microseconds(it->second.lax); Priority prio = current < laxity ? it->second.lax_prio : it->second.crit_prio; if (prio > highest_prio) highest_prio = prio; @@ -34,20 +59,48 @@ void LaxityPolicy::Sched(const Status &status) } } - // suspend all other xqueues - for (auto &status : status.xqueue_status) { + for (const auto &process : status.process_status) { + PID pid = process.first; + + auto earliest_ready = (std::chrono::system_clock::time_point::max)(); + bool any_ready = false; + for (auto handle : process.second->running_xqueues) { + auto it = status.xqueue_status.find(handle); + if (it == status.xqueue_status.end()) continue; + if (it->second->ready) { + any_ready = true; + if (it->second->ready_time < earliest_ready) + earliest_ready = it->second->ready_time; + } + } + for (auto handle : process.second->suspended_xqueues) { + auto it = status.xqueue_status.find(handle); + if (it == status.xqueue_status.end()) continue; + if (it->second->ready) { + any_ready = true; + if (it->second->ready_time < earliest_ready) + earliest_ready = it->second->ready_time; + } + } + Priority prio = PRIORITY_DEFAULT; - XQueueHandle handle = status.second->handle; - auto it = laxity_infos_.find(handle); - if (it != laxity_infos_.end()) { - auto laxity = status.second->ready_time + std::chrono::microseconds(it->second.lax); - prio = !status.second->ready || current < laxity - ? it->second.lax_prio - : it->second.crit_prio; + auto lit = laxity_infos_.find(pid); + if (lit != laxity_infos_.end() && any_ready) { + auto laxity = earliest_ready + std::chrono::microseconds(lit->second.lax); + prio = current < laxity ? lit->second.lax_prio : lit->second.crit_prio; } - if (prio < highest_prio) this->Suspend(handle); - else this->Resume(handle); + if (prio < highest_prio) { + std::list running; + for (const auto &xq : process.second->running_xqueues) + running.push_back(xq); + for (const auto xq : running) this->Suspend(xq); + } else { + std::list suspended; + for (const auto &xq : process.second->suspended_xqueues) + suspended.push_back(xq); + for (const auto xq : suspended) this->Resume(xq); + } } if (has_laxity) { @@ -64,7 +117,10 @@ void LaxityPolicy::RecvHint(std::shared_ptr hint) Laxity lax = h->Lax(); Priority lax_prio = h->LaxPrio(); Priority crit_prio = h->CritPrio(); - laxity_infos_[h->Handle()] = { + + auto pid_it = handle_to_pid_.find(h->Handle()); + PID pid = (pid_it != handle_to_pid_.end()) ? pid_it->second : (PID)h->Handle(); + laxity_infos_[pid] = { .lax = lax < 0 ? NO_LAXITY : lax, .lax_prio = lax_prio, .crit_prio = crit_prio diff --git a/sched/src/policy/mlfq.cpp b/sched/src/policy/mlfq.cpp index 9a2e734a..029b12df 100644 --- a/sched/src/policy/mlfq.cpp +++ b/sched/src/policy/mlfq.cpp @@ -1,6 +1,7 @@ #include "xsched/sched/policy/mlfq.h" #include "xsched/utils/log.h" #include +#include using namespace xsched::sched; @@ -9,68 +10,60 @@ void MultiLevelFeedbackQueuePolicy::Sched(const Status &status) auto now = std::chrono::system_clock::now(); bool has_ready_tasks = false; - // Maintain variables: i_a (idle time) and q_a (request time) through edge detection for (auto &st : status.xqueue_status) { - XQueueHandle handle = st.second->handle; + PID pid = st.second->pid; bool is_ready_now = st.second->ready; - // Initialize if not present - if (mlfq_infos_.find(handle) == mlfq_infos_.end()) { - mlfq_infos_[handle] = MLFQNode(); - mlfq_infos_[handle].p_a = now; - mlfq_infos_[handle].i_a = now; - mlfq_infos_[handle].q_a = now; - mlfq_infos_[handle].last_pending_start = now; - mlfq_infos_[handle].last_resume_time = now; - XINFO("MLFQ: Initialized new XQueue 0x" FMT_64X, handle); + if (mlfq_infos_.find(pid) == mlfq_infos_.end()) { + mlfq_infos_[pid] = MLFQNode(); + mlfq_infos_[pid].p_a = now; + mlfq_infos_[pid].i_a = now; + mlfq_infos_[pid].q_a = now; + mlfq_infos_[pid].last_pending_start = now; + mlfq_infos_[pid].last_resume_time = now; + XINFO("MLFQ: Initialized new PID " FMT_PID, pid); } - auto &node = mlfq_infos_[handle]; + auto &node = mlfq_infos_[pid]; if (is_ready_now && !node.was_ready_last_tick) { - // Task just woke up (idle -> ready) node.q_a = now; node.last_pending_start = now; } else if (!is_ready_now && node.was_ready_last_tick) { - // Task just finished executing (ready -> idle) node.i_a = now; } node.was_ready_last_tick = is_ready_now; } - // Count number of tasks (N) per priority level on each device std::map> N_count; for (auto &st : status.xqueue_status) { if (st.second->ready) { - int prio = mlfq_infos_[st.second->handle].priority; + int prio = mlfq_infos_[st.second->pid].priority; N_count[st.second->device][prio]++; } } - // Update pending times and check for priority recovery / demotion for (auto &st : status.xqueue_status) { - XQueueHandle handle = st.second->handle; + PID pid = st.second->pid; XDevice device = st.second->device; - auto &node = mlfq_infos_[handle]; + auto &node = mlfq_infos_[pid]; if (!st.second->ready) continue; has_ready_tasks = true; - // Pending time & Run time updates if (!node.is_running) { auto pending_duration = std::chrono::duration_cast(now - node.last_pending_start); node.accumulated_pending_time += pending_duration; - node.last_pending_start = now; // reset start point + node.last_pending_start = now; } else { auto run_duration = std::chrono::duration_cast(now - node.last_resume_time); node.time_slice_used += run_duration; node.last_resume_time = now; } - // a) Soft Priority Recovery int N = N_count[device][node.priority]; - double R = (N > 1) ? (0.9 / N) : 1.0; // Dynamic discount factor R < 1/N + double R = (N > 1) ? (0.9 / N) : 1.0; auto time_since_last_update = std::chrono::duration_cast(now - node.p_a).count(); double discounted_pending = node.accumulated_pending_time.count() * R; @@ -79,88 +72,90 @@ void MultiLevelFeedbackQueuePolicy::Sched(const Status &status) if (effective_time > recovery_threshold_.count()) { if (node.priority > 0) { node.priority--; - XINFO("MLFQ: Priority Recovery (Promotion) for 0x" FMT_64X " to %d", handle, node.priority); + XINFO("MLFQ: Priority Recovery (Promotion) for PID " FMT_PID " to %d", pid, node.priority); } - // Record time of last priority update node.p_a = now; node.accumulated_pending_time = std::chrono::microseconds(0); node.time_slice_used = std::chrono::microseconds(0); } - // b) Time Slice Exhaustion (Demotion) if (node.is_running && node.time_slice_used >= get_time_slice(node.priority)) { if (node.priority < max_priority_) { node.priority++; - XINFO("MLFQ: Time Slice Exhausted (Demotion) for 0x" FMT_64X " to %d", handle, node.priority); + XINFO("MLFQ: Time Slice Exhausted (Demotion) for PID " FMT_PID " to %d", pid, node.priority); } node.p_a = now; node.time_slice_used = std::chrono::microseconds(0); node.accumulated_pending_time = std::chrono::microseconds(0); - // Force it to yield node.is_running = false; - this->Suspend(handle); - node.last_pending_start = now; // Starts pending from here + node.last_pending_start = now; } } - // Select the highest priority task per device to execute - std::map best_handles; + std::map best_pids; std::map best_prios; for (auto &st : status.xqueue_status) { if (!st.second->ready) continue; XDevice device = st.second->device; - XQueueHandle handle = st.second->handle; - auto &node = mlfq_infos_[handle]; + PID pid = st.second->pid; + auto &node = mlfq_infos_[pid]; - // Find highest priority (lowest number) if (best_prios.find(device) == best_prios.end() || node.priority < best_prios[device]) { best_prios[device] = node.priority; - best_handles[device] = handle; + best_pids[device] = pid; } - // Tie breaker: longest pending time (acts as Round-Robin or FCFS within priority) else if (node.priority == best_prios[device]) { - if (node.accumulated_pending_time > mlfq_infos_[best_handles[device]].accumulated_pending_time) { - best_handles[device] = handle; + if (node.accumulated_pending_time > mlfq_infos_[best_pids[device]].accumulated_pending_time) { + best_pids[device] = pid; } } } - // Suspend / Resume execution based on the selection + std::set running_pids; + for (const auto &pair : best_pids) { + PID best_pid = pair.second; + auto &node = mlfq_infos_[best_pid]; + if (!node.is_running) { + node.is_running = true; + node.last_resume_time = now; + node.last_pending_start = now; + } else { + node.last_resume_time = now; + } + running_pids.insert(best_pid); + } + for (auto &st : status.xqueue_status) { - XDevice device = st.second->device; - XQueueHandle handle = st.second->handle; - auto &node = mlfq_infos_[handle]; + PID pid = st.second->pid; + auto &node = mlfq_infos_[pid]; if (!st.second->ready) { node.is_running = false; continue; } - if (best_handles[device] == handle) { - // The chosen one - if (!node.is_running) { - this->Resume(handle); - node.is_running = true; - node.last_resume_time = now; - node.last_pending_start = now; // no longer pending - } else { - // already running - node.last_resume_time = now; - } - } else { - // Not chosen, must wait + if (!running_pids.count(pid)) { if (node.is_running) { - this->Suspend(handle); node.is_running = false; node.last_pending_start = now; } } } - // Schedule next timer interrupt to keep evaluating time slices + for (auto &st : status.xqueue_status) { + PID pid = st.second->pid; + if (!st.second->ready) continue; + + if (running_pids.count(pid)) { + this->Resume(st.first); + } else { + this->Suspend(st.first); + } + } + if (has_ready_tasks) { this->AddTimer(now + default_tick_); } @@ -169,6 +164,4 @@ void MultiLevelFeedbackQueuePolicy::Sched(const Status &status) void MultiLevelFeedbackQueuePolicy::RecvHint(std::shared_ptr hint) { (void)hint; - // Not strictly needed for autonomous Soft Priority Recovery, - // but can be implemented to support manual user priorities if required. -} \ No newline at end of file +} diff --git a/sched/src/policy/up.cpp b/sched/src/policy/up.cpp index 52783fed..ed5c3ccc 100644 --- a/sched/src/policy/up.cpp +++ b/sched/src/policy/up.cpp @@ -5,85 +5,53 @@ using namespace xsched::sched; void UtilizationPartitionPolicy::Sched(const Status &status) { - // delete all destroyed xqueues and their utilizations for (auto it = utils_.begin(); it != utils_.end();) { - XQueueHandle xqueue = it->first; - if (status.xqueue_status.find(xqueue) == status.xqueue_status.end()) { + PID pid = it->first; + if (status.process_status.find(pid) == status.process_status.end()) { it = utils_.erase(it); } else { ++it; } } - // delete destroyed current running xqueue + for (const auto &xq : status.xqueue_status) { + handle_to_pid_[xq.first] = xq.second->pid; + } + if (cur_running_ != 0 && - status.xqueue_status.find(cur_running_) == status.xqueue_status.end()) { + status.process_status.find(cur_running_) == status.process_status.end()) { cur_running_ = 0; } if (utils_.empty()) return; if (cur_running_ == 0) { - // nothing is running SwitchToAny(status); return; } - auto xit = status.xqueue_status.find(cur_running_); - if (xit == status.xqueue_status.end()) { - // current xqueue is not found - auto bit = utils_.find(cur_running_); - XASSERT(bit != utils_.end(), - "utilization of XQueue 0x" FMT_64X " not found", cur_running_); - utils_.erase(bit); + auto now = std::chrono::system_clock::now(); + if (now < cur_end_ && ProcessReady(cur_running_, status)) return; + + auto bit = utils_.find(cur_running_); + if (bit == utils_.end()) { SwitchToAny(status); return; } - auto now = std::chrono::system_clock::now(); - if (now < cur_end_ && xit->second->ready) return; - - auto bit = utils_.find(cur_running_); - XASSERT(bit != utils_.end(), - "utilization of XQueue 0x" FMT_64X " not found", cur_running_); - - // current xqueue has finished its time slice - // select the next xqueue to run - for (++bit; bit != utils_.end();) { - auto xit = status.xqueue_status.find(bit->first); - if (xit == status.xqueue_status.end()) { - // xqueue not found - utils_.erase(bit++); - continue; - } - if (!xit->second->ready) { - // not running check the next one - ++bit; - continue; - } - SwitchTo(bit->first, bit->second, status); + for (++bit; bit != utils_.end(); ++bit) { + if (!ProcessReady(bit->first, status)) continue; + SwitchProcess(bit->first, bit->second, status); return; } - for (bit = utils_.begin(); bit != utils_.end();) { + for (bit = utils_.begin(); bit != utils_.end(); ++bit) { if (bit->first == cur_running_) break; - - auto xit = status.xqueue_status.find(bit->first); - if (xit == status.xqueue_status.end()) { - // xqueue not found - utils_.erase(bit++); - continue; - } - if (!xit->second->ready) { - // not running check the next one - ++bit; - continue; - } - SwitchTo(bit->first, bit->second, status); + if (!ProcessReady(bit->first, status)) continue; + SwitchProcess(bit->first, bit->second, status); return; } - // checked a round, no xqueue to run cur_running_ = 0; } @@ -100,7 +68,15 @@ void UtilizationPartitionPolicy::RecvHint(std::shared_ptr hint) XWARN("invalid utilization %d", util); break; } - utils_[h->Handle()] = h->Util(); + PID pid = h->Pid(); + if (pid != 0) { + utils_[pid] = util; + } else { + auto it = handle_to_pid_.find(h->Handle()); + if (it != handle_to_pid_.end()) { + utils_[it->second] = util; + } + } break; } case kHintTypeTimeslice: @@ -120,45 +96,74 @@ std::chrono::microseconds UtilizationPartitionPolicy::GetBudget(Utilization util { Utilization total_util = 0; int64_t totalUs = timeslice_.count(); - for (const auto &xqueue : utils_) { total_util += xqueue.second; } + for (const auto &process : utils_) { total_util += process.second; } if(total_util == 0) return std::chrono::microseconds(TIMESLICE_DEFAULT); return std::chrono::microseconds(totalUs * util / total_util); } +bool UtilizationPartitionPolicy::ProcessReady(PID pid, const Status &status) +{ + auto it = status.process_status.find(pid); + if (it == status.process_status.end()) return false; + for (auto xq : it->second->running_xqueues) { + auto xit = status.xqueue_status.find(xq); + if (xit != status.xqueue_status.end() && xit->second->ready) return true; + } + for (auto xq : it->second->suspended_xqueues) { + auto xit = status.xqueue_status.find(xq); + if (xit != status.xqueue_status.end() && xit->second->ready) return true; + } + return false; +} + void UtilizationPartitionPolicy::SwitchToAny(const Status &status) { cur_running_ = 0; - bool selected = false; - for (const auto &xqueue : status.xqueue_status) { - if (selected || !xqueue.second->ready) { - this->Suspend(xqueue.first); + for (const auto &process : status.process_status) { + PID pid = process.first; + if (!ProcessReady(pid, status)) { + for (const auto xq : process.second->running_xqueues) + this->Suspend(xq); + for (const auto xq : process.second->suspended_xqueues) + this->Suspend(xq); continue; } - auto it = utils_.find(xqueue.first); + auto it = utils_.find(pid); if (it == utils_.end()) { - this->Suspend(xqueue.first); + for (const auto xq : process.second->running_xqueues) + this->Suspend(xq); + for (const auto xq : process.second->suspended_xqueues) + this->Suspend(xq); continue; } - selected = true; - this->Resume(xqueue.first); - cur_running_ = xqueue.first; - cur_end_ = std::chrono::system_clock::now() + GetBudget(it->second); - this->AddTimer(cur_end_); + SwitchProcess(pid, it->second, status); + return; } } void -UtilizationPartitionPolicy::SwitchTo(XQueueHandle handle, Utilization util, const Status &status) +UtilizationPartitionPolicy::SwitchProcess(PID pid, Utilization util, const Status &status) { - for (const auto &xqueue : status.xqueue_status) { - if (xqueue.first == handle) continue; - this->Suspend(xqueue.first); + for (const auto &process : status.process_status) { + const auto it = status.process_status.find(process.first); + if (it == status.process_status.end()) continue; + + if (process.first == pid) { + std::list suspended_xqueues; + for (const auto xq : it->second->suspended_xqueues) + suspended_xqueues.push_back(xq); + for (const auto xq : suspended_xqueues) { this->Resume(xq); } + } else { + std::list running_xqueues; + for (const auto xq : it->second->running_xqueues) + running_xqueues.push_back(xq); + for (const auto xq : running_xqueues) { this->Suspend(xq); } + } } - this->Resume(handle); - cur_running_ = handle; + cur_running_ = pid; cur_end_ = std::chrono::system_clock::now() + GetBudget(util); this->AddTimer(cur_end_); } diff --git a/sched/src/scheduler/local.cpp b/sched/src/scheduler/local.cpp index 9028451e..3be9e0df 100644 --- a/sched/src/scheduler/local.cpp +++ b/sched/src/scheduler/local.cpp @@ -106,6 +106,8 @@ void LocalScheduler::SetPolicy(XPolicyType type) policy_->SetResumeFunc(std::bind(&LocalScheduler::Resume, this, std::placeholders::_1)); policy_->SetAddTimerFunc(std::bind(&LocalScheduler::AddTimer, this, std::placeholders::_1)); for (auto &status : status_.xqueue_status) Resume(status.first); + policy_->Sched(status_); + ExecuteOperations(); this->Run(); XINFO("policy changed from %s to %s", old.c_str(), GetPolicyTypeName(policy_type_).c_str()); }