Simulation.cpp

Go to the documentation of this file.

 
#include "Simulation.h"
 
#include <algorithm>
#include <chrono>
#include <cmath>
#include <ctime>
#include <iomanip>
#include <ostream>
#include <sstream>
#include <string>
 
namespace {
 
const char* const kRule = "==============================================================================";
const char* const kThinRule = "------------------------------------------------------------------------------";
 
std::string fixed(double value, int precision) {
    std::ostringstream text;
    text << std::fixed << std::setprecision(precision) << value;
    return text.str();
}
 
void heading(std::ostream& out, const std::string& title) {
    out << '\n' << kThinRule << '\n' << title << '\n' << kThinRule << '\n';
}
 
void row(std::ostream& out, const std::string& label, const std::string& value) {
    out << "  " << std::left << std::setw(46) << label << value << '\n';
}
 
const char* kindName(ScaleEvent::Kind kind) {
    switch (kind) {
        case ScaleEvent::Kind::Added: return "+ added";
        case ScaleEvent::Kind::Reactivated: return "+ reactivated";
        case ScaleEvent::Kind::Removed: return "- removed idle";
        case ScaleEvent::Kind::Retiring: return "- retiring busy";
        case ScaleEvent::Kind::Retired: return "  retired (done)";
    }
    return "";
}
 
Firewall buildFirewall(const Config& config) {
    Firewall firewall;
    for (const std::string& range : config.blockedRanges) {
        firewall.addRange(range);
    }
    return firewall;
}
 
ScalingPolicy buildPolicy(const Config& config) {
    ScalingPolicy policy;
    policy.upperFactor = config.upperFactor;
    policy.lowerFactor = config.lowerFactor;
    policy.cooldownCycles = config.cooldown;
    policy.minServers = config.minServers;
    policy.maxServers = config.maxServers;
    return policy;
}
 
GeneratorSettings buildGeneratorSettings(const Config& config) {
    GeneratorSettings settings;
    settings.minTaskCycles = config.minTask;
    settings.maxTaskCycles = config.maxTask;
    settings.arrivalRate = config.arrivalRate;
    settings.waveAmplitude = config.waveAmplitude;
    settings.wavePeriod = config.autoWavePeriod ? std::max<unsigned long>(4, config.cycles / 4) : config.wavePeriod;
    settings.streamingShare = config.streamingShare;
    return settings;
}
 
}  // namespace
 
Simulation::Simulation(const Config& config, Logger& logger)
    : config_(config),
      logger_(logger),
      generator_(buildGeneratorSettings(config), config.seed),
      statusEvery_(config.statusEvery),
      snapshotEvery_(std::max<unsigned long>(1, config.cycles / 20)) {
    if (statusEvery_ == 0) {
        statusEvery_ = config.verbose ? 100 : std::max<unsigned long>(1, config.cycles / 20);
    }
    const Firewall firewall = buildFirewall(config);
    const ScalingPolicy policy = buildPolicy(config);
    if (config.useSwitch) {
        // Split servers in proportion to the expected job mix; each side keeps at least one.
        std::size_t streamingServers = static_cast<std::size_t>(std::lround(config.servers * config.streamingShare));
        streamingServers = std::clamp<std::size_t>(streamingServers, 1, config.servers - 1);
        const std::size_t processingServers = config.servers - streamingServers;
        switch_ = std::make_unique<Switch>(
            std::make_unique<LoadBalancer>("LB-P", processingServers, policy, firewall, config.maxQueue, logger),
            std::make_unique<LoadBalancer>("LB-S", streamingServers, policy, firewall, config.maxQueue, logger));
    } else {
        single_ = std::make_unique<LoadBalancer>("LB", config.servers, policy, firewall, config.maxQueue, logger);
    }
}
 
void Simulation::run() {
    const auto started = std::chrono::steady_clock::now();
    const std::size_t initialCount = static_cast<std::size_t>(config_.servers) * config_.queuePerServer;
    logger_.log(LogCategory::Info, 0, "SIM",
                "generating initial queue of " + std::to_string(initialCount) + " requests (seed " +
                    std::to_string(config_.seed) + ")");
    for (const Request& request : generator_.makeInitialBatch(initialCount, 0)) {
        if (switch_) {
            switch_->route(request, 0);
        } else {
            single_->admit(request, 0);
        }
    }
    if (switch_) {
        switch_->start();
    } else {
        single_->start();
    }
    for (LoadBalancer* balancer : mutableLoadBalancers()) {
        balancer->recordSnapshot(0);
        logger_.log(LogCategory::Status, 0, balancer->name(), balancer->statusLine(0));
    }
 
    for (unsigned long cycle = 1; cycle <= config_.cycles; ++cycle) {
        const std::vector<Request> arrivals = generator_.arrivalsFor(cycle);
        if (switch_) {
            switch_->processCycle(cycle, arrivals);
        } else {
            single_->processCycle(cycle, arrivals);
        }
        const bool snapshotDue = cycle % snapshotEvery_ == 0 || cycle == config_.cycles;
        const bool statusDue = cycle % statusEvery_ == 0 || cycle == config_.cycles;
        for (LoadBalancer* balancer : mutableLoadBalancers()) {
            if (snapshotDue) {
                balancer->recordSnapshot(cycle);
            }
            if (statusDue) {
                logger_.log(LogCategory::Status, cycle, balancer->name(), balancer->statusLine(cycle));
            }
        }
    }
    runSeconds_ = std::chrono::duration<double>(std::chrono::steady_clock::now() - started).count();
}
 
std::vector<const LoadBalancer*> Simulation::loadBalancers() const {
    if (switch_) {
        return {&switch_->processing(), &switch_->streaming()};
    }
    return {single_.get()};
}
 
std::vector<LoadBalancer*> Simulation::mutableLoadBalancers() {
    if (switch_) {
        return {&switch_->processing(), &switch_->streaming()};
    }
    return {single_.get()};
}
 
const RequestGenerator& Simulation::generator() const {
    return generator_;
}
 
void Simulation::writeSettings(std::ostream& out) const {
    heading(out, "1. RUN SETTINGS (user input)");
    row(out, "Initial web servers", std::to_string(config_.servers));
    row(out, "Clock cycles simulated", std::to_string(config_.cycles));
    row(out, "Output mode", config_.verbose ? "verbose" : "quiet");
    row(out, "Routing", config_.useSwitch ? "Switch: P jobs -> LB-P, S jobs -> LB-S (bonus)" : "single load balancer");
    row(out, "Random seed", std::to_string(config_.seed));
    row(out, "Task time range (uniform)", std::to_string(config_.minTask) + " to " + std::to_string(config_.maxTask) + " cycles");
    row(out, "Arrivals per cycle", "Poisson, average " + fixed(config_.arrivalRate, 2));
    const unsigned long wavePeriod = generator_.settings().wavePeriod;
    if (wavePeriod > 0 && config_.waveAmplitude > 0.0) {
        row(out, "Traffic wave", "rate " + fixed(config_.arrivalRate * (1.0 - config_.waveAmplitude), 2) + " to " +
                                     fixed(config_.arrivalRate * (1.0 + config_.waveAmplitude), 2) + " per cycle, period " +
                                     std::to_string(wavePeriod) + " cycles");
    } else {
        row(out, "Traffic wave", "off (steady rate)");
    }
    row(out, "Streaming (S) job share", fixed(config_.streamingShare * 100.0, 0) + "%");
    row(out, "Initial queue", std::to_string(config_.queuePerServer) + " x servers = " +
                                  std::to_string(static_cast<unsigned long>(config_.servers) * config_.queuePerServer) +
                                  " requests");
    row(out, "Scale up when", "queue > " + fixed(config_.upperFactor, 0) + " x active servers");
    row(out, "Scale down when", "queue < " + fixed(config_.lowerFactor, 0) + " x active servers");
    row(out, "Scaling check interval", "every " + std::to_string(config_.cooldown) + " cycles, one server at a time");
    row(out, "Server limits", std::to_string(config_.minServers) + " to " + std::to_string(config_.maxServers));
    row(out, "Queue capacity", config_.maxQueue == 0 ? "unlimited" : std::to_string(config_.maxQueue));
    std::string ranges;
    for (const std::string& range : config_.blockedRanges) {
        ranges += (ranges.empty() ? "" : ", ") + range;
    }
    row(out, "Firewall blocked source ranges", ranges.empty() ? "none" : ranges);
}
 
void Simulation::writeRandomness(std::ostream& out) const {
    const GeneratorStats& stats = generator_.stats();
    heading(out, "3. RANDOMNESS IN REQUEST GENERATION");
    row(out, "Requests generated", std::to_string(stats.total) + " (" + std::to_string(stats.initial) +
                                       " initial + " + std::to_string(stats.total - stats.initial) + " during run)");
    const double total = stats.total == 0 ? 1.0 : static_cast<double>(stats.total);
    row(out, "Processing (P) / Streaming (S) jobs",
        std::to_string(stats.processing) + " (" + fixed(100.0 * stats.processing / total, 1) + "%) / " +
            std::to_string(stats.streaming) + " (" + fixed(100.0 * stats.streaming / total, 1) + "%)");
    row(out, "Observed task time min / mean / max",
        std::to_string(stats.minTaskSeen) + " / " + fixed(static_cast<double>(stats.taskCycleSum) / total, 1) + " / " +
            std::to_string(stats.maxTaskSeen) + " cycles");
    const double runCycles = config_.cycles == 0 ? 1.0 : static_cast<double>(config_.cycles);
    row(out, "Observed arrivals per cycle (mean / max)",
        fixed((stats.total - stats.initial) / runCycles, 3) + " / " + std::to_string(stats.maxArrivalsInCycle));
 
    out << "\n  Task time histogram (cycles)\n";
    const GeneratorSettings& settings = generator_.settings();
    const unsigned long span = settings.maxTaskCycles - settings.minTaskCycles + 1UL;
    const unsigned long largest = *std::max_element(stats.taskHistogram.begin(), stats.taskHistogram.end());
    for (std::size_t bin = 0; bin < GeneratorStats::kTaskBins; ++bin) {
        const unsigned long low = settings.minTaskCycles + bin * span / GeneratorStats::kTaskBins;
        const unsigned long high = settings.minTaskCycles + (bin + 1) * span / GeneratorStats::kTaskBins - 1;
        if (high < low) {
            continue;  // empty bin when the range is narrower than the bin count
        }
        const unsigned long count = stats.taskHistogram[bin];
        const std::size_t bar = largest == 0 ? 0 : static_cast<std::size_t>(30.0 * count / largest + 0.5);
        std::ostringstream label;
        label << std::setw(4) << low << "-" << std::left << std::setw(4) << high;
        out << "    " << label.str() << " " << std::right << std::setw(7) << count << "  " << std::string(bar, '#') << '\n';
    }
 
    out << "\n  New arrivals per cycle (Poisson)\n";
    const char* labels[GeneratorStats::kArrivalBins] = {"0", "1", "2", "3", "4+"};
    for (std::size_t bin = 0; bin < GeneratorStats::kArrivalBins; ++bin) {
        const unsigned long count = stats.arrivalHistogram[bin];
        out << "    " << std::left << std::setw(3) << labels[bin] << std::right << std::setw(8) << count
            << " cycles (" << fixed(100.0 * count / runCycles, 2) << "%)\n";
    }
 
    out << "\n  First generated requests (sample)\n";
    out << "    " << std::left << std::setw(6) << "id" << std::setw(17) << "source IP" << std::setw(17)
        << "destination IP" << std::setw(6) << "type" << "task cycles\n";
    for (const Request& request : generator_.samples()) {
        out << "    " << std::left << std::setw(6) << request.id << std::setw(17) << request.sourceIp
            << std::setw(17) << request.destinationIp << std::setw(6) << jobTypeCode(request.jobType)
            << request.taskCycles << '\n';
    }
}
 
void Simulation::writeSnapshots(std::ostream& out, const LoadBalancer& balancer) const {
    out << "\n  " << balancer.name() << " status every " << snapshotEvery_ << " cycles\n";
    out << "    " << std::right << std::setw(7) << "cycle" << std::setw(8) << "queue" << std::setw(8) << "live"
        << std::setw(7) << "busy" << std::setw(7) << "idle" << std::setw(7) << "retire" << std::setw(11)
        << "completed" << std::setw(10) << "rejected" << '\n';
    for (const Snapshot& snapshot : balancer.snapshots()) {
        out << "    " << std::setw(7) << snapshot.cycle << std::setw(8) << snapshot.queueSize << std::setw(8)
            << snapshot.liveServers << std::setw(7) << snapshot.busyServers << std::setw(7) << snapshot.idleServers
            << std::setw(7) << snapshot.retiringServers << std::setw(11) << snapshot.completed << std::setw(10)
            << snapshot.rejected << '\n';
    }
}
 
void Simulation::writeScaleTimeline(std::ostream& out, const LoadBalancer& balancer) const {
    const std::vector<ScaleEvent>& events = balancer.scaleEvents();
    out << "\n  " << balancer.name() << ": " << events.size() << " server pool changes\n";
    if (events.empty()) {
        return;
    }
    // Print every change for normal runs; very long runs keep the first and last changes.
    const std::size_t head = events.size() <= 500 ? events.size() : 300;
    const std::size_t tail = events.size() <= 500 ? 0 : 100;
    const auto print = [&out](const ScaleEvent& event) {
        out << "    t=" << std::left << std::setw(8) << event.cycle << std::setw(17) << kindName(event.kind)
            << "server " << std::setw(5) << event.serverId << "queue " << std::setw(7) << event.queueSize
            << "active servers now " << event.activeServers << '\n';
    };
    for (std::size_t i = 0; i < events.size() && i < head; ++i) {
        print(events[i]);
    }
    if (events.size() > head + tail) {
        out << "    ... " << (events.size() - head - tail)
            << " more changes omitted (see the status table above or run with --verbose) ...\n";
    }
    for (std::size_t i = std::max(head, events.size() > tail ? events.size() - tail : 0); i < events.size(); ++i) {
        print(events[i]);
    }
}
 
void Simulation::writeReport(std::ostream& out) const {
    const std::time_t now = std::time(nullptr);
    char stamp[32];
    std::strftime(stamp, sizeof(stamp), "%Y-%m-%d %H:%M:%S", std::localtime(&now));
 
    out << kRule << '\n'
        << "  AGGIE STUDENT TOOLKIT - LOAD BALANCER LOG SUMMARY\n"
        << "  CSCE 412 Project 4 | Arnav Fotedar | generated " << stamp << '\n'
        << kRule << '\n';
 
    writeSettings(out);
 
    const std::vector<const LoadBalancer*> balancers = loadBalancers();
    std::size_t startingQueue = 0;
    std::size_t endingQueue = 0;
    for (const LoadBalancer* balancer : balancers) {
        startingQueue += balancer->stats().startingQueue;
        endingQueue += balancer->queueSize();
    }
    heading(out, "2. REQUIRED SUMMARY");
    row(out, "Starting queue size", std::to_string(startingQueue) + " requests (t=0, after firewall, before dispatch)");
    row(out, "Ending queue size", std::to_string(endingQueue) + " requests (t=" + std::to_string(config_.cycles) + ")");
    row(out, "Range for task times", std::to_string(config_.minTask) + " to " + std::to_string(config_.maxTask) +
                                         " clock cycles");
 
    writeRandomness(out);
 
    heading(out, "4. END STATUS AND STATISTICS");
    if (switch_) {
        row(out, "Requests routed to LB-P / LB-S",
            std::to_string(switch_->routedProcessing()) + " / " + std::to_string(switch_->routedStreaming()));
        out << '\n';
    }
    for (const LoadBalancer* balancer : balancers) {
        balancer->summarize(out);
        out << '\n';
    }
    row(out, "Simulation wall-clock time", fixed(runSeconds_, 2) + " s");
 
    heading(out, "5. SERVERS AT THE END OF THE RUN");
    for (const LoadBalancer* balancer : balancers) {
        out << "  " << balancer->name() << '\n';
        out << "    " << std::left << std::setw(8) << "server" << std::setw(10) << "state" << std::setw(26)
            << "current request" << std::setw(11) << "remaining" << std::setw(11) << "completed"
            << "online since\n";
        for (const WebServer& server : balancer->servers()) {
            std::string state = server.isIdle() ? "idle" : (server.isRetiring() ? "retiring" : "busy");
            std::string current = "-";
            if (server.current()) {
                current = "req#" + std::to_string(server.current()->id) + " (" +
                          std::string(1, jobTypeCode(server.current()->jobType)) + ", " +
                          std::to_string(server.current()->taskCycles) + " cyc)";
            }
            out << "    " << std::left << std::setw(8) << server.id() << std::setw(10) << state << std::setw(26)
                << current << std::setw(11) << server.remainingCycles() << std::setw(11) << server.completedCount()
                << "t=" << server.createdCycle() << '\n';
        }
        out << '\n';
    }
 
    heading(out, "6. DYNAMIC SERVER COUNT AND QUEUE OVER TIME");
    for (const LoadBalancer* balancer : balancers) {
        writeSnapshots(out, *balancer);
    }
 
    heading(out, "7. SCALING TIMELINE (server pool changes)");
    for (const LoadBalancer* balancer : balancers) {
        writeScaleTimeline(out, *balancer);
    }
    out << '\n' << kRule << "\n  END OF LOG\n" << kRule << '\n';
}