marimo-team/marimo-learn
3
1# /// script2# requires-python = ">=3.13"3# dependencies = [4# "altair",5# "asimpy",6# "marimo",7# "polars==1.24.0",8# ]9# ///10 11import marimo12 13__generated_with = "0.20.4"14app = marimo.App(width="medium")15 16 17@app.cell(hide_code=True)18def _():19 import marimo as mo20 import random21 import statistics22 23 import altair as alt24 import polars as pl25 26 from asimpy import Environment, Process, Queue27 28 return Environment, Process, Queue, alt, mo, pl, random, statistics29 30 31@app.cell(hide_code=True)32def _(mo):33 mo.md(r"""34 # Priority Starvation35 36 ## *When High-Priority Traffic Crowds Out Low-Priority Jobs*37 38 A single server processes two job classes:39 40 - High-priority jobs (class H) arrive frequently and are served quickly.41 - Low-priority jobs (class L) arrive rarely and take longer to serve.42 43 The server always picks the highest-priority job available. Total server utilization $\rho = \rho_H + \rho_L < 1$, so the server has spare capacity on average. Yet low-priority jobs can wait far longer than the utilization level suggests they should.44 45 ### Static Priority: Starvation at Moderate Load46 47 With a static priority queue, high-priority jobs *never* yield to low-priority ones. Even when $\rho_H < 1$, high-priority bursts can lock out low-priority jobs for extended periods. The mean wait for low-priority jobs under a static non-preemptive priority queue is:48 49 $$W_L = \frac{\overline{s}_L}{1-\rho_H} \cdot \frac{1}{1-\rho_H-\rho_L}$$50 51 This diverges as $\rho_H \to 1$ independently of $\rho_L$. As $\rho_H$ approaches 100%, low-priority jobs wait arbitrarily long, even if only a few low-priority jobs ever arrive.52 53 ### Aging: Solving Starvation Creates Oscillation54 55 The standard remedy for starvation is *priority aging*: a waiting job's priority improves over time until it eventually beats even high-priority arrivals. This guarantees finite wait for all jobs.56 57 However, aging introduces a new pathology. When aged low-priority jobs finally burst through, they occupy the server and leave a backlog of high-priority jobs waiting. The high-priority queue then drains, and the cycle repeats — producing oscillating bursts rather than smooth, uniform service.58 59 ### Intuition60 61 Suppose H jobs arrive in random bursts. During a burst, the server never pauses for L jobs. An L job unlucky enough to arrive at the start of a long burst must wait for every H job in that burst to be served before getting its turn. As bursts grow more frequent (larger $\rho_H$), the expected burst length grows, and with it the expected wait for that unlucky L job. The math confirms: starvation is a real risk at moderate $\rho_H$, not just at extreme loads.62 63 ### What aging does64 65 Aging assigns each waiting L job a maximum patience time $T_{\max}$. After waiting $T_{\max}$, the job is promoted to high priority. This caps the worst-case wait: no L job can wait longer than $T_{\max}$ plus one service time. Mathematically, the effective $W_L$ is bounded by $T_{\max} + 1/\mu_L$.66 67 ## Practical Implications68 69 Priority queues appear throughout computing:70 71 - OS scheduling: interactive processes (high priority) vs. batch jobs (low priority). Linux uses dynamic priority aging (nice values + sleep bonuses) to avoid starvation.72 - Network QoS: real-time traffic (VoIP, video) vs. bulk data. Traffic shaping with Deficit Round Robin (DRR) or Weighted Fair Queuing (WFQ) guarantees bandwidth shares without starvation.73 - Database query planning: short OLTP queries vs. long OLAP queries. Resource groups and query timeouts implement a form of aging.74 """)75 return76 77 78@app.cell(hide_code=True)79def _(mo):80 mo.md(r"""81 ## Implementation82 83 Two runs are compared:84 85 1. Static priority: H jobs are inserted as `(0, ...)` and L jobs as `(1, ...)` into a `Queue(priority=True)`. The server always picks the smallest key first, so H jobs are always served before L jobs.86 2. Aging: an `Ager` process wakes up every `AGING_INTERVAL` time units, inspects waiting L jobs, and promotes sufficiently old ones by reducing their priority key until it falls below the H threshold and they move into the server's feed queue.87 """)88 return89 90 91@app.cell(hide_code=True)92def _(mo):93 sim_time_slider = mo.ui.slider(94 start=0,95 stop=100_000,96 step=1_000,97 value=20_000,98 label="Simulation time",99 )100 101 aging_threshold_slider = mo.ui.slider(102 start=1.0,103 stop=60.0,104 step=1.0,105 value=15.0,106 label="Aging threshold",107 )108 109 seed_input = mo.ui.number(110 value=192,111 step=1,112 label="Random seed",113 )114 115 run_button = mo.ui.run_button(label="Run simulation")116 117 mo.vstack([118 sim_time_slider,119 aging_threshold_slider,120 seed_input,121 run_button,122 ])123 return aging_threshold_slider, seed_input, sim_time_slider124 125 126@app.cell127def _(aging_threshold_slider, seed_input, sim_time_slider):128 SIM_TIME = int(sim_time_slider.value)129 AGING_THRESHOLD = float(aging_threshold_slider.value)130 SEED = int(seed_input.value)131 SERVICE_RATE_HI = 2.0132 SERVICE_RATE_LO = 1.0133 ARRIVAL_RATE_LO = 0.2134 return (135 AGING_THRESHOLD,136 ARRIVAL_RATE_LO,137 SEED,138 SERVICE_RATE_HI,139 SERVICE_RATE_LO,140 SIM_TIME,141 )142 143 144@app.cell145def _(Process):146 class StaticPriorityServer(Process):147 def init(self, hi_q, lo_q, sojourn_hi, sojourn_lo):148 self.hi_q = hi_q149 self.lo_q = lo_q150 self.sojourn_hi = sojourn_hi151 self.sojourn_lo = sojourn_lo152 153 async def _serve(self, arrival, svc, record):154 await self.timeout(svc)155 record.append(self.now - arrival)156 157 async def run(self):158 while True:159 if not self.hi_q.is_empty():160 arrival, svc = await self.hi_q.get()161 await self._serve(arrival, svc, self.sojourn_hi)162 elif not self.lo_q.is_empty():163 arrival, svc = await self.lo_q.get()164 await self._serve(arrival, svc, self.sojourn_lo)165 else:166 await self.timeout(0.01)167 168 return (StaticPriorityServer,)169 170 171@app.cell172def _(AGING_THRESHOLD, Process):173 class AgingServer(Process):174 def init(self, hi_q, lo_q, sojourn_hi, sojourn_lo):175 self.hi_q = hi_q176 self.lo_q = lo_q177 self.sojourn_hi = sojourn_hi178 self.sojourn_lo = sojourn_lo179 180 async def run(self):181 while True:182 lo_aged = (183 not self.lo_q.is_empty()184 and self.now - self.lo_q._items[0][0] >= AGING_THRESHOLD185 )186 if lo_aged:187 arrival, svc = await self.lo_q.get()188 await self.timeout(svc)189 self.sojourn_lo.append(self.now - arrival)190 elif not self.hi_q.is_empty():191 arrival, svc = await self.hi_q.get()192 await self.timeout(svc)193 self.sojourn_hi.append(self.now - arrival)194 elif not self.lo_q.is_empty():195 arrival, svc = await self.lo_q.get()196 await self.timeout(svc)197 self.sojourn_lo.append(self.now - arrival)198 else:199 await self.timeout(0.01)200 201 return (AgingServer,)202 203 204@app.cell205def _(Process, SERVICE_RATE_HI, random):206 class HiSource(Process):207 def init(self, rate, q):208 self.rate = rate209 self.q = q210 211 async def run(self):212 while True:213 await self.timeout(random.expovariate(self.rate))214 svc = random.expovariate(SERVICE_RATE_HI)215 await self.q.put((self.now, svc))216 217 return (HiSource,)218 219 220@app.cell221def _(ARRIVAL_RATE_LO, Process, SERVICE_RATE_LO, random):222 class LoSource(Process):223 def init(self, q):224 self.q = q225 226 async def run(self):227 while True:228 await self.timeout(random.expovariate(ARRIVAL_RATE_LO))229 svc = random.expovariate(SERVICE_RATE_LO)230 await self.q.put((self.now, svc))231 232 return (LoSource,)233 234 235@app.cell236def _(237 AgingServer,238 Environment,239 HiSource,240 LoSource,241 Queue,242 SIM_TIME,243 StaticPriorityServer,244 statistics,245):246 def simulate(arrival_rate_hi, use_aging):247 env = Environment()248 hi_q = Queue(env)249 lo_q = Queue(env)250 sojourn_hi = []251 sojourn_lo = []252 HiSource(env, arrival_rate_hi, hi_q)253 LoSource(env, lo_q)254 if use_aging:255 AgingServer(env, hi_q, lo_q, sojourn_hi, sojourn_lo)256 else:257 StaticPriorityServer(env, hi_q, lo_q, sojourn_hi, sojourn_lo)258 env.run(until=SIM_TIME)259 return sojourn_hi, sojourn_lo260 261 def mean_or_none(lst):262 return statistics.mean(lst) if lst else None263 264 def pct_or_none(lst, p):265 if not lst:266 return None267 return sorted(lst)[int(p * len(lst))]268 269 return mean_or_none, pct_or_none, simulate270 271 272@app.cell273def _(274 ARRIVAL_RATE_LO,275 SEED,276 SERVICE_RATE_HI,277 SERVICE_RATE_LO,278 mean_or_none,279 pl,280 random,281 simulate,282):283 def sweep():284 sweep_rows = []285 for rho_hi in [0.10, 0.20, 0.40, 0.60, 0.70, 0.80]:286 rate_hi = rho_hi * SERVICE_RATE_HI287 hi, lo = simulate(rate_hi, use_aging=False)288 rho_total = rho_hi + ARRIVAL_RATE_LO / SERVICE_RATE_LO289 sweep_rows.append({290 "rho_hi": rho_hi,291 "rho_total": rho_total,292 "mean_W_hi": mean_or_none(hi),293 "mean_W_lo": mean_or_none(lo),294 })295 return pl.DataFrame(sweep_rows)296 297 random.seed(SEED)298 df_sweep = sweep()299 return (df_sweep,)300 301 302@app.cell303def _(304 ARRIVAL_RATE_LO,305 SERVICE_RATE_HI,306 SERVICE_RATE_LO,307 mean_or_none,308 pct_or_none,309 pl,310 simulate,311):312 FIXED_RHO_HI = 0.70313 rho_total = FIXED_RHO_HI + ARRIVAL_RATE_LO / SERVICE_RATE_LO314 def compare():315 rate_hi = FIXED_RHO_HI * SERVICE_RATE_HI316 hi_static, lo_static = simulate(rate_hi, use_aging=False)317 hi_aging, lo_aging = simulate(rate_hi, use_aging=True)318 319 compare_rows = [320 {321 "policy": "static", "class": "hi", "n": len(hi_static),322 "mean_W": mean_or_none(hi_static),323 "p95": pct_or_none(hi_static, 0.95),324 "p99": pct_or_none(hi_static, 0.99),325 },326 {327 "policy": "static", "class": "lo", "n": len(lo_static),328 "mean_W": mean_or_none(lo_static),329 "p95": pct_or_none(lo_static, 0.95),330 "p99": pct_or_none(lo_static, 0.99),331 },332 {333 "policy": "aging", "class": "hi", "n": len(hi_aging),334 "mean_W": mean_or_none(hi_aging),335 "p95": pct_or_none(hi_aging, 0.95),336 "p99": pct_or_none(hi_aging, 0.99),337 },338 {339 "policy": "aging", "class": "lo", "n": len(lo_aging),340 "mean_W": mean_or_none(lo_aging),341 "p95": pct_or_none(lo_aging, 0.95),342 "p99": pct_or_none(lo_aging, 0.99),343 },344 ]345 return pl.DataFrame(compare_rows)346 347 df_compare = compare()348 return (df_compare, FIXED_RHO_HI, rho_total,)349 350 351@app.cell(hide_code=True)352def _(AGING_THRESHOLD, ARRIVAL_RATE_LO, SERVICE_RATE_LO, mo):353 mo.md(f"""354 ## Part 1 — Static Priority: Effect of Hi-Priority Load on Lo-Priority Wait355 356 Lo-priority: arrival rate {ARRIVAL_RATE_LO}, mean service {1 / SERVICE_RATE_LO:.1f},357 ρ_lo = {ARRIVAL_RATE_LO / SERVICE_RATE_LO:.2f}358 359 Aging threshold: {AGING_THRESHOLD} time units360 """)361 return362 363 364@app.cell365def _(df_sweep):366 df_sweep367 return368 369 370@app.cell(hide_code=True)371def _(FIXED_RHO_HI, mo, rho_total):372 mo.md(f"""373 ## Part 2 — Static vs. Aging at ρ_hi = {FIXED_RHO_HI:.2f}, ρ_total = {rho_total:.2f}374 """)375 return376 377 378@app.cell379def _(df_compare):380 df_compare381 return382 383 384@app.cell385def _(alt, df_compare, df_sweep):386 df_plot = df_sweep.unpivot(387 on=["mean_W_hi", "mean_W_lo"],388 index=["rho_hi", "rho_total"],389 variable_name="job_class",390 value_name="mean_W",391 )392 sweep_chart = (393 alt.Chart(df_plot)394 .mark_line(point=True)395 .encode(396 x=alt.X("rho_hi:Q", title="Hi-priority utilization (ρ_hi)"),397 y=alt.Y("mean_W:Q", title="Mean sojourn time (W)"),398 color=alt.Color("job_class:N", title="Job class"),399 tooltip=["rho_hi:Q", "job_class:N", "mean_W:Q"],400 )401 .properties(title="Priority Starvation: Effect of Hi-Priority Load")402 )403 compare_chart = (404 alt.Chart(df_compare)405 .mark_bar()406 .encode(407 x=alt.X("class:N", title="Job class"),408 y=alt.Y("mean_W:Q", title="Mean sojourn time (W)"),409 color=alt.Color("policy:N", title="Policy"),410 xOffset="policy:N",411 tooltip=["policy:N", "class:N", "mean_W:Q", "p99:Q"],412 )413 .properties(title="Static Priority vs. Aging")414 )415 (sweep_chart | compare_chart)416 return417 418 419@app.cell(hide_code=True)420def _(mo):421 mo.md(r"""422 ## Understanding the Math423 424 ### Mean wait for two-priority queues425 426 Let $\lambda_i$, $\mu_i$, and $\rho_i = \lambda_i / \mu_i$ be the arrival rate, service rate, and utilization of class $i \in \{H, L\}$. For a non-preemptive priority queue:427 428 $$W_H = \frac{R_0}{1 - \rho_H}$$429 430 $$W_L = \frac{R_0}{(1 - \rho_H)(1 - \rho_H - \rho_L)}$$431 432 where $R_0 = \tfrac{1}{2}(\lambda_H \overline{s_H^2} + \lambda_L \overline{s_L^2})$ is the mean residual work seen by an arriving customer. The ratio $W_L / W_H = 1/(1 - \rho_H)$ grows without bound as $\rho_H \to 1$.433 434 ### Utilization of each class435 436 Let $\lambda_H$ be the arrival rate of high-priority jobs (H) and $\mu_H$ be their service rate. The utilization contributed by H jobs alone is $\rho_H = \lambda_H / \mu_H$ — the fraction of server time that H jobs would consume if they were the only class. Similarly, $\rho_L = \lambda_L / \mu_L$ for low-priority jobs. The total utilization is $\rho = \rho_H + \rho_L$. Requiring $\rho < 1$ means the server has enough capacity for both classes on average.437 438 ### Why "on average" is not enough439 440 Even when $\rho < 1$, randomness creates bursts of H arrivals. During a burst, the server is continuously occupied by H jobs, and L jobs must wait in the background. The mean wait for low-priority jobs in a non-preemptive priority queue is:441 442 $$W_L = \frac{R_0}{(1 - \rho_H)(1 - \rho_H - \rho_L)}$$443 444 where $R_0$ is the mean residual work in the system when a job arrives. The critical observation is the factor $(1 - \rho_H)$ in the denominator. As $\rho_H \to 1$, this factor approaches zero and $W_L \to \infty$ — even if $\rho_L$ stays small and the total load $\rho$ is comfortably below 1.445 446 ### The trade-off447 448 Without aging, $W_L$ can be infinite when $\rho_H$ is large. With aging, $W_L \leq T_{\max} + 1/\mu_L$, but during promotion events the effective $\rho_H$ spikes temporarily, increasing $W_H$. Choosing $T_{\max}$ is a design decision: a small $T_{\max}$ protects L jobs but forces more promotions and penalizes H jobs more often; a large $T_{\max}$ is kinder to H jobs but allows L jobs to wait longer. There is no setting that simultaneously minimizes both — the trade-off is fundamental.449 """)450 return451 452 453if __name__ == "__main__":454 app.run()455