"""
Native Python implementation of LQNS solver.
This module provides a native Python wrapper for the lqns and lqsim command-line
tools that analyze layered queueing networks, and, on a flat Network, for the
qnsolver tool of the same distribution (see qnsolver.py).
"""
import numpy as np
import subprocess
import tempfile
import shutil
import os
import platform
import xml.etree.ElementTree as ET
import json
import urllib.request
import urllib.error
from dataclasses import dataclass, field
from ....constants import default_verbose
from typing import Optional, Dict, Any, List, Tuple
import pandas as pd
from ....api.io.logging import line_debug, line_ack
from ...base import NetworkSolver
from ....layered import LayeredNetworkElement
from .qnsolver import (QnsolverNetworkMixin, LQNSNetworkResult, QNS_METHODS,
is_qns_method, qns_multiserver, qns_feature_set,
qns_immfeed_refusal, qns_multiserver_refusal)
from line_solver.api.io.errors import LineExternalError, LineUnsupportedError
# The kinds lqn.type stores, under the names parseXMLResults.m uses for them.
LQN_HOST = int(LayeredNetworkElement.PROCESSOR)
LQN_TASK = int(LayeredNetworkElement.TASK)
LQN_ENTRY = int(LayeredNetworkElement.ENTRY)
LQN_ACTIVITY = int(LayeredNetworkElement.ACTIVITY)
LQN_CALL = int(LayeredNetworkElement.CALL)
@dataclass
class LQNSOptions:
"""Options for the LQNS solver."""
# LayeredNetwork: default, lqns, srvn, exactmva, srvn.exactmva, sim, lqsim;
# Network: default, qns, qns.conway, qns.rolia, qns.zhou, qns.suri, qns.reiser, qns.schmidt
method: str = 'default'
multiserver: str = 'rolia' # rolia, conway, etc.
samples: int = 10000 # For simulation methods
verbose: bool = field(default_factory=default_verbose)
keep: bool = True # Keep temporary files
seed: int = 23000
remote: bool = False # Enable remote execution via REST API
remote_url: str = 'http://localhost:8080' # URL of lqns-rest server
# see _kb/06-solver-catalog.md (Wrappers) for the lang toggle semantics
lang: str = field(default_factory=lambda: os.environ.get('LINE_SOLVER_LANG', 'python'))
@dataclass
class LQNSResult:
"""Result from LQNS solver."""
PN: np.ndarray = field(default_factory=lambda: np.array([])) # Processor utilization
SN: np.ndarray = field(default_factory=lambda: np.array([])) # Phase 1 service times
TN: np.ndarray = field(default_factory=lambda: np.array([])) # Throughputs
UN: np.ndarray = field(default_factory=lambda: np.array([])) # Utilizations
RN: np.ndarray = field(default_factory=lambda: np.array([])) # Processor waiting
QN: np.ndarray = field(default_factory=lambda: np.array([])) # Queue lengths
AN: np.ndarray = field(default_factory=lambda: np.array([])) # Arrival rates
WN: np.ndarray = field(default_factory=lambda: np.array([])) # Residence times
runtime: float = 0.0
method: str = 'default'
iterations: int = 0
def _snap_to_tenth(values):
"""Snap values that are within CoarseTol (relative) of an exact tenth.
lqns reports quantities that are analytically an exact multiple of 0.1
(a deterministic 0.4 service time, say) with a few digits of iteration
noise, as 0.39990. The MATLAB LQNS wrapper removes that noise in
getAvgTable before tabulating, and the Python row must apply the same
step or the two rows disagree on values neither solver actually computed
differently.
References:
MATLAB: matlab/src/solvers/wrappers/LQNS/@SolverLQNS/SolverLQNS.m
"""
from ....constants import GlobalConstants
arr = np.asarray(values, dtype=float).copy()
if arr.size == 0:
return arr
scaled = arr * 10.0
snapped = np.round(scaled)
with np.errstate(invalid='ignore'):
# The tolerance is relative to the scaled value, exactly as in MATLAB;
# NaN and non-positive entries fail the comparison and pass through.
mask = np.abs(scaled - snapped) < GlobalConstants.CoarseTol * scaled
arr[mask] = snapped[mask] / 10.0
return arr
[docs]
class SolverLQNS(QnsolverNetworkMixin, NetworkSolver):
"""
Native Python LQNS solver using external lqns/lqsim/qnsolver tools.
This solver wraps the LQNS (Layered Queueing Network Solver) command-line tools
to analyze layered queueing networks, and the qnsolver tool of the same
distribution to analyze flat queueing networks, the way SolverJMT reaches
JMVA.
Supported methods on a LayeredNetwork:
- default/lqns: Standard LQNS analytical solver
- srvn: SRVN layering
- exactmva: Exact MVA algorithm
- srvn.exactmva: SRVN with exact MVA
- sim/lqsim: Simulation-based solver
Supported methods on a Network (see QnsolverNetworkMixin):
- default, qns: qnsolver if product-form or open, otherwise QN2LQN + lqns,
with the rolia multiserver approximation
- qns.conway, qns.rolia, qns.zhou, qns.suri, qns.reiser, qns.schmidt: the
same routing under that multiserver approximation
Requirements:
The 'lqns' and 'lqsim' commands must be available in the system PATH.
Install from: http://www.sce.carleton.ca/rads/lqns/
LINE ships no LQNS binary and runs none from a container image: LQNS is
distributed under an evaluation agreement that forbids redistribution.
Alternatively, point LINE at a host that already runs LQNS:
options.config.remote = True; options.config.remote_url = 'http://localhost:8080'
Note:
Model serialization to LQNX format
method. The native aspect is in the CLI execution and result parsing.
Example:
>>> lqn = LayeredNetwork('model')
>>> # ... build model ...
>>> solver = SolverLQNS(lqn, LQNSOptions(method='lqns'))
>>> result = solver.runAnalyzer()
"""
[docs]
def __init__(self, model, options: Optional[LQNSOptions] = None, **kwargs):
"""
Initialize the LQNS solver.
Args:
model: LayeredNetwork model, or a Network (or NetworkStruct) for
the qns methods
options: Optional LQNSOptions configuration, or a method name
**kwargs: Additional parameters (keep, etc.) for compatibility
"""
if isinstance(options, str):
kwargs['method'] = options
options = None
self._layered = _is_layered(model)
if not self._layered:
self._qns_init(model, options, kwargs)
return
self.model = model
if options is not None:
self.options = options
else:
# Route recognized kwargs into options; unknown ones are ignored.
import dataclasses
valid = {f.name for f in dataclasses.fields(LQNSOptions)}
self.options = LQNSOptions(**{k: v for k, v in kwargs.items() if k in valid})
self._result: Optional[LQNSResult] = None
self._lqn = None # Will be set during analysis
self._keep = kwargs.get('keep', True) # For compatibility
[docs]
@staticmethod
def isAvailable() -> bool:
"""
Check if lqns can be run: a native binary is on the PATH.
LINE never runs LQNS from a container image, because its licence forbids
redistribution. To exercise a containerised build, put a shim on the PATH
with run-tests.sh --lqns-docker.
Returns:
True if a native lqns binary is available, False otherwise
"""
return SolverLQNS._has_native_lqns()
@staticmethod
def _has_native_lqns() -> bool:
"""Check if a native lqns binary is available in the system PATH."""
try:
if platform.system() == 'Windows':
process = subprocess.Popen(
['lqns', '--help'],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE
)
else:
process = subprocess.Popen(
['lqns', '--help'],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE
)
try:
stdout, stderr = process.communicate(timeout=5)
output = (stdout + stderr).decode().lower()
return 'lqns' in output or 'usage' in output
except subprocess.TimeoutExpired:
process.kill()
return False
except (FileNotFoundError, OSError):
return False
# The Network path's result goes through the NetworkSolver near-zero mask,
# as it did before the qns methods moved here; the layered result is
# element-indexed and is stored as lqns returned it.
@property
def _result(self):
return self.__dict__.get('_result_store')
@_result.setter
def _result(self, value):
if self.__dict__.get('_layered', True):
self.__dict__['_result_store'] = value
else:
self.__dict__['_result_store'] = self._maskAvgResult(value)
[docs]
def listValidMethods(self=None, model=None) -> List[str]:
"""List valid methods: the layered ones on a LayeredNetwork, the qns
ones on a flat Network. Callable on the class (layered methods) or on
an instance; a model argument asks about that model instead."""
if self is not None and not isinstance(self, SolverLQNS) and model is None:
model = self # called on the class with the model as argument
elif model is None and isinstance(self, SolverLQNS):
model = self.model if self._layered else getattr(self, 'sn', None)
if model is not None and not _is_layered(model):
return list(QNS_METHODS)
# the names and their order are this solver's row of the method registry
# (doc/line-methods.json, line_solver.solvers.method_registry)
from line_solver.solvers import method_registry
return [m for m in method_registry.methods('LQNS') if not (m == 'qns' or m.startswith('qns.'))]
def getName(self) -> str:
"""Get solver name."""
return 'LQNS'
get_name = getName
[docs]
def isStochasticMethod(self, method):
"""The lqsim simulator is stochastic; the analytical lqns/srvn methods
are deterministic.
"""
return str(method).lower() in ('sim', 'lqsim') if method else False
is_stochastic_method = isStochasticMethod
def getStruct(self):
"""Get the LayeredNetworkStruct, or the NetworkStruct on a flat Network."""
if not self._layered:
return self.sn
if hasattr(self.model, 'getStruct'):
return self.model.getStruct()
return None
[docs]
def runAnalyzer(self) -> LQNSResult:
"""
Run the LQNS analysis.
Returns:
LQNSResult containing performance metrics
Raises:
RuntimeError: If lqns is not available or fails
"""
if not self._layered:
return self._qns_run_analyzer()
line_ack('LQNS', self.options.verbose)
# LINE's own sentence before the .lqnx is written, not the binary's parse error
ok, reason = lqns_method_refusal(self.model, self.options.method)
if not ok:
raise LineUnsupportedError(reason, error_id='LINE:unsupported:Structure')
multiserver = self.options.multiserver or 'default'
# Solver console: the layered path never reaches the shared run hook
# in solvers/base.py (getAvg below does not go through
# _ensureAvgResults), so it opens its own run here, as the MATLAB
# wrapper does.
from line_solver.api.io import console as _console
with _console.run_scope(self, self.options):
return self._runAnalyzerBody(multiserver)
def _runAnalyzerBody(self, multiserver: str) -> LQNSResult:
"""Write the model, run the binary, parse what came back."""
from line_solver.api.io import console as _console
line_debug("LQNS: starting (method=%s, multiserver=%s)", self.options.method, multiserver, options=self.options)
import time
start_time = time.time()
# LINE_WORKSPACE_ROOT relocates the staging dir; run-tests.sh sets it when
# a solver is wrapped in a container, so the path is bind-mountable.
workspace_root = os.environ.get('LINE_WORKSPACE_ROOT', '').strip()
if workspace_root:
base = os.path.join(workspace_root, 'line_workspace', 'lqns')
os.makedirs(base, exist_ok=True)
temp_dir = tempfile.mkdtemp(prefix='lqns_', dir=base)
else:
temp_dir = tempfile.mkdtemp(prefix='lqns_')
try:
# Write model to LQNX format
lqnx_file = os.path.join(temp_dir, 'model.lqnx')
# Check if model is native LayeredNetwork
model_type = type(self.model).__name__
_console.step('writing the LQN model to %s', lqnx_file)
# standard schema: lqns/lqsim reject the LINE dialect <call-group>
# and read the ungrouped twin from the member calls instead
if model_type == 'LayeredNetwork' or hasattr(self.model, 'processors'):
# Use native writeXML
self.model.writeXML(lqnx_file, False, True)
elif hasattr(self.model, 'writeXML'):
# Use LayeredNetwork writeXML
self.model.writeXML(lqnx_file, False, True)
else:
raise LineUnsupportedError(f"Model type {model_type} does not support writeXML", error_id='LINE:unsupported:Format')
# Check for remote execution
if self.options.remote:
line_debug("LQNS: using remote execution at %s", self.options.remote_url, options=self.options)
if self.options.verbose and not _console.is_active():
print(f"Using remote LQNS at: {self.options.remote_url}")
_console.step('running the lqns service at %s', self.options.remote_url)
self._run_remote_lqns(lqnx_file)
else:
# Build and execute command locally
cmd = self._build_command(lqnx_file)
line_debug("LQNS: using local execution, command: %s", cmd, options=self.options)
# the console already reports the command through the routed
# line_debug above, so printing it again would duplicate it
if self.options.verbose and not _console.is_active():
print(f"LQNS command: {cmd}")
_console.step('running the lqns binary as a subprocess')
# Execute
if platform.system() == 'Windows':
process = subprocess.Popen(
cmd.split(),
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT
)
else:
process = subprocess.Popen(
cmd.split(),
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT
)
stdout, _ = process.communicate()
output = stdout.decode()
# a model lqns rejects may still exit 0 and leave no .lqxo; report lqns's own message either way
if process.returncode != 0 or not os.path.exists(lqnx_file.replace('.lqnx', '.lqxo')):
raise LineExternalError(
f"The LQNS binary (exit status {process.returncode}) produced no results for this model. "
f"Its output was:\n{output.strip()}"
, error_id='LINE:external:LQNS')
# Parse results from .lqxo file
lqxo_file = lqnx_file.replace('.lqnx', '.lqxo')
_console.step('parsing the lqns XML results')
self._parse_xml_results(lqxo_file)
finally:
# Clean up
if not self.options.keep:
shutil.rmtree(temp_dir, ignore_errors=True)
runtime = time.time() - start_time
if self._result:
self._result.runtime = runtime
from .qnsolver import _print_lqns_banner
_print_lqns_banner(self.options, self.options.method, runtime)
return self._result
def _build_command(self, filename: str) -> str:
"""Build the LQNS/LQSIM command line."""
method = self.options.method.lower()
# Verbose flags
verbose_flag = '' if self.options.verbose else '-a -w'
# Multiserver policy
praqma_flag = ''
if method not in ('lqsim', 'sim'):
pol = self.options.multiserver or 'rolia'
praqma_flag = f'-Pmultiserver={pol}'
common = f'{verbose_flag} {praqma_flag} -Pstop-on-message-loss=false -x {filename}'
if method == 'srvn':
cmd = f'lqns {common} -Playering=srvn'
elif method == 'exactmva':
cmd = f'lqns {common} -Pmva=exact-mva'
elif method == 'srvn.exactmva':
cmd = f'lqns {common} -Playering=srvn -Pmva=exact-mva'
elif method in ('sim', 'lqsim'):
cmd = f'lqsim {common} -A {self.options.samples}'
elif method == 'lqns.default':
cmd = f'lqns {verbose_flag} {praqma_flag} -x {filename}'
else: # default or lqns
cmd = f'lqns {common}'
# Clean up multiple spaces
cmd = ' '.join(cmd.split())
return cmd
def _run_remote_lqns(self, lqnx_file: str) -> None:
"""
Execute LQNS via remote REST API.
Args:
lqnx_file: Path to the LQNX model file
"""
# Read LQNX model file
with open(lqnx_file, 'r', encoding='utf-8') as f:
model_content = f.read()
# Determine endpoint based on method
base_url = self.options.remote_url.rstrip('/')
method = self.options.method.lower()
if method in ('sim', 'lqsim'):
endpoint = f"{base_url}/api/v1/solve/lqsim"
else:
endpoint = f"{base_url}/api/v1/solve/lqns"
# Build request
request_data = {
'model': {
'content': model_content,
'base64': False
},
'options': {
'include_raw_output': True
}
}
# Add pragmas based on method
if method not in ('sim', 'lqsim'):
multiserver = self.options.multiserver or 'rolia'
if multiserver == 'default':
multiserver = 'rolia'
request_data['options']['pragmas'] = {
'multiserver': multiserver,
'stop_on_message_loss': False
}
# Method-specific pragmas
if method == 'srvn':
request_data['options']['pragmas']['layering'] = 'srvn'
elif method == 'exactmva':
request_data['options']['pragmas']['mva'] = 'exact'
elif method == 'srvn.exactmva':
request_data['options']['pragmas']['layering'] = 'srvn'
request_data['options']['pragmas']['mva'] = 'exact'
else:
# LQSIM options
request_data['options']['blocks'] = 30
if self.options.samples > 0:
request_data['options']['run_time'] = self.options.samples
# Make HTTP request
json_data = json.dumps(request_data).encode('utf-8')
req = urllib.request.Request(
endpoint,
data=json_data,
headers={
'Content-Type': 'application/json',
'Accept': 'application/json'
},
method='POST'
)
try:
with urllib.request.urlopen(req, timeout=300) as response:
response_data = json.loads(response.read().decode('utf-8'))
except urllib.error.HTTPError as e:
error_body = e.read().decode('utf-8') if e.fp else ''
raise LineExternalError(f"Remote LQNS returned error: {e.code} - {error_body}", error_id='LINE:external:LQNS')
except urllib.error.URLError as e:
raise LineExternalError(f"Remote LQNS connection failed: {e.reason}", error_id='LINE:external:LQNS')
# Check response status
status = response_data.get('status', '')
if status in ('error', 'failed'):
error_msg = response_data.get('error', 'Unknown error')
raise LineExternalError(f"Remote solver returned error: {error_msg}", error_id='LINE:external:LQNS')
# Extract and write LQXO content
raw_output = response_data.get('raw_output', {})
lqxo_content = raw_output.get('lqxo', '') if isinstance(raw_output, dict) else ''
if lqxo_content:
lqxo_file = lqnx_file.replace('.lqnx', '.lqxo')
with open(lqxo_file, 'w', encoding='utf-8') as f:
f.write(lqxo_content)
else:
raise LineExternalError("Remote solver did not return LQXO output", error_id='LINE:external:LQNS')
def _parse_xml_results(self, filename: str) -> None:
"""
Parse LQNS XML output file (.lqxo).
Extracts utilization, service times, throughputs, etc. from the output.
"""
# Get LayeredNetworkStruct for node info
self._lqn = self.getStruct()
if self._lqn is None:
raise RuntimeError("Cannot get LayeredNetworkStruct from model")
# Wait for file to exist
import time
max_wait = 10 # seconds
waited = 0
while not os.path.exists(filename) and waited < max_wait:
time.sleep(0.01)
waited += 0.01
if not os.path.exists(filename):
raise LineExternalError(f"LQNS output file not found: {filename}", error_id='LINE:external:LQNS')
# Parse XML
tree = ET.parse(filename)
root = tree.getroot()
# Try to get names from model first
names = self._get_node_names()
types = self._get_node_types()
# If names is empty, build names mapping from the XML itself
if not names:
names, types = self._build_names_from_xml(root)
# Get structure info
if self._lqn is not None:
# Native Python struct has nidx and ncalls as attributes
num_nodes = int(self._lqn.nidx)
num_calls = int(self._lqn.ncalls) if hasattr(self._lqn, 'ncalls') else 0
else:
# Use the number of names we found
num_nodes = max(names.keys()) if names else 100
num_calls = 100
# Initialize result matrices
utilization = np.full(num_nodes, np.nan)
phase1_util = np.full(num_nodes, np.nan)
phase2_util = np.full(num_nodes, np.nan)
phase1_st = np.full(num_nodes, np.nan)
phase2_st = np.full(num_nodes, np.nan)
throughput = np.full(num_nodes, np.nan)
proc_waiting = np.full(num_nodes, np.nan)
proc_util = np.full(num_nodes, np.nan)
edges_waiting = np.full(num_calls, np.nan)
# host residence of ONE execution of an activity, input to ResidT below
resid_visit = np.full(num_nodes, np.nan)
iterations = 0
# Store names for later use in getAvgTable
self._parsed_names = names
# Parse solver-params for iterations
for solver_params in root.findall('.//solver-params'):
for result_general in solver_params.findall('result-general'):
iter_str = result_general.get('iterations', '0')
try:
iterations = int(iter_str)
except ValueError:
pass
# Parse processors
for proc_elem in root.findall('.//processor'):
proc_name = proc_elem.get('name')
proc_pos = self._find_lqn_elem(names, types, proc_name, LQN_HOST)
# Get processor utilization
for proc_result in proc_elem.findall('result-processor'):
util_str = proc_result.get('utilization', '')
if util_str and proc_pos >= 0 and proc_pos < len(proc_util):
proc_util[proc_pos] = float(util_str)
# Parse tasks
for task_elem in proc_elem.findall('task'):
task_name = task_elem.get('name')
task_pos = self._find_lqn_elem(names, types, task_name, LQN_TASK)
for task_result in task_elem.findall('result-task'):
if task_pos >= 0 and task_pos < len(utilization):
utilization[task_pos] = float(task_result.get('utilization', 0))
p1u = task_result.get('phase1-utilization', '')
if p1u:
phase1_util[task_pos] = float(p1u)
p2u = task_result.get('phase2-utilization', '')
if p2u:
phase2_util[task_pos] = float(p2u)
throughput[task_pos] = float(task_result.get('throughput', 0))
proc_util[task_pos] = float(task_result.get('proc-utilization', 0))
# Parse entries
for entry_elem in task_elem.findall('.//entry'):
entry_name = entry_elem.get('name')
entry_pos = self._find_lqn_elem(names, types, entry_name, LQN_ENTRY)
for entry_result in entry_elem.findall('result-entry'):
if entry_pos >= 0 and entry_pos < len(utilization):
utilization[entry_pos] = float(entry_result.get('utilization', 0))
p1u = entry_result.get('phase1-utilization', '')
if p1u:
phase1_util[entry_pos] = float(p1u)
p2u = entry_result.get('phase2-utilization', '')
if p2u:
phase2_util[entry_pos] = float(p2u)
p1st = entry_result.get('phase1-service-time', '')
if p1st:
phase1_st[entry_pos] = float(p1st)
p2st = entry_result.get('phase2-service-time', '')
if p2st:
phase2_st[entry_pos] = float(p2st)
throughput[entry_pos] = float(entry_result.get('throughput', 0))
proc_util[entry_pos] = float(entry_result.get('proc-utilization', 0))
# Parse entry-phase-activities (PH1PH2 format): fill the
# phase activity rows, otherwise left NaN (JLINE parity)
entry_tput = throughput[entry_pos] if 0 <= entry_pos < len(throughput) else np.nan
for epa in entry_elem.findall('entry-phase-activities'):
for activity in epa.findall('activity'):
act_name = activity.get('name')
act_pos = self._find_lqn_elem(names, types, act_name, LQN_ACTIVITY)
if act_pos < 0 or act_pos >= len(utilization):
continue
for act_result in activity.findall('result-activity'):
u = act_result.get('utilization', '')
if u:
utilization[act_pos] = float(u)
st = act_result.get('service-time', '')
if st:
phase1_st[act_pos] = float(st)
pw = act_result.get('proc-waiting', '')
if pw:
proc_waiting[act_pos] = float(pw)
resid_visit[act_pos] = _activity_resid_visit(activity, proc_waiting[act_pos],
phase1_st[act_pos])
t = act_result.get('throughput', '')
if t:
throughput[act_pos] = float(t)
else:
# LQNS omits throughput here; each phase executes once per entry invocation
throughput[act_pos] = entry_tput
pu = act_result.get('proc-utilization', '')
hd = activity.get('host-demand-mean', '')
if pu:
proc_util[act_pos] = float(pu)
elif hd and not np.isnan(entry_tput):
# LQNS omits proc-util here: per-phase value = entry throughput * phase host demand
proc_util[act_pos] = entry_tput * float(hd)
# Parse task-activities
for task_acts in root.findall('.//task-activities'):
for activity in task_acts.findall('activity'):
# Activities under task-activities (already filtered by findall)
act_name = activity.get('name')
act_pos = self._find_lqn_elem(names, types, act_name, LQN_ACTIVITY)
for act_result in activity.findall('result-activity'):
if act_pos >= 0 and act_pos < len(utilization):
utilization[act_pos] = float(act_result.get('utilization', 0))
st_raw = act_result.get('service-time', '')
# absent means UNREPORTED, not zero -- MATLAB's str2double('')
# is NaN and the unanimity rule below must read the same thing
phase1_st[act_pos] = float(st_raw) if st_raw else np.nan
throughput[act_pos] = float(act_result.get('throughput', 0))
pw = act_result.get('proc-waiting', '')
if pw:
proc_waiting[act_pos] = float(pw)
proc_util[act_pos] = float(act_result.get('proc-utilization', 0))
resid_visit[act_pos] = _activity_resid_visit(activity, proc_waiting[act_pos],
phase1_st[act_pos])
# Processor utilization of an entry, aggregated from its activity graph.
# lqns credits host work to whichever level carries the host demand: in
# the activity-graph form an entry declares none, so lqns reports
# result-entry proc-utilization as a literal 0 and the work sits on the
# result-activity rows. The entry value is the sum over the activities
# reachable from the entry within its own task, which is what actsof
# holds. In PH1PH2 form the same sum runs over the phase activities and
# reproduces the value lqns reports there, so no form test is needed. An
# entry with no activities, or any activity lqns left unreported, keeps
# the raw attribute rather than a partial sum.
lqn = self._lqn
if lqn is not None and getattr(lqn, 'actsof', None):
for eoff in range(int(lqn.nentries)):
eidx = int(lqn.eshift) + eoff
acts = [a for a in lqn.actsof.get(eidx, []) if a < num_nodes]
if not acts or len(acts) != len(lqn.actsof.get(eidx, [])):
continue
pu_acts = proc_util[acts]
if not np.any(np.isnan(pu_acts)):
proc_util[eidx] = float(np.sum(pu_acts))
# Phase-1 service time of an entry lqns never invoked.
# lqns omits phase1-service-time from result-entry exactly when the entry's
# throughput is zero: nothing was served, so there is no per-invocation mean
# to report. LINE then carried a NaN where the table says an entry HAS a
# response time and every other solver reports one, breaking the NaN mask --
# see _kb/06-solver-catalog.md. The value is taken from the activity rows,
# and ONLY where they are unanimous: if every activity reachable from the
# entry reports a zero service time then every aggregation law agrees on
# zero -- the serial sum, the branch-weighted mean of an OrFork, the order
# statistic of an AndFork -- so the derivation does not depend on which one
# applies.
# It is deliberately NOT generalised the way proc_util is above.
# Utilizations add over an activity graph; response times do not. Measured
# over the example corpus, sum(actsof) reproduces phase1-service-time on
# serial chains only and misses it wherever the graph branches
# (lqn_workflows `Entry`: 12.5667 reported against 8.5667 summed,
# lqn_fork_open_arrival `SE`: 0.841667 against 1.0), so a summed fallback
# would answer with a number lqns contradicts. An entry whose activities are
# unreported, absent, or not all zero keeps NaN.
if lqn is not None and getattr(lqn, 'actsof', None):
for eoff in range(int(lqn.nentries)):
eidx = int(lqn.eshift) + eoff
if not np.isnan(phase1_st[eidx]):
continue
acts = [a for a in lqn.actsof.get(eidx, []) if a < num_nodes]
if not acts or len(acts) != len(lqn.actsof.get(eidx, [])):
continue
st_acts = phase1_st[acts]
if not np.any(np.isnan(st_acts)) and np.all(st_acts == 0):
phase1_st[eidx] = 0.0
# Residence time, on SolverLN's definition (its residt): the time an
# activity holds its host per cycle of its task, i.e. one execution's
# host-demand-mean + proc-waiting (no think time, as in SolverLN) weighted
# by the executions per task cycle X_a/X_t, and for a task the sum over its activities. lqns
# reports no such column, but the .lqxo carries every term of it. A task
# lqns never invoked (X_t = 0) holds its host for no time. Entries and
# processors have no ResidT in any solver and stay NaN, as does any row
# with a term unreported.
from ....constants import GlobalConstants
resid_t = np.full(num_nodes, np.nan)
if lqn is not None:
parent = np.asarray(lqn.parent).flatten()
ashift, nacts = int(lqn.ashift), int(lqn.nacts)
tshift, ntasks = int(lqn.tshift), int(lqn.ntasks)
act_range = range(ashift, min(ashift + nacts, num_nodes))
for aidx in act_range:
tidx = int(parent[aidx])
xa, xt, rv = throughput[aidx], throughput[tidx], resid_visit[aidx]
if np.isnan(rv) or np.isnan(xa) or np.isnan(xt):
continue
resid_t[aidx] = xa / xt * rv if xt > GlobalConstants.FineTol else 0.0
for tidx in range(tshift, tshift + ntasks):
acts = [a for a in act_range if int(parent[a]) == tidx]
if acts and not np.any(np.isnan(resid_t[acts])):
resid_t[tidx] = float(np.sum(resid_t[acts]))
# Build result
self._result = LQNSResult(
PN=proc_util,
SN=phase1_st,
TN=throughput,
UN=utilization,
RN=proc_waiting,
QN=np.full_like(proc_waiting, np.nan),
AN=np.full(num_nodes, np.nan),
WN=resid_t,
runtime=0.0,
method=self.options.method,
iterations=iterations
)
def _build_names_from_xml(self, root) -> Tuple[Dict[int, str], Dict[int, int]]:
"""
Build the node name and node kind mappings by walking the .lqxo itself.
This is the fallback used when the struct carries no names. The kind is
the tag being walked, so it is known exactly here and is returned
alongside the name; see _find_lqn_elem for why the pair is the key.
"""
names = {}
types = {}
idx = 0 # 0-based, in step with the struct index space
# Parse processors
for proc_elem in root.findall('.//processor'):
proc_name = proc_elem.get('name')
if proc_name:
names[idx] = proc_name
types[idx] = LQN_HOST
idx += 1
# Parse tasks
for task_elem in proc_elem.findall('task'):
task_name = task_elem.get('name')
if task_name:
names[idx] = task_name
types[idx] = LQN_TASK
idx += 1
# Parse entries
for entry_elem in task_elem.findall('.//entry'):
entry_name = entry_elem.get('name')
if entry_name:
names[idx] = entry_name
types[idx] = LQN_ENTRY
idx += 1
# Parse activities from task-activities
for task_acts in task_elem.findall('task-activities'):
for activity in task_acts.findall('activity'):
act_name = activity.get('name')
if act_name:
names[idx] = act_name
types[idx] = LQN_ACTIVITY
idx += 1
return names, types
def _get_node_names(self) -> Dict[int, str]:
"""Get mapping of node index to node name."""
names = {}
lqn = self._lqn
if lqn is None:
return names
# Native Python struct has names as numpy array
if hasattr(lqn, 'names') and lqn.names is not None:
# THE STRUCT INDEX SPACE IS 0-BASED (b94cccb83), so element 0 is
# a real node -- the first processor -- and skipping it drops
# that node from every result and shifts each remaining one down
# a slot. On lqn_twotasks that lost the P1 row outright and left
# a table that still READ correctly, because the same map labels
# the rows it mis-indexes.
for idx in range(len(lqn.names)):
name = lqn.names[idx]
if name is not None and name != '':
names[idx] = str(name)
return names
def _get_node_types(self) -> Dict[int, int]:
"""Get mapping of node index to LayeredNetworkElement kind."""
types = {}
lqn = self._lqn
if lqn is None or not hasattr(lqn, 'type') or lqn.type is None:
return types
# lqn.type is 0-based, in step with lqn.names.
for idx in range(len(lqn.type)):
types[idx] = int(lqn.type[idx])
return types
def _find_lqn_elem(self, names: Dict[int, str], types: Dict[int, int],
target: str, elem_type: int) -> int:
"""
0-based position of the element called TARGET whose kind is ELEM_TYPE,
or -1 when no such element exists.
THE KIND IS PART OF THE KEY, and has to be. A LINE-generated layered
model routinely gives a processor, its task and that task's entry the
SAME name, and lqn.names holds all three, so a name-only lookup returns
whichever one it meets first and the .lqxo rows for the other two are
written into it: one result file then yields three different wrong
answers. The document states which kind each row describes -- it is the
tag being read -- so the ambiguity does not have to exist. On a model
whose names are unique this agrees element for element with the
name-only lookup it replaces.
"""
for pos, name in names.items():
if name == target and types.get(pos) == elem_type:
return pos
return -1
[docs]
def getAvg(self):
"""Get average performance metrics: an LQNSResult on a LayeredNetwork,
the (Q, U, R, T, A, W) tuple on a flat Network."""
if not self._layered:
return self._qns_get_avg()
if self._result is None:
# runAnalyzer, NOT the NetworkSolver _ensureAvgResults funnel:
# its MAP/MMPP random-environment gate has no meaning for a
# LayeredNetwork.
self.runAnalyzer()
# Copy result and swap QN/UN/RN
result = LQNSResult(
QN=self._result.UN.copy(),
UN=self._result.PN.copy(),
RN=self._result.SN.copy(),
TN=self._result.TN.copy(),
PN=self._result.PN.copy(),
SN=self._result.SN.copy(),
AN=self._result.AN.copy(),
WN=self._result.WN.copy(),
runtime=self._result.runtime,
method=self._result.method,
iterations=self._result.iterations
)
# UN is lqns' proc-utilization, aggregated over the activity graph for
# entries, which lqns itself reports as 0 in the activity-graph form.
# lqns sums it over the host's busy servers; it is divided by the servers
# the work can reach (lsn.hostReachMult), the divisor of SolverLN and
# SolverLDES, so the three Util columns agree (P150).
lsn = self.model.getStruct()
UN = np.asarray(result.UN, dtype=float)
for i in range(min(UN.size, int(lsn.nidx))):
UN.flat[i] = UN.flat[i] / lsn.hostReachMult(i)
result.UN = UN
return result
[docs]
def getAvgTable(self) -> pd.DataFrame:
"""
Get average performance metrics table.
Returns:
pandas.DataFrame with layered network performance metrics, or the
station table on a flat Network
"""
if not self._layered:
return self._qns_get_avg_table()
# see _kb/06-solver-catalog.md (Wrappers: "lang='java' opt-in JAR delegation")
if getattr(self.options, 'lang', 'python') == 'java':
from ...jar_dispatch import ln_avg_table_via_jar
return ln_avg_table_via_jar(self)
result = self.getAvg()
# see _kb/06-solver-catalog.md (Wrappers: "LQNS snap-to-tenth")
QN = _snap_to_tenth(result.QN)
UN = _snap_to_tenth(result.UN)
RN = _snap_to_tenth(result.RN)
TN = _snap_to_tenth(result.TN)
WN = _snap_to_tenth(result.WN)
# Get node info - prefer parsed names from XML
if hasattr(self, '_parsed_names') and self._parsed_names:
names = self._parsed_names
else:
names = self._get_node_names()
# Build rows for all nodes (preserves NaN for values not computed)
rows = []
for idx, name in names.items():
i = idx
if i < len(QN):
# Determine node type from the model structure
node_type = self._get_node_type(idx)
rows.append({
'Node': name,
'NodeType': node_type,
'QLen': QN[i],
'Util': UN[i],
'RespT': RN[i],
'ResidT': WN[i],
'ArvR': np.nan, # LQNS doesn't compute ArvR
'Tput': TN[i],
})
df = pd.DataFrame(rows)
if not self._table_silent and len(df) > 0:
print(df.to_string(index=False))
# IndexedTable gives MATLAB-style 5-sig-fig formatting; a bare
# DataFrame's fixed 5-decimal repr drops a sig fig below 0.1.
from line_solver.indexed_table import IndexedTable
return IndexedTable(df)
# avg_table / get_avg_table are inherited from NetworkSolver: they wrap a
# DataFrame in an IndexedTable and return an IndexedTable as it is.
[docs]
def getLNAvgTable(self) -> pd.DataFrame:
"""The per-element table of a LayeredNetwork, under the name SolverLDES gives it."""
if not self._layered:
raise LineUnsupportedError(
"getLNAvgTable needs a LayeredNetwork and this SolverLQNS holds a flat Network; use getAvgTable().",
error_id='LINE:unsupported:Analysis')
return self.getAvgTable()
get_ln_avg_table = getLNAvgTable
def _get_node_type(self, idx: int) -> str:
"""Get node type name for a given index."""
lqn = self.getStruct()
if lqn is None:
return 'Unknown'
# Check if model has type information
# Note: lqn.type is 0-based, matching lqn.names
if hasattr(lqn, 'type') and idx < len(lqn.type):
elem_type = int(lqn.type[idx])
# Native Python uses integer values:
# 0=PROCESSOR, 1=TASK, 2=ENTRY, 3=ACTIVITY
if elem_type == 0:
return 'Processor'
elif elem_type == 1:
if hasattr(lqn, 'isref') and lqn.isref[idx, 0]:
return 'RefTask'
return 'Task'
elif elem_type == 2:
return 'Entry'
elif elem_type == 3:
return 'Activity'
elif elem_type == 4:
return 'Call'
return 'Unknown'
[docs]
def getRawAvgTables(self) -> Tuple[pd.DataFrame, pd.DataFrame]:
"""
Get raw average tables including call metrics.
Returns:
Tuple of (avg_table, call_avg_table) DataFrames
"""
result = self.getAvg()
# Get node info
if hasattr(self, '_parsed_names') and self._parsed_names:
names = self._parsed_names
else:
names = self._get_node_names()
# Build average table rows (preserves NaN for values not computed)
rows = []
for idx, name in names.items():
i = idx
if i < len(result.QN):
node_type = self._get_node_type(idx)
rows.append({
'Node': name,
'NodeType': node_type,
'QLen': result.QN[i],
'Util': result.UN[i],
'RespT': result.RN[i],
'ResidT': result.WN[i],
'ArvR': np.nan, # LQNS doesn't compute ArvR
'Tput': result.TN[i],
})
avg_table = pd.DataFrame(rows)
# Build call average table (inter-entry calls)
# This table shows call statistics between entries
call_rows = []
# For now, return an empty call table - full implementation would
# parse call statistics from LQNS output
call_avg_table = pd.DataFrame(call_rows, columns=['From', 'To', 'CallRate', 'Wait'])
return avg_table, call_avg_table
# Alias
get_raw_avg_tables = getRawAvgTables
[docs]
@staticmethod
def getFeatureSet() -> set:
"""Get set of features supported on a flat Network (the qns methods).
Mirrors MATLAB SolverLQNS.getFeatureSet and the JAR
SolverLQNS.getFeatureSet; getLayeredFeatureSet is the per-layer set.
"""
return qns_feature_set()
[docs]
@staticmethod
def getLayeredFeatureSet() -> set:
"""Get set of features supported per layer by the LQNS solver.
Returns the canonical feature names (mirrors MATLAB
SolverLQNS.supports and the JAR SolverLQNS.getLayeredFeatureSet).
"""
return {
'Sink', 'Source', 'Queue',
'Coxian', 'Erlang', 'Exp', 'HyperExp',
'Buffer', 'Server', 'JobSink', 'RandomSource', 'ServiceTunnel',
'SchedStrategy_PS', 'SchedStrategy_FCFS',
'ClosedClass',
}
[docs]
@staticmethod
def supports(model) -> bool:
"""Check if model is supported.
Mirrors MATLAB SolverLQNS.supports: the qns feature set on a flat
Network, the per-layer set otherwise.
"""
from ...base import supports_via_featureset
if model is not None and not _is_layered(model):
return supports_via_featureset(SolverLQNS, model)
return supports_via_featureset(_LayeredFeatures, model)
[docs]
def getMethodFeatureSet(self, method):
"""The qns feature envelope on a flat Network (it does not vary by
method), so the gate can NAME the offending features; None on a
LayeredNetwork, which has no used-feature record and keeps the coarse
supports(model) path. Mirrors @SolverLQNS/getMethodFeatureSet."""
if self._layered:
return None
model = getattr(self, 'model', None)
if model is None or not hasattr(model, 'get_used_lang_features'):
return None
return qns_feature_set()
[docs]
def supportsModelMethod(self, method):
"""Method-aware gate. A LayeredNetwork asks lqns_method_refusal, which refuses the
qns methods by name since qnsolver takes a flat Network; a Network runs the qns gate."""
if not self._layered:
return self._qns_supports_model_method(method)
ok, reason = lqns_method_refusal(self.model, method)
if not ok:
return ok, reason
return NetworkSolver.supportsModelMethod(self, method)
supports_model_method = supportsModelMethod
[docs]
@staticmethod
def defaultOptions() -> 'LQNSOptions':
"""Get default solver options.
Returns:
LQNSOptions with default configuration
"""
return LQNSOptions()
# Aliases
run_analyzer = runAnalyzer
is_available = isAvailable
list_valid_methods = listValidMethods
default_options = defaultOptions
get_struct = getStruct
class _LayeredFeatures:
"""The per-layer feature set, in the shape supports_via_featureset reads."""
getFeatureSet = staticmethod(SolverLQNS.getLayeredFeatureSet)
def _activity_resid_visit(activity, proc_waiting, service_time) -> float:
"""Host residence of one execution of an activity: its host-demand-mean plus
the wait for the host. NaN when lqns left either unreported. proc-waiting is a
part of service-time, so one that is negative or exceeds it is not a
measurement: lqns 6.2.31 writes uninitialised values there for some phase
activities (6.96e+252 in its own 76-distributions model, 1.44e+161 in
94-5101-a2-q2b1) while service-time stays sane. Such a row reports NaN."""
hd = activity.get('host-demand-mean', '')
if not hd:
return np.nan
if proc_waiting < 0 or (not np.isnan(service_time) and proc_waiting > service_time):
return np.nan
return float(hd) + float(proc_waiting)
def _is_layered(model) -> bool:
"""True for a LayeredNetwork (anything that is not a Network or a NetworkStruct)."""
from ....api.sn import NetworkStruct
if isinstance(model, NetworkStruct):
return False
from ....lang.network import Network
return not isinstance(model, Network)
def _lqns_unsupported_constructs(model) -> List[str]:
"""The LQN-level constructs neither lqns nor lqsim models, named one by one.
The .lqnx writer carries them in the LINE dialect, so lqns would otherwise
stop at its own parse error ("Unexpected element <setup ...>") for what is a
modelling limit of the binary. Mirrors MATLAB SolverLQNS.unsupportedLNConstructs.
"""
from ....constants import GlobalConstants
from ....layered import _get_dist_mean
reasons = []
tasks = list(getattr(model, 'tasks', []) or [])
for task in tasks:
if type(task).__name__ == 'CacheTask':
reasons.append("task '%s' is a CacheTask, and neither lqns nor lqsim models a cache" % task.name)
for entry in getattr(task, 'entries', []) or []:
if type(entry).__name__ == 'ItemEntry':
reasons.append("entry '%s' on task '%s' is an ItemEntry, and neither lqns nor lqsim models "
"an item reference stream" % (entry.name, task.name))
for attr, what in (('setup_time', 'a setup time'), ('delay_off_time', 'a delay-off time')):
dist = getattr(task, attr, None)
mean = _get_dist_mean(dist) if dist is not None else 0.0
if mean > GlobalConstants.FineTol:
reasons.append("task '%s' declares %s (%g), which neither lqns nor lqsim charges"
% (task.name, what, mean))
# the schema takes think-time on reference tasks only, so write_xml drops it here with a warning
sched = getattr(task, 'sched_strategy', None)
tt = getattr(task, 'think_time', None)
tt_mean = _get_dist_mean(tt) if tt is not None and not isinstance(tt, (int, float)) else float(tt or 0.0)
if getattr(sched, 'name', str(sched)).upper() != 'REF' and tt_mean > GlobalConstants.FineTol:
reasons.append("task '%s' is not a reference task and declares a think time (%g), which the LQN XML "
"schema accepts on reference tasks only" % (task.name, tt_mean))
for elem in list(getattr(model, 'processors', []) or []) + tasks:
if hasattr(elem, 'hasServerPools') and elem.hasServerPools():
reasons.append("'%s' declares heterogeneous server pools with a class-compatibility graph, which "
"neither lqns nor lqsim models: their multiserver is homogeneous" % elem.name)
for attr, what in (('lld_scaling', 'load-dependent'), ('lcd_scaling', 'class-dependent'),
('ljd_scaling', 'joint-dependent')):
if getattr(elem, attr, None) is not None:
reasons.append("'%s' declares a %s service rate, which neither lqns nor lqsim models"
% (elem.name, what))
return reasons
def lqns_method_refusal(model, method: str = '') -> Tuple[bool, str]:
"""Whether lqns or lqsim can serve this LayeredNetwork under ``method``.
One predicate for the gate (supportsModelMethod) and for runAnalyzer, which
asks again before the .lqnx is written. Mirrors MATLAB lqns_method_refusal.
A qns* method is refused first: it reaches qnsolver, which takes a flat Network.
"""
from ....constants import SchedStrategy
if is_qns_method(method):
return False, ("SolverLQNS: the method '%s' reaches qnsolver, which takes a flat "
"Network; on a LayeredNetwork use default, lqns, srvn, exactmva, "
"srvn.exactmva, sim, lqsim or lqns.default." % method)
reasons = _lqns_unsupported_constructs(model)
if reasons:
return False, ("SolverLQNS cannot serve this model: %s. Use SolverLN (the native layered solver), "
"which models these constructs." % '; '.join(reasons))
lqnx_sched = ('fcfs', 'ps', 'inf', 'hol', 'pri', 'ref')
elems = list(getattr(model, 'processors', []) or []) + list(getattr(model, 'tasks', []) or [])
for elem in elems:
sched = getattr(elem, 'sched_strategy', None)
if sched is None:
continue
sched = sched if isinstance(sched, SchedStrategy) else next(
(s for s in SchedStrategy if s.value == getattr(sched, 'value', sched)), None)
if sched is None:
continue
text = 'pri' if sched == SchedStrategy.FCFSPRPRIO else sched.name.lower()
if text not in lqnx_sched:
return False, ("'%s' is scheduled %s, which the LQN XML schema does not name: lqns and lqsim "
"accept fcfs, ps, inf, hol, pri (preemptive priority) and ref only. Use SolverLN "
"or SolverLDES." % (elem.name, text.upper()))
sim_method = str(method or '').lower() in ('sim', 'lqsim')
if not sim_method:
# lqns stops at "The quorum feature is not supported in this version"; lqsim honours the quorum
from ....layered import PrecedenceType
import numpy as np
for task in list(getattr(model, 'tasks', []) or []):
for prec in list(getattr(task, 'precedences', []) or []):
npre = len(getattr(prec, 'pre_activities', []) or [])
pre_params = getattr(prec, 'pre_params', None)
if prec.prec_type != PrecedenceType.PARALLEL or npre < 2 or pre_params is None \
or np.size(pre_params) != 1:
continue
q = int(round(float(np.ravel(pre_params)[0])))
if 1 <= q < npre:
return False, ("task '%s' declares a quorum AND-join (%d of %d), which lqns does not "
"support. Use method sim (lqsim), SolverLN or SolverLDES." % (task.name, q, npre))
if sim_method:
for elem in elems:
repl = elem.getReplication() if hasattr(elem, 'getReplication') else 1
if repl is not None and repl > 1:
return False, ("'%s' is replicated %d times, which lqsim refuses to simulate. Use an lqns "
"method, SolverLN or SolverLDES, which materialise the replicas."
% (elem.name, int(repl)))
return True, ''