lang.workflow

class WorkflowActivity

Bases: Element

A computational activity in a Workflow.

WorkflowActivity represents a stage of computation in a standalone workflow model. Unlike Activity in LayeredNetwork, it carries no call list: an external call is represented as an activity whose host demand is the law of the call response time, so that a synchronous call and a local computation compose in the same way. An asynchronous call blocks the caller for no time and is simply left out of the workflow.

Each activity has a service time distribution that can be any phase-type compatible distribution (Exp, Erlang, APH, HyperExp, etc.).

Constructor Summary

WorkflowActivity(workflow, name, hostDemand)

WORKFLOWACTIVITY Create a workflow activity

OBJ = WORKFLOWACTIVITY(WORKFLOW, NAME, HOSTDEMAND)

Parameters:

workflow - Parent Workflow object
name - Activity name (string)
hostDemand - Service time distribution or numeric mean

If hostDemand is numeric, it is treated as the mean of an exponential distribution.

Property Summary

hostDemand: Distribution object (Exp, APH, Erlang, etc.)

hostDemandMean: double - mean service time

hostDemandSCV: double - squared coefficient of variation

index: Index in the workflow’s activity list

metadata: Optional struct for external metadata (e.g., WfCommons)

workflow: Reference to parent Workflow

Method Summary

getNumberOfPhases()

GETNUMBEROFPHASES Get the number of phases

N = GETNUMBEROFPHASES(OBJ)

Returns the number of phases in the PH representation.

getPHRepresentation()

GETPHREPRESENTATION Get the phase-type representation

[ALPHA, T] = GETPHREPRESENTATION(OBJ)

Returns:: alpha - Initial probability vector (1 x n) T - Subgenerator matrix (n x n)

For Markovian distributions, extracts the PH representation. For other distributions, fits an APH from the first two moments.

setHostDemand(hostDemand)

SETHOSTDEMAND Set the service time distribution

OBJ = SETHOSTDEMAND(OBJ, HOSTDEMAND)

Parameters:: hostDemand - Distribution object or numeric mean

setHostDemandMean(meanValue)

SETHOSTDEMANDMEAN Change the mean, preserving the shape

OBJ = SETHOSTDEMANDMEAN(OBJ, MEANVALUE)

Scales the current law in time rather than refitting it, so the SCV, the skewness and the order are preserved and the cached series-parallel tree keeps its shape. Falls back to a plain exponential when the current law has no usable mean.

class Workflow

Bases: Model

Workflow - A computational workflow that can be converted to a PH distribution.

Workflow allows declaring computational workflows using the same activity graph syntax as LayeredNetwork. The workflow can then be converted to an equivalent phase-type (PH) distribution via explicit CTMC construction.

Supported precedence patterns:

Serial: Sequential execution
AndFork/AndJoin: Parallel execution with synchronization
OrFork/OrJoin: Probabilistic branching
Loop: Geometric repetition, with the same semantics as the LayeredNetwork POST_LOOP precedence: the loop body is executed a geometric number of times of mean COUNT when COUNT>=1, and is executed at most once, with probability COUNT, when COUNT<1

A workflow whose precedence graph is series-parallel is reduced exactly, by recursive composition of the series-parallel tree; the reduction handles arbitrary nesting (a fork inside a loop, a branch of a fork that is itself a fork-join). Graphs that are not series-parallel fall back to the block-based composition, which is a heuristic.

Activities model both local computation and, when the workflow stands for an LQN entry, a synchronous call: the call is an activity whose host demand is the fitted call-response law. Asynchronous calls contribute no time and are simply omitted by the builder.

Example

wf = Workflow(‘myWorkflow’); A = wf.addActivity(‘A’, Exp.fitMean(1.0)); B = wf.addActivity(‘B’, Exp.fitMean(2.0)); wf.addPrecedence(Workflow.Serial(A, B)); ph = wf.toPH();

Constructor Summary

Workflow(name)

WORKFLOW Create a new Workflow instance

SELF = WORKFLOW(NAME)

Parameters:: name - Name for the workflow

Property Summary

activities: Cell array of WorkflowActivity objects

activityMap: Map from activity name to index

precedences: Array of ActivityPrecedence objects

Method Summary

static AndFork(preAct, postActs)

ANDFORK Create AND-fork precedence

AP = WORKFLOW.ANDFORK(PREACT, {POSTACT1, POSTACT2, …})

static AndJoin(preActs, postAct, quorum)

ANDJOIN Create AND-join precedence

AP = WORKFLOW.ANDJOIN({PREACT1, PREACT2, …}, POSTACT) AP = WORKFLOW.ANDJOIN({PREACT1, PREACT2, …}, POSTACT, QUORUM)

static Loop(preAct, postActs, counts)

LOOP Create loop precedence

AP = WORKFLOW.LOOP(PREACT, {LOOPACT, ENDACT}, COUNT) AP = WORKFLOW.LOOP(PREACT, LOOPACT, ENDACT, COUNT)

static OrFork(preAct, postActs, probs)

ORFORK Create OR-fork (probabilistic) precedence

AP = WORKFLOW.ORFORK(PREACT, {POSTACT1, POSTACT2, …}, [P1, P2, …])

static OrJoin(preActs, postAct)

ORJOIN Create OR-join precedence

AP = WORKFLOW.ORJOIN({PREACT1, PREACT2, …}, POSTACT)

static Serial(varargin)

SERIAL Create serial precedence

AP = WORKFLOW.SERIAL(A1, A2, …)

static Xor(preAct, postActs, probs)

XOR Create XOR precedence (alias for OrFork)

AP = WORKFLOW.XOR(PREACT, {POSTACT1, POSTACT2, …}, [P1, P2, …])

addActivity(name, hostDemand)

ADDACTIVITY Add an activity to the workflow

ACT = ADDACTIVITY(SELF, NAME, HOSTDEMAND)

Parameters:

name - Activity name (string)
hostDemand - Service time distribution or numeric mean

Returns:

act - WorkflowActivity object

addPrecedence(prec)

ADDPRECEDENCE Add precedence constraints to the workflow

SELF = ADDPRECEDENCE(SELF, PREC)

Parameters:: prec - ActivityPrecedence object or cell array (from Serial)

buildCTMC()

BUILDCTMC Build the CTMC representation of the workflow

[ALPHA, T] = BUILDCTMC(SELF)

Returns:: alpha - Initial probability vector T - Subgenerator matrix

static composeLoopGeometric(alpha, T, count)

COMPOSELOOPGEOMETRIC Geometric repetition of a PH law

[ALPHA_OUT, T_OUT] = COMPOSELOOPGEOMETRIC(ALPHA, T, COUNT)

Implements the LayeredNetwork POST_LOOP semantics, in which the number of executions of the loop body is geometric of mean COUNT. For COUNT>=1 the body runs at least once and repeats on absorption with probability P = 1-1/COUNT, so

T_OUT = T + P/D * (-T*e)*ALPHA, ALPHA_OUT = ALPHA/D

with D = 1 - P*(1-ALPHA*e) the correction for an atom at zero in ALPHA. The order of the law is that of the body, unlike the COUNT-fold convolution COMPOSEREPEAT, and the mean is COUNT times the mean of the body in both cases.

For COUNT<1 the body is executed at most once, with probability COUNT, which is how a fractional loop count is read when the LQN activity graph is built; the skipped branch is an immediate phase, the representation of a zero-time activity used throughout this class.

static composeMixture(alphas, Ts, probs)

COMPOSEMIXTURE Probabilistic mixture of several PH laws

[ALPHA_OUT, T_OUT] = COMPOSEMIXTURE(ALPHAS, TS, PROBS)

Block-diagonal generator whose initial vector picks branch I with probability PROBS(I). This is aph_simplify pattern 3 generalised to any number of branches.

static composeParallel(alpha1, T1, alpha2, T2)

COMPOSEPARALLEL Parallel fork-join composition

Models two PH distributions running in parallel with synchronization at completion (AND-join). The resulting PH represents the time until BOTH activities complete (i.e., the maximum).

State space:

States (i,j) where both are active: i=1..n1, j=1..n2
States where only activity 1 is active (2 completed): i=1..n1
States where only activity 2 is active (1 completed): j=1..n2

Absorption occurs only when both have completed.

static composeRepeat(alpha, T, count)

COMPOSEREPEAT Repeat a PH distribution count times (convolution)

Equivalent to serial composition of the same PH count times, so COUNT is a deterministic number of executions. The POST_LOOP precedence of an activity graph is instead geometric; use COMPOSELOOPGEOMETRIC for it.

static composeSerial(alpha1, T1, alpha2, T2)

COMPOSESERIAL Serial composition of two PH distributions

The second PH starts when the first one absorbs.

T_serial = [T1, -T1*e*alpha2]: [0, T2 ]

static fromWfCommons(jsonFile, options)

FROMWFCOMMONS Load a workflow from a WfCommons JSON file.

WF = WORKFLOW.FROMWFCOMMONS(JSONFILE) WF = WORKFLOW.FROMWFCOMMONS(JSONFILE, OPTIONS)

Loads a workflow trace from the WfCommons format (https://github.com/wfcommons/workflow-schema) into a LINE Workflow object for queueing analysis.

Parameters:

jsonFile - Path to WfCommons JSON file
options - Optional struct with – .distributionType - ‘exp’ (default), ‘det’, ‘aph’, ‘hyperexp’ .defaultSCV - Default SCV for APH/HyperExp (default: 1.0) .defaultRuntime - Default runtime when missing (default: 1.0) .useExecutionData - Use execution data if available (default: true) .storeMetadata - Store WfCommons metadata (default: true)

Returns:

wf - Workflow object

Example

wf = Workflow.fromWfCommons(‘montage-workflow.json’); ph = wf.toPH(); fprintf(‘Mean execution time: %.2fn’, ph.getMean());

getActivity(name)

GETACTIVITY Get activity by name

ACT = GETACTIVITY(SELF, NAME)

Parameters:: name - Activity name (string)
Returns:: act - WorkflowActivity object

getActivityIndex(name)

GETACTIVITYINDEX Get activity index by name

IDX = GETACTIVITYINDEX(SELF, NAME)

static invalidateBranch(tree, k): INVALIDATEBRANCH Invalidate a node and all its ancestors

static isAcyclicGenerator(T)

ISACYCLICGENERATOR True if the phase graph of T has no cycle

TF = ISACYCLICGENERATOR(T)

A geometric loop over a body of two or more phases closes a cycle, so the composed law is a PH and not an APH.

refreshPH()

REFRESHPH Recompose the workflow law after a demand change

PH = REFRESHPH(SELF)

Recomposes only the series-parallel nodes on the path from a dirty leaf to the root; nodes whose subtree is unchanged keep their cached (alpha,T). The topology is not rebuilt. When the workflow is not series-parallel this degrades to a full recomposition through the block path.

setActivityDemand(name, hostDemand)

SETACTIVITYDEMAND Change the host demand of one activity

SELF = SETACTIVITYDEMAND(SELF, NAME, HOSTDEMAND)

Marks only that leaf dirty, so that the next TOPH/REFRESHPH recomposes the path from the leaf to the root and reuses every other cached block. This is the entry point used by iterative solvers that update call-response laws at each iteration.

setActivityDemandMean(name, meanValue)

SETACTIVITYDEMANDMEAN Change only the mean of one activity

SELF = SETACTIVITYDEMANDMEAN(SELF, NAME, MEANVALUE)

Rescales the activity law in time, so its SCV and its whole shape are preserved. The leaf keeps its order and its initial probability vector, and the cached series-parallel tree keeps its shape.

static spExecutionCounts(tree)

SPEXECUTIONCOUNTS Expected executions of each node per run

EXECS = SPEXECUTIONCOUNTS(TREE)

Serial and parallel children inherit the count of their parent, an OR branch is weighted by its probability, and a loop body is weighted by the loop count. This is the weight by which a layer result is split back over entries, activities and calls.

toPH()

TOPH Convert workflow to phase-type distribution

PH = TOPH(SELF)

Returns:

ph - APH distribution representing the workflow execution: time, or PH when the composed generator is cyclic, which a geometric loop over a multi-phase body makes it

validate()

VALIDATE Validate workflow structure

[ISVALID, MSG] = VALIDATE(SELF)

Returns:: isValid - true if workflow is valid msg - Error message if invalid