SimPy Interrupts Explained: Handling the Unexpected

Published on February 09, 2026

Real systems have interruptions. Machines break. Priorities change. Shifts end. SimPy lets you model all of this with interrupts.

What Is an Interrupt?

An interrupt forces a waiting process to stop waiting and handle the interruption:

import simpy

def worker(env):
    try:
        print(f"Working at {env.now}")
        yield env.timeout(100)  # Long task
        print(f"Finished at {env.now}")
    except simpy.Interrupt as interrupt:
        print(f"Interrupted at {env.now}: {interrupt.cause}")

def interrupter(env, worker_process):
    yield env.timeout(10)
    worker_process.interrupt("Break time!")

env = simpy.Environment()
worker_proc = env.process(worker(env))
env.process(interrupter(env, worker_proc))
env.run()

Output:

Working at 0
Interrupted at 10: Break time!

When to Use Interrupts

Interrupts model: - Machine breakdowns - Failure during operation - Preemption - Higher priority takes over - Cancellation - Customer leaves, order cancelled - Timeouts - Giving up after waiting too long - Shift changes - Worker goes home

Basic Interrupt Pattern

def interruptible_process(env):
    while True:
        try:
            print(f"Starting task at {env.now}")
            yield env.timeout(10)
            print(f"Completed task at {env.now}")
        except simpy.Interrupt:
            print(f"Task interrupted at {env.now}")

Always wrap potentially-interrupted waits in try/except.

Interrupt with Cause

Pass information about why the interrupt happened:

def maintenance(env, machine_process):
    yield env.timeout(50)
    machine_process.interrupt(cause={
        'type': 'maintenance',
        'duration': 20
    })

def machine(env):
    try:
        yield env.timeout(100)
    except simpy.Interrupt as i:
        if i.cause['type'] == 'maintenance':
            print(f"Maintenance for {i.cause['duration']} units")
            yield env.timeout(i.cause['duration'])

Resuming After Interrupt

Continue where you left off:

def resilient_worker(env, total_work):
    remaining = total_work

    while remaining > 0:
        try:
            start = env.now
            yield env.timeout(remaining)
            remaining = 0  # Completed
            print(f"Work complete at {env.now}")
        except simpy.Interrupt:
            elapsed = env.now - start
            remaining -= elapsed
            print(f"Interrupted. {remaining} remaining")

Machine Breakdown Example

import random

def machine(env, name, repair_time_mean):
    """Machine that occasionally breaks down."""
    while True:
        # Working
        try:
            processing_time = random.expovariate(1/10)
            print(f"{name} processing for {processing_time:.1f}")
            yield env.timeout(processing_time)
            print(f"{name} completed part at {env.now:.1f}")
        except simpy.Interrupt:
            print(f"{name} broke down at {env.now:.1f}")
            repair_time = random.expovariate(1/repair_time_mean)
            yield env.timeout(repair_time)
            print(f"{name} repaired at {env.now:.1f}")

def breakdown_generator(env, machine_process, mean_time_to_fail):
    """Generates breakdowns at random intervals."""
    while True:
        yield env.timeout(random.expovariate(1/mean_time_to_fail))
        if not machine_process.is_alive:
            continue  # Machine process ended
        machine_process.interrupt("Breakdown!")

env = simpy.Environment()
machine_proc = env.process(machine(env, "Lathe", repair_time_mean=5))
env.process(breakdown_generator(env, machine_proc, mean_time_to_fail=30))
env.run(until=100)

Customer Reneging with Interrupt

Customer gives up waiting:

def customer(env, name, server, patience):
    """Customer who will leave if wait is too long."""
    with server.request() as req:
        # Wait for service or give up
        result = yield req | env.timeout(patience)

        if req in result:
            # Got service
            yield env.timeout(random.expovariate(1/5))
            print(f"{name} served at {env.now:.1f}")
        else:
            # Patience ran out
            print(f"{name} left at {env.now:.1f} (waited too long)")

This uses event composition, not explicit interrupts—but achieves the same goal.

Interrupt vs Event Composition

Two ways to handle conditional waits:

Interrupt approach:

def waiter(env):
    try:
        yield env.timeout(100)
    except simpy.Interrupt:
        print("Interrupted!")

def interrupter(env, proc):
    yield env.timeout(10)
    proc.interrupt()

Event composition approach:

def process(env):
    condition = env.timeout(10)
    long_wait = env.timeout(100)

    result = yield condition | long_wait
    if condition in result:
        print("Condition met first")

Use interrupts when: - Another process decides to interrupt - You need to model external intervention

Use event composition when: - The process itself decides based on conditions - Simpler logic suffices

Checking If Process Is Alive

Before interrupting:

if process.is_alive:
    process.interrupt("cause")

Interrupting a dead process raises an error.

Nested Try/Except

Handle multiple interrupt sources:

def complex_worker(env):
    while True:
        try:
            yield env.timeout(10)
        except simpy.Interrupt as i:
            if i.cause == 'break':
                yield env.timeout(5)  # Take break
            elif i.cause == 'emergency':
                yield env.timeout(20)  # Handle emergency
            elif i.cause == 'shutdown':
                return  # Exit completely

Interrupt Timing

Interrupts occur immediately when called:

def printer(env):
    try:
        yield env.timeout(10)
    except simpy.Interrupt:
        print(f"Interrupted at exactly {env.now}")

def interrupter(env, proc):
    yield env.timeout(5.5)
    proc.interrupt()

# Interrupt happens at exactly t=5.5

Common Pitfalls

Not Handling Interrupts

# Bad - unhandled interrupt crashes simulation
def fragile(env):
    yield env.timeout(100)

# Good - always handle if interrupts possible
def robust(env):
    try:
        yield env.timeout(100)
    except simpy.Interrupt:
        pass  # Handle appropriately

Interrupting Dead Process

# Bad - raises RuntimeError
proc.interrupt()  # But proc already finished

# Good - check first
if proc.is_alive:
    proc.interrupt()

Infinite Interrupt Loop

# Bad - immediately re-interrupted forever
def bad_handler(env):
    while True:
        try:
            yield env.timeout(10)
        except simpy.Interrupt:
            pass  # Should yield something!

# Good - yield after handling
def good_handler(env):
    while True:
        try:
            yield env.timeout(10)
        except simpy.Interrupt:
            yield env.timeout(1)  # Recovery time

Summary

Interrupts: - Model external interventions - Use process.interrupt(cause) to trigger - Handle with try/except simpy.Interrupt - Check process.is_alive before interrupting - Can include information via cause

When the unexpected happens, interrupt.

Next Steps

Discover the Power of Simulation

Want to become a go-to expert in simulation with Python? The Complete Simulation Bootcamp will show you how simulation can transform your career and your projects.

Explore the Bootcamp