LINE Solver (C++)
Templated C++ port of the LINE queueing solver
Loading...
Searching...
No Matches
workflow.h
Go to the documentation of this file.
1/*
2 * Copyright (c) 2012-2026, QORE Lab, Imperial College London
3 * All rights reserved.
4 */
5#ifndef LINE_LANG_WORKFLOW_WORKFLOW_H
6#define LINE_LANG_WORKFLOW_WORKFLOW_H
7
8/**
9 * @file
10 * @ingroup line_lang
11 * An activity workflow reduced to one phase-type law.
12 *
13 * Port of matlab/src/lang/workflow/Workflow.m and WorkflowActivity.m, and of
14 * their twins jline.lang.workflow.Workflow (JAR) and
15 * line_solver.lang.workflow.Workflow (Python). A workflow is a precedence
16 * graph over activities, each carrying a host demand; `to_ph` composes those
17 * laws into a single phase-type distribution.
18 *
19 * WHAT THIS IS NOT. `include/line/api/wf/` holds the workflow pattern
20 * DETECTORS of `jline.api.wf`, which read a link matrix and report sequences,
21 * branches, parallel blocks and loops. This header is the algebra those
22 * patterns feed: the composition that turns a precedence graph into an
23 * (alpha, T) pair. An LQN fork-join is also a different object -- its branches
24 * contend for a host, so it yields a response time under contention rather
25 * than the order statistic of independent activity times returned here.
26 *
27 * A precedence graph that is SERIES-PARALLEL is reduced exactly, by recursive
28 * composition of the series-parallel tree, which handles arbitrary nesting (a
29 * fork inside a loop, a branch that is itself a fork-join). Any other graph
30 * falls back to the block composition, which is a heuristic and is documented
31 * as one.
32 *
33 * A LOOP repeats its body a GEOMETRIC number of times of mean COUNT, the
34 * POST_LOOP semantics of an activity graph (`getStruct.m:528-567`): a count of
35 * at least one runs the body once and takes the back edge with probability
36 * 1-1/COUNT, and a fractional count runs the body at most once, with
37 * probability COUNT. The COUNT-fold convolution is a different law with the
38 * same mean; it stays available as `compose_repeat` for a caller that wants
39 * exactly that.
40 *
41 * An EXTERNAL CALL has no separate representation: an activity whose host
42 * demand is the law of the call response time composes exactly like a local
43 * computation, and an asynchronous call blocks the caller for no time and is
44 * simply left out of the workflow.
45 *
46 * A QUORUM (partial) AND-join is REFUSED by name rather than served as a full
47 * join, because the first k of n branches to finish is not their maximum.
48 */
49
50#include <algorithm>
51#include <map>
52#include <string>
53#include <vector>
54
60#include "line/util/error.h"
61#include "line/util/matrix.h"
62
63namespace line {
64namespace workflow {
65
66using lang::Distrib;
68// The precedence kinds of `ActivityPrecedenceType.m`, shared with the LQN layer
71
72/**
73 * One precedence of the activity graph.
74 *
75 * `pre_params` carries the AND-join quorum when there is one; `post_params`
76 * carries the OR-fork probabilities or the loop count.
77 */
78template <class T>
79struct Precedence {
80 std::vector<std::string> pre_acts;
81 std::vector<std::string> post_acts;
82 PrecedenceType pre_type = PrecedenceType::PRE_SEQ;
83 PrecedenceType post_type = PrecedenceType::POST_SEQ;
84 std::vector<T> pre_params;
85 std::vector<T> post_params;
86};
87
88/** A phase-type law as the composition rules pass it around. */
89template <class T>
90struct PhLaw {
91 std::vector<T> alpha;
93};
94
95/**
96 * A computational activity.
97 *
98 * Carries no call list: an external call is an activity whose host demand is
99 * the law of the call response time.
100 */
101template <class T>
103public:
104 WorkflowActivity() = default;
105 WorkflowActivity(const std::string& name, const Distrib<T>& host_demand)
106 : name_(name), host_demand_(host_demand) {}
107
108 const std::string& name() const { return name_; }
109
110 const Distrib<T>& host_demand() const { return host_demand_; }
111 void set_host_demand(const Distrib<T>& d) { host_demand_ = d; }
112
113 T host_demand_mean() const { return host_demand_.mean; }
114 T host_demand_scv() const { return host_demand_.scv; }
115
116 /**
117 * The (alpha, T) pair of this activity.
118 *
119 * A zero-time activity is one immediate phase, the representation used
120 * throughout this class; a Markovian law hands back its own pair; anything
121 * else is fitted to an acyclic phase-type on its first two moments, as the
122 * MATLAB and JAR twins do.
123 */
125 const T zero = num_traits<T>::from_int(0);
126 PhLaw<T> out;
127
128 if (host_demand_.type == ProcessType::IMMEDIATE ||
130 out.alpha.assign(1, num_traits<T>::from_int(1));
131 out.S = Matrix<T>(1, 1, zero);
133 return out;
134 }
135
136 if (host_demand_.has_map()) {
137 out.S = host_demand_.D0;
138 out.alpha = init_prob_of(host_demand_);
139 return out;
140 }
141
142 T scv = host_demand_.scv;
145 const Distrib<T> aph = lang::aph_fit_mean_scv(host_demand_.mean, scv);
146 out.S = aph.D0;
147 out.alpha = init_prob_of(aph);
148 return out;
149 }
150
151 std::size_t num_phases() const { return ph_representation().S.rows(); }
152
153 /**
154 * Optional external metadata, MATLAB's `act.metadata` and the JAR's `getMetadata()`.
155 *
156 * Set by `io::wfcommons_load`. Keyed by field name; each value is the JSON TEXT of
157 * the field (`"4"`, `"\"mem\""`, `["a","b"]`), which keeps arrays and objects
158 * intact without making the lang layer depend on a JSON library. It plays no part
159 * in `to_ph()`.
160 */
161 const std::map<std::string, std::string>& metadata() const { return metadata_; }
162 void set_metadata(const std::map<std::string, std::string>& m) { metadata_ = m; }
163 bool has_metadata() const { return !metadata_.empty(); }
164
165private:
166 /**
167 * The initial vector of a Markovian law.
168 *
169 * `Distrib::phase_type` stores alpha in `params`, so a PH/APH/ME hands it
170 * back directly. For every other Markovian family D1 = (-D0 e) alpha, so
171 * ANY row with a positive exit rate recovers alpha -- and it cannot be row
172 * 0, since a canonical bidiagonal APH never completes from phase 1.
173 */
174 static std::vector<T> init_prob_of(const Distrib<T>& d) {
175 const std::size_t n = d.D0.rows();
176 const T zero = num_traits<T>::from_int(0);
177 if ((d.type == ProcessType::PH || d.type == ProcessType::APH ||
178 d.type == ProcessType::ME) &&
179 d.params.size() == n) {
180 return std::vector<T>(d.params.begin(), d.params.begin() + static_cast<long>(n));
181 }
182 for (std::size_t i = 0; i < n; ++i) {
183 T out = zero;
184 for (std::size_t j = 0; j < n; ++j) out += d.D1(i, j);
185 if (!(out > zero)) continue;
186 std::vector<T> alpha(n);
187 for (std::size_t j = 0; j < n; ++j) alpha[j] = T(d.D1(i, j) / out);
188 return alpha;
189 }
190 // No phase completes: start in phase 0, which is what a degenerate
191 // representation leaves as the only defensible reading
192 std::vector<T> alpha(n, zero);
193 if (n > 0) alpha[0] = num_traits<T>::from_int(1);
194 return alpha;
195 }
196
197 std::string name_;
198 Distrib<T> host_demand_;
199 std::map<std::string, std::string> metadata_;
200};
201
202/** A node of the series-parallel tree. */
203enum class SPNodeType { LEAF, SERIAL, PAR, OR, LOOP };
204
205template <class T>
206struct SPNode {
208 /** Activity index for a LEAF, npos otherwise. */
209 std::size_t act = static_cast<std::size_t>(-1);
210 std::vector<std::size_t> kids;
211 std::size_t parent = static_cast<std::size_t>(-1);
212 /** Branch probabilities for an OR node. */
213 std::vector<T> probs;
214 /** Loop count for a LOOP node. */
217 bool valid = false;
218};
219
220/**
221 * The flat series-parallel tree.
222 *
223 * `execs` carries the expected number of executions of each node per workflow
224 * execution, which is the weight by which an LQN metric reconstruction splits
225 * a layer result back over entries, activities and calls.
226 */
227template <class T>
228struct SPTree {
229 std::vector<SPNode<T>> nodes;
230 std::size_t root = static_cast<std::size_t>(-1);
231 /** Node index of each activity's leaf, npos when the activity has none. */
232 std::vector<std::size_t> leaf_of;
233 std::vector<T> execs;
234};
235
236template <class T>
237class Workflow {
238public:
239 static constexpr std::size_t npos = static_cast<std::size_t>(-1);
240
241 explicit Workflow(const std::string& name) : name_(name) {}
242
243 const std::string& name() const { return name_; }
244
245 /** Add an activity; the name must be unique. */
246 std::size_t add_activity(const std::string& name, const Distrib<T>& host_demand) {
247 if (activity_map_.find(name) != activity_map_.end())
248 throw InputError("Workflow: activity '" + name + "' is already declared");
249 activities_.push_back(WorkflowActivity<T>(name, host_demand));
250 const std::size_t idx = activities_.size() - 1;
251 activity_map_[name] = idx;
253 return idx;
254 }
255
256 /** Add an activity with an exponential host demand of the given mean. */
257 std::size_t add_activity(const std::string& name, const T& mean) {
259 }
260
261 void add_precedence(const Precedence<T>& prec) {
262 precedences_.push_back(prec);
264 }
265
266 std::size_t num_activities() const { return activities_.size(); }
267 const std::vector<WorkflowActivity<T>>& activities() const { return activities_; }
268 const std::vector<Precedence<T>>& precedences() const { return precedences_; }
269
270 std::size_t activity_index(const std::string& name) const {
271 auto it = activity_map_.find(name);
272 return it == activity_map_.end() ? npos : it->second;
273 }
274
275 WorkflowActivity<T>& activity(const std::string& name) {
276 const std::size_t i = activity_index(name);
277 if (i == npos) throw InputError("Workflow: activity '" + name + "' not found");
278 return activities_[i];
279 }
280
281 /** Activity by index, the form a series-parallel LEAF node names it in. */
282 const WorkflowActivity<T>& activity_at(std::size_t i) const {
283 if (i >= activities_.size()) throw InputError("Workflow: activity index out of range");
284 return activities_[i];
285 }
286
287 /**
288 * Validate the workflow, throwing on the first defect.
289 *
290 * A quorum AND-join is refused here rather than composed as a full join.
291 */
292 void validate() const {
293 if (activities_.empty())
294 throw InputError("Workflow must have at least one activity.");
295
296 for (const Precedence<T>& prec : precedences_) {
297 for (const std::string& nm : prec.pre_acts)
298 if (activity_index(nm) == npos)
299 throw InputError("Activity '" + nm +
300 "' referenced in precedence not found in workflow.");
301 for (const std::string& nm : prec.post_acts)
302 if (activity_index(nm) == npos)
303 throw InputError("Activity '" + nm +
304 "' referenced in precedence not found in workflow.");
305 }
306
307 for (const Precedence<T>& prec : precedences_) {
308 if (prec.post_type == PrecedenceType::POST_OR) {
309 if (prec.post_params.empty())
310 throw InputError("OR-fork must have probabilities specified.");
311 T total = num_traits<T>::from_int(0);
312 for (const T& p : prec.post_params) total += p;
313 const double gap =
314 std::abs(num_traits<T>::to_double(total) - 1.0);
315 if (gap > GlobalConstants::FineTol)
316 throw InputError("OR-fork probabilities must sum to 1.");
317 }
318 if (prec.post_type == PrecedenceType::POST_LOOP) {
319 if (prec.post_params.size() != 1)
320 throw InputError("Loop count must be a single positive number.");
321 if (!(prec.post_params[0] > num_traits<T>::from_int(0)))
322 throw InputError("Loop count must be a positive number.");
323 }
324 if (prec.pre_type == PrecedenceType::PRE_AND && !prec.pre_params.empty()) {
325 const double quorum = num_traits<T>::to_double(prec.pre_params[0]);
326 const double nb = static_cast<double>(prec.pre_acts.size());
327 if (quorum > 0.0 && quorum < nb)
328 throw UnsupportedError(
329 "AND-join with quorum " + std::to_string(static_cast<long>(quorum)) +
330 " of " + std::to_string(static_cast<long>(nb)) +
331 " is not supported by Workflow: a partial join is not the maximum of the "
332 "branches. Use a full join, or SolverLN with method='default', which "
333 "routes the join explicitly.");
334 }
335 }
336 }
337
338 /**
339 * The composed law of the workflow.
340 *
341 * The generator is acyclic unless a geometric loop closes a cycle over a
342 * multi-phase body, which `is_acyclic_generator` reports; the returned
343 * Distrib is typed APH or PH accordingly.
344 */
346 if (cached_valid_) return cached_ph_;
347 validate();
348
349 PhLaw<T> law;
350 if (!compose_series_parallel(law)) {
351 // Not series-parallel: fall back to the block composition
352 law = build_ctmc();
353 }
354
355 cached_ph_ = Distrib<T>::phase_type(law.alpha, law.S, is_acyclic_generator(law.S));
356 cached_valid_ = true;
357 return cached_ph_;
358 }
359
360 /**
361 * Recompose the law after a demand change.
362 *
363 * Only the series-parallel nodes on the path from a dirty leaf to the root
364 * recompose; nodes whose subtree is unchanged keep their cached law. The
365 * topology is not rebuilt.
366 */
368
369 /**
370 * Draw `n` workflow execution times from the composed law.
371 *
372 * MATLAB samples `wf.toPH().sample(n)`, i.e. `map_sample` on the renewal
373 * MAP (D0 = S, D1 = (-S e) alpha), and native Python's `Workflow.sample`
374 * walks the same absorbing chain; this is that walk, starting each draw in
375 * a phase drawn from alpha (the renewal MAP's embedded stationary law).
376 * The generator follows the tree's Monte Carlo convention, a
377 * `pfqn::McRng` advanced by the call, so draws match MATLAB and numpy in
378 * distribution only, never digit for digit.
379 */
380 std::vector<T> sample(std::size_t n, pfqn::McRng& rng) {
381 const Distrib<T> d = to_ph();
382 mam::Map<T> m;
383 m.D0 = d.D0;
384 m.D1 = d.D1;
385 return mam::map_sample(m, n, rng, d.params);
386 }
387
388 /**
389 * Change the host demand of one activity.
390 *
391 * Marks only that leaf dirty, which is the entry point an iterative solver
392 * uses when it updates call-response laws at each iteration.
393 */
394 void set_activity_demand(const std::string& name, const Distrib<T>& host_demand) {
395 const std::size_t i = activity_index(name);
396 if (i == npos) throw InputError("Workflow: activity '" + name + "' not found");
397 activities_[i].set_host_demand(host_demand);
399 }
400
401 /**
402 * Change only the mean of one activity, preserving its shape.
403 *
404 * The law is scaled in time rather than refitted, so its SCV, its skewness
405 * and its order are preserved and the cached tree keeps its shape.
406 */
407 void set_activity_demand_mean(const std::string& name, const T& mean_value) {
408 if (!(mean_value > num_traits<T>::from_int(0)))
409 throw InputError("The activity mean must be a positive finite scalar.");
410 const std::size_t i = activity_index(name);
411 if (i == npos) throw InputError("Workflow: activity '" + name + "' not found");
412
413 const Distrib<T>& d = activities_[i].host_demand();
414 const T old_mean = d.mean;
415 if (d.type == ProcessType::IMMEDIATE ||
417 activities_[i].set_host_demand(Distrib<T>::exp_mean(mean_value));
419 return;
420 }
421
422 const T factor = T(old_mean / mean_value);
423 activities_[i].set_host_demand(lang::dist_scale_rate(d, factor));
424 // The SCV is invariant under a time scaling
425 rescale_activity_leaf(i, factor);
426 }
427
428 /** Discard the cached law and the decomposition. */
430 cached_valid_ = false;
431 sp_tree_.nodes.clear();
432 sp_tree_.root = npos;
433 sp_tree_.leaf_of.clear();
434 sp_tree_.execs.clear();
435 sp_built_ = false;
436 sp_failed_ = false;
437 }
438
439 /** Mark one activity law dirty, keeping every other cached block. */
440 void invalidate_activity(std::size_t act_idx) {
441 cached_valid_ = false;
442 if (!sp_built_) return;
443 if (act_idx >= sp_tree_.leaf_of.size() || sp_tree_.leaf_of[act_idx] == npos) {
444 // Activity outside the decomposition: rebuild it entirely
446 return;
447 }
448 invalidate_branch(sp_tree_, sp_tree_.leaf_of[act_idx]);
449 }
450
451 /**
452 * Time-scale a cached leaf in place: S -> S*FACTOR with alpha fixed.
453 *
454 * Ancestors still recompose, because they mix phases of several leaves,
455 * but the tree keeps its shape and no acyclic phase-type is refitted.
456 */
457 void rescale_activity_leaf(std::size_t act_idx, const T& factor) {
458 cached_valid_ = false;
459 if (!sp_built_) return;
460 if (act_idx >= sp_tree_.leaf_of.size() || sp_tree_.leaf_of[act_idx] == npos) {
462 return;
463 }
464 const std::size_t k = sp_tree_.leaf_of[act_idx];
465 invalidate_branch(sp_tree_, k);
466 SPNode<T>& node = sp_tree_.nodes[k];
467 if (node.law.S.rows() > 0) {
468 for (std::size_t i = 0; i < node.law.S.rows(); ++i)
469 for (std::size_t j = 0; j < node.law.S.cols(); ++j)
470 node.law.S(i, j) = T(node.law.S(i, j) * factor);
471 node.valid = true;
472 }
473 }
474
475 /**
476 * The cached series-parallel decomposition, or null when the precedence
477 * graph is not series-parallel.
478 */
480 if (!sp_built_ && !sp_failed_) build_sp_tree();
481 return sp_built_ ? &sp_tree_ : nullptr;
482 }
483
484 // -----------------------------------------------------------------
485 // Composition rules
486 // -----------------------------------------------------------------
487
488 /**
489 * Serial composition: the second law starts when the first absorbs.
490 *
491 * S = [S1, (-S1 e) alpha2; 0, S2]
492 *
493 * A defective alpha1 carries an atom at zero, which starts the second law
494 * immediately; this is `aph_simplify` pattern 1.
495 */
496 static PhLaw<T> compose_serial(const PhLaw<T>& a, const PhLaw<T>& b) {
497 const T zero = num_traits<T>::from_int(0);
498 const std::size_t n1 = a.S.rows(), n2 = b.S.rows();
499
500 PhLaw<T> out;
501 out.S = Matrix<T>(n1 + n2, n1 + n2, zero);
502 for (std::size_t i = 0; i < n1; ++i)
503 for (std::size_t j = 0; j < n1; ++j) out.S(i, j) = a.S(i, j);
504 for (std::size_t i = 0; i < n2; ++i)
505 for (std::size_t j = 0; j < n2; ++j) out.S(n1 + i, n1 + j) = b.S(i, j);
506
507 for (std::size_t i = 0; i < n1; ++i) {
508 T rate = zero;
509 for (std::size_t j = 0; j < n1; ++j) rate += a.S(i, j);
510 rate = T(-rate);
511 for (std::size_t j = 0; j < n2; ++j) out.S(i, n1 + j) = T(rate * b.alpha[j]);
512 }
513
514 T defect = num_traits<T>::from_int(1);
515 for (const T& v : a.alpha) defect -= v;
516 out.alpha.assign(n1 + n2, zero);
517 for (std::size_t i = 0; i < n1; ++i) out.alpha[i] = a.alpha[i];
518 for (std::size_t j = 0; j < n2; ++j) out.alpha[n1 + j] = T(defect * b.alpha[j]);
519 return out;
520 }
521
522 /**
523 * Parallel (AND-fork/join) composition: the time until BOTH complete.
524 *
525 * States (i,j) with both active carry the Kronecker sum S1 (+) S2; when one
526 * branch absorbs the chain moves into that branch's own survivor block, so
527 * the law is the maximum of the two.
528 */
529 static PhLaw<T> compose_parallel(const PhLaw<T>& a, const PhLaw<T>& b) {
530 const T zero = num_traits<T>::from_int(0);
531 const std::size_t n1 = a.S.rows(), n2 = b.S.rows();
532 const std::size_t nboth = n1 * n2;
533 const std::size_t ntot = nboth + n1 + n2;
534
535 std::vector<T> abs1(n1, zero), abs2(n2, zero);
536 for (std::size_t i = 0; i < n1; ++i) {
537 T r = zero;
538 for (std::size_t j = 0; j < n1; ++j) r += a.S(i, j);
539 abs1[i] = T(-r);
540 }
541 for (std::size_t i = 0; i < n2; ++i) {
542 T r = zero;
543 for (std::size_t j = 0; j < n2; ++j) r += b.S(i, j);
544 abs2[i] = T(-r);
545 }
546
547 PhLaw<T> out;
548 out.S = Matrix<T>(ntot, ntot, zero);
549
550 // Kronecker sum on the block where both are still running
551 for (std::size_t i = 0; i < n1; ++i)
552 for (std::size_t j = 0; j < n2; ++j) {
553 const std::size_t r = i * n2 + j;
554 for (std::size_t ii = 0; ii < n1; ++ii)
555 out.S(r, ii * n2 + j) += a.S(i, ii);
556 for (std::size_t jj = 0; jj < n2; ++jj)
557 out.S(r, i * n2 + jj) += b.S(j, jj);
558 // one branch absorbs, the other keeps running
559 out.S(r, nboth + i) += abs2[j];
560 out.S(r, nboth + n1 + j) += abs1[i];
561 }
562
563 for (std::size_t i = 0; i < n1; ++i)
564 for (std::size_t j = 0; j < n1; ++j) out.S(nboth + i, nboth + j) = a.S(i, j);
565 for (std::size_t i = 0; i < n2; ++i)
566 for (std::size_t j = 0; j < n2; ++j)
567 out.S(nboth + n1 + i, nboth + n1 + j) = b.S(i, j);
568
569 out.alpha.assign(ntot, zero);
570 for (std::size_t i = 0; i < n1; ++i)
571 for (std::size_t j = 0; j < n2; ++j)
572 out.alpha[i * n2 + j] = T(a.alpha[i] * b.alpha[j]);
573 return out;
574 }
575
576 /**
577 * Probabilistic mixture: a block-diagonal generator whose initial vector
578 * picks branch i with probability PROBS[i]. `aph_simplify` pattern 3,
579 * generalised to any number of branches.
580 */
581 static PhLaw<T> compose_mixture(const std::vector<PhLaw<T>>& laws,
582 const std::vector<T>& probs) {
583 const T zero = num_traits<T>::from_int(0);
584 std::size_t total = 0;
585 for (const PhLaw<T>& l : laws) total += l.S.rows();
586
587 PhLaw<T> out;
588 out.S = Matrix<T>(total, total, zero);
589 out.alpha.assign(total, zero);
590
591 std::size_t off = 0;
592 for (std::size_t b = 0; b < laws.size(); ++b) {
593 const std::size_t nb = laws[b].S.rows();
594 for (std::size_t i = 0; i < nb; ++i) {
595 for (std::size_t j = 0; j < nb; ++j) out.S(off + i, off + j) = laws[b].S(i, j);
596 out.alpha[off + i] = T(probs[b] * laws[b].alpha[i]);
597 }
598 off += nb;
599 }
600 return out;
601 }
602
603 /**
604 * Geometric repetition of a phase-type law, the POST_LOOP semantics.
605 *
606 * For COUNT >= 1 the body runs at least once and repeats on absorption
607 * with probability P = 1-1/COUNT, so
608 *
609 * S_out = S + P/D (-S e) alpha, alpha_out = alpha / D
610 *
611 * with D = 1 - P(1 - alpha e) the correction for an atom at zero in alpha.
612 * The ORDER is that of the body, unlike the COUNT-fold convolution, and the
613 * mean is COUNT times the body mean in both cases.
614 *
615 * For COUNT < 1 the body runs at most once, with probability COUNT; the
616 * skipped branch is an immediate phase.
617 */
618 static PhLaw<T> compose_loop_geometric(const PhLaw<T>& body, const T& count) {
619 const T zero = num_traits<T>::from_int(0);
620 const T one = num_traits<T>::from_int(1);
621 const std::size_t n = body.S.rows();
622 const double c = num_traits<T>::to_double(count);
623
624 PhLaw<T> out;
625 if (!(c > 0.0)) {
626 out.alpha.assign(1, one);
627 out.S = Matrix<T>(1, 1, zero);
629 return out; // zero-time branch
630 }
631
632 if (std::abs(c - 1.0) <= GlobalConstants::FineTol) return body;
633
634 if (c < 1.0) {
635 // Executed with probability COUNT, skipped otherwise
636 out.alpha.assign(n + 1, zero);
637 for (std::size_t i = 0; i < n; ++i) out.alpha[i] = T(count * body.alpha[i]);
638 out.alpha[n] = T(one - count);
639 out.S = Matrix<T>(n + 1, n + 1, zero);
640 for (std::size_t i = 0; i < n; ++i)
641 for (std::size_t j = 0; j < n; ++j) out.S(i, j) = body.S(i, j);
643 return out; // zero-time skip branch
644 }
645
646 const T p = T(one - one / count);
647 T defect = one;
648 for (const T& v : body.alpha) defect -= v;
649 const T denom = T(one - p * defect);
650
651 out.alpha.assign(n, zero);
652 for (std::size_t i = 0; i < n; ++i) out.alpha[i] = T(body.alpha[i] / denom);
653
654 out.S = body.S;
655 for (std::size_t i = 0; i < n; ++i) {
656 T rate = zero;
657 for (std::size_t j = 0; j < n; ++j) rate += body.S(i, j);
658 rate = T(-rate * p / denom);
659 for (std::size_t j = 0; j < n; ++j) out.S(i, j) += T(rate * body.alpha[j]);
660 }
661 return out;
662 }
663
664 /**
665 * COUNT-fold convolution of a phase-type law.
666 *
667 * The DETERMINISTIC repetition, kept for a caller that genuinely wants an
668 * exact number of executions. POST_LOOP is geometric and uses
669 * `compose_loop_geometric` instead.
670 */
671 static PhLaw<T> compose_repeat(const PhLaw<T>& body, long count) {
672 if (count <= 0) {
673 PhLaw<T> out;
674 out.alpha.assign(1, num_traits<T>::from_int(1));
675 out.S = Matrix<T>(1, 1, num_traits<T>::from_int(0));
677 return out;
678 }
679 PhLaw<T> out = body;
680 for (long i = 1; i < count; ++i) out = compose_serial(out, body);
681 return out;
682 }
683
684 /**
685 * True when the phase graph of S has no cycle.
686 *
687 * A geometric loop over a body of two or more phases closes a cycle, so the
688 * composed law is a PH and not an APH.
689 */
690 static bool is_acyclic_generator(const Matrix<T>& S) {
691 const std::size_t n = S.rows();
692 std::vector<std::vector<bool>> A(n, std::vector<bool>(n, false));
693 std::vector<std::size_t> in_deg(n, 0);
694 for (std::size_t i = 0; i < n; ++i)
695 for (std::size_t j = 0; j < n; ++j) {
696 if (i == j) continue;
697 if (std::abs(num_traits<T>::to_double(S(i, j))) > GlobalConstants::ArcTol) {
698 A[i][j] = true;
699 ++in_deg[j];
700 }
701 }
702
703 std::vector<std::size_t> queue;
704 for (std::size_t i = 0; i < n; ++i)
705 if (in_deg[i] == 0) queue.push_back(i);
706 std::size_t visited = 0, head = 0;
707 while (head < queue.size()) {
708 const std::size_t cur = queue[head++];
709 ++visited;
710 for (std::size_t j = 0; j < n; ++j) {
711 if (!A[cur][j]) continue;
712 if (--in_deg[j] == 0) queue.push_back(j);
713 }
714 }
715 return visited == n;
716 }
717
718 // -----------------------------------------------------------------
719 // Precedence factories, named as in MATLAB and the JAR
720 // -----------------------------------------------------------------
721
722 static Precedence<T> Serial(const std::string& pre, const std::string& post) {
724 p.pre_acts.push_back(pre);
725 p.post_acts.push_back(post);
726 p.pre_type = PrecedenceType::PRE_SEQ;
727 p.post_type = PrecedenceType::POST_SEQ;
728 return p;
729 }
730
731 static std::vector<Precedence<T>> SerialSequence(const std::vector<std::string>& acts) {
732 std::vector<Precedence<T>> out;
733 for (std::size_t i = 0; i + 1 < acts.size(); ++i)
734 out.push_back(Serial(acts[i], acts[i + 1]));
735 return out;
736 }
737
738 static Precedence<T> AndFork(const std::string& pre,
739 const std::vector<std::string>& posts) {
741 p.pre_acts.push_back(pre);
742 p.post_acts = posts;
743 p.pre_type = PrecedenceType::PRE_SEQ;
744 p.post_type = PrecedenceType::POST_AND;
745 return p;
746 }
747
748 /** An empty QUORUM means a full join; a partial one is refused by validate. */
749 static Precedence<T> AndJoin(const std::vector<std::string>& pres,
750 const std::string& post,
751 const std::vector<T>& quorum = std::vector<T>()) {
753 p.pre_acts = pres;
754 p.post_acts.push_back(post);
755 p.pre_type = PrecedenceType::PRE_AND;
756 p.post_type = PrecedenceType::POST_SEQ;
757 p.pre_params = quorum;
758 return p;
759 }
760
761 static Precedence<T> OrFork(const std::string& pre, const std::vector<std::string>& posts,
762 const std::vector<T>& probs) {
764 p.pre_acts.push_back(pre);
765 p.post_acts = posts;
766 p.pre_type = PrecedenceType::PRE_SEQ;
767 p.post_type = PrecedenceType::POST_OR;
768 p.post_params = probs;
769 return p;
770 }
771
772 static Precedence<T> OrJoin(const std::vector<std::string>& pres,
773 const std::string& post) {
775 p.pre_acts = pres;
776 p.post_acts.push_back(post);
777 p.pre_type = PrecedenceType::PRE_OR;
778 p.post_type = PrecedenceType::POST_SEQ;
779 return p;
780 }
781
782 /**
783 * A loop: PRE runs once, then POSTS[0..n-2] repeat a geometric number of
784 * times of mean COUNT and POSTS[n-1] continues after the loop.
785 */
786 static Precedence<T> Loop(const std::string& pre, const std::vector<std::string>& posts,
787 const T& count) {
789 p.pre_acts.push_back(pre);
790 p.post_acts = posts;
791 p.pre_type = PrecedenceType::PRE_SEQ;
792 p.post_type = PrecedenceType::POST_LOOP;
793 p.post_params.push_back(count);
794 return p;
795 }
796
797private:
798 // -----------------------------------------------------------------
799 // Series-parallel decomposition
800 // -----------------------------------------------------------------
801
802 /** How a parsed sequence terminated. */
803 enum class ParseStatus { END, STOP, JOIN, FAIL };
804
805 struct ParseState {
806 /** Index of the precedence each activity heads / is reached by. */
807 std::vector<std::size_t> out_p, in_p;
808 std::vector<bool> consumed;
809 };
810
811 bool compose_series_parallel(PhLaw<T>& out) {
812 if (!sp_built_) {
813 if (sp_failed_) return false;
814 build_sp_tree();
815 if (!sp_built_) return false;
816 }
817 out = compose_node(sp_tree_.root);
818 return true;
819 }
820
821 /**
822 * Decompose the precedence graph into a series-parallel tree.
823 *
824 * The decomposition is attempted once per topology: on failure `sp_failed_`
825 * is set and the block path takes over.
826 */
827 bool build_sp_tree() {
828 sp_tree_.nodes.clear();
829 sp_tree_.root = npos;
830 sp_tree_.leaf_of.clear();
831 sp_tree_.execs.clear();
832 sp_built_ = false;
833 sp_failed_ = true;
834
835 const std::size_t n = activities_.size();
836 if (n == 0) return false;
837
838 ParseState S;
839 S.out_p.assign(n, npos);
840 S.in_p.assign(n, npos);
841 S.consumed.assign(n, false);
842
843 // An activity may head at most one precedence and be reached by at most
844 // one precedence; otherwise the graph is not series-parallel
845 for (std::size_t p = 0; p < precedences_.size(); ++p) {
846 for (const std::string& nm : precedences_[p].pre_acts) {
847 const std::size_t i = activity_index(nm);
848 if (i == npos || S.out_p[i] != npos) return false;
849 S.out_p[i] = p;
850 }
851 for (const std::string& nm : precedences_[p].post_acts) {
852 const std::size_t j = activity_index(nm);
853 if (j == npos || S.in_p[j] != npos) return false;
854 S.in_p[j] = p;
855 }
856 }
857
858 std::vector<std::size_t> starts;
859 for (std::size_t i = 0; i < n; ++i)
860 if (S.in_p[i] == npos) starts.push_back(i);
861 if (starts.size() != 1) return false;
862
863 std::vector<std::size_t> kids;
864 std::size_t stop_at = npos;
865 const ParseStatus st = sp_parse_seq(S, starts[0], std::vector<std::size_t>(), kids, stop_at);
866 if (st != ParseStatus::END) return false;
867 for (std::size_t i = 0; i < n; ++i)
868 if (!S.consumed[i]) return false;
869
870 const std::size_t root = sp_serial_node(kids);
871 if (root == npos) return false;
872
873 sp_tree_.root = root;
874 sp_tree_.leaf_of.assign(n, npos);
875 for (std::size_t k = 0; k < sp_tree_.nodes.size(); ++k)
876 if (sp_tree_.nodes[k].type == SPNodeType::LEAF)
877 sp_tree_.leaf_of[sp_tree_.nodes[k].act] = k;
878 sp_tree_.execs = sp_execution_counts(sp_tree_);
879
880 sp_built_ = true;
881 sp_failed_ = false;
882 return true;
883 }
884
885 /** Parse a maximal sequence of blocks starting at CUR. */
886 ParseStatus sp_parse_seq(ParseState& S, std::size_t cur,
887 const std::vector<std::size_t>& stop_set,
888 std::vector<std::size_t>& kids, std::size_t& stop_at) {
889 stop_at = npos;
890 while (true) {
891 if (cur == npos) return ParseStatus::END;
892 if (std::find(stop_set.begin(), stop_set.end(), cur) != stop_set.end()) {
893 stop_at = cur;
894 return ParseStatus::STOP;
895 }
896 if (S.consumed[cur]) return ParseStatus::FAIL;
897 S.consumed[cur] = true;
898 kids.push_back(sp_add_node(SPNodeType::LEAF, cur, std::vector<std::size_t>(),
899 std::vector<T>(), num_traits<T>::from_int(0)));
900
901 const std::size_t p = S.out_p[cur];
902 if (p == npos) return ParseStatus::END;
903 const Precedence<T>& prec = precedences_[p];
904 if (prec.pre_acts.size() > 1) {
905 // CUR is the tail of a branch: the caller composes the join
906 stop_at = p;
907 return ParseStatus::JOIN;
908 }
909
910 std::vector<std::size_t> post_inds = sp_indices_of(prec.post_acts);
911 for (std::size_t i : post_inds)
912 if (i == npos) return ParseStatus::FAIL;
913
914 std::size_t knode = npos, next_act = npos;
915 switch (prec.post_type) {
916 case PrecedenceType::POST_AND:
917 if (!sp_parse_fork(S, post_inds, std::vector<T>(), stop_set, true, knode,
918 next_act))
919 return ParseStatus::FAIL;
920 kids.push_back(knode);
921 cur = next_act;
922 break;
923 case PrecedenceType::POST_OR:
924 if (prec.post_params.size() != post_inds.size()) return ParseStatus::FAIL;
925 if (!sp_parse_fork(S, post_inds, prec.post_params, stop_set, false, knode,
926 next_act))
927 return ParseStatus::FAIL;
928 kids.push_back(knode);
929 cur = next_act;
930 break;
931 case PrecedenceType::POST_LOOP:
932 if (!sp_parse_loop(S, post_inds, prec.post_params, stop_set, knode, next_act))
933 return ParseStatus::FAIL;
934 kids.push_back(knode);
935 cur = next_act;
936 break;
937 case PrecedenceType::POST_SEQ:
938 if (post_inds.size() != 1) return ParseStatus::FAIL;
939 cur = post_inds[0];
940 break;
941 default:
942 // POST_CACHE and any other pattern is not a workflow
943 // composition rule
944 return ParseStatus::FAIL;
945 }
946 }
947 }
948
949 /** Parse the branches of a fork and their join. */
950 bool sp_parse_fork(ParseState& S, const std::vector<std::size_t>& branch_heads,
951 const std::vector<T>& probs, const std::vector<std::size_t>& stop_set,
952 bool is_and, std::size_t& knode, std::size_t& next_act) {
953 const std::size_t nb = branch_heads.size();
954 std::vector<std::size_t> branch_nodes(nb, npos), bstop(nb, npos);
955 std::vector<ParseStatus> bstatus(nb, ParseStatus::FAIL);
956
957 for (std::size_t b = 0; b < nb; ++b) {
958 std::vector<std::size_t> bkids;
959 std::size_t sa = npos;
960 const ParseStatus st = sp_parse_seq(S, branch_heads[b], stop_set, bkids, sa);
961 if (st == ParseStatus::FAIL) return false;
962 const std::size_t bn = sp_serial_node(bkids);
963 if (bn == npos) return false;
964 branch_nodes[b] = bn;
965 bstatus[b] = st;
966 bstop[b] = sa;
967 }
968
969 const bool all_join =
970 std::all_of(bstatus.begin(), bstatus.end(),
971 [](ParseStatus s) { return s == ParseStatus::JOIN; });
972 const bool all_end = std::all_of(bstatus.begin(), bstatus.end(),
973 [](ParseStatus s) { return s == ParseStatus::END; });
974 const bool all_stop = std::all_of(bstatus.begin(), bstatus.end(),
975 [](ParseStatus s) { return s == ParseStatus::STOP; });
976 const bool same_stop =
977 std::all_of(bstop.begin(), bstop.end(),
978 [&bstop](std::size_t x) { return x == bstop[0]; });
979
980 if (all_join) {
981 if (!same_stop) return false;
982 const Precedence<T>& join_prec = precedences_[bstop[0]];
983 if (join_prec.pre_acts.size() != nb) return false;
984 if (is_and) {
985 if (join_prec.pre_type != PrecedenceType::PRE_AND) return false;
986 } else {
987 if (join_prec.pre_type != PrecedenceType::PRE_OR) return false;
988 }
989 const std::vector<std::size_t> post_inds = sp_indices_of(join_prec.post_acts);
990 if (post_inds.size() != 1 || post_inds[0] == npos) return false;
991 next_act = post_inds[0];
992 } else if (all_end) {
993 // Branches terminate the workflow. An AND-fork with no join still
994 // synchronises at the end of the workflow
995 next_act = npos;
996 } else if (!is_and && all_stop && same_stop) {
997 next_act = bstop[0];
998 } else {
999 return false;
1000 }
1001
1002 knode = sp_add_node(is_and ? SPNodeType::PAR : SPNodeType::OR, npos, branch_nodes, probs,
1003 num_traits<T>::from_int(0));
1004 return true;
1005 }
1006
1007 /**
1008 * Parse a loop block: the last post activity continues after the loop, the
1009 * others form the body.
1010 */
1011 bool sp_parse_loop(ParseState& S, const std::vector<std::size_t>& post_inds,
1012 const std::vector<T>& counts, const std::vector<std::size_t>& stop_set,
1013 std::size_t& knode, std::size_t& next_act) {
1014 if (counts.size() != 1) return false;
1015 const T count = counts[0];
1016
1017 std::vector<std::size_t> body_acts;
1018 std::size_t end_act = npos;
1019 if (post_inds.size() >= 2) {
1020 body_acts.assign(post_inds.begin(), post_inds.end() - 1);
1021 end_act = post_inds.back();
1022 } else {
1023 body_acts.push_back(post_inds[0]);
1024 }
1025
1026 std::vector<std::size_t> loop_stop = stop_set;
1027 loop_stop.insert(loop_stop.end(), body_acts.begin(), body_acts.end());
1028 if (end_act != npos) loop_stop.push_back(end_act);
1029
1030 std::vector<std::size_t> body_kids;
1031 std::size_t j = 0;
1032 while (j < body_acts.size()) {
1033 const std::size_t a = body_acts[j];
1034 if (S.consumed[a]) {
1035 ++j;
1036 continue;
1037 }
1038 std::vector<std::size_t> this_stop;
1039 for (std::size_t x : loop_stop)
1040 if (x != a) this_stop.push_back(x);
1041
1042 std::vector<std::size_t> kk;
1043 std::size_t sa = npos;
1044 const ParseStatus st = sp_parse_seq(S, a, this_stop, kk, sa);
1045 if (st == ParseStatus::FAIL) return false;
1046 body_kids.insert(body_kids.end(), kk.begin(), kk.end());
1047
1048 if (st == ParseStatus::END) {
1049 ++j;
1050 } else if (st == ParseStatus::STOP) {
1051 const auto it = std::find(body_acts.begin(), body_acts.end(), sa);
1052 if (it != body_acts.end()) {
1053 j = static_cast<std::size_t>(it - body_acts.begin());
1054 } else if (end_act != npos && sa == end_act) {
1055 j = body_acts.size();
1056 } else {
1057 return false;
1058 }
1059 } else {
1060 // A join reached from inside the body crosses the loop
1061 // boundary, so the graph is not series-parallel
1062 return false;
1063 }
1064 }
1065
1066 const std::size_t body_node = sp_serial_node(body_kids);
1067 if (body_node == npos) return false;
1068
1069 knode = sp_add_node(SPNodeType::LOOP, npos, std::vector<std::size_t>(1, body_node),
1070 std::vector<T>(), count);
1071 next_act = end_act;
1072 return true;
1073 }
1074
1075 /** Wrap a list of nodes in a serial node; a single node is returned as is. */
1076 std::size_t sp_serial_node(const std::vector<std::size_t>& kids) {
1077 if (kids.empty()) return npos;
1078 if (kids.size() == 1) return kids[0];
1079 return sp_add_node(SPNodeType::SERIAL, npos, kids, std::vector<T>(),
1080 num_traits<T>::from_int(0));
1081 }
1082
1083 std::size_t sp_add_node(SPNodeType type, std::size_t act,
1084 const std::vector<std::size_t>& kids, const std::vector<T>& probs,
1085 const T& count) {
1086 SPNode<T> node;
1087 node.type = type;
1088 node.act = act;
1089 node.kids = kids;
1090 node.probs = probs;
1091 node.count = count;
1092 sp_tree_.nodes.push_back(node);
1093 const std::size_t k = sp_tree_.nodes.size() - 1;
1094 for (std::size_t c : kids) sp_tree_.nodes[c].parent = k;
1095 return k;
1096 }
1097
1098 std::vector<std::size_t> sp_indices_of(const std::vector<std::string>& names) const {
1099 std::vector<std::size_t> out(names.size(), npos);
1100 for (std::size_t i = 0; i < names.size(); ++i) out[i] = activity_index(names[i]);
1101 return out;
1102 }
1103
1104 /**
1105 * Compose one node. A cached node is returned untouched, so a demand change
1106 * only recomposes the path from the dirty leaf to the root.
1107 */
1108 PhLaw<T> compose_node(std::size_t k) {
1109 SPNode<T>& node = sp_tree_.nodes[k];
1110 if (node.valid) return node.law;
1111
1112 PhLaw<T> law;
1113 switch (node.type) {
1114 case SPNodeType::LEAF:
1115 law = activities_[node.act].ph_representation();
1116 break;
1117 case SPNodeType::SERIAL: {
1118 const std::vector<std::size_t> kids = node.kids;
1119 law = compose_node(kids[0]);
1120 for (std::size_t i = 1; i < kids.size(); ++i)
1121 law = compose_serial(law, compose_node(kids[i]));
1122 break;
1123 }
1124 case SPNodeType::PAR: {
1125 const std::vector<std::size_t> kids = node.kids;
1126 law = compose_node(kids[0]);
1127 for (std::size_t i = 1; i < kids.size(); ++i)
1128 law = compose_parallel(law, compose_node(kids[i]));
1129 break;
1130 }
1131 case SPNodeType::OR: {
1132 const std::vector<std::size_t> kids = node.kids;
1133 std::vector<PhLaw<T>> laws;
1134 for (std::size_t c : kids) laws.push_back(compose_node(c));
1135 law = compose_mixture(laws, sp_tree_.nodes[k].probs);
1136 break;
1137 }
1138 case SPNodeType::LOOP: {
1139 const std::size_t child = node.kids[0];
1140 law = compose_loop_geometric(compose_node(child), sp_tree_.nodes[k].count);
1141 break;
1142 }
1143 }
1144
1145 sp_tree_.nodes[k].law = law;
1146 sp_tree_.nodes[k].valid = true;
1147 return law;
1148 }
1149
1150 static void invalidate_branch(SPTree<T>& tree, std::size_t k) {
1151 while (k != npos) {
1152 tree.nodes[k].valid = false;
1153 k = tree.nodes[k].parent;
1154 }
1155 }
1156
1157 /**
1158 * Expected executions of each node per workflow run.
1159 *
1160 * Serial and parallel children inherit the count of their parent, an OR
1161 * branch is weighted by its probability, and a loop body by the loop count.
1162 */
1163 static std::vector<T> sp_execution_counts(const SPTree<T>& tree) {
1164 std::vector<T> execs(tree.nodes.size(), num_traits<T>::from_int(0));
1165 if (tree.root == npos) return execs;
1166 execs[tree.root] = num_traits<T>::from_int(1);
1167 std::vector<std::size_t> stack(1, tree.root);
1168 while (!stack.empty()) {
1169 const std::size_t k = stack.back();
1170 stack.pop_back();
1171 const SPNode<T>& node = tree.nodes[k];
1172 for (std::size_t i = 0; i < node.kids.size(); ++i) {
1173 const std::size_t c = node.kids[i];
1174 if (node.type == SPNodeType::OR) {
1175 execs[c] = T(execs[k] * node.probs[i]);
1176 } else if (node.type == SPNodeType::LOOP) {
1177 execs[c] = T(execs[k] * node.count);
1178 } else {
1179 execs[c] = execs[k];
1180 }
1181 stack.push_back(c);
1182 }
1183 }
1184 return execs;
1185 }
1186
1187 // -----------------------------------------------------------------
1188 // Block composition, the fallback for a graph that is not series-parallel
1189 // -----------------------------------------------------------------
1190
1191 struct ForkInfo {
1192 bool is_and = true;
1193 std::size_t pre_act = npos;
1194 std::vector<std::size_t> post_acts;
1195 std::vector<T> probs;
1196 };
1197
1198 struct JoinInfo {
1199 bool is_and = true;
1200 std::vector<std::size_t> pre_acts;
1201 std::size_t post_act = npos;
1202 };
1203
1204 struct LoopInfo {
1205 std::size_t pre_act = npos;
1206 std::vector<std::size_t> body_acts;
1207 std::size_t end_act = npos;
1208 T count = num_traits<T>::from_int(1);
1209 };
1210
1211 struct Structure {
1212 std::vector<std::vector<std::size_t>> adj;
1213 std::vector<std::size_t> in_deg, out_deg;
1214 std::vector<ForkInfo> forks;
1215 std::vector<JoinInfo> joins;
1216 std::vector<LoopInfo> loops;
1217 };
1218
1219 Structure analyze_structure() const {
1220 const std::size_t n = activities_.size();
1221 Structure st;
1222 st.adj.assign(n, std::vector<std::size_t>());
1223 st.in_deg.assign(n, 0);
1224 st.out_deg.assign(n, 0);
1225
1226 for (const Precedence<T>& prec : precedences_) {
1227 const std::vector<std::size_t> pre = sp_indices_of(prec.pre_acts);
1228 const std::vector<std::size_t> post = sp_indices_of(prec.post_acts);
1229 for (std::size_t i : pre)
1230 for (std::size_t j : post) {
1231 st.adj[i].push_back(j);
1232 ++st.out_deg[i];
1233 ++st.in_deg[j];
1234 }
1235
1236 if (prec.post_type == PrecedenceType::POST_AND) {
1237 ForkInfo f;
1238 f.is_and = true;
1239 f.pre_act = pre[0];
1240 f.post_acts = post;
1241 st.forks.push_back(f);
1242 } else if (prec.post_type == PrecedenceType::POST_OR) {
1243 ForkInfo f;
1244 f.is_and = false;
1245 f.pre_act = pre[0];
1246 f.post_acts = post;
1247 f.probs = prec.post_params;
1248 st.forks.push_back(f);
1249 } else if (prec.post_type == PrecedenceType::POST_LOOP) {
1250 LoopInfo l;
1251 l.pre_act = pre[0];
1252 if (post.size() > 1) {
1253 l.body_acts.assign(post.begin(), post.end() - 1);
1254 l.end_act = post.back();
1255 } else {
1256 l.body_acts = post;
1257 }
1258 l.count = prec.post_params.empty() ? num_traits<T>::from_int(1)
1259 : prec.post_params[0];
1260 st.loops.push_back(l);
1261 }
1262
1263 if (prec.pre_type == PrecedenceType::PRE_AND ||
1264 prec.pre_type == PrecedenceType::PRE_OR) {
1265 JoinInfo j;
1266 j.is_and = prec.pre_type == PrecedenceType::PRE_AND;
1267 j.pre_acts = pre;
1268 j.post_act = post[0];
1269 st.joins.push_back(j);
1270 }
1271 }
1272 return st;
1273 }
1274
1275 std::vector<std::size_t> topological_sort(
1276 const std::vector<std::vector<std::size_t>>& adj) const {
1277 const std::size_t n = activities_.size();
1278 std::vector<std::size_t> in_deg(n, 0);
1279 for (const std::vector<std::size_t>& nbrs : adj)
1280 for (std::size_t j : nbrs) ++in_deg[j];
1281
1282 std::vector<std::size_t> queue, order;
1283 for (std::size_t i = 0; i < n; ++i)
1284 if (in_deg[i] == 0) queue.push_back(i);
1285 std::size_t head = 0;
1286 while (head < queue.size()) {
1287 const std::size_t cur = queue[head++];
1288 order.push_back(cur);
1289 for (std::size_t nx : adj[cur])
1290 if (--in_deg[nx] == 0) queue.push_back(nx);
1291 }
1292 std::vector<bool> seen(n, false);
1293 for (std::size_t i : order) seen[i] = true;
1294 for (std::size_t i = 0; i < n; ++i)
1295 if (!seen[i]) order.push_back(i);
1296 return order;
1297 }
1298
1299 /**
1300 * The block composition.
1301 *
1302 * A HEURISTIC, and the reason the series-parallel reduction exists: it
1303 * folds each fork/join and loop block independently and then chains what is
1304 * left in topological order, so a block nested inside another is not
1305 * reduced exactly.
1306 */
1307 PhLaw<T> build_ctmc() {
1308 const std::size_t n = activities_.size();
1309 if (n == 1) return activities_[0].ph_representation();
1310
1311 const Structure st = analyze_structure();
1312
1313 std::vector<PhLaw<T>> block(n);
1314 std::vector<bool> absorbed(n, false);
1315 for (std::size_t i = 0; i < n; ++i) block[i] = activities_[i].ph_representation();
1316
1317 for (const LoopInfo& loop : st.loops) {
1318 PhLaw<T> body = activities_[loop.body_acts[0]].ph_representation();
1319 for (std::size_t j = 1; j < loop.body_acts.size(); ++j)
1320 body = compose_serial(body, activities_[loop.body_acts[j]].ph_representation());
1321
1322 PhLaw<T> res = compose_serial(block[loop.pre_act],
1323 compose_loop_geometric(body, loop.count));
1324 if (loop.end_act != npos) {
1325 res = compose_serial(res, activities_[loop.end_act].ph_representation());
1326 absorbed[loop.end_act] = true;
1327 }
1328 block[loop.pre_act] = res;
1329 for (std::size_t idx : loop.body_acts) absorbed[idx] = true;
1330 }
1331
1332 for (const ForkInfo& fork : st.forks) {
1333 const JoinInfo* join = find_matching_join(fork.post_acts, st.joins, fork.is_and);
1334 if (fork.is_and && join == nullptr) continue;
1335
1336 PhLaw<T> inner;
1337 if (fork.is_and) {
1338 inner = block[fork.post_acts[0]];
1339 for (std::size_t i = 1; i < fork.post_acts.size(); ++i)
1340 inner = compose_parallel(inner, block[fork.post_acts[i]]);
1341 } else {
1342 std::vector<PhLaw<T>> laws;
1343 for (std::size_t idx : fork.post_acts) laws.push_back(block[idx]);
1344 inner = compose_mixture(laws, fork.probs);
1345 }
1346
1347 PhLaw<T> res = compose_serial(block[fork.pre_act], inner);
1348 if (join != nullptr && !absorbed[join->post_act]) {
1349 res = compose_serial(res, block[join->post_act]);
1350 absorbed[join->post_act] = true;
1351 }
1352 block[fork.pre_act] = res;
1353 for (std::size_t idx : fork.post_acts) absorbed[idx] = true;
1354 }
1355
1356 const std::vector<std::size_t> order = topological_sort(st.adj);
1357 bool started = false;
1358 PhLaw<T> out;
1359 for (std::size_t idx : order) {
1360 if (absorbed[idx]) continue;
1361 if (!started) {
1362 out = block[idx];
1363 started = true;
1364 } else {
1365 out = compose_serial(out, block[idx]);
1366 }
1367 }
1368 if (!started) out = activities_[0].ph_representation();
1369 return out;
1370 }
1371
1372 static const JoinInfo* find_matching_join(const std::vector<std::size_t>& post_acts,
1373 const std::vector<JoinInfo>& joins, bool is_and) {
1374 std::vector<std::size_t> want = post_acts;
1375 std::sort(want.begin(), want.end());
1376 for (const JoinInfo& j : joins) {
1377 if (j.is_and != is_and) continue;
1378 std::vector<std::size_t> have = j.pre_acts;
1379 std::sort(have.begin(), have.end());
1380 if (have == want) return &j;
1381 }
1382 return nullptr;
1383 }
1384
1385 std::string name_;
1386 std::vector<WorkflowActivity<T>> activities_;
1387 std::map<std::string, std::size_t> activity_map_;
1388 std::vector<Precedence<T>> precedences_;
1389
1390 Distrib<T> cached_ph_;
1391 bool cached_valid_ = false;
1392 SPTree<T> sp_tree_;
1393 bool sp_built_ = false;
1394 bool sp_failed_ = false;
1395};
1396
1397} // namespace workflow
1398} // namespace line
1399
1400#endif // LINE_LANG_WORKFLOW_WORKFLOW_H
InputError(const std::string &what)
Definition error.h:39
std::size_t rows() const
Definition matrix.h:89
UnsupportedError(const std::string &what)
Definition error.h:51
Workflow(const std::string &name)
Definition workflow.h:241
A computational activity.
Definition workflow.h:102
void set_metadata(const std::map< std::string, std::string > &m)
Definition workflow.h:162
WorkflowActivity(const std::string &name, const Distrib< T > &host_demand)
Definition workflow.h:105
void set_host_demand(const Distrib< T > &d)
Definition workflow.h:111
const std::string & name() const
Definition workflow.h:108
std::size_t num_phases() const
Definition workflow.h:151
const std::map< std::string, std::string > & metadata() const
Optional external metadata, MATLAB's act.metadata and the JAR's getMetadata().
Definition workflow.h:161
PhLaw< T > ph_representation() const
The (alpha, T) pair of this activity.
Definition workflow.h:124
const Distrib< T > & host_demand() const
Definition workflow.h:110
static bool is_acyclic_generator(const Matrix< T > &S)
True when the phase graph of S has no cycle.
Definition workflow.h:690
static Precedence< T > Loop(const std::string &pre, const std::vector< std::string > &posts, const T &count)
A loop: PRE runs once, then POSTS[0..n-2] repeat a geometric number of times of mean COUNT and POSTS[...
Definition workflow.h:786
static Precedence< T > AndJoin(const std::vector< std::string > &pres, const std::string &post, const std::vector< T > &quorum=std::vector< T >())
An empty QUORUM means a full join; a partial one is refused by validate.
Definition workflow.h:749
const SPTree< T > * sp_tree()
The cached series-parallel decomposition, or null when the precedence graph is not series-parallel.
Definition workflow.h:479
std::vector< T > sample(std::size_t n, pfqn::McRng &rng)
Draw n workflow execution times from the composed law.
Definition workflow.h:380
void invalidate_topology()
Discard the cached law and the decomposition.
Definition workflow.h:429
void set_activity_demand(const std::string &name, const Distrib< T > &host_demand)
Change the host demand of one activity.
Definition workflow.h:394
static PhLaw< T > compose_repeat(const PhLaw< T > &body, long count)
COUNT-fold convolution of a phase-type law.
Definition workflow.h:671
std::size_t add_activity(const std::string &name, const Distrib< T > &host_demand)
Add an activity; the name must be unique.
Definition workflow.h:246
static PhLaw< T > compose_loop_geometric(const PhLaw< T > &body, const T &count)
Geometric repetition of a phase-type law, the POST_LOOP semantics.
Definition workflow.h:618
void add_precedence(const Precedence< T > &prec)
Definition workflow.h:261
const std::string & name() const
Definition workflow.h:243
std::size_t add_activity(const std::string &name, const T &mean)
Add an activity with an exponential host demand of the given mean.
Definition workflow.h:257
const std::vector< WorkflowActivity< T > > & activities() const
Definition workflow.h:267
static std::vector< Precedence< T > > SerialSequence(const std::vector< std::string > &acts)
Definition workflow.h:731
static constexpr std::size_t npos
Definition workflow.h:239
void invalidate_activity(std::size_t act_idx)
Mark one activity law dirty, keeping every other cached block.
Definition workflow.h:440
static Precedence< T > AndFork(const std::string &pre, const std::vector< std::string > &posts)
Definition workflow.h:738
static Precedence< T > OrFork(const std::string &pre, const std::vector< std::string > &posts, const std::vector< T > &probs)
Definition workflow.h:761
void set_activity_demand_mean(const std::string &name, const T &mean_value)
Change only the mean of one activity, preserving its shape.
Definition workflow.h:407
Distrib< T > refresh_ph()
Recompose the law after a demand change.
Definition workflow.h:367
std::size_t num_activities() const
Definition workflow.h:266
void validate() const
Validate the workflow, throwing on the first defect.
Definition workflow.h:292
void rescale_activity_leaf(std::size_t act_idx, const T &factor)
Time-scale a cached leaf in place: S -> S*FACTOR with alpha fixed.
Definition workflow.h:457
std::size_t activity_index(const std::string &name) const
Definition workflow.h:270
static PhLaw< T > compose_parallel(const PhLaw< T > &a, const PhLaw< T > &b)
Parallel (AND-fork/join) composition: the time until BOTH complete.
Definition workflow.h:529
static PhLaw< T > compose_serial(const PhLaw< T > &a, const PhLaw< T > &b)
Serial composition: the second law starts when the first absorbs.
Definition workflow.h:496
static Precedence< T > OrJoin(const std::vector< std::string > &pres, const std::string &post)
Definition workflow.h:772
const std::vector< Precedence< T > > & precedences() const
Definition workflow.h:268
Distrib< T > to_ph()
The composed law of the workflow.
Definition workflow.h:345
Workflow(const std::string &name)
Definition workflow.h:241
static Precedence< T > Serial(const std::string &pre, const std::string &post)
Definition workflow.h:722
static PhLaw< T > compose_mixture(const std::vector< PhLaw< T > > &laws, const std::vector< T > &probs)
Probabilistic mixture: a block-diagonal generator whose initial vector picks branch i with probabilit...
Definition workflow.h:581
WorkflowActivity< T > & activity(const std::string &name)
Definition workflow.h:275
const WorkflowActivity< T > & activity_at(std::size_t i) const
Activity by index, the form a series-parallel LEAF node names it in.
Definition workflow.h:282
The moment fitters the reference distributions carry as STATIC FACTORIES: Erlang.fitMeanAndOrder,...
Rate-scaled copy of a distribution, preserving its shape.
What refreshProcessRepresentations and refreshLST compute FROM a distribution: the (D0,...
The exception types the port throws.
Enumerations and the minimal distribution descriptor shared by the model layer of the C++ port.
Sample the inter-arrival times of a MAP, a RAP or a matrix exponential.
Dense matrix and non-owning view.
const Element * child(const Element *e, const std::string &tag)
The first direct child with the given tag, or null.
Definition jmva_reader.h:89
PrecedenceType
Activity precedence kinds, with the values of MATLAB ActivityPrecedenceType.
Definition lang_types.h:472
Distrib< T > aph_fit_mean_scv(const T &mean, const T &scv)
APH.fitMeanAndSCV(MEAN, SCV), through mam::aph_fit_mean_scv.
ProcessType
Distribution kinds, with the values of MATLAB ProcessType.
Definition lang_types.h:485
Distrib< T > dist_scale_rate(const Distrib< T > &d, const T &factor)
The law of X / factor, in the same family as d.
std::vector< T > map_sample(const Map< T > &m, std::size_t n, pfqn::McRng &rng, const std::vector< T > &pie0=std::vector< T >(), SampleTrace *trace=0)
Sample the inter-arrival times of a MAP, a RAP or a matrix exponential.
Definition map_sample.h:98
std::mt19937_64 McRng
The generator type every Monte Carlo entry point in this tree accepts.
SPNodeType
A node of the series-parallel tree.
Definition workflow.h:203
Conservation laws of a layered queueing network, enumerated from its structure.
Definition aoi_dist2ph.h:52
Matrix< T > D0
The (D0,D1) pair when the type carries one directly.
Definition lang_types.h:853
std::vector< T > params
Constructor arguments, in MATLAB getParam order.
Definition lang_types.h:826
The MATLAB GlobalConstants, as reported by lineStart at its defaults.
Definition lang_types.h:759
A MAP as the pair of matrices (D0, D1).
Definition map_moment.h:53
Matrix< T > D1
Definition map_moment.h:55
Matrix< T > D0
Definition map_moment.h:54
static Distrib phase_type(const std::vector< T > &alpha, const Matrix< T > &A, bool acyclic)
PH / APH given by (alpha, A): D0 = A and D1 = (-A e) alpha.
static Distrib exp_mean(const T &m)
Definition lang_types.h:930
static constexpr double Immediate
Rate of an Immediate distribution; its mean is 1/Immediate = 1e-8.
Definition lang_types.h:766
static constexpr double FineTol
Definition lang_types.h:760
static constexpr double ArcTol
Below this an off-diagonal entry is NO ARC of the phase / state graph.
Definition lang_types.h:764
A phase-type law as the composition rules pass it around.
Definition workflow.h:90
std::vector< T > alpha
Definition workflow.h:91
One precedence of the activity graph.
Definition workflow.h:79
std::vector< T > pre_params
Definition workflow.h:84
PrecedenceType pre_type
Definition workflow.h:82
PrecedenceType post_type
Definition workflow.h:83
std::vector< T > post_params
Definition workflow.h:85
std::vector< std::string > post_acts
Definition workflow.h:81
std::vector< std::string > pre_acts
Definition workflow.h:80
T count
Loop count for a LOOP node.
Definition workflow.h:215
std::vector< T > probs
Branch probabilities for an OR node.
Definition workflow.h:213
std::vector< std::size_t > kids
Definition workflow.h:210
std::size_t act
Activity index for a LEAF, npos otherwise.
Definition workflow.h:209
The flat series-parallel tree.
Definition workflow.h:228
std::vector< SPNode< T > > nodes
Definition workflow.h:229
std::vector< T > execs
Definition workflow.h:233
std::vector< std::size_t > leaf_of
Node index of each activity's leaf, npos when the activity has none.
Definition workflow.h:232