LoadBalancer.cpp

Go to the documentation of this file.

 
#include "LoadBalancer.h"
 
#include <algorithm>
#include <iomanip>
#include <ostream>
#include <sstream>
#include <stdexcept>
 
LoadBalancer::LoadBalancer(std::string name, std::size_t initialServers, const ScalingPolicy& policy,
                           const Firewall& firewall, std::size_t maxQueue, Logger& logger)
    : name_(std::move(name)), policy_(policy), firewall_(firewall), maxQueue_(maxQueue), logger_(logger) {
    if (initialServers == 0) {
        throw std::invalid_argument("LoadBalancer needs at least one server");
    }
    for (std::size_t i = 0; i < initialServers; ++i) {
        servers_.emplace_back(nextServerId_++, 0);
    }
    stats_.initialServers = initialServers;
    stats_.minActiveServers = initialServers;
    stats_.peakLiveServers = initialServers;
}
 
std::string LoadBalancer::describe(const Request& request) const {
    std::ostringstream text;
    text << "req#" << request.id << " " << request.sourceIp << " -> " << request.destinationIp << " ("
         << jobTypeCode(request.jobType) << ", " << request.taskCycles << " cycles)";
    return text.str();
}
 
bool LoadBalancer::admit(const Request& request, unsigned long cycle) {
    ++stats_.received;
    if (firewall_.isBlocked(request.sourceIp)) {
        ++stats_.rejected;
        logger_.log(LogCategory::Reject, cycle, name_, describe(request) + " blocked by firewall");
        return false;
    }
    if (maxQueue_ != 0 && pending_.size() >= maxQueue_) {
        ++stats_.discarded;
        logger_.log(LogCategory::Discard, cycle, name_, describe(request) + " dropped, queue full");
        return false;
    }
    pending_.push(request);
    ++stats_.admitted;
    if (pending_.size() > stats_.peakQueue) {
        stats_.peakQueue = pending_.size();
        stats_.peakQueueCycle = cycle;
    }
    if (cycle > 0) {  // the initial queue is summarized instead of logged one by one
        logger_.log(LogCategory::Arrival, cycle, name_,
                    describe(request) + " queued, queue=" + std::to_string(pending_.size()));
    }
    return true;
}
 
void LoadBalancer::start() {
    stats_.startingQueue = pending_.size();
    logger_.log(LogCategory::Info, 0, name_,
                "starting with " + std::to_string(servers_.size()) + " servers and " +
                    std::to_string(pending_.size()) + " queued requests");
    dispatch(0);
    stats_.queueAfterFirstDispatch = pending_.size();
}
 
void LoadBalancer::processCycle(unsigned long cycle, const std::vector<Request>& arrivals) {
    completeWork(cycle);
    for (const Request& request : arrivals) {
        admit(request, cycle);
    }
    adjustCapacity(cycle);
    dispatch(cycle);
    updateExtremes(cycle);
}
 
void LoadBalancer::completeWork(unsigned long cycle) {
    stats_.serverCycles += servers_.size();
    for (WebServer& server : servers_) {
        if (server.isIdle()) {
            continue;
        }
        ++stats_.busyCycles;
        if (server.tick()) {
            ++stats_.completed;
            logger_.log(LogCategory::Complete, cycle, name_,
                        "server " + std::to_string(server.id()) + " finished " +
                            describe(server.lastCompleted()));
        }
    }
    // Remove retiring servers that have now finished their last request.
    for (auto it = servers_.begin(); it != servers_.end();) {
        if (it->isRetiring() && it->isIdle()) {
            const unsigned int id = it->id();
            it = servers_.erase(it);
            recordScale(ScaleEvent::Kind::Retired, cycle, id);
        } else {
            ++it;
        }
    }
}
 
void LoadBalancer::adjustCapacity(unsigned long cycle) {
    if (policy_.cooldownCycles == 0 || cycle % policy_.cooldownCycles != 0) {
        return;
    }
    const double queue = static_cast<double>(pending_.size());
    const std::size_t active = activeServers();
 
    if (queue > policy_.upperFactor * static_cast<double>(active)) {
        // Prefer returning a retiring server to service: it already exists.
        for (WebServer& server : servers_) {
            if (server.isRetiring()) {
                server.setRetiring(false);
                ++stats_.scaleUps;
                recordScale(ScaleEvent::Kind::Reactivated, cycle, server.id());
                return;
            }
        }
        if (servers_.size() < policy_.maxServers) {
            addServer(cycle);
        }
        return;
    }
 
    if (queue < policy_.lowerFactor * static_cast<double>(active) && active > policy_.minServers) {
        // Prefer removing an idle server immediately (search from the newest).
        for (auto it = servers_.rbegin(); it != servers_.rend(); ++it) {
            if (it->isIdle() && !it->isRetiring()) {
                const unsigned int id = it->id();
                servers_.erase(std::next(it).base());
                ++stats_.scaleDowns;
                recordScale(ScaleEvent::Kind::Removed, cycle, id);
                return;
            }
        }
        // Otherwise retire the busy server that will finish soonest.
        WebServer* candidate = nullptr;
        for (WebServer& server : servers_) {
            if (!server.isRetiring() &&
                (candidate == nullptr || server.remainingCycles() < candidate->remainingCycles())) {
                candidate = &server;
            }
        }
        if (candidate != nullptr) {
            candidate->setRetiring(true);
            ++stats_.scaleDowns;
            recordScale(ScaleEvent::Kind::Retiring, cycle, candidate->id());
        }
    }
}
 
void LoadBalancer::addServer(unsigned long cycle) {
    servers_.emplace_back(nextServerId_++, cycle);
    ++stats_.scaleUps;
    recordScale(ScaleEvent::Kind::Added, cycle, servers_.back().id());
}
 
void LoadBalancer::recordScale(ScaleEvent::Kind kind, unsigned long cycle, unsigned int serverId) {
    ScaleEvent event;
    event.cycle = cycle;
    event.kind = kind;
    event.serverId = serverId;
    event.queueSize = pending_.size();
    event.activeServers = activeServers();
    scaleEvents_.push_back(event);
 
    std::string action;
    LogCategory category = LogCategory::ScaleDown;
    switch (kind) {
        case ScaleEvent::Kind::Added:
            action = "added server ";
            category = LogCategory::ScaleUp;
            break;
        case ScaleEvent::Kind::Reactivated:
            action = "reactivated retiring server ";
            category = LogCategory::ScaleUp;
            break;
        case ScaleEvent::Kind::Removed: action = "removed idle server "; break;
        case ScaleEvent::Kind::Retiring: action = "retiring busy server "; break;
        case ScaleEvent::Kind::Retired: action = "retired server "; break;
    }
    std::ostringstream text;
    text << action << serverId << " | queue=" << pending_.size() << " active=" << activeServers()
         << " live=" << servers_.size();
    logger_.log(category, cycle, name_, text.str());
}
 
void LoadBalancer::dispatch(unsigned long cycle) {
    for (WebServer& server : servers_) {
        if (pending_.empty()) {
            return;
        }
        if (!server.isIdle() || server.isRetiring()) {
            continue;
        }
        const Request request = pending_.front();
        pending_.pop();
        const unsigned long wait = cycle - request.arrivalCycle;
        stats_.totalWait += wait;
        stats_.maxWait = std::max(stats_.maxWait, wait);
        ++stats_.dispatched;
        server.assign(request, cycle);
        logger_.log(LogCategory::Assign, cycle, name_,
                    describe(request) + " -> server " + std::to_string(server.id()) + " after waiting " +
                        std::to_string(wait) + " cycles");
    }
}
 
void LoadBalancer::updateExtremes(unsigned long /*cycle*/) {
    stats_.minActiveServers = std::min(stats_.minActiveServers, activeServers());
    stats_.peakLiveServers = std::max(stats_.peakLiveServers, servers_.size());
}
 
void LoadBalancer::recordSnapshot(unsigned long cycle) {
    Snapshot snapshot;
    snapshot.cycle = cycle;
    snapshot.queueSize = pending_.size();
    snapshot.liveServers = servers_.size();
    snapshot.busyServers = busyServers();
    snapshot.idleServers = idleServers();
    snapshot.retiringServers = retiringServers();
    snapshot.completed = stats_.completed;
    snapshot.rejected = stats_.rejected;
    snapshots_.push_back(snapshot);
}
 
std::string LoadBalancer::statusLine(unsigned long /*cycle*/) const {
    std::ostringstream text;
    text << "queue=" << std::setw(5) << pending_.size() << " | servers=" << std::setw(3) << servers_.size()
         << " (busy " << busyServers() << ", idle " << idleServers() << ", retiring " << retiringServers()
         << ") | completed=" << stats_.completed << " rejected=" << stats_.rejected
         << " discarded=" << stats_.discarded << " | scale +" << stats_.scaleUps << "/-" << stats_.scaleDowns;
    return text.str();
}
 
void LoadBalancer::summarize(std::ostream& out) const {
    const auto row = [&out](const std::string& label, const std::string& value) {
        out << "  " << std::left << std::setw(46) << label << value << '\n';
    };
    const auto fixed = [](double value, int precision) {
        std::ostringstream text;
        text << std::fixed << std::setprecision(precision) << value;
        return text.str();
    };
 
    out << "Load balancer " << name_ << '\n';
    out << "  -- Queue --\n";
    row("Starting queue size (t=0, before dispatch)", std::to_string(stats_.startingQueue));
    row("Queue size after first dispatch (t=0)", std::to_string(stats_.queueAfterFirstDispatch));
    row("Ending queue size", std::to_string(pending_.size()));
    row("Peak queue size", std::to_string(stats_.peakQueue) + " (cycle " + std::to_string(stats_.peakQueueCycle) + ")");
 
    out << "  -- Servers --\n";
    row("Initial servers", std::to_string(stats_.initialServers));
    row("Final live servers", std::to_string(servers_.size()));
    row("  Active (busy) servers at end", std::to_string(busyServers() - std::min(busyServers(), retiringServers())));
    row("  Inactive (idle) servers at end", std::to_string(idleServers()));
    row("  Retiring servers at end (still busy)", std::to_string(retiringServers()));
    row("Fewest active / most live servers", std::to_string(stats_.minActiveServers) + " / " +
                                                  std::to_string(stats_.peakLiveServers));
    row("Scale-up events (added + reactivated)", std::to_string(stats_.scaleUps));
    row("Scale-down events (removed + retiring)", std::to_string(stats_.scaleDowns));
    const double utilization =
        stats_.serverCycles == 0 ? 0.0 : 100.0 * static_cast<double>(stats_.busyCycles) / stats_.serverCycles;
    row("Server utilization (busy / available)", fixed(utilization, 1) + "%");
 
    out << "  -- Requests --\n";
    row("Received (initial queue + arrivals)", std::to_string(stats_.received));
    row("Admitted to queue", std::to_string(stats_.admitted));
    row("Rejected by firewall", std::to_string(stats_.rejected));
    row("Discarded (queue full)", std::to_string(stats_.discarded));
    row("Dispatched to servers", std::to_string(stats_.dispatched));
    row("Completed", std::to_string(stats_.completed));
    row("In service at end (unfinished)", std::to_string(inService()));
    row("Remaining in queue at end", std::to_string(pending_.size()));
    const double averageWait =
        stats_.dispatched == 0 ? 0.0 : static_cast<double>(stats_.totalWait) / stats_.dispatched;
    row("Average / longest wait before dispatch", fixed(averageWait, 1) + " / " + std::to_string(stats_.maxWait) + " cycles");
 
    const bool receivedBalanced = stats_.received == stats_.admitted + stats_.rejected + stats_.discarded;
    const bool admittedBalanced = stats_.admitted == stats_.completed + pending_.size() + inService();
    out << "  -- Consistency checks --\n";
    row("received = admitted + rejected + discarded",
        std::to_string(stats_.received) + " = " + std::to_string(stats_.admitted) + " + " +
            std::to_string(stats_.rejected) + " + " + std::to_string(stats_.discarded) +
            (receivedBalanced ? "  PASS" : "  FAIL"));
    row("admitted = completed + queued + in service",
        std::to_string(stats_.admitted) + " = " + std::to_string(stats_.completed) + " + " +
            std::to_string(pending_.size()) + " + " + std::to_string(inService()) +
            (admittedBalanced ? "  PASS" : "  FAIL"));
}
 
const std::string& LoadBalancer::name() const { return name_; }
 
std::size_t LoadBalancer::queueSize() const { return pending_.size(); }
 
std::size_t LoadBalancer::liveServers() const { return servers_.size(); }
 
std::size_t LoadBalancer::activeServers() const {
    return static_cast<std::size_t>(std::count_if(servers_.begin(), servers_.end(),
                                                  [](const WebServer& s) { return !s.isRetiring(); }));
}
 
std::size_t LoadBalancer::busyServers() const {
    return static_cast<std::size_t>(std::count_if(servers_.begin(), servers_.end(),
                                                  [](const WebServer& s) { return !s.isIdle(); }));
}
 
std::size_t LoadBalancer::idleServers() const { return servers_.size() - busyServers(); }
 
std::size_t LoadBalancer::retiringServers() const {
    return static_cast<std::size_t>(std::count_if(servers_.begin(), servers_.end(),
                                                  [](const WebServer& s) { return s.isRetiring(); }));
}
 
std::size_t LoadBalancer::inService() const { return busyServers(); }
 
const LoadBalancerStats& LoadBalancer::stats() const { return stats_; }
 
const std::vector<ScaleEvent>& LoadBalancer::scaleEvents() const { return scaleEvents_; }
 
const std::vector<Snapshot>& LoadBalancer::snapshots() const { return snapshots_; }
 
const std::vector<WebServer>& LoadBalancer::servers() const { return servers_; }
 
const ScalingPolicy& LoadBalancer::policy() const { return policy_; }
 
const Firewall& LoadBalancer::firewall() const { return firewall_; }