"node_modules/immer/images/performance.png" n'existait pas sur "ab2e672331b8de251d943723c185d0fca108623a"
Newer
Older
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
"""
Tests for analytical module (Jackson's theorem).
"""
import pytest
import math
from src.analytics.jackson import JacksonAnalyzer, QueueAnalytics
class TestJacksonAnalyzer:
"""Tests for Jackson's theorem analyzer."""
def test_simple_mm1_queue(self):
"""Test analysis of a simple M/M/1 queue."""
# Single queue: λ=0.8, μ=1.0, ρ=0.8
analyzer = JacksonAnalyzer(
external_arrival_rate=0.8,
coordinator_service_rate=1.0,
coordinator_exit_prob=1.0, # Always exit (no servers)
server_service_rates=[],
server_routing_probs=[]
)
results = analyzer.analyze()
# Check stability
assert results.is_stable
# Check coordinator metrics
coord = results.coordinator
assert coord.utilization == pytest.approx(0.8, abs=0.01)
assert coord.average_customers == pytest.approx(4.0, abs=0.01) # L = 0.8/(1-0.8) = 4
assert coord.average_time == pytest.approx(5.0, abs=0.01) # W = L/λ = 4/0.8 = 5
def test_unstable_queue(self):
"""Test detection of unstable queue (ρ ≥ 1)."""
# λ=1.0, μ=0.8, ρ=1.25 > 1
analyzer = JacksonAnalyzer(
external_arrival_rate=1.0,
coordinator_service_rate=0.8,
coordinator_exit_prob=1.0,
server_service_rates=[],
server_routing_probs=[]
)
results = analyzer.analyze()
# Check instability
assert not results.is_stable
assert results.instability_reason is not None
assert "coordinator" in results.instability_reason
# Metrics should be infinite
assert results.coordinator.average_customers == math.inf
assert results.coordinator.average_time == math.inf
def test_network_with_servers(self):
"""Test network with coordinator and servers."""
# λ=0.1, μc=1.0, p=0.5, q1=0.5, μ1=0.5
analyzer = JacksonAnalyzer(
external_arrival_rate=0.1,
coordinator_service_rate=1.0,
coordinator_exit_prob=0.5,
server_service_rates=[0.5],
server_routing_probs=[0.5]
)
results = analyzer.analyze()
# Check stability
assert results.is_stable
# Coordinator: λc=0.1, μc=1.0, ρc=0.1
coord = results.coordinator
assert coord.utilization == pytest.approx(0.1, abs=0.01)
# Server 1: λ1=0.1*0.5=0.05, μ1=0.5, ρ1=0.1
server1 = results.servers["server_1"]
assert server1.arrival_rate == pytest.approx(0.05, abs=0.01)
assert server1.utilization == pytest.approx(0.1, abs=0.01)
def test_littles_law(self):
"""Test that Little's Law holds: L = λ * W."""
analyzer = JacksonAnalyzer(
external_arrival_rate=0.5,
coordinator_service_rate=1.0,
coordinator_exit_prob=0.7,
server_service_rates=[0.6, 0.4],
server_routing_probs=[0.2, 0.1]
)
results = analyzer.analyze()
if results.is_stable:
# Check Little's Law for coordinator
coord = results.coordinator
L_from_formula = coord.average_customers
L_from_littles = coord.arrival_rate * coord.average_time
assert L_from_formula == pytest.approx(L_from_littles, abs=0.01)
# Check for each server
for server in results.servers.values():
L_from_formula = server.average_customers
L_from_littles = server.arrival_rate * server.average_time
assert L_from_formula == pytest.approx(L_from_littles, abs=0.01)
# Check for entire network
total_L = results.total_average_customers
total_W = results.total_average_time
lambda_0 = analyzer.lambda_0
assert total_L == pytest.approx(lambda_0 * total_W, abs=0.01)
def test_probability_conservation(self):
"""Test that probabilities must sum to 1.0."""
with pytest.raises(ValueError, match="must sum to 1.0"):
JacksonAnalyzer(
external_arrival_rate=0.1,
coordinator_service_rate=1.0,
coordinator_exit_prob=0.5,
server_service_rates=[0.5],
server_routing_probs=[0.3] # 0.5 + 0.3 = 0.8 ≠ 1.0
)
def test_effective_arrival_rates(self):
"""Test calculation of effective arrival rates."""
analyzer = JacksonAnalyzer(
external_arrival_rate=1.0,
coordinator_service_rate=2.0,
coordinator_exit_prob=0.4,
server_service_rates=[1.0, 1.0, 1.0],
server_routing_probs=[0.2, 0.2, 0.2]
)
arrival_rates = analyzer.calculate_effective_arrival_rates()
# Coordinator gets all external arrivals
assert arrival_rates["coordinator"] == pytest.approx(1.0)
# Each server gets λ * qi
assert arrival_rates["server_1"] == pytest.approx(0.2)
assert arrival_rates["server_2"] == pytest.approx(0.2)
assert arrival_rates["server_3"] == pytest.approx(0.2)
def test_multiple_servers_stability(self):
"""Test stability with multiple servers."""
# All servers stable
analyzer = JacksonAnalyzer(
external_arrival_rate=0.3,
coordinator_service_rate=1.0,
coordinator_exit_prob=0.25,
server_service_rates=[1.0, 1.0, 1.0],
server_routing_probs=[0.25, 0.25, 0.25]
)
results = analyzer.analyze()
assert results.is_stable
# One server unstable
analyzer2 = JacksonAnalyzer(
external_arrival_rate=0.9,
coordinator_service_rate=2.0,
coordinator_exit_prob=0.1,
server_service_rates=[0.5, 1.0, 1.0], # First server will be unstable
server_routing_probs=[0.6, 0.15, 0.15] # λ1 = 0.9*0.6 = 0.54, μ1 = 0.5, ρ1 > 1
)
results2 = analyzer2.analyze()
assert not results2.is_stable
assert not results2.servers["server_1"].is_stable