"""
Single queueing system analysis functions.
This module provides analytical formulas and approximations for single
queueing systems, including classical models like M/M/1, M/M/k, M/G/1,
and various G/G/1 approximations.
Supported queueing systems:
- qsys_mm1: M/M/1 queue (Poisson arrivals, exponential service)
- qsys_mmk: M/M/k queue (Poisson arrivals, k exponential servers)
- qsys_gm1: G/M/1 queue (general arrivals, exponential service)
- qsys_mg1: M/G/1 queue (Poisson arrivals, general service)
- Various G/G/1 approximations (Whitt, Allen-Cunneen, Kingman, etc.)
These functions compute exact results where available, or high-quality
approximations for more general cases.
"""
import jpype
import numpy as np
from line_solver import jlineMatrixToArray, jlineMatrixFromArray
[docs]
def qsys_mm1(lambda_val, mu):
"""
Analyze M/M/1 queue (Poisson arrivals, exponential service).
Args:
lambda_val (float): Arrival rate.
mu (float): Service rate.
Returns:
dict: Performance measures including:
- L: Mean number in system
- Lq: Mean number in queue
- W: Mean response time
- Wq: Mean waiting time
- rho: Utilization (λ/μ)
"""
result = jpype.JPackage('jline').api.qsys.Qsys_mm1Kt.qsys_mm1(
jpype.JDouble(lambda_val), jpype.JDouble(mu)
)
return {
'L': result.L,
'Lq': result.Lq,
'W': result.W,
'Wq': result.Wq,
'rho': result.rho
}
[docs]
def qsys_mmk(lambda_val, mu, k):
"""
Analyze M/M/k queue (Poisson arrivals, k exponential servers).
Args:
lambda_val (float): Arrival rate.
mu (float): Service rate per server.
k (int): Number of parallel servers.
Returns:
dict: Performance measures including L, Lq, W, Wq, rho, P0.
"""
result = jpype.JPackage('jline').api.qsys.Qsys_mmkKt.qsys_mmk(
jpype.JDouble(lambda_val), jpype.JDouble(mu), jpype.JInt(k)
)
return {
'L': result.L,
'Lq': result.Lq,
'W': result.W,
'Wq': result.Wq,
'rho': result.rho,
'P0': result.P0
}
[docs]
def qsys_gm1(lambda_val, mu, sigma_s_squared):
"""
Analyze G/M/1 queue (general arrivals, exponential service).
Args:
lambda_val (float): Arrival rate.
mu (float): Service rate.
sigma_s_squared (float): Variance of service times.
Returns:
dict: Performance measures including L, Lq, W, Wq, rho.
"""
result = jpype.JPackage('jline').api.qsys.Qsys_gm1Kt.qsys_gm1(
jpype.JDouble(lambda_val), jpype.JDouble(mu), jpype.JDouble(sigma_s_squared)
)
return {
'L': result.L,
'Lq': result.Lq,
'W': result.W,
'Wq': result.Wq,
'rho': result.rho
}
[docs]
def qsys_mg1(lambda_val, mu, sigma_s_squared):
"""
Analyze M/G/1 queue (Poisson arrivals, general service).
Args:
lambda_val (float): Arrival rate.
mu (float): Mean service rate.
sigma_s_squared (float): Variance of service times.
Returns:
dict: Performance measures including L, Lq, W, Wq, rho.
"""
result = jpype.JPackage('jline').api.qsys.Qsys_mg1Kt.qsys_mg1(
jpype.JDouble(lambda_val), jpype.JDouble(mu), jpype.JDouble(sigma_s_squared)
)
return {
'L': result.L,
'Lq': result.Lq,
'W': result.W,
'Wq': result.Wq,
'rho': result.rho
}
[docs]
def qsys_gig1_approx_lin(lambda_val, mu, ca_squared, cs_squared):
"""
G/G/1 queue approximation using linear interpolation.
Args:
lambda_val (float): Arrival rate.
mu (float): Service rate.
ca_squared (float): Squared coefficient of variation of arrivals.
cs_squared (float): Squared coefficient of variation of service.
Returns:
dict: Approximate performance measures.
"""
result = jpype.JPackage('jline').api.qsys.Qsys_gig1_approx_linKt.qsys_gig1_approx_lin(
jpype.JDouble(lambda_val), jpype.JDouble(mu),
jpype.JDouble(ca_squared), jpype.JDouble(cs_squared)
)
return {
'L': result.L,
'Lq': result.Lq,
'W': result.W,
'Wq': result.Wq,
'rho': result.rho
}
[docs]
def qsys_gig1_approx_kk(lambda_val, mu, ca_squared, cs_squared):
"""
G/G/1 queue approximation using Kraemer-Langenbach-Belz method.
Args:
lambda_val (float): Arrival rate.
mu (float): Service rate.
ca_squared (float): Squared coefficient of variation of arrivals.
cs_squared (float): Squared coefficient of variation of service.
Returns:
dict: Approximate performance measures.
"""
result = jpype.JPackage('jline').api.qsys.Qsys_gig1_approx_kkKt.qsys_gig1_approx_kk(
jpype.JDouble(lambda_val), jpype.JDouble(mu),
jpype.JDouble(ca_squared), jpype.JDouble(cs_squared)
)
return {
'L': result.L,
'Lq': result.Lq,
'W': result.W,
'Wq': result.Wq,
'rho': result.rho
}
[docs]
def qsys_gig1_approx_whitt(lambda_val, mu, ca_squared, cs_squared):
"""
G/G/1 queue approximation using Whitt's method.
Args:
lambda_val (float): Arrival rate.
mu (float): Service rate.
ca_squared (float): Squared coefficient of variation of arrivals.
cs_squared (float): Squared coefficient of variation of service.
Returns:
dict: Approximate performance measures.
"""
result = jpype.JPackage('jline').api.qsys.Qsys_gig1_approx_whittKt.qsys_gig1_approx_whitt(
jpype.JDouble(lambda_val), jpype.JDouble(mu),
jpype.JDouble(ca_squared), jpype.JDouble(cs_squared)
)
return {
'L': result.L,
'Lq': result.Lq,
'W': result.W,
'Wq': result.Wq,
'rho': result.rho
}
[docs]
def qsys_gig1_approx_allencunneen(lambda_val, mu, ca_squared, cs_squared):
"""
G/G/1 queue approximation using Allen-Cunneen method.
Args:
lambda_val: Arrival rate
mu: Service rate
ca_squared: Squared coefficient of variation of arrivals
cs_squared: Squared coefficient of variation of service
Returns:
dict: Performance measures (W, rho)
"""
result = jpype.JPackage('jline').api.qsys.Qsys_gig1_approx_allencunneenKt.qsys_gig1_approx_allencunneen(
jpype.JDouble(lambda_val), jpype.JDouble(mu),
jpype.JDouble(ca_squared), jpype.JDouble(cs_squared)
)
return {
'W': result.W,
'rho': result.rho
}
[docs]
def qsys_gig1_approx_heyman(lambda_val, mu, ca_squared, cs_squared):
"""
G/G/1 queue approximation using Heyman method.
Args:
lambda_val: Arrival rate
mu: Service rate
ca_squared: Squared coefficient of variation of arrivals
cs_squared: Squared coefficient of variation of service
Returns:
dict: Performance measures (W, rho)
"""
result = jpype.JPackage('jline').api.qsys.Qsys_gig1_approx_heymanKt.qsys_gig1_approx_heyman(
jpype.JDouble(lambda_val), jpype.JDouble(mu),
jpype.JDouble(ca_squared), jpype.JDouble(cs_squared)
)
return {
'W': result.W,
'rho': result.rho
}
[docs]
def qsys_gig1_approx_kobayashi(lambda_val, mu, ca_squared, cs_squared):
"""
G/G/1 queue approximation using Kobayashi method.
Args:
lambda_val: Arrival rate
mu: Service rate
ca_squared: Squared coefficient of variation of arrivals
cs_squared: Squared coefficient of variation of service
Returns:
dict: Performance measures (W, rho)
"""
result = jpype.JPackage('jline').api.qsys.Qsys_gig1_approx_kobayashiKt.qsys_gig1_approx_kobayashi(
jpype.JDouble(lambda_val), jpype.JDouble(mu),
jpype.JDouble(ca_squared), jpype.JDouble(cs_squared)
)
return {
'W': result.W,
'rho': result.rho
}
[docs]
def qsys_gig1_approx_marchal(lambda_val, mu, ca_squared, cs_squared):
"""
G/G/1 queue approximation using Marchal method.
Args:
lambda_val: Arrival rate
mu: Service rate
ca_squared: Squared coefficient of variation of arrivals
cs_squared: Squared coefficient of variation of service
Returns:
dict: Performance measures (W, rho)
"""
result = jpype.JPackage('jline').api.qsys.Qsys_gig1_approx_marchalKt.qsys_gig1_approx_marchal(
jpype.JDouble(lambda_val), jpype.JDouble(mu),
jpype.JDouble(ca_squared), jpype.JDouble(cs_squared)
)
return {
'W': result.W,
'rho': result.rho
}
[docs]
def qsys_gig1_ubnd_kingman(lambda_val, mu, ca_squared, cs_squared):
"""
G/G/1 queue upper bound using Kingman method.
Args:
lambda_val: Arrival rate
mu: Service rate
ca_squared: Squared coefficient of variation of arrivals
cs_squared: Squared coefficient of variation of service
Returns:
dict: Upper bound performance measures (W, rho)
"""
result = jpype.JPackage('jline').api.qsys.Qsys_gig1_ubnd_kingmanKt.qsys_gig1_ubnd_kingman(
jpype.JDouble(lambda_val), jpype.JDouble(mu),
jpype.JDouble(ca_squared), jpype.JDouble(cs_squared)
)
return {
'W': result.W,
'rho': result.rho
}
[docs]
def qsys_gigk_approx(lambda_val, mu, ca_squared, cs_squared, k):
"""
G/G/k queue approximation for k servers.
Args:
lambda_val: Arrival rate
mu: Service rate per server
ca_squared: Squared coefficient of variation of arrivals
cs_squared: Squared coefficient of variation of service
k: Number of servers
Returns:
dict: Performance measures (W, rho)
"""
result = jpype.JPackage('jline').api.qsys.Qsys_gigk_approxKt.qsys_gigk_approx(
jpype.JDouble(lambda_val), jpype.JDouble(mu),
jpype.JDouble(ca_squared), jpype.JDouble(cs_squared), jpype.JInt(k)
)
return {
'W': result.W,
'rho': result.rho
}
[docs]
def qsys_gigk_approx_kingman(lambda_val, mu, ca_squared, cs_squared, k):
"""
G/G/k queue approximation using Kingman method.
Args:
lambda_val: Arrival rate
mu: Service rate per server
ca_squared: Squared coefficient of variation of arrivals
cs_squared: Squared coefficient of variation of service
k: Number of servers
Returns:
dict: Performance measures (W, rho)
"""
result = jpype.JPackage('jline').api.qsys.Qsys_gigk_approx_kingmanKt.qsys_gigk_approx_kingman(
jpype.JDouble(lambda_val), jpype.JDouble(mu),
jpype.JDouble(ca_squared), jpype.JDouble(cs_squared), jpype.JInt(k)
)
return {
'W': result.W,
'rho': result.rho
}
def qsys_gig1_approx_klb(lambda_val, mu, ca_squared, cs_squared):
"""
G/G/1 queue approximation using Kraemer-Langenbach-Belz method.
Args:
lambda_val: Arrival rate
mu: Service rate
ca_squared: Squared coefficient of variation of arrivals
cs_squared: Squared coefficient of variation of service
Returns:
dict: Performance measures (L, Lq, W, Wq, rho)
"""
result = jpype.JPackage('jline').api.qsys.Qsys_gig1_approx_klbKt.qsys_gig1_approx_klb(
jpype.JDouble(lambda_val), jpype.JDouble(mu),
jpype.JDouble(ca_squared), jpype.JDouble(cs_squared)
)
return {
'L': result.L,
'Lq': result.Lq,
'W': result.W,
'Wq': result.Wq,
'rho': result.rho
}
# Additional QSYS functions for complete API coverage
[docs]
def qsys_gg1(arrival_mean, arrival_scv, service_mean, service_scv):
"""
Analyze G/G/1 queueing system.
Args:
arrival_mean: Mean inter-arrival time
arrival_scv: Squared coefficient of variation of inter-arrival times
service_mean: Mean service time
service_scv: Squared coefficient of variation of service times
Returns:
dict: Performance metrics including utilization, queue length, and waiting time
"""
result = jpype.JPackage('jline').api.qsys.Qsys_gg1Kt.qsys_gg1(
float(arrival_mean), float(arrival_scv),
float(service_mean), float(service_scv)
)
return {
'utilization': float(result.getUtilization()),
'queue_length': float(result.getQueueLength()),
'waiting_time': float(result.getWaitingTime()),
'response_time': float(result.getResponseTime())
}
[docs]
def qsys_gig1_approx_gelenbe(arrival_mean, arrival_scv, service_mean, service_scv):
"""
G/G/1 approximation using Gelenbe's method.
Args:
arrival_mean: Mean inter-arrival time
arrival_scv: Squared coefficient of variation of inter-arrival times
service_mean: Mean service time
service_scv: Squared coefficient of variation of service times
Returns:
dict: Approximate performance metrics
"""
result = jpype.JPackage('jline').api.qsys.Qsys_gig1_approx_gelenbeKt.qsys_gig1_approx_gelenbe(
float(arrival_mean), float(arrival_scv),
float(service_mean), float(service_scv)
)
return {
'queue_length': float(result.getQueueLength()),
'waiting_time': float(result.getWaitingTime())
}
[docs]
def qsys_gig1_approx_kimura(arrival_mean, arrival_scv, service_mean, service_scv):
"""
G/G/1 approximation using Kimura's method.
Args:
arrival_mean: Mean inter-arrival time
arrival_scv: Squared coefficient of variation of inter-arrival times
service_mean: Mean service time
service_scv: Squared coefficient of variation of service times
Returns:
dict: Approximate performance metrics
"""
result = jpype.JPackage('jline').api.qsys.Qsys_gig1_approx_kimuraKt.qsys_gig1_approx_kimura(
float(arrival_mean), float(arrival_scv),
float(service_mean), float(service_scv)
)
return {
'queue_length': float(result.getQueueLength()),
'waiting_time': float(result.getWaitingTime())
}
[docs]
def qsys_gig1_approx_klb(arrival_mean, arrival_scv, service_mean, service_scv):
"""
G/G/1 approximation using Kraemer-Langenbach-Belz formula.
Args:
arrival_mean: Mean inter-arrival time
arrival_scv: Squared coefficient of variation of inter-arrival times
service_mean: Mean service time
service_scv: Squared coefficient of variation of service times
Returns:
dict: Approximate performance metrics
"""
result = jpype.JPackage('jline').api.qsys.Qsys_gig1_approx_klbKt.qsys_gig1_approx_klb(
float(arrival_mean), float(arrival_scv),
float(service_mean), float(service_scv)
)
return {
'queue_length': float(result.getQueueLength()),
'waiting_time': float(result.getWaitingTime())
}
[docs]
def qsys_gig1_approx_myskja(arrival_mean, arrival_scv, service_mean, service_scv):
"""
G/G/1 approximation using Myskja's method.
Args:
arrival_mean: Mean inter-arrival time
arrival_scv: Squared coefficient of variation of inter-arrival times
service_mean: Mean service time
service_scv: Squared coefficient of variation of service times
Returns:
dict: Approximate performance metrics
"""
result = jpype.JPackage('jline').api.qsys.Qsys_gig1_approx_myskjaKt.qsys_gig1_approx_myskja(
float(arrival_mean), float(arrival_scv),
float(service_mean), float(service_scv)
)
return {
'queue_length': float(result.getQueueLength()),
'waiting_time': float(result.getWaitingTime())
}
[docs]
def qsys_gig1_approx_myskja2(arrival_mean, arrival_scv, service_mean, service_scv):
"""
G/G/1 approximation using Myskja's refined method.
Args:
arrival_mean: Mean inter-arrival time
arrival_scv: Squared coefficient of variation of inter-arrival times
service_mean: Mean service time
service_scv: Squared coefficient of variation of service times
Returns:
dict: Approximate performance metrics
"""
result = jpype.JPackage('jline').api.qsys.Qsys_gig1_approx_myskja2Kt.qsys_gig1_approx_myskja2(
float(arrival_mean), float(arrival_scv),
float(service_mean), float(service_scv)
)
return {
'queue_length': float(result.getQueueLength()),
'waiting_time': float(result.getWaitingTime())
}
[docs]
def qsys_gig1_lbnd(arrival_mean, arrival_scv, service_mean, service_scv):
"""
Lower bound for G/G/1 queue performance.
Args:
arrival_mean: Mean inter-arrival time
arrival_scv: Squared coefficient of variation of inter-arrival times
service_mean: Mean service time
service_scv: Squared coefficient of variation of service times
Returns:
dict: Lower bounds on performance metrics
"""
result = jpype.JPackage('jline').api.qsys.Qsys_gig1_lbndKt.qsys_gig1_lbnd(
float(arrival_mean), float(arrival_scv),
float(service_mean), float(service_scv)
)
return {
'queue_length_lower': float(result.getQueueLengthLower()),
'waiting_time_lower': float(result.getWaitingTimeLower())
}
[docs]
def qsys_gigk_approx_whitt(arrival_mean, arrival_scv, service_mean, service_scv, num_servers):
"""
G/G/k approximation using Whitt's method.
Args:
arrival_mean: Mean inter-arrival time
arrival_scv: Squared coefficient of variation of inter-arrival times
service_mean: Mean service time
service_scv: Squared coefficient of variation of service times
num_servers: Number of servers
Returns:
dict: Approximate performance metrics
"""
result = jpype.JPackage('jline').api.qsys.Qsys_gigk_approx_whittKt.qsys_gigk_approx_whitt(
float(arrival_mean), float(arrival_scv),
float(service_mean), float(service_scv),
int(num_servers)
)
return {
'queue_length': float(result.getQueueLength()),
'waiting_time': float(result.getWaitingTime()),
'utilization': float(result.getUtilization())
}
[docs]
def qsys_mg1k_loss(arrival_rate, service_mean, service_scv, buffer_size):
"""
M/G/1/k queue with finite buffer - compute loss probability.
Args:
arrival_rate: Arrival rate
service_mean: Mean service time
service_scv: Squared coefficient of variation of service times
buffer_size: Buffer capacity
Returns:
dict: Performance metrics including loss probability
"""
result = jpype.JPackage('jline').api.qsys.Qsys_mg1k_lossKt.qsys_mg1k_loss(
float(arrival_rate), float(service_mean),
float(service_scv), int(buffer_size)
)
return {
'loss_prob': float(result.getLossProbability()),
'queue_length': float(result.getQueueLength()),
'utilization': float(result.getUtilization())
}
[docs]
def qsys_mmkk(arrival_rate, service_rate, num_servers, buffer_size):
"""
M/M/k/k Erlang loss system.
Args:
arrival_rate: Arrival rate
service_rate: Service rate per server
num_servers: Number of servers
buffer_size: System capacity (including servers)
Returns:
dict: Performance metrics including blocking probability
"""
result = jpype.JPackage('jline').api.qsys.Qsys_mmkkKt.qsys_mmkk(
float(arrival_rate), float(service_rate),
int(num_servers), int(buffer_size)
)
return {
'blocking_prob': float(result.getBlockingProbability()),
'utilization': float(result.getUtilization()),
'mean_customers': float(result.getMeanCustomers())
}
[docs]
def qsys_mmm(arrival_rate, service_rate, num_servers):
"""
M/M/m infinite buffer multi-server queue.
Args:
arrival_rate: Arrival rate
service_rate: Service rate per server
num_servers: Number of servers
Returns:
dict: Performance metrics
"""
result = jpype.JPackage('jline').api.qsys.Qsys_mmmKt.qsys_mmm(
float(arrival_rate), float(service_rate), int(num_servers)
)
return {
'utilization': float(result.getUtilization()),
'queue_length': float(result.getQueueLength()),
'waiting_time': float(result.getWaitingTime()),
'response_time': float(result.getResponseTime())
}
[docs]
def qsys_mminf(arrival_rate, service_rate):
"""
M/M/∞ infinite server queue.
Args:
arrival_rate: Arrival rate
service_rate: Service rate per server
Returns:
dict: Performance metrics
"""
result = jpype.JPackage('jline').api.qsys.Qsys_mminfKt.qsys_mminf(
float(arrival_rate), float(service_rate)
)
return {
'mean_customers': float(result.getMeanCustomers()),
'variance_customers': float(result.getVarianceCustomers())
}
[docs]
def qsys_mginf(arrival_rate, service_mean, service_scv):
"""
Analyzes M/G/∞ queue (infinite servers).
Computes performance metrics for a queue with Poisson arrivals,
general service times, and infinite servers.
Args:
arrival_rate: Arrival rate.
service_mean: Mean service time.
service_scv: Squared coefficient of variation of service time.
Returns:
dict: Performance metrics.
"""
result = jpype.JPackage('jline').api.qsys.Qsys_mginfKt.qsys_mginf(
float(arrival_rate), float(service_mean), float(service_scv)
)
return {
'mean_customers': float(result.getMeanCustomers()),
'variance_customers': float(result.getVarianceCustomers()),
'throughput': float(result.getThroughput())
}
[docs]
def qsys_mm1k_loss(arrival_rate, service_rate, buffer_size):
"""
Analyzes M/M/1/K queue with finite buffer.
Computes performance metrics and loss probability for a single-server
queue with Poisson arrivals, exponential service, and finite capacity.
Args:
arrival_rate: Arrival rate.
service_rate: Service rate.
buffer_size: Maximum system capacity (including server).
Returns:
dict: Performance metrics including loss probability.
"""
result = jpype.JPackage('jline').api.qsys.Qsys_mm1k_lossKt.qsys_mm1k_loss(
float(arrival_rate), float(service_rate), int(buffer_size)
)
return {
'loss_probability': float(result.getLossProbability()),
'mean_queue_length': float(result.getMeanQueueLength()),
'mean_response_time': float(result.getMeanResponseTime()),
'effective_arrival_rate': float(result.getEffectiveArrivalRate()),
'utilization': float(result.getUtilization())
}
[docs]
def qsys_mg1k_loss_mgs(arrival_rate, service_mean, service_scv, buffer_size):
"""
Analyzes M/G/1/K queue using modified generating function method.
Computes loss probability and performance metrics for finite capacity
queues with general service time distributions using the MGS
(Modified Generating Function with Spectral) method.
Args:
arrival_rate: Arrival rate.
service_mean: Mean service time.
service_scv: Squared coefficient of variation of service time.
buffer_size: Maximum system capacity.
Returns:
dict: Performance metrics including loss probability.
"""
result = jpype.JPackage('jline').api.qsys.Qsys_mg1k_loss_mgsKt.qsys_mg1k_loss_mgs(
float(arrival_rate), float(service_mean), float(service_scv), int(buffer_size)
)
return {
'loss_probability': float(result.getLossProbability()),
'mean_queue_length': float(result.getMeanQueueLength()),
'mean_waiting_time': float(result.getMeanWaitingTime()),
'effective_arrival_rate': float(result.getEffectiveArrivalRate())
}