LoadBalancer.cpp

Go to the documentation of this file.

 
#include "LoadBalancer.h"
#include "Logger.h"
#include <iostream>
#include <sstream>
#include <iomanip>
#include <algorithm>

const char PROCESS_JOB = 'P';
const char STREAMING_JOB = 'S';
 
namespace {
    const char* RESET = "\033[0m";
    const char* CYAN = "\033[36m";
    const char* GREEN = "\033[32m";
    const char* YELLOW = "\033[33m";
    const char* MAGENTA = "\033[35m";
    const char* BOLD = "\033[1m";
    const char* RED = "\033[31m";
}
 
void LoadBalancer::log(const char* colorCode, const char* msg) const {
    std::ostringstream oss;
    if (config_.colorsEnabled) {
        oss << CYAN << "[" << RESET << BOLD << currentTime_ << RESET << CYAN << "]" << RESET
            << " " << colorCode << msg << RESET << "\n";
    } else {
        oss << "[" << currentTime_ << "] " << msg << "\n";
    }
    Logger::log(oss.str());
}
 
 
LoadBalancer::LoadBalancer(int initialServerCount, const LoadBalancerConfig& config)
 : config_(config), 
 currentTime_(0), 
 minQueueThreshold_(config.minThreshold), 
 maxQueueThreshold_(config.maxThreshold), 
 lastScaleTime_(0), 
 nextServerId_(1),
 rng_(std::random_device()()), 
 requestDist_(config.requestProbability), 
 timeDist_(config.minProcessingTime, config.maxProcessingTime), 
 octetDist_(0, 255), 
 jobTypeDist_(0, 1),
 totalProcessed_(0),
 totalBlocked_(0),
 requestsAddedToQueue_(0),
 serversAdded_(0),
 serversRemoved_(0)
  {
    for (int i = 0; i < initialServerCount; ++i) {
        servers_.push_back(std::make_unique<WebServer>(nextServerId_++));
    }
    minQueueThreshold_ = config.minThreshold * static_cast<int>(servers_.size()); 
    maxQueueThreshold_ = config.maxThreshold * static_cast<int>(servers_.size()); 
}
 
std::string LoadBalancer::generateRandomIP() {
    std::ostringstream oss;
    oss << octetDist_(rng_) << "." << octetDist_(rng_) << "." 
    << octetDist_(rng_) << "." << octetDist_(rng_);
    return oss.str();
}
 
Request LoadBalancer::createRandomRequest() {
    static const char jobTypes[] = {PROCESS_JOB, STREAMING_JOB};
    Request request;
    request.ipIn = generateRandomIP();
    request.ipOut = generateRandomIP();
    request.jobType = jobTypes[jobTypeDist_(rng_)];
    request.timeRequired = timeDist_(rng_);
    return request;
}
 
void LoadBalancer::prefillQueue(int count) {
    for (int i = 0; i < count; ++i) {
        requestQueue_.push(createRandomRequest());
    }
    if (count > 0) {
        std::ostringstream oss;
        oss << "Prefilled queue with " << count << " requests";
        log(GREEN, oss.str().c_str());
    }
}
 
void LoadBalancer::generateRequest() {
    if (requestDist_(rng_)) {
        Request req = createRandomRequest();
 
        std::vector<int> blockedOctets;
        if (!config_.blockedOctets.empty()) {
            blockedOctets = config_.blockedOctets;
        } else {
            blockedOctets = {192, 255};
        }
 
        auto isBlockedFirstOctet = [](int octet, const std::vector<int>& blocked) -> bool {
            return std::find(blocked.begin(), blocked.end(), octet) != blocked.end();
        };
 
        int firstOctet = std::stoi(req.ipIn.substr(0, req.ipIn.find('.')));
        if (isBlockedFirstOctet(firstOctet, blockedOctets)) {
            std::ostringstream oss;
            oss << "Request blocked (bad octet) from " << req.ipIn;
            ++totalBlocked_;
            log(RED, oss.str().c_str());
            return;
        }
       
        else {
            requestQueue_.push(req);
            ++requestsAddedToQueue_;
            std::ostringstream oss;
            oss << "New request added (" << req.jobType << ") from " << req.ipIn;
            log(GREEN, oss.str().c_str());
        }
    }
    
 
}
 
void LoadBalancer::assignRequests() {
    for (auto& server : servers_) {
        if (!server->isBusy() && !requestQueue_.empty()) {
            Request req = requestQueue_.front();
            requestQueue_.pop();
            server->assignRequest(req);
        }
    }
}
 
void LoadBalancer::scaleServers() {
    int numServers = static_cast<int>(servers_.size());
    int queueSize = static_cast<int>(requestQueue_.size());
    minQueueThreshold_ = config_.minThreshold * numServers;
    maxQueueThreshold_ = config_.maxThreshold * numServers;
 
    if (currentTime_ - lastScaleTime_ < config_.scaleCooldown) return;
 
    if (queueSize < minQueueThreshold_ && numServers > 1) {
        auto it = std::find_if(servers_.begin(), servers_.end(),
            [](const std::unique_ptr<WebServer>& s) { return !s->isBusy(); });
        if (it != servers_.end()) {
            servers_.erase(it);
            lastScaleTime_ = currentTime_;
            ++serversRemoved_;
            std::ostringstream oss;
            oss << "Removed server. Active: " << static_cast<int>(servers_.size())
                << " | Added: " << serversAdded_ << " | Removed: " << serversRemoved_
                << " (Queue: " << queueSize << ")";
            log(YELLOW, oss.str().c_str());
        }
    } else if (queueSize > maxQueueThreshold_) {
        servers_.push_back(std::make_unique<WebServer>(nextServerId_++));
        lastScaleTime_ = currentTime_;
        ++serversAdded_;
        std::ostringstream oss;
        oss << "Added server. Active: " << static_cast<int>(servers_.size())
            << " | Added: " << serversAdded_ << " | Removed: " << serversRemoved_
            << " (Queue: " << queueSize << ")";
        log(YELLOW, oss.str().c_str());
    }
}
 
void LoadBalancer::run(int totalClockCycles) {
    for (; currentTime_ < totalClockCycles; ++currentTime_) {
        generateRequest();
 
        assignRequests();
 
        scaleServers();
 
        if (config_.statusInterval > 0 && currentTime_ % config_.statusInterval == 0) {
            std::ostringstream oss;
            int active = static_cast<int>(servers_.size());
            if (config_.colorsEnabled) {
                oss << CYAN << "[" << RESET << BOLD << currentTime_ << RESET << CYAN << "]" << RESET
                    << " " << BOLD << "Active: " << active << RESET << " | "
                    << GREEN << "Added: " << serversAdded_ << RESET << " | "
                    << YELLOW << "Removed: " << serversRemoved_ << RESET << "\n";
            } else {
                oss << "[" << currentTime_ << "] Active: " << active
                    << " | Added: " << serversAdded_ << " | Removed: " << serversRemoved_ << "\n";
            }
            Logger::log(oss.str());
        }
 
        for (auto& server : servers_) {
            server->process();
            if (server->isFinished()) {
                std::ostringstream oss;
                oss << "Server " << server->getId() << " finished request";
                log(MAGENTA, oss.str().c_str());
                ++totalProcessed_;
                server->clear();
            }
        }
 
        
    }
 
}
 
 
 
 
void LoadBalancer::printSummary(int initialServerCount) const {
    int finalServerCount = servers_.size();
    std::ostringstream oss;
    if (config_.colorsEnabled) {
        oss << "\n" << BOLD << "Simulation Summary:" << RESET << "\n"
            << "Task time range: " << config_.minProcessingTime << " - " << config_.maxProcessingTime << "\n"
            << MAGENTA << "Total Requests Processed: " << totalProcessed_ << RESET << "\n"
            << YELLOW << "Total Requests still in Queue: " << requestQueue_.size() << RESET << "\n"
            << GREEN << "Total Requests Added to Queue: " << requestsAddedToQueue_ << RESET << "\n"
            << RED << "Total Requests Blocked: " << totalBlocked_ << RESET << "\n"
            << GREEN << "Servers Added: " << serversAdded_ << RESET << "\n"
            << YELLOW << "Servers Removed: " << serversRemoved_ << RESET << "\n"
            << BOLD << "Started with " << initialServerCount << " servers" << RESET << "\n"
            << BOLD << "Ended with " << finalServerCount << " servers" << RESET << "\n\n";
    } else {
        oss << "\nSimulation Summary:\n"
            << "Task time range: " << config_.minProcessingTime << " - " << config_.maxProcessingTime << "\n"
            << "Total Requests Processed: " << totalProcessed_ << "\n"
            << "Total Requests still in Queue: " << requestQueue_.size() << "\n"
            << "Total Requests Added to Queue: " << requestsAddedToQueue_ << "\n"
            << "Total Requests Blocked: " << totalBlocked_ << "\n"
            << "Servers Added: " << serversAdded_ << "\n"
            << "Servers Removed: " << serversRemoved_ << "\n"
            << "Started with " << initialServerCount << " servers\n"
            << "Ended with " << finalServerCount << " servers\n\n";
    }
    Logger::log(oss.str());
}