Common Heap Patterns & Problem Types

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

import heapq
from dataclasses import dataclass, field
from typing import Callable, Any
from abc import ABC, abstractmethod
 
@dataclass(order=True)
class Event:
    """A simulation event."""
    time: float
    priority: int = 0  # Tie-breaker for same-time events
    handler: Callable = field(compare=False)
    data: Any = field(default=None, compare=False)
 
class Simulation:
    """
    Core discrete event simulation engine.
    
    The event queue is a min-heap ordered by (time, priority).
    Each event handler can schedule future events.
    """
    
    def __init__(self):
        self.clock = 0.0
        self.event_queue: list[Event] = []
        self.running = False
    
    def schedule(self, delay: float, handler: Callable, 
                 data: Any = None, priority: int = 0) -> Event:
        """Schedule event to occur after delay from current time."""
        event = Event(
            time=self.clock + delay,
            priority=priority,
            handler=handler,
            data=data
        )
        heapq.heappush(self.event_queue, event)
        return event
    
    def schedule_at(self, time: float, handler: Callable,
                    data: Any = None, priority: int = 0) -> Event:
        """Schedule event at absolute time."""
        event = Event(time=time, priority=priority, handler=handler, data=data)
        heapq.heappush(self.event_queue, event)
        return event
    
    def run(self, until: float = float('inf')) -> None:
        """Run simulation until time limit or no more events."""
        self.running = True
        
        while self.running and self.event_queue:
            event = heapq.heappop(self.event_queue)
            
            if event.time > until:
                heapq.heappush(self.event_queue, event)
                break
            
            self.clock = event.time
            event.handler(self, event.data)
        
        self.running = False
    
    def stop(self) -> None:
        """Stop simulation."""
        self.running = False

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

import random
from collections import deque
 
class MM1Queue:
    """
    M/M/1 Queue Simulation
    
    - Poisson arrivals (exponential inter-arrival times)
    - Exponential service times
    - Single server
    
    Uses discrete event simulation with heap-based event queue.
    """
    
    def __init__(self, arrival_rate: float, service_rate: float):
        self.arrival_rate = arrival_rate
        self.service_rate = service_rate
        self.sim = Simulation()
        
        # State
        self.queue = deque()
        self.server_busy = False
        self.customers_served = 0
        
        # Statistics
        self.total_wait_time = 0.0
        self.total_system_time = 0.0
    
    def run(self, duration: float) -> dict:
        """Run simulation and return statistics."""
        # Schedule first arrival
        self.sim.schedule(
            self._exp_random(self.arrival_rate),
            self._handle_arrival
        )
        
        self.sim.run(until=duration)
        
        return {
            'customers_served': self.customers_served,
            'avg_wait': self.total_wait_time / max(1, self.customers_served),
            'avg_system_time': self.total_system_time / max(1, self.customers_served),
            'utilization': self.service_rate / self.arrival_rate if self.arrival_rate else 0
        }
    
    def _handle_arrival(self, sim: Simulation, data):
        """Customer arrives."""
        arrival_time = sim.clock
        
        if not self.server_busy:
            # Start service immediately
            self.server_busy = True
            service_time = self._exp_random(self.service_rate)
            sim.schedule(service_time, self._handle_departure, arrival_time)
        else:
            # Join queue
            self.queue.append(arrival_time)
        
        # Schedule next arrival
        sim.schedule(
            self._exp_random(self.arrival_rate),
            self._handle_arrival
        )
    
    def _handle_departure(self, sim: Simulation, arrival_time: float):
        """Customer departs after service."""
        wait_time = sim.clock - arrival_time - self._exp_random(self.service_rate)
        self.total_wait_time += max(0, wait_time)
        self.total_system_time += sim.clock - arrival_time
        self.customers_served += 1
        
        if self.queue:
            # Serve next customer
            next_arrival = self.queue.popleft()
            service_time = self._exp_random(self.service_rate)
            sim.schedule(service_time, self._handle_departure, next_arrival)
        else:
            self.server_busy = False
    
    def _exp_random(self, rate: float) -> float:
        """Exponential random variable."""
        return random.expovariate(rate)
 
 
# Run simulation
queue = MM1Queue(arrival_rate=0.8, service_rate=1.0)
stats = queue.run(duration=10000)
print(f"Customers served: {stats['customers_served']}")
print(f"Average wait time: {stats['avg_wait']:.2f}")
print(f"Average system time: {stats['avg_system_time']:.2f}")

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

from dataclasses import dataclass
from typing import Dict, List
import random
 
@dataclass
class Packet:
    id: int
    source: str
    destination: str
    size: int  # bytes
    created_at: float
 
class NetworkSimulator:
    """
    Simple network simulation with routers and links.
    
    Events: packet_arrival, packet_departure, link_free
    Demonstrates: DES for network modeling
    """
    
    def __init__(self):
        self.sim = Simulation()
        self.links: Dict[tuple, dict] = {}  # (from, to) -> {bandwidth, delay, queue}
        self.delivered = []
        self.packet_id = 0
    
    def add_link(self, from_node: str, to_node: str, 
                 bandwidth: float, delay: float):
        """Add directional link between nodes."""
        self.links[(from_node, to_node)] = {
            'bandwidth': bandwidth,  # bytes/sec
            'delay': delay,          # seconds
            'queue': [],
            'busy': False
        }
    
    def send_packet(self, source: str, dest: str, size: int = 1000):
        """Create and send a packet."""
        self.packet_id += 1
        packet = Packet(self.packet_id, source, dest, size, self.sim.clock)
        
        if (source, dest) in self.links:
            self._enqueue_packet(source, dest, packet)
    
    def _enqueue_packet(self, from_node: str, to_node: str, packet: Packet):
        """Add packet to link queue, start transmission if idle."""
        link = self.links[(from_node, to_node)]
        link['queue'].append(packet)
        
        if not link['busy']:
            self._start_transmission(from_node, to_node)
    
    def _start_transmission(self, from_node: str, to_node: str):
        """Start transmitting next packet on link."""
        link = self.links[(from_node, to_node)]
        if not link['queue']:
            link['busy'] = False
            return
        
        link['busy'] = True
        packet = link['queue'].pop(0)
        
        # Transmission time = size / bandwidth
        tx_time = packet.size / link['bandwidth']
        total_delay = tx_time + link['delay']
        
        self.sim.schedule(total_delay, self._handle_arrival, 
                         (to_node, packet))
    
    def _handle_arrival(self, sim: Simulation, data):
        """Packet arrives at destination."""
        node, packet = data
        
        if node == packet.destination:
            latency = sim.clock - packet.created_at
            self.delivered.append((packet, latency))
        # Could route to next hop here for multi-hop networks
    
    def run(self, duration: float) -> dict:
        self.sim.run(until=duration)
        
        if self.delivered:
            latencies = [lat for _, lat in self.delivered]
            return {
                'packets_delivered': len(self.delivered),
                'avg_latency': sum(latencies) / len(latencies),
                'max_latency': max(latencies)
            }
        return {'packets_delivered': 0}