Elevator System
Multiple elevators with pluggable scheduling. Each elevator runs its own state machine while a central dispatcher assigns requests using the LOOK algorithm to prevent starvation.
Key Abstractions
Facade that accepts floor requests and dispatches them to the best elevator
Manages its own state, current floor, direction, and pending stop queue
Captures a floor number and desired direction from a hall call
Strategy for assigning incoming requests to elevators. Swappable.
Enum: IDLE, MOVING_UP, MOVING_DOWN, DOOR_OPEN. Governs valid transitions.
Class Diagram
Why This Design
Elevator scheduling is a classic problem because it forces you to separate "who decides" from "who moves." A naive approach puts all the logic in one place: the elevator itself figures out where to go next. That works fine with a single elevator, but the moment you add a second one, you need coordination. Two elevators shouldn't both rush to floor 7 when one could handle floor 3 instead.
Splitting the scheduler from the elevator gives you two independently testable units. The elevator is a simple state machine. It knows its current floor, its direction, and its queue of stops. That's it. The scheduler looks at all elevators and picks the best one for each new request. Swap the scheduler, and every elevator benefits without changing a line of elevator code.
Requirements
Functional
- Accept hall calls from any floor with a direction (up or down)
- Dispatch requests to the best available elevator
- Each elevator services its stops in LOOK order (sweep up, then sweep down)
- Support multiple concurrent elevators operating independently
Non-Functional
- Scheduling must be O(E) where E is the number of elevators, not O(F) where F is floors
- Thread-safe stop queues since requests arrive from multiple sources
- Scheduling algorithm must be swappable without modifying elevator internals
Design Decisions
Why LOOK over simple nearest-elevator?
Nearest-elevator sounds reasonable until you picture this: elevator at floor 1, requests at floors 2, 8, 3, 9. Nearest keeps flip-flopping between close floors and never reaches 8 or 9. LOOK commits to a direction and sweeps. Floors 2, 3, 8, 9 all get served in one pass. Starvation disappears because every floor in the sweep direction gets visited before reversing.
Why each elevator is independent with its own lock?
If elevators shared a global lock, elevator 1 opening its doors would block elevator 2 from processing its queue. Independent locks mean true parallelism. Each elevator runs its state machine without waiting on others. The only shared coordination point is the scheduler, which only runs briefly when a new request arrives.
Why separate Scheduler from Elevator?
Single responsibility. An elevator shouldn't care about other elevators. It shouldn't know how many exist or where they are. Its job is to move between floors and open doors. Assignment logic belongs in a component that has visibility across all elevators. This also makes testing straightforward: unit test elevators with hardcoded stops, unit test schedulers with mock elevator states.
Why TreeSet for the stop queues?
Sorted order matters. When moving up, you want the next stop above you, not a random one from the set. TreeSet gives O(log n) insertion and removal with guaranteed ordering. Two sets (up and down) let the elevator grab the lowest up-stop or highest down-stop without filtering.
Interview Follow-ups
- "How would you handle emergency stops?" Add an EMERGENCY state to the enum. When triggered, the elevator stops at the nearest floor, opens doors, and ignores all pending stops until reset. The scheduler removes it from the available pool.
- "What about VIP/express elevators?" New scheduler strategy that reserves certain elevators for specific floor ranges. The elevator class stays the same.
- "How would you add weight limits?" Track current passenger count or weight in the Elevator. The scheduler factors remaining capacity into its assignment scoring.
- "What if an elevator breaks down?" The scheduler needs a health check. Mark it OUT_OF_SERVICE, redistribute its pending stops to other elevators, and exclude it from future assignments.
Code Implementation
1 from __future__ import annotations
2 from abc import ABC, abstractmethod
3 from enum import Enum
4 from threading import Lock
5 from sortedcontainers import SortedList
6
7
8 class Direction(Enum):
9 UP = "UP"
10 DOWN = "DOWN"
11
12
13 class ElevatorState(Enum):
14 IDLE = "IDLE"
15 MOVING_UP = "MOVING_UP"
16 MOVING_DOWN = "MOVING_DOWN"
17 DOOR_OPEN = "DOOR_OPEN"
18
19
20 class Request:
21 """A hall call: someone at a floor wants to go in a direction."""
22
23 __slots__ = ("floor", "direction")
24
25 def __init__(self, floor: int, direction: Direction):
26 self.floor = floor
27 self.direction = direction
28
29 def __repr__(self) -> str:
30 return f"Request(floor={self.floor}, dir={self.direction.value})"
31
32
33 class Elevator:
34 """Independent elevator with its own state machine and stop queue."""
35
36 def __init__(self, elevator_id: int, min_floor: int = 0, max_floor: int = 10):
37 self.id = elevator_id
38 self.current_floor = 0
39 self.state = ElevatorState.IDLE
40 self.min_floor = min_floor
41 self.max_floor = max_floor
42 self._up_stops: SortedList = SortedList()
43 self._down_stops: SortedList = SortedList()
44 self._lock = Lock()
45
46 def add_stop(self, floor: int) -> None:
47 with self._lock:
48 if floor > self.current_floor or (
49 floor == self.current_floor and self.state in (ElevatorState.MOVING_UP, ElevatorState.IDLE)
50 ):
51 if floor not in self._up_stops:
52 self._up_stops.add(floor)
53 else:
54 if floor not in self._down_stops:
55 self._down_stops.add(floor)
56
57 def step(self) -> str | None:
58 """Advance one floor tick. Returns a message if something interesting happens."""
59 with self._lock:
60 return self._step_internal()
61
62 def _step_internal(self) -> str | None:
63 if self.state == ElevatorState.IDLE:
64 if self._up_stops:
65 self.state = ElevatorState.MOVING_UP
66 elif self._down_stops:
67 self.state = ElevatorState.MOVING_DOWN
68 else:
69 return None
70
71 if self.state == ElevatorState.MOVING_UP:
72 self.current_floor += 1
73 if self.current_floor in self._up_stops:
74 self._up_stops.remove(self.current_floor)
75 self.state = ElevatorState.DOOR_OPEN
76 return f"Elevator {self.id}: doors open at floor {self.current_floor}"
77 if not self._up_stops:
78 # Reverse direction if we have down stops, otherwise idle
79 if self._down_stops:
80 self.state = ElevatorState.MOVING_DOWN
81 else:
82 self.state = ElevatorState.IDLE
83 return None
84
85 if self.state == ElevatorState.MOVING_DOWN:
86 self.current_floor -= 1
87 if self.current_floor in self._down_stops:
88 self._down_stops.remove(self.current_floor)
89 self.state = ElevatorState.DOOR_OPEN
90 return f"Elevator {self.id}: doors open at floor {self.current_floor}"
91 if not self._down_stops:
92 if self._up_stops:
93 self.state = ElevatorState.MOVING_UP
94 else:
95 self.state = ElevatorState.IDLE
96 return None
97
98 if self.state == ElevatorState.DOOR_OPEN:
99 # Close doors, pick next direction
100 if self._up_stops:
101 self.state = ElevatorState.MOVING_UP
102 elif self._down_stops:
103 self.state = ElevatorState.MOVING_DOWN
104 else:
105 self.state = ElevatorState.IDLE
106 return f"Elevator {self.id}: doors closed at floor {self.current_floor}"
107
108 return None
109
110 def is_idle(self) -> bool:
111 return self.state == ElevatorState.IDLE
112
113 def pending_count(self) -> int:
114 return len(self._up_stops) + len(self._down_stops)
115
116 def __repr__(self) -> str:
117 return (
118 f"Elevator(id={self.id}, floor={self.current_floor}, "
119 f"state={self.state.value}, up={list(self._up_stops)}, down={list(self._down_stops)})"
120 )
121
122
123 class Scheduler(ABC):
124 """Strategy interface for dispatching requests to elevators."""
125
126 @abstractmethod
127 def assign(self, request: Request, elevators: list[Elevator]) -> Elevator: ...
128
129
130 class NearestScheduler(Scheduler):
131 """Picks the closest idle or same-direction elevator."""
132
133 def assign(self, request: Request, elevators: list[Elevator]) -> Elevator:
134 best = None
135 best_distance = float("inf")
136 for elev in elevators:
137 dist = abs(elev.current_floor - request.floor)
138 if elev.is_idle() and dist < best_distance:
139 best = elev
140 best_distance = dist
141 # Fallback: least loaded elevator if none idle
142 if best is None:
143 best = min(elevators, key=lambda e: e.pending_count())
144 return best
145
146
147 class LookScheduler(Scheduler):
148 """LOOK algorithm: prefers an elevator already moving toward the request floor."""
149
150 def assign(self, request: Request, elevators: list[Elevator]) -> Elevator:
151 best = None
152 best_score = float("inf")
153
154 for elev in elevators:
155 dist = abs(elev.current_floor - request.floor)
156 if elev.is_idle():
157 score = dist
158 elif (
159 request.direction == Direction.UP
160 and elev.state == ElevatorState.MOVING_UP
161 and elev.current_floor <= request.floor
162 ):
163 # Moving up and hasn't passed this floor yet
164 score = dist * 0.5
165 elif (
166 request.direction == Direction.DOWN
167 and elev.state == ElevatorState.MOVING_DOWN
168 and elev.current_floor >= request.floor
169 ):
170 score = dist * 0.5
171 else:
172 # Wrong direction or already passed: penalize
173 score = dist * 2 + elev.pending_count()
174
175 if score < best_score:
176 best_score = score
177 best = elev
178
179 return best
180
181
182 class ElevatorSystem:
183 """Facade: accepts hall requests and coordinates multiple elevators."""
184
185 def __init__(self, num_elevators: int = 3, scheduler: Scheduler | None = None):
186 self._elevators = [Elevator(i) for i in range(num_elevators)]
187 self._scheduler = scheduler or LookScheduler()
188
189 def request_elevator(self, floor: int, direction: Direction) -> None:
190 req = Request(floor, direction)
191 chosen = self._scheduler.assign(req, self._elevators)
192 chosen.add_stop(floor)
193 print(f" Dispatched {req} -> Elevator {chosen.id}")
194
195 def step(self) -> list[str]:
196 """Advance all elevators by one tick. Returns list of events."""
197 events = []
198 for elev in self._elevators:
199 event = elev.step()
200 if event:
201 events.append(event)
202 return events
203
204 def run_until_idle(self, max_steps: int = 100) -> None:
205 """Keep stepping until all elevators are idle or we hit the limit."""
206 for _ in range(max_steps):
207 events = self.step()
208 for e in events:
209 print(f" {e}")
210 if all(elev.is_idle() for elev in self._elevators):
211 break
212
213 def status(self) -> None:
214 for elev in self._elevators:
215 print(f" {elev}")
216
217
218 if __name__ == "__main__":
219 print("=== Elevator System Demo ===\n")
220 system = ElevatorSystem(num_elevators=3, scheduler=LookScheduler())
221
222 print("Initial status:")
223 system.status()
224
225 print("\n--- Batch 1: Multiple hall calls ---")
226 system.request_elevator(5, Direction.UP)
227 system.request_elevator(3, Direction.DOWN)
228 system.request_elevator(7, Direction.UP)
229
230 print("\nRunning elevators...")
231 system.run_until_idle()
232
233 print("\nStatus after batch 1:")
234 system.status()
235
236 print("\n--- Batch 2: Concurrent requests ---")
237 system.request_elevator(1, Direction.UP)
238 system.request_elevator(9, Direction.DOWN)
239 system.request_elevator(4, Direction.UP)
240 system.request_elevator(6, Direction.DOWN)
241
242 print("\nRunning elevators...")
243 system.run_until_idle()
244
245 print("\nFinal status:")
246 system.status()
247 print("\nDone.")Common Mistakes
- ✗Putting scheduling logic inside Elevator. Now every elevator duplicates the same decision code, and swapping algorithms means editing every elevator.
- ✗Using a flat list of pending floors without direction awareness. That leads to unnecessary direction changes and wasted trips.
- ✗Not handling concurrent requests. Two people pressing buttons at the same time can corrupt the stop queue without proper locking.
- ✗Forgetting the IDLE state. An elevator with no pending requests should stop moving, not keep sweeping endlessly.
Key Points
- ✓LOOK algorithm over naive nearest-elevator: it services requests in one sweep direction before reversing, preventing starvation of distant floors
- ✓Each elevator is an independent unit with its own lock and stop queue. Parallelism comes naturally.
- ✓Scheduler is decoupled from Elevator. Single responsibility: one picks, the other moves.
- ✓State enum makes transitions explicit. An elevator can't go from DOOR_OPEN to MOVING_DOWN without closing first.