scheduler.cpp

#include "scheduler.h"
#include <sstream>
#include <iomanip>
#include "config.h"
#include "VMstat.h"


Scheduler::Scheduler(int count, const std::string& schedulerType, int timeQuantum, int delayPerInstruction, IMemoryAllocator* allocator)
    : coreCount(count),
      schedulerType(schedulerType),
      timeQuantum(timeQuantum),
      delayPerInstruction(delayPerInstruction),
      allocator(allocator),
      cores(count, nullptr)
{}

//override scheduler constructor to handle case when no timeQuantum is given if type is fcfs
Scheduler::Scheduler(int count, const std::string& schedulerType, int delayPerInstruction, IMemoryAllocator* allocator) 
    : coreCount(count), 
      schedulerType(schedulerType), 
      delayPerInstruction(delayPerInstruction),
      cores(count, nullptr),
      allocator(allocator)
{}


void Scheduler::addProcess(Process* p) 
{
    std::unique_lock<std::mutex> lock(mtx); 
    processQueue.push(p);
    cv.notify_all();
}

// Starts the scheduler by launching a thread for each CPU core.
void Scheduler::start() 
{
    for (int i = 0; i < coreCount; ++i) 
    {
        coreThreads.emplace_back(&Scheduler::coreWorker, this, i);
    }
}

// The main function executed by each CPU core thread.
// Implements the First-Come, First-Served (FCFS) scheduling logic.
void Scheduler::coreWorker(int coreId)
{
    while (true)
    {
        Process* currentProcess = nullptr;
        {
            std::unique_lock<std::mutex> lock(mtx);
            cv.wait(lock, [this] {
                return !processQueue.empty() || shutdown;
            });

            if (shutdown && processQueue.empty()) {
                if (cores[coreId] != nullptr) {
                    // If this core is still running a process, deallocate it
                    allocator->deallocate(cores[coreId]->getPID());
                }

                cores[coreId] = nullptr;
                
                // check if ALL cores are now idle
                if (getUsedCores() == 0) {
                    break; // safe to exit this core
                } else {
                    // wait again for the last one to exit
                    std::this_thread::sleep_for(std::chrono::milliseconds(50));
                    continue;
                }
            }

            // If queue is not empty, assign a process
            if (!processQueue.empty()) {
                currentProcess = processQueue.front();
                processQueue.pop();
                

                // Allocate memory if needed
                //check if all pages are already there?
                void* mem = nullptr;
                if (currentProcess->getMemoryPtr() == nullptr && schedulerType == "fcfs") {
                    if (allocator->getFreeFrameList().size() == 0){
                        currentProcess->setMemoryPtr(nullptr); 
                        processQueue.push(currentProcess);
                        {
                        //update vmstat
                        std::lock_guard<std::mutex> lock(vmstatMutex);
                        vmstats.idleCPUticks += 1; // Increment active CPU ticks
                        }
                        continue;
                    }
                    mem = allocator->allocate(currentProcess);
                } else{
                    mem = allocator->allocate(currentProcess);
                }

                if (mem == nullptr) {
                    {
                        //update vmstat
                        std::lock_guard<std::mutex> lock(vmstatMutex);
                        vmstats.idleCPUticks += 1; // Increment active CPU ticks
                    }
                    processQueue.push(currentProcess);
                    cores[coreId] = nullptr;
                    continue;
                }else {
                    //std::cout << "Core " << coreId << " running: " << currentProcess->getPID() << std::endl;
                    //when can it not run for rr? 
                        //when the processes inside the cores take up all the frames 
                    if (allocator->checkProcessFrames(currentProcess) == false){
                        processQueue.push(currentProcess);
                        cores[coreId] = nullptr;
                        {
                        //update vmstat
                        std::lock_guard<std::mutex> lock(vmstatMutex);
                        vmstats.idleCPUticks += 1; // Increment active CPU ticks
                        }
                        continue;
                    }else{
                        cores[coreId] = currentProcess;
                        currentProcess->setMemoryPtr(mem);
                    }   
                    
                }
            }
        }

        if (currentProcess->getMemoryPtr())
        {
            // Mark as running
            currentProcess->setState(Process::RUNNING);

            if (schedulerType == "fcfs")
            {
                while (!currentProcess->isFinished())
                {
                    if (shutdown) break;
                    currentProcess->executeCurrentCommand(coreId);
                    currentProcess->moveToNextLine();
                    std::this_thread::sleep_for(std::chrono::milliseconds(delayPerInstruction));
                    {
                        //update vmstat
                        std::lock_guard<std::mutex> lock(vmstatMutex);
                        vmstats.activeCPUticks += 1; // Increment active CPU ticks
                    }
                }
            }
            else if (schedulerType == "rr")
            {
                for (int i = 0; i < timeQuantum && !currentProcess->isFinished(); ++i)
                {
                    if (shutdown) {
                        break;
                    }
                    //std::cout << "Core " << coreId << " executing process: " << currentProcess->getName() << std::endl;
                    if (allocator->checkProcessFrames(currentProcess) == false) 
                    {
                        break; // skip execution if process is not in memory
                    }
                    currentProcess->executeCurrentCommand(coreId);
                    currentProcess->moveToNextLine();
                    std::this_thread::sleep_for(std::chrono::milliseconds(delayPerInstruction));
                    //std::cout << cores[coreId]->getPID() << " running on core " << coreId << std::endl;
                    //std::this_thread::sleep_for(std::chrono::milliseconds(3000)); // Simulate some delay for output clarity
                }

                {
                    std::unique_lock<std::mutex> lock(mtx);
                    quantumCounter++;
                }
                {
                        //update vmstat
                        std::lock_guard<std::mutex> lock(vmstatMutex);
                        vmstats.activeCPUticks += 1; // Increment active CPU ticks
                }
            }

            std::unique_lock<std::mutex> lock(mtx);
            if (currentProcess->isFinished()) 
            {
                currentProcess->setState(Process::FINISHED);
                currentProcess->setFinishTime(currentTimestamp());
                finishedProcesses.push_back(currentProcess);
                allocator->deallocate(currentProcess->getPID());
            }
            else
            {
                currentProcess->setState(Process::READY);
                processQueue.push(currentProcess);  // RR puts it back if not finished
            }

            
            cores[coreId] = nullptr;
            //update vmstat
           
        }else{
                std::lock_guard<std::mutex> lock(vmstatMutex); 
                vmstats.idleCPUticks+= 1; // Increment active CPU ticks
        }
    }
}

void Scheduler::stop() 
{
    {
        std::unique_lock<std::mutex> lock(mtx); // Acquire lock
        this->shutdown = true; // Set shutdown flag
    } // Lock is released here
    cv.notify_all(); // Notify all waiting core threads
}

// Waits for all core threads to finish their execution.
void Scheduler::wait() 
{
    for (auto& t : coreThreads) 
    {
        if (t.joinable()) t.join(); // Join each thread
    }
}

// Returns the current timestamp formatted as YYYY-MM-DD HH:MM:SS.
std::string Scheduler::currentTimestamp() 
{
    auto now = std::chrono::system_clock::now();
    auto in_time_t = std::chrono::system_clock::to_time_t(now);

    std::stringstream ss;
    // Use std::put_time for formatting the time
    ss << std::put_time(std::localtime(&in_time_t), "(%m/%d/%Y %H:%M:%S %p)");
    return ss.str();
}

// Returns a vector of strings, each representing the status of a CPU core.
std::vector<std::string> Scheduler::getCoreStatus()
{
    std::vector<std::string> status;
    std::unique_lock<std::mutex> lock(mtx); // Acquire lock to access cores vector

    for (int i = 0; i < coreCount; ++i)
    {
        if (cores[i] != nullptr)
        {
            status.push_back("Core " + std::to_string(i) + ": Running " + cores[i]->getName());
        }
        else
        {
            status.push_back("Core " + std::to_string(i) + ": Idle");
        }
    }
    return status;
}

std::vector<std::string> Scheduler::getRunningProcesses()
{
    std::vector<std::string> status;
    std::unique_lock<std::mutex> lock(mtx); // Lock to safely access cores

    for (int i = 0; i < cores.size(); ++i)
    {
        if (cores[i])  // if core is working on a process
        {
            Process* p = cores[i];
            int done = p->getCommandCounter();
            int total = p->getLinesOfCode();
            std::string time_now = currentTimestamp();

            status.push_back(p->getName() + "   " + time_now + "   Core: " + std::to_string(i) + "   " +
                             std::to_string(done) + "/" + std::to_string(total));
        }
    }
    return status;
}

std::unordered_map<int, std::string> Scheduler::getRunningProcessDetails()
{
    std::unique_lock<std::mutex> lock(mtx); // Lock to safely access cores
    std::unordered_map<int, std::string> runningDetails;

    for (int i = 0; i < cores.size(); ++i)
    {
        if (cores[i])  // if core is working on a process
        {
            Process* p = cores[i];
            runningDetails[p->getPID()] = p->getName(); 
        }
    }
    return runningDetails; 
}


// Returns a vector of strings, each representing a finished process.
std::vector<std::string> Scheduler::getFinishedProcesses()
{
    std::vector<std::string> status;
    std::unique_lock<std::mutex> lock(mtx); // Acquire lock

    for (const auto& p : finishedProcesses)
    {
        int total = p->getLinesOfCode();
        status.push_back(p->getName() + "   " + p->getFinishTime() + "   Finished   " +
                         std::to_string(total) + "/" + std::to_string(total));
    }

    return status;
}

int Scheduler::getUsedCores() {
    std::unique_lock<std::mutex> lock(mtx);
    int used = 0;
    for (const auto& p : cores) {
        if (p != nullptr) used++;
    }
    return used;
}