ln_zero_capacity.h

Go to the documentation of this file.

/*
 * Copyright (c) 2012-2026, QORE Lab, Imperial College London
 * All rights reserved.
 */
#ifndef LINE_SOLVERS_LN_LN_ZERO_CAPACITY_H
#define LINE_SOLVERS_LN_LN_ZERO_CAPACITY_H
 
/**
 * @file
 * @ingroup line_solvers
 * Reference tasks trapped at a zero-capacity element of a layered network.
 *
 * Port of matlab/src/solvers/LN/ln_zero_capacity.m.
 *
 * A processor or a non-reference task of multiplicity 0 never serves a request.
 * A reference task that reaches one through synchronous calls with positive
 * probability ends, in steady state, with all of its users stuck on that call
 * path (nested blocking): every throughput it drives is 0, the response times
 * along the path are Inf, and each task, entry and activity on the path holds
 * the requests stuck below it, the dead task holding those queued at it. A
 * trapped request holds task threads only, never a processor, so the other
 * reference tasks are solved as if the trapped ones were absent.
 *
 * Every vector is indexed by LQN element, 1-based, (nidx+1) long:
 *   dead     zero-capacity processors and tasks
 *   trapped  reference tasks whose users are all trapped
 *   onroute  elements on a blocked call path
 *   held     requests each element holds in the trapped state
 *   ignore   elements no live reference task or open arrival reaches, to drop from the layers
 *   live     true when some reference task or open arrival is still served
 *   reason   non-empty when the trapped state is ambiguous or does not exist
 */
 
#include <algorithm>
#include <cmath>
#include <cstddef>
#include <cstdio>
#include <limits>
#include <string>
#include <vector>
 
#include "line/lang/lqn/lqn_struct.h"
#include "line/num/number.h"
 
namespace line {
namespace ln {
 
/** The structural answer of ln_zero_capacity; see the file header. */
struct LnZeroCapacity {
    std::vector<bool> dead, trapped, onroute, ignore;
    std::vector<double> held;
    bool live = true;
    std::string reason;
 
    bool any_trapped() const {
        for (bool b : trapped)
            if (b) return true;
        return false;
    }
};
 
namespace zc_detail {
 
inline std::string fmt_g(double v) {
    if (std::isinf(v)) return v > 0 ? "Inf" : "-Inf";
    char buf[64];
    std::snprintf(buf, sizeof(buf), "%g", v);
    return buf;
}
 
template <class T>
class ZeroCapacity {
public:
    explicit ZeroCapacity(const lqn::LqnStruct<T>& l) : lqn(l), N(l.nidx) {
        zc.dead.assign(N + 1, false);
        zc.trapped.assign(N + 1, false);
        zc.onroute.assign(N + 1, false);
        zc.ignore.assign(N + 1, false);
        zc.held.assign(N + 1, 0.0);
    }
 
    LnZeroCapacity run() {
        using lang::CallType;
        using lang::LqnElement;
        const std::size_t NT = lqn.tshift + lqn.ntasks;
        bool anydead = false;
        for (std::size_t idx = 1; idx <= NT; ++idx) {
            zc.dead[idx] = lqn.mult[idx] == 0.0 && lqn.sched[idx] != lang::SchedStrategy::INF && !lqn.isref[idx];
            anydead = anydead || zc.dead[idx];
        }
        if (!anydead) return zc;
 
        // Entries from which a zero-capacity element is reached through synchronous calls.
        // blockAct[e] is the activity a request to e is stuck at, next[e] the entry it waits
        // on there, -1 when the activity waits for its dead processor, 0 when e's task is dead.
        state.assign(N + 1, 0);
        reach.assign(N + 1, false);
        blockAct.assign(N + 1, 0);
        next.assign(N + 1, 0);
        for (std::size_t e = lqn.eshift + 1; e <= lqn.eshift + lqn.nentries; ++e) {
            entry_route(e);
            if (!zc.reason.empty()) return zc;
        }
 
        // Requests that no trapped state can absorb
        for (std::size_t c = 1; c <= lqn.ncalls; ++c) {
            const std::size_t tgt = lqn.callpair_dst[c];
            if (lqn.calltype[c] == CallType::FWD && reach[tgt]) {
                zc.reason = "Entry " + lqn.names[lqn.callpair_src[c]] + " forwards requests to " + lqn.names[tgt] +
                            ", which reaches a zero-capacity element: forwarding into a blocked call path is "
                            "not supported.";
                return zc;
            }
        }
        std::vector<bool> arv(N + 1, false);
        for (std::size_t e = lqn.eshift + 1; e <= lqn.eshift + lqn.nentries && e < lqn.has_arrival.size(); ++e) {
            arv[e] = lqn.has_arrival[e];
            if (arv[e] && reach[e]) {
                zc.reason = "Entry " + lqn.names[e] + " has an open arrival stream and reaches a zero-capacity "
                            "element, where its requests accumulate without bound: the model has no steady state.";
                return zc;
            }
        }
 
        // Trapped reference tasks and the requests held along their call paths
        std::vector<std::size_t> refs;
        for (std::size_t t = 1; t <= lqn.ntasks; ++t)
            if (lqn.isref[lqn.tshift + t]) refs.push_back(lqn.tshift + t);
        std::vector<double> incoming(N + 1, 0.0), nroutes(N + 1, 0.0);
        for (std::size_t t : refs) {
            std::vector<std::size_t> ents;
            for (std::size_t e : lqn.entriesof[t])
                if (reach[e]) ents.push_back(e);
            if (ents.empty()) continue;
            if (ents.size() > 1) {
                zc.reason = "Reference task " + lqn.names[t] + " reaches a zero-capacity element from " +
                            std::to_string(ents.size()) +
                            " of its entries: how its users divide between the blocked paths depends on the "
                            "initial state.";
                return zc;
            }
            zc.trapped[t] = true;
            double h = lqn.mult[t];
            mark(t, h);
            std::size_t e = ents[0];
            bool first = true;
            while (true) {
                const std::size_t tt = lqn.parent[e];
                if (zc.dead[tt]) {
                    mark(e, h);  // requests queued at a task with no threads
                    mark(tt, h);
                    break;
                }
                if (!first) {
                    incoming[tt] += h;
                    nroutes[tt] += 1.0;
                    h = std::min(h, task_cap(tt));
                    mark(tt, h);
                }
                first = false;
                mark(e, h);
                mark(blockAct[e], h);
                if (next[e] < 0) {
                    zc.onroute[lqn.parent[tt]] = true;  // its dead processor
                    break;
                }
                e = std::size_t(next[e]);
            }
        }
        for (std::size_t tt = 1; tt <= N; ++tt)
            if (nroutes[tt] > 1.0 && incoming[tt] > task_cap(tt)) {
                zc.reason = "Task " + lqn.names[tt] + " has " + fmt_g(lqn.mult[tt]) +
                            " threads and is on the blocked call paths of several reference tasks that send it " +
                            fmt_g(incoming[tt]) + " requests: which of them hold its threads depends on the "
                            "initial state.";
                return zc;
            }
        for (std::size_t idx = 1; idx <= NT; ++idx)
            if (zc.onroute[idx] && lqn.repl[idx] > 1.0) {
                zc.reason = lqn.names[idx] +
                            " is replicated and on a call path blocked at a zero-capacity element, which is not "
                            "supported.";
                return zc;
            }
 
        // What the reference tasks that are not trapped, and the open arrivals, still reach
        std::vector<std::size_t> starts;
        for (std::size_t t : refs)
            if (!zc.trapped[t]) starts.push_back(t);
        for (std::size_t e = 1; e <= N; ++e)
            if (arv[e]) starts.push_back(e);
        const std::vector<bool> live = reach_from(starts);
        for (std::size_t c = 1; c <= lqn.ncalls; ++c) {
            const std::size_t tgt = lqn.callpair_dst[c];
            if (lqn.calltype[c] == CallType::ASYNC && live[lqn.callpair_src[c]] &&
                (reach[tgt] || zc.dead[lqn.parent[tgt]])) {
                zc.reason = "Activity " + lqn.names[lqn.callpair_src[c]] + " calls " + lqn.names[tgt] +
                            " asynchronously, which reaches a zero-capacity element where its requests accumulate "
                            "without bound: the model has no steady state.";
                return zc;
            }
        }
        for (std::size_t idx = 1; idx <= N; ++idx)
            if (live[idx] && zc.dead[idx]) {
                zc.reason = lqn.names[idx] +
                            " has zero capacity and is reached by a reference task that is not trapped there, "
                            "through a call that cannot block it.";
                return zc;
            }
        for (std::size_t tt = lqn.tshift + 1; tt <= NT; ++tt)
            if (live[tt] && zc.held[tt] > 0.0 && std::isfinite(task_cap(tt)) && !lqn.isref[tt]) {
                zc.reason = "Task " + lqn.names[tt] +
                            " is reached by a reference task that is not trapped, while " + fmt_g(zc.held[tt]) +
                            " of its " + fmt_g(lqn.mult[tt]) +
                            " threads are held forever by requests blocked at a zero-capacity element.";
                return zc;
            }
        zc.live = false;
        for (std::size_t i = 1; i <= N; ++i) zc.live = zc.live || live[i];
 
        // Drop from the layers the processors and tasks only the trapped part reaches
        std::vector<std::size_t> tstarts;
        for (std::size_t t : refs)
            if (zc.trapped[t]) tstarts.push_back(t);
        const std::vector<bool> fromTrapped = reach_from(tstarts);
        std::vector<bool> drop(N + 1, false);
        for (std::size_t i = 1; i <= NT; ++i) drop[i] = (fromTrapped[i] || zc.dead[i]) && !live[i];
        for (std::size_t idx = lqn.eshift + 1; idx <= lqn.ashift + lqn.nacts; ++idx) drop[idx] = drop[lqn.parent[idx]];
        zc.ignore = drop;
        return zc;
    }
 
private:
    const lqn::LqnStruct<T>& lqn;
    const std::size_t N;
    LnZeroCapacity zc;
    std::vector<int> state;  // 0 unvisited, 1 on the stack, 2 done
    std::vector<bool> reach;
    std::vector<std::size_t> blockAct;
    std::vector<long> next;
 
    void mark(std::size_t idx, double h) {
        zc.onroute[idx] = true;
        zc.held[idx] += h;
    }
 
    /** Threads of a task, Inf at an infinite server or for anything that is not a task. */
    double task_cap(std::size_t tt) const {
        if (tt > lqn.tshift && tt <= lqn.tshift + lqn.ntasks && lqn.sched[tt] != lang::SchedStrategy::INF)
            return lqn.mult[tt];
        return std::numeric_limits<double>::infinity();
    }
 
    static bool has(const std::vector<std::size_t>& v, std::size_t x) {
        return std::find(v.begin(), v.end(), x) != v.end();
    }
 
    /** True when a request to entry e is trapped at a zero-capacity element. */
    bool entry_route(std::size_t e) {
        using lang::CallType;
        if (state[e] == 2) return reach[e];
        bool r = false;
        if (state[e] == 1 || !zc.reason.empty()) return false;  // a call cycle cannot reach anything new
        state[e] = 1;
        const std::size_t tk = lqn.parent[e];
        if (zc.dead[tk]) {
            r = true;
        } else {
            const std::vector<std::size_t> starts = acts_after(e);
            const std::vector<std::size_t> acts = act_closure(starts, {});
            if (zc.dead[lqn.parent[tk]]) {
                // the first activity never gets the processor
                if (starts.size() != 1) {
                    zc.reason = "Entry " + lqn.names[e] + " starts " + std::to_string(starts.size()) +
                                " activities on zero-capacity processor " + lqn.names[lqn.parent[tk]] +
                                ": how its requests divide between them depends on the initial state.";
                } else {
                    r = true;
                    blockAct[e] = starts[0];
                    next[e] = -1;
                }
            } else {
                std::vector<std::size_t> cand;
                std::vector<std::vector<std::size_t>> candCall;
                for (std::size_t a : acts) {
                    std::vector<std::size_t> dc;
                    if (a < lqn.callsof.size())
                        for (std::size_t ci : lqn.callsof[a])
                            if (lqn.calltype[ci] == CallType::SYNC &&
                                num_traits<T>::to_double(lqn.callproc_mean[ci]) > 0.0 &&
                                entry_route(lqn.callpair_dst[ci]))
                                dc.push_back(ci);
                    if (!dc.empty()) {
                        cand.push_back(a);
                        candCall.push_back(dc);
                    }
                }
                // an activity behind another blocked one on every path is never reached
                std::vector<std::size_t> c2;
                std::vector<std::vector<std::size_t>> cc2;
                for (std::size_t k = 0; k < cand.size(); ++k) {
                    std::vector<std::size_t> others;
                    for (std::size_t j = 0; j < cand.size(); ++j)
                        if (j != k) others.push_back(cand[j]);
                    if (has(act_closure(starts, others), cand[k])) {
                        c2.push_back(cand[k]);
                        cc2.push_back(candCall[k]);
                    }
                }
                if (c2.size() > 1 || (c2.size() == 1 && cc2[0].size() > 1)) {
                    zc.reason = "Entry " + lqn.names[e] +
                                " reaches a zero-capacity element along more than one call path: how its "
                                "requests divide between them depends on the initial state.";
                } else if (c2.size() == 1) {
                    r = true;
                    blockAct[e] = c2[0];
                    next[e] = long(lqn.callpair_dst[cc2[0][0]]);
                }
            }
        }
        reach[e] = r;
        state[e] = 2;
        return r;
    }
 
    /** Activities that directly follow entry or activity i in its own task. */
    std::vector<std::size_t> acts_after(std::size_t i) const {
        std::vector<std::size_t> s;
        for (std::size_t j : lqn.graph.succ(i))
            if (j <= N && lqn.type[j] == lang::LqnElement::ACTIVITY && lqn.parent[j] == lqn.parent[i])
                s.push_back(j);
        return s;
    }
 
    /** Activities reached from `starts` without passing through `blocked`. */
    std::vector<std::size_t> act_closure(const std::vector<std::size_t>& starts,
                                         const std::vector<std::size_t>& blocked) const {
        std::vector<std::size_t> seen, stack;
        for (std::size_t s : starts)
            if (!has(blocked, s)) stack.push_back(s);
        while (!stack.empty()) {
            const std::size_t a = stack.back();
            stack.pop_back();
            if (has(seen, a)) continue;
            seen.push_back(a);
            for (std::size_t b : acts_after(a))
                if (!has(seen, b) && !has(blocked, b)) stack.push_back(b);
        }
        return seen;
    }
 
    /** Elements a request from the tasks or entries in `starts` can visit. */
    std::vector<bool> reach_from(const std::vector<std::size_t>& starts) const {
        std::vector<bool> seen(N + 1, false);
        std::vector<std::size_t> stack = starts;
        while (!stack.empty()) {
            const std::size_t i = stack.back();
            stack.pop_back();
            if (seen[i]) continue;
            seen[i] = true;
            std::vector<std::size_t> nxt;
            switch (lqn.type[i]) {
                case lang::LqnElement::TASK:
                    nxt.push_back(lqn.parent[i]);
                    if (lqn.isref[i]) nxt.insert(nxt.end(), lqn.entriesof[i].begin(), lqn.entriesof[i].end());
                    break;
                case lang::LqnElement::ENTRY:
                case lang::LqnElement::ACTIVITY: {
                    nxt.push_back(lqn.parent[i]);
                    for (std::size_t j : lqn.graph.succ(i))
                        if (j <= N) nxt.push_back(j);
                    break;
                }
                default:
                    break;
            }
            for (std::size_t j : nxt)
                if (j >= 1 && !seen[j]) stack.push_back(j);
        }
        return seen;
    }
};
 
}  // namespace zc_detail
 
/** Port of ln_zero_capacity.m; see the file header. */
template <class T>
LnZeroCapacity ln_zero_capacity(const lqn::LqnStruct<T>& lqn) {
    return zc_detail::ZeroCapacity<T>(lqn).run();
}
 
}  // namespace ln
}  // namespace line
 
#endif  // LINE_SOLVERS_LN_LN_ZERO_CAPACITY_H