---

name: simpy-discrete-event-simulation description: "Process-based discrete-event simulation. Model queues, shared resources, timed events: manufacturing, service ops, network traffic, logistics. Processes are Python generators yielding events. Resources: capacity-limited (Resource/Priority/Preemptive), bulk (Container), objects (Store, FilterStore). For continuous use SciPy ODEs; for agent-based use Mesa." license: MIT

SimPy — Discrete-Event Simulation

Overview

SimPy is a process-based discrete-event simulation framework using standard Python generators. Model systems where entities (customers, vehicles, packets) interact with shared resources (servers, machines, bandwidth) over time, with event-driven scheduling and optional real-time synchronization.

When to Use

• Modeling queue-based systems with resource contention (servers, machines, staff)
• Manufacturing process simulation (production lines, scheduling, bottleneck analysis)
• Network simulation (packet routing, bandwidth allocation, latency analysis)
• Capacity planning (determining optimal resource levels for target throughput)
• Healthcare operations (ER patient flow, staff allocation, bed management)
• Logistics and transportation (warehouse operations, vehicle routing)
• For continuous-time ODE systems → use SciPy solve_ivp
• For agent-based modeling → use Mesa

Prerequisites

# pip install simpy
import simpy
import random

Quick Start

import simpy
import random

def customer(env, name, server):
    """Customer arrives, waits for server, gets served, departs."""
    arrival = env.now
    with server.request() as req:
        yield req  # Wait in queue
        wait = env.now - arrival
        yield env.timeout(random.expovariate(1/3))  # Service time
        print(f'{name}: waited {wait:.1f}, served at {env.now:.1f}')

def arrivals(env, server):
    for i in range(20):
        yield env.timeout(random.expovariate(1/2))  # Inter-arrival
        env.process(customer(env, f'C{i}', server))

env = simpy.Environment()
server = simpy.Resource(env, capacity=2)
env.process(arrivals(env, server))
env.run(until=50)

Core API

1. Environment & Processes

import simpy

# Standard environment
env = simpy.Environment(initial_time=0)

# Processes are Python generators that yield events
def machine(env, name, repair_time):
    while True:
        yield env.timeout(random.expovariate(1/10))  # Time to failure
        print(f'{name} broke at {env.now:.1f}')
        yield env.timeout(repair_time)
        print(f'{name} repaired at {env.now:.1f}')

# Start processes — returns a Process event
proc = env.process(machine(env, 'Machine-1', repair_time=2))

# Run until time limit or no events remain
env.run(until=100)
# env.run()  # Run until no more events

# Current simulation time
print(f'Final time: {env.now}')

# Processes can return values and be awaited
def subtask(env, duration):
    yield env.timeout(duration)
    return f'completed in {duration}'

def main_task(env):
    # Sequential: wait for one process
    result = yield env.process(subtask(env, 5))
    print(f'Subtask {result} at {env.now}')

    # Parallel: wait for ALL (AllOf)
    t1 = env.process(subtask(env, 3))
    t2 = env.process(subtask(env, 4))
    results = yield t1 & t2  # AllOf — resumes when both done
    print(f'Both done at {env.now}')

    # Race: wait for ANY (AnyOf)
    t3 = env.process(subtask(env, 2))
    t4 = env.process(subtask(env, 6))
    result = yield t3 | t4  # AnyOf — resumes when first completes
    print(f'First done at {env.now}')

env = simpy.Environment()
env.process(main_task(env))
env.run()

2. Resources

import simpy

env = simpy.Environment()

# Basic resource — capacity-limited (e.g., 2 servers)
server = simpy.Resource(env, capacity=2)
print(f'Capacity: {server.capacity}, In use: {server.count}, Queue: {len(server.queue)}')

# Priority resource — lower number = higher priority
priority_server = simpy.PriorityResource(env, capacity=1)

def vip_customer(env, res):
    with res.request(priority=1) as req:  # Higher priority
        yield req
        yield env.timeout(3)

def regular_customer(env, res):
    with res.request(priority=10) as req:  # Lower priority
        yield req
        yield env.timeout(3)

# Preemptive resource — high priority interrupts low priority
preemptive = simpy.PreemptiveResource(env, capacity=1)

def urgent_job(env, res):
    with res.request(priority=0, preempt=True) as req:
        yield req  # May interrupt current user
        yield env.timeout(1)

# Container — bulk material (fuel, water, inventory)
tank = simpy.Container(env, capacity=100, init=50)

def refuel(env, tank):
    yield tank.put(30)  # Add 30 units
    print(f'Tank level: {tank.level}/{tank.capacity}')

def consume(env, tank):
    yield tank.get(20)  # Remove 20 units
    print(f'Tank level: {tank.level}/{tank.capacity}')

# Store — FIFO object storage
warehouse = simpy.Store(env, capacity=10)

def producer(env, store):
    for i in range(5):
        yield env.timeout(2)
        yield store.put(f'Item-{i}')

def consumer(env, store):
    while True:
        item = yield store.get()
        print(f'Got {item} at {env.now}')
        yield env.timeout(3)

# FilterStore — selective retrieval
parts = simpy.FilterStore(env, capacity=20)

def picker(env, store):
    # Get specific item matching condition
    item = yield store.get(lambda x: x['color'] == 'red')
    print(f'Found red item: {item}')

3. Events & Synchronization

import simpy

env = simpy.Environment()

# Basic event — manual trigger for signaling between processes
signal = env.event()

def waiter(env, event):
    print(f'Waiting at {env.now}')
    value = yield event  # Blocks until triggered
    print(f'Got signal "{value}" at {env.now}')

def sender(env, event):
    yield env.timeout(5)
    event.succeed(value='go')  # Trigger with value

env.process(waiter(env, signal))
env.process(sender(env, signal))
env.run()
# Output: Waiting at 0, Got signal "go" at 5

# Timeout — most common event
yield env.timeout(delay=5)

# Process interruption
def interruptible(env, name):
    try:
        yield env.timeout(10)
    except simpy.Interrupt as interrupt:
        print(f'{name} interrupted: {interrupt.cause} at {env.now}')

def interruptor(env, proc):
    yield env.timeout(3)
    proc.interrupt('maintenance')

proc = env.process(interruptible(env, 'Worker'))
env.process(interruptor(env, proc))

# Barrier synchronization — wait for N processes
class Barrier:
    def __init__(self, env, n):
        self.env = env
        self.n = n
        self.count = 0
        self.event = env.event()

    def wait(self):
        self.count += 1
        if self.count >= self.n:
            self.event.succeed()
        return self.event

def phase_worker(env, name, barrier):
    yield env.timeout(random.uniform(1, 5))  # Phase work
    print(f'{name} reached barrier at {env.now:.1f}')
    yield barrier.wait()  # Wait for all workers
    print(f'{name} passed barrier at {env.now:.1f}')

env = simpy.Environment()
barrier = Barrier(env, n=3)
for i in range(3):
    env.process(phase_worker(env, f'W{i}', barrier))
env.run()

4. Monitoring & Statistics

import simpy

# Inline statistics collection
class Stats:
    def __init__(self):
        self.wait_times = []
        self.queue_lengths = []

    def report(self):
        if self.wait_times:
            avg_wait = sum(self.wait_times) / len(self.wait_times)
            max_wait = max(self.wait_times)
            print(f'Avg wait: {avg_wait:.2f}, Max wait: {max_wait:.2f}')
            print(f'Customers served: {len(self.wait_times)}')

def customer(env, name, server, stats):
    arrival = env.now
    with server.request() as req:
        yield req
        wait = env.now - arrival
        stats.wait_times.append(wait)
        stats.queue_lengths.append(len(server.queue))
        yield env.timeout(random.expovariate(1/3))

env = simpy.Environment()
server = simpy.Resource(env, capacity=2)
stats = Stats()

def gen(env, server, stats):
    for i in range(100):
        yield env.timeout(random.expovariate(1/2))
        env.process(customer(env, f'C{i}', server, stats))

env.process(gen(env, server, stats))
env.run(until=200)
stats.report()

# Resource monitoring via monkey-patching
def patch_resource(resource, data):
    """Patch resource to log request/release events."""
    original_request = resource.request
    original_release = resource.release

    def monitored_request(*args, **kwargs):
        req = original_request(*args, **kwargs)
        data.append((resource._env.now, 'request', resource.count, len(resource.queue)))
        return req

    def monitored_release(*args, **kwargs):
        result = original_release(*args, **kwargs)
        data.append((resource._env.now, 'release', resource.count, len(resource.queue)))
        return result

    resource.request = monitored_request
    resource.release = monitored_release

log = []
patch_resource(server, log)
# After simulation: analyze log for utilization, queue dynamics

5. Real-Time Simulation

import simpy.rt

# Real-time environment — synchronized with wall clock
env = simpy.rt.RealtimeEnvironment(factor=1.0)  # 1 sim unit = 1 second
# factor=0.1 → 10x faster (1 sim unit = 0.1 seconds)
# factor=60 → 1 sim unit = 1 minute

# Strict mode raises RuntimeError if simulation can't keep up
env_strict = simpy.rt.RealtimeEnvironment(factor=1.0, strict=True)

# Non-strict mode (default) allows slower-than-real-time execution
env_relaxed = simpy.rt.RealtimeEnvironment(factor=1.0, strict=False)

def periodic_task(env, interval):
    while True:
        print(f'Tick at sim time {env.now:.1f}')
        yield env.timeout(interval)

env = simpy.rt.RealtimeEnvironment(factor=1.0)
env.process(periodic_task(env, 2.0))
env.run(until=10)
# Prints "Tick" every ~2 real seconds

Key Concepts

Resource Selection Guide

| Need | Resource Type | Key Feature | |------|--------------|-------------| | Limited servers/machines | Resource | FIFO queue, capacity limit | | Priority queuing | PriorityResource | Lower number = higher priority | | Preemptive scheduling | PreemptiveResource | High priority interrupts current user | | Bulk material (fuel, water) | Container | put(amount) / get(amount), continuous level | | Object queue (FIFO) | Store | put(item) / get(), ordered retrieval | | Conditional retrieval | FilterStore | get(lambda x: condition) | | Priority-ordered items | PriorityStore | Items sorted by priority |

Process Interaction Mechanisms

| Mechanism | Use When | Code Pattern | |-----------|----------|-------------| | Event signaling | Broadcast to multiple waiters | event = env.event() → yield event / event.succeed() | | Process yield | Sequential or parallel execution | yield env.process(func()) or yield p1 & p2 | | Interruption | Preemption, maintenance, cancellation | proc.interrupt(cause) + try/except simpy.Interrupt | | Timeout racing | Timeout with cancellation | yield event | env.timeout(limit) |

Common Workflows

1. Manufacturing Line Simulation

import simpy
import random

def part(env, name, machines, buffer, stats):
    """Part flows through sequential machines with intermediate buffer."""
    for i, machine in enumerate(machines):
        with machine.request() as req:
            yield req
            process_time = random.triangular(1, 3, 2)
            yield env.timeout(process_time)

    if buffer.level < buffer.capacity:
        yield buffer.put(1)
        stats['produced'] += 1

def part_generator(env, machines, buffer, stats):
    i = 0
    while True:
        yield env.timeout(random.expovariate(1/2))
        env.process(part(env, f'Part-{i}', machines, buffer, stats))
        i += 1

random.seed(42)
env = simpy.Environment(