ldes_station.h

Go to the documentation of this file.

/*
 * Copyright (c) 2012-2026, QORE Lab, Imperial College London
 * All rights reserved.
 */
#ifndef LINE_SOLVERS_LDES_LDES_STATION_H
#define LINE_SOLVERS_LDES_LDES_STATION_H
 
/**
 * @file
 * @ingroup line_solvers
 * The scheduling disciplines of the native LDES engine.
 *
 * TWO FAMILIES, and they are structurally different simulators sharing one
 * station:
 *
 *  - THE BUFFERED disciplines (FCFS, LCFS, SIRO, the priority variants, SJF,
 *    LJF, SEPT, LEPT) hold waiting jobs in an ORDER and hand the head to a
 *    free server. The discipline IS the order, so it lives entirely in
 *    `WaitCmp` below; everything downstream is the same code.
 *  - THE SHARING disciplines (PS, DPS, GPS, their priority variants and LPS)
 *    have no order at all. Every job in service advances at its own share of
 *    the server, so the station has no head and no service-start event: the
 *    residual work of EVERY job must be integrated forward and EVERY departure
 *    rescheduled whenever the population changes. `PsStation` does that.
 *
 * THE SERVICE TIME IS SAMPLED AT ARRIVAL, not at service start. This is the
 * reference's choice (`customer.serviceTime = generateServiceTime(...)` in the
 * arrival handlers) and it is forced: SJF, LJF and the size-based orders
 * compare service times of jobs that have not started, so the sample must
 * exist before the job is ordered. It is invisible under FCFS, where arrival
 * and service order coincide, and visible under LCFS with a MAP service
 * process, whose phase then advances in ARRIVAL order.
 *
 * PRIORITY ORDERS ASCENDING: a lower value is more urgent, 0 highest. That is
 * LINE's convention throughout (`SaveHandlers` inverts it when exporting to
 * JMT, which orders the other way), and every comparator here follows it.
 */
 
#include <algorithm>
#include <cmath>
#include <cstddef>
#include <cstdint>
#include <limits>
#include <set>
#include <vector>
 
#include "line/lang/lang_types.h"
 
namespace line {
namespace ldes {
namespace engine {
 
/**
 * A job held by a station.
 *
 * `service`, `remaining` and `elapsed` are the three quantities the size-based
 * orders need and are NOT redundant. `service` is the ORIGINAL requirement
 * PSJF orders by; `remaining` is the residual SRPT and LRPT order by, equal to
 * `service` until the job is first preempted; `elapsed` is the attained
 * service FB and SETF order by. The reference keeps them in a side table
 * (`preemptedJobHistory`, keyed by the arrival instants) because its wait
 * queue holds a plain `Customer`; carrying them on the job is the same
 * information without the lookup, and without the key collision two jobs
 * arriving at the same instant would produce.
 */
struct Job {
    std::size_t cls = 0;
    int priority = 0;
    double t_arr = 0.0;      ///< arrival instant at this station
    double t_sys = 0.0;      ///< instant the current passage started
    double service = 0.0;    ///< the sampled requirement, drawn at arrival
    double remaining = 0.0;  ///< residual work; equals `service` before any preemption
    double elapsed = 0.0;    ///< attained service across all its service intervals
    double rank = 0.0;       ///< SIRO's random key, drawn at arrival
    double deadline = std::numeric_limits<double>::infinity();  ///< EDD/EDF key
    double vft = 0.0;        ///< FSP's virtual finish time, recomputed on demand
    /**
     * Identity of this job while it waits, so a reneging timer can find it.
     *
     * A timer names the job it was armed for, and the job may by then have
     * entered service, been preempted back into the queue, or already left. An
     * identity is the only thing that survives all three; the arrival instant
     * cannot serve as one, because two jobs admitted at the same instant --
     * routine on a closed model placed at t=0 -- would share it.
     */
    std::uint64_t id = 0;
    /**
     * The fork synchronization this job is a sibling of, 0 when it is not one.
     *
     * It rides on the JOB and not on the station because siblings of different
     * parents interleave freely at every station between the Fork and the Join.
     */
    std::uint64_t parent = 0;
    /**
     * How many times this job has already retried from an orbit.
     *
     * It rides on the job because `maxAttempts` bounds the retries of ONE job,
     * not of the station: a per-station counter would cap the orbit's total
     * traffic instead and would let a job retry forever as long as others gave
     * up.
     */
    int attempts = 0;
    /**
     * The SYNCHRONOUS CALL this job belongs to, 0 when it belongs to none.
     *
     * It rides on the job because the call is answered by a REPLY that comes
     * back from somewhere else entirely: this identity is the only link between
     * that reply and the server still held for it, and `id` cannot serve as one,
     * being reassigned at every admission.
     */
    std::uint64_t call = 0;
};
 
/** True for the disciplines that interrupt a job already in service. */
inline bool is_preemptive(lang::SchedStrategy s) {
    switch (s) {
        case lang::SchedStrategy::LCFSPR:
        case lang::SchedStrategy::LCFSPI:
        case lang::SchedStrategy::LCFSPRPRIO:
        case lang::SchedStrategy::LCFSPIPRIO:
        case lang::SchedStrategy::FCFSPR:
        case lang::SchedStrategy::FCFSPI:
        case lang::SchedStrategy::FCFSPRPRIO:
        case lang::SchedStrategy::FCFSPIPRIO:
        case lang::SchedStrategy::SRPT:
        case lang::SchedStrategy::SRPTPRIO:
        case lang::SchedStrategy::PSJF:
        case lang::SchedStrategy::FB:
        case lang::SchedStrategy::LRPT:
        case lang::SchedStrategy::SETF:
        case lang::SchedStrategy::FSP:
            return true;
        default:
            return false;
    }
}
 
/**
 * PREEMPTIVE RESUME, as against preemptive restart.
 *
 * A resumed job continues from its residual work; a restarted one draws a
 * FRESH requirement and discards what it had done. That distinction is the
 * whole of the PR/PI split and it is not cosmetic: under exponential service
 * the two agree in distribution (memorylessness) and under any other law they
 * do not, so a model that uses PI to represent lost work is silently served
 * as PR on a Coxian. Every size-based policy is RESUME -- restarting a job
 * would make its own scheduling key meaningless.
 */
inline bool is_preemptive_resume(lang::SchedStrategy s) {
    switch (s) {
        case lang::SchedStrategy::LCFSPI:
        case lang::SchedStrategy::LCFSPIPRIO:
        case lang::SchedStrategy::FCFSPI:
        case lang::SchedStrategy::FCFSPIPRIO:
            return false;
        default:
            return is_preemptive(s);
    }
}
 
/** True for the orders whose key is the job's size, residual or attained work. */
inline bool is_size_based(lang::SchedStrategy s) {
    switch (s) {
        case lang::SchedStrategy::SRPT:
        case lang::SchedStrategy::SRPTPRIO:
        case lang::SchedStrategy::PSJF:
        case lang::SchedStrategy::FB:
        case lang::SchedStrategy::LRPT:
        case lang::SchedStrategy::SETF:
        case lang::SchedStrategy::FSP:
            return true;
        default:
            return false;
    }
}
 
/** True for the disciplines that share the server instead of ordering a queue. */
inline bool is_ps_family(lang::SchedStrategy s) {
    switch (s) {
        case lang::SchedStrategy::PS:
        case lang::SchedStrategy::DPS:
        case lang::SchedStrategy::GPS:
        case lang::SchedStrategy::PSPRIO:
        case lang::SchedStrategy::DPSPRIO:
        case lang::SchedStrategy::GPSPRIO:
        case lang::SchedStrategy::LPS:
            return true;
        default:
            return false;
    }
}
 
/**
 * The waiting-room order of a buffered discipline.
 *
 * Used as a `std::priority_queue` comparator, which pops the GREATEST element,
 * so every rule below is inverted relative to the reference's `Comparator`
 * (which orders ascending and polls the least). Getting that backwards turns
 * FCFS into LCFS silently -- both run, both conserve jobs, and only the
 * response-time variance tells them apart.
 */
struct WaitCmp {
    lang::SchedStrategy sched = lang::SchedStrategy::FCFS;
    const std::vector<double>* class_mean = 0;  ///< mean service per class, for SEPT/LEPT
 
    bool operator()(const Job& a, const Job& b) const {
        switch (sched) {
            case lang::SchedStrategy::LCFS:
            case lang::SchedStrategy::LCFSPR:
            case lang::SchedStrategy::LCFSPI:
                return a.t_arr < b.t_arr;  // latest first
            case lang::SchedStrategy::SIRO:
                return a.rank > b.rank;
            case lang::SchedStrategy::HOL:
            case lang::SchedStrategy::FCFSPR:
            case lang::SchedStrategy::FCFSPI:
            case lang::SchedStrategy::FCFSPRPRIO:
            case lang::SchedStrategy::FCFSPIPRIO:
                if (a.priority != b.priority) return a.priority > b.priority;
                return a.t_arr > b.t_arr;
            case lang::SchedStrategy::LCFSPRIO:
            case lang::SchedStrategy::LCFSPRPRIO:
            case lang::SchedStrategy::LCFSPIPRIO:
                if (a.priority != b.priority) return a.priority > b.priority;
                return a.t_arr < b.t_arr;
            case lang::SchedStrategy::SJF:
                if (a.service != b.service) return a.service > b.service;
                return a.t_arr > b.t_arr;
            case lang::SchedStrategy::LJF:
                if (a.service != b.service) return a.service < b.service;
                return a.t_arr > b.t_arr;
            case lang::SchedStrategy::SEPT: {
                const double ma = mean_of(a), mb = mean_of(b);
                if (ma != mb) return ma > mb;  // shortest expected first
                return a.t_arr > b.t_arr;
            }
            case lang::SchedStrategy::LEPT: {
                const double ma = mean_of(a), mb = mean_of(b);
                if (ma != mb) return ma < mb;
                return a.t_arr > b.t_arr;
            }
            case lang::SchedStrategy::EDD:
            case lang::SchedStrategy::EDF:
                if (a.deadline != b.deadline) return a.deadline > b.deadline;
                return a.t_arr > b.t_arr;
            case lang::SchedStrategy::SRPT:
                if (a.remaining != b.remaining) return a.remaining > b.remaining;
                return a.t_arr > b.t_arr;
            case lang::SchedStrategy::SRPTPRIO:
                if (a.priority != b.priority) return a.priority > b.priority;
                if (a.remaining != b.remaining) return a.remaining > b.remaining;
                return a.t_arr > b.t_arr;
            case lang::SchedStrategy::LRPT:
                if (a.remaining != b.remaining) return a.remaining < b.remaining;
                return a.service < b.service;
            case lang::SchedStrategy::PSJF:
                // The ORIGINAL requirement, not the residual: a job that has
                // already been served keeps the rank its full size gave it.
                if (a.service != b.service) return a.service > b.service;
                return a.t_arr > b.t_arr;
            case lang::SchedStrategy::FB:
            case lang::SchedStrategy::SETF:
                // Least attained service first. FB and SETF share this order;
                // they differ in WHEN the running job is interrupted, not in
                // how the waiting ones are ranked.
                if (a.elapsed != b.elapsed) return a.elapsed > b.elapsed;
                return a.t_arr > b.t_arr;
            case lang::SchedStrategy::FSP:
                if (a.vft != b.vft) return a.vft > b.vft;
                return a.t_arr > b.t_arr;
            default:  // FCFS and anything ordered like it
                return a.t_arr > b.t_arr;
        }
    }
 
private:
    double mean_of(const Job& j) const {
        if (class_mean == 0 || j.cls >= class_mean->size()) return 0.0;
        return (*class_mean)[j.cls];
    }
};
 
/** A job being served by a sharing discipline, with the work it still owes. */
struct PsJob {
    std::size_t cls = 0;
    int priority = 0;
    double t_arr = 0.0;
    double t_sys = 0.0;
    double total = 0.0;      ///< the sampled requirement
    double remaining = 0.0;  ///< residual work, in service-time units
    std::uint64_t tag = 0;   ///< identifies the departure event scheduled for it
    /**
     * The fork the job belongs to, carried across the sharing station.
     *
     * A buffered station keeps the whole `Job` in its server slot and hands it
     * back at the completion, so the identity survives; a sharing station holds
     * this reduced record instead and would hand back a DEFAULT-CONSTRUCTED one.
     * A sibling leaving with parent 0 matches no fork at the Join and is
     * discarded there, which drains a closed fork-join model after one pass.
     */
    std::uint64_t parent = 0;
};
 
/**
 * The per-job shares of a sharing discipline, in units of one server.
 *
 * `c` is the station's effective server count: with n <= c every job runs at
 * rate 1 and the station is an infinite server; above that the capacity is
 * split. Each rule is capped at 1 because ONE job cannot consume more than ONE
 * server however the weights fall -- dropping that cap is what makes a
 * two-class DPS station report a utilization above one.
 *
 * Transcribed from `calculatePSRates` and the five rules under it.
 */
inline std::vector<double> ps_shares(lang::SchedStrategy sched, const std::vector<PsJob>& jobs,
                                     double c, const std::vector<double>& weight,
                                     std::size_t nclasses) {
    const std::size_t n = jobs.size();
    std::vector<double> rates(n, 0.0);
    if (n == 0) return rates;
 
    const bool prio = (sched == lang::SchedStrategy::PSPRIO ||
                       sched == lang::SchedStrategy::DPSPRIO ||
                       sched == lang::SchedStrategy::GPSPRIO);
    const bool weighted_by_job = (sched == lang::SchedStrategy::DPS ||
                                  sched == lang::SchedStrategy::DPSPRIO);
    const bool weighted_by_class = (sched == lang::SchedStrategy::GPS ||
                                    sched == lang::SchedStrategy::GPSPRIO);
 
    if (!prio) {
        if (weighted_by_job) {
            double tot = 0.0;
            for (const PsJob& j : jobs) tot += weight[j.cls];
            if (tot > 0.0) {
                for (std::size_t i = 0; i < n; ++i)
                    rates[i] = std::min(1.0, weight[jobs[i].cls] * c / tot);
                return rates;
            }
        } else if (weighted_by_class) {
            std::vector<int> per_class(nclasses, 0);
            for (const PsJob& j : jobs) ++per_class[j.cls];
            double tot = 0.0;
            for (std::size_t k = 0; k < nclasses; ++k)
                if (per_class[k] > 0) tot += weight[k];
            if (tot > 0.0) {
                for (std::size_t i = 0; i < n; ++i) {
                    const std::size_t k = jobs[i].cls;
                    rates[i] = std::min(1.0, (weight[k] / tot / per_class[k]) * c);
                }
                return rates;
            }
        }
        // PS, LPS, and the degenerate all-zero-weight case: equal shares.
        const double share = std::min(1.0, c / static_cast<double>(n));
        for (std::size_t i = 0; i < n; ++i) rates[i] = share;
        return rates;
    }
 
    // The priority variants give the whole station to the most urgent class
    // present, and only what is left over to the next -- so an uncongested
    // station (n <= c) has nothing to arbitrate and every job runs at 1.
    if (static_cast<double>(n) <= c) {
        for (std::size_t i = 0; i < n; ++i) rates[i] = 1.0;
        return rates;
    }
    std::set<int> prios;
    for (const PsJob& j : jobs) prios.insert(j.priority);
    double remaining = c;
    for (std::set<int>::const_iterator it = prios.begin(); it != prios.end(); ++it) {
        if (remaining <= 0.0) break;
        const int p = *it;
        int count = 0;
        double tot_job_weight = 0.0;
        std::vector<int> per_class(nclasses, 0);
        for (const PsJob& j : jobs)
            if (j.priority == p) {
                ++count;
                tot_job_weight += weight[j.cls];
                ++per_class[j.cls];
            }
        const double allocated = std::min(remaining, static_cast<double>(count));
        double tot_class_weight = 0.0;
        for (std::size_t k = 0; k < nclasses; ++k)
            if (per_class[k] > 0) tot_class_weight += weight[k];
 
        for (std::size_t i = 0; i < n; ++i) {
            if (jobs[i].priority != p) continue;
            const std::size_t k = jobs[i].cls;
            if (sched == lang::SchedStrategy::DPSPRIO && tot_job_weight > 0.0)
                rates[i] = std::min(1.0, weight[k] * allocated / tot_job_weight);
            else if (sched == lang::SchedStrategy::GPSPRIO && tot_class_weight > 0.0)
                rates[i] = std::min(1.0, (weight[k] / tot_class_weight / per_class[k]) * allocated);
            else
                rates[i] = allocated / count;
        }
        remaining -= allocated;
    }
    return rates;
}
 
/**
 * FSP's virtual finish time: when `target` would finish if the station ran
 * processor sharing from now on.
 *
 * Sort every residual work at the station, then walk it: while n jobs remain
 * each advances at min(1, c/n), so the wall time to clear the k-th smallest
 * residual is the sum of (w_k - w_{k-1}) / rate over the ranks below it. FSP
 * then serves in that order, which is what makes it dominate PS job by job.
 * Transcribes `computeFSPVirtualFinishTime`.
 */
inline double fsp_virtual_finish(const std::vector<double>& residuals, double target_work,
                                 double c, double now) {
    std::vector<double> works = residuals;
    std::sort(works.begin(), works.end());
    const std::size_t n = works.size();
    double cumulative = 0.0, prev = 0.0;
    for (std::size_t rank = 0; rank < n; ++rank) {
        const double w = works[rank];
        const double rate = std::min(1.0, c / static_cast<double>(n - rank));
        if (rate > 0.0) cumulative += (w - prev) / rate;
        if (w >= target_work) return now + cumulative;
        prev = w;
    }
    return now + cumulative;
}
 
/**
 * Which job in service `arriving` displaces, or `in_service.size()` for none.
 *
 * ONE RULE PER FAMILY, transcribed from `shouldPreemptForLCFS`,
 * `shouldPreemptForFCFS` and `shouldPreemptForSizeBasedPolicy` together with
 * their `findJobToPreempt*` counterparts. The two halves are fused here because
 * they ask the same question twice in the reference -- once to decide and once
 * to choose -- and answering it once cannot disagree with itself.
 *
 * THE PRIORITY VARIANTS PREEMPT ONLY STRICTLY LOWER PRIORITY. `LCFSPR` without
 * the PRIO suffix preempts unconditionally (it is last-come-first-served with
 * no class ordering at all), while `LCFSPRPRIO` preempts a job whose priority
 * value is at least the arrival's, and the FCFS family only one strictly
 * greater. Collapsing those three tests into one turns a priority station into
 * a thrashing one, or into a non-preemptive one, depending which way it falls.
 */
inline std::size_t preemption_victim(lang::SchedStrategy sched, const std::vector<Job>& in_service,
                                     const Job& arriving, double c, double now) {
    const std::size_t none = in_service.size();
    if (in_service.empty()) return none;
 
    switch (sched) {
        case lang::SchedStrategy::LCFSPR:
        case lang::SchedStrategy::LCFSPI: {
            // Displace the OLDEST job in service: the newcomer takes the head
            // of a last-come-first-served station unconditionally.
            std::size_t best = none;
            double oldest = std::numeric_limits<double>::infinity();
            for (std::size_t j = 0; j < in_service.size(); ++j)
                if (in_service[j].t_arr < oldest) {
                    oldest = in_service[j].t_arr;
                    best = j;
                }
            return best;
        }
        case lang::SchedStrategy::LCFSPRPRIO:
        case lang::SchedStrategy::LCFSPIPRIO: {
            std::size_t best = none;
            double oldest = std::numeric_limits<double>::infinity();
            for (std::size_t j = 0; j < in_service.size(); ++j)
                if (in_service[j].priority >= arriving.priority && in_service[j].t_arr < oldest) {
                    oldest = in_service[j].t_arr;
                    best = j;
                }
            return best;
        }
        case lang::SchedStrategy::FCFSPR:
        case lang::SchedStrategy::FCFSPI:
        case lang::SchedStrategy::FCFSPRPRIO:
        case lang::SchedStrategy::FCFSPIPRIO: {
            // The least urgent job in service, tie-broken by the LATEST service
            // start: among equals, the one that has invested least is displaced.
            std::size_t best = none;
            for (std::size_t j = 0; j < in_service.size(); ++j) {
                if (in_service[j].priority <= arriving.priority) continue;
                if (best == none || in_service[j].priority > in_service[best].priority ||
                    (in_service[j].priority == in_service[best].priority &&
                     in_service[j].t_arr > in_service[best].t_arr))
                    best = j;
            }
            return best;
        }
        case lang::SchedStrategy::SRPT:
        case lang::SchedStrategy::SRPTPRIO: {
            std::size_t worst = 0;
            for (std::size_t j = 1; j < in_service.size(); ++j)
                if (in_service[j].remaining > in_service[worst].remaining) worst = j;
            if (sched == lang::SchedStrategy::SRPTPRIO &&
                in_service[worst].priority < arriving.priority)
                return none;
            return (arriving.remaining < in_service[worst].remaining) ? worst : none;
        }
        case lang::SchedStrategy::PSJF: {
            std::size_t largest = 0;
            for (std::size_t j = 1; j < in_service.size(); ++j)
                if (in_service[j].service > in_service[largest].service) largest = j;
            return (arriving.service < in_service[largest].service) ? largest : none;
        }
        case lang::SchedStrategy::FB:
        case lang::SchedStrategy::SETF: {
            // A fresh arrival has attained nothing, so it displaces whichever
            // job has attained the most -- provided that is more than nothing.
            std::size_t most = 0;
            for (std::size_t j = 1; j < in_service.size(); ++j)
                if (in_service[j].elapsed > in_service[most].elapsed) most = j;
            return (in_service[most].elapsed > arriving.elapsed) ? most : none;
        }
        case lang::SchedStrategy::LRPT: {
            std::size_t least = 0;
            for (std::size_t j = 1; j < in_service.size(); ++j)
                if (in_service[j].remaining < in_service[least].remaining) least = j;
            return (arriving.remaining > in_service[least].remaining) ? least : none;
        }
        case lang::SchedStrategy::FSP: {
            std::vector<double> works;
            for (const Job& j : in_service) works.push_back(j.remaining);
            works.push_back(arriving.remaining);
            const double arrival_vft = fsp_virtual_finish(works, arriving.remaining, c, now);
            std::size_t worst = 0;
            double worst_vft = -std::numeric_limits<double>::infinity();
            for (std::size_t j = 0; j < in_service.size(); ++j) {
                const double v = fsp_virtual_finish(works, in_service[j].remaining, c, now);
                if (v > worst_vft) {
                    worst_vft = v;
                    worst = j;
                }
            }
            return (arrival_vft < worst_vft) ? worst : none;
        }
        default:
            return none;
    }
}
 
}  // namespace engine
}  // namespace ldes
}  // namespace line
 
#endif  // LINE_SOLVERS_LDES_LDES_STATION_H