Design an elevator system¶
TL;DR¶
- You build a controller that receives hall calls, picks a car with a pluggable policy, and drives every car one tick at a time so the whole thing is deterministic and testable.
- Three decisions carry the interview: the car owns its state machine (five states, two stop sets), the controller owns the conversation (Mediator: landings and cars never address each other), and dispatch is a Strategy you can swap and measure.
- Patterns that earn their place: State, Strategy, Mediator, Command, Observer. Singleton is discussed and deliberately not used.
Problem statement¶
"Design the software that runs a bank of elevators. A building has N cars and M floors. People press up or down at a landing and press a floor number inside a car; the system decides which car answers each landing call, moves the cars, opens and closes doors, and drives the indicator displays. Cars have a rated capacity, can be stopped in an emergency and taken into maintenance. Make the scheduling policy pluggable and show me how you would compare two policies."
Requirements¶
Functional
- N cars serving M floors, each car starting wherever you place it.
- Hall calls (a floor plus a direction) and cabin requests (a floor pressed inside a car).
- A dispatcher selects the car that answers each hall call; a cabin request needs no decision.
- Movement one floor at a time, with doors that open, dwell and close, and a rated capacity per car.
- Emergency stop and maintenance mode: a car out of service takes no new stops, and the hall calls it still owed are re-dispatched.
- Direction and floor displays that update without polling the cars.
- Pluggable scheduling: FCFS, nearest-car, SCAN/LOOK and destination dispatch, comparable on the same workload.
- A tick-based simulation: no threads sleeping, no wall-clock time.
Non-functional and constraints
- Correct when calls arrive while cars are moving: the same landing button must never be assigned to two cars.
- Deterministic: identical inputs produce identical output, so a scheduling comparison means something.
- In-memory and single-process; the hardware (motors, sensors, door edges) sits behind method calls you could replace with a driver.
Out of scope: fire-service recall codes, physically accurate acceleration curves, building access control, the hardware bus.
Clarifying questions and assumptions¶
| Question to ask | Assumption taken here |
|---|---|
| What is a tick worth? | One floor of travel, one step of the door timer, or one decision. An idle car spends its first tick choosing a direction, as a real one spends it closing up. |
| Can a landing button be pressed twice? | Yes, and the second press is absorbed: the lamp is already lit and the call already has a car. |
| Does the landing panel know where you are going? | Only in the destination-dispatch variant. HallRequest.destination is optional, so three of the four strategies ignore it. |
| What happens when the chosen car fills up? | The passenger stays on the landing and the call is re-dispatched; if every car is full, it is retried on the next tick. |
| Are hall calls and cabin requests the same thing? | Both are commands, but only a hall call needs a dispatch decision. A cabin request goes straight to the car that is already carrying you. |
| Does the interviewer want threads? | Assume the simulation is single-threaded but the API is not: calls can arrive from any thread, so state is locked. |
Core entities and relationships¶
- Elevator — one car: current floor,
ElevatorState,Direction, aDoor, a load, and two stop sets (_up_stops,_down_stops). It owns the lock that makes its own state atomic and exposes an immutableCarStatussnapshot. - Door — a dwell timer rather than a motor:
openstarts it,tickruns it down, an obstruction restarts it. - Floor — a landing with two lamps. Pressing a lit lamp is absorbed here, so the controller never sees a duplicate.
- ElevatorController — the mediator: many cars, many landings, one assignment map, the event log, and the served-call statistics. Built once and injected; the demo builds three of them side by side.
- HallRequest / CabinRequest — button presses as command objects, each carrying the time it was made, so waiting time is measurable rather than guessed.
- DispatchStrategy —
FcfsDispatch,NearestCarDispatch,LookDispatch,DestinationDispatch. Each scoresCarStatussnapshots and names a car. - CarStatus / ServedCall — immutable value objects: the first is what strategies and displays read, the second is one answered call with its wait.
- Display — an observer of car status; SimulationClock — the injected clock the whole package shares.
Multiplicities: controller 1 -> * cars, controller 1 -> * landings, car 1 -> 1 door, controller 1 -> * assignments, one assignment 1 -> 1 car.
Class diagram¶
Structure: the controller, the cars it drives, and the value objects that cross between them.
classDiagram
class ElevatorController {
-_cars: Dict~str,Elevator~
-_assignments: Dict~tuple,AssignedCall~
-_lock: Lock
+hall_call(floor: int, direction: Direction, destination: int) str
+cabin_request(car_id: str, floor: int) str
+tick() int
+emergency_stop(car_id: str) List~str~
+average_wait() float
+total_travel() int
}
class Elevator {
+id: str
+floor: int
+state: ElevatorState
+direction: Direction
-_up_stops: Set~int~
-_down_stops: Set~int~
-_lock: Lock
+add_stop(floor: int, direction: Direction) None
+step() bool
+board(passengers: int) None
+emergency_stop() List~int~
+status() CarStatus
}
class Door {
+state: DoorState
+hold_ticks: int
+open() None
+obstruct() None
+tick() bool
}
class Floor {
+number: int
+lit: Set~Direction~
+press(direction: Direction) bool
+clear(direction: Direction) None
}
class CarStatus {
+car_id: str
+floor: int
+stops: Tuple~int~
+load: int
+is_idle() bool
+distance_to(floor: int) int
}
class AssignedCall {
+request: HallRequest
+car_id: str
}
class ServedCall {
+floor: int
+car_id: str
+wait() float
}
class SimulationClock {
+ticks: int
+now() float
+tick() int
}
class ElevatorState {
<<enumeration>>
IDLE
MOVING_UP
MOVING_DOWN
DOOR_OPEN
MAINTENANCE
}
class Direction {
<<enumeration>>
UP
DOWN
IDLE
}
class DoorState {
<<enumeration>>
CLOSED
OPEN
OBSTRUCTED
}
ElevatorController "1" o-- "many" Elevator
ElevatorController "1" *-- "many" Floor
ElevatorController "1" o-- "many" AssignedCall
ElevatorController --> SimulationClock
ElevatorController ..> ServedCall : records
Elevator "1" *-- "1" Door
Elevator ..> CarStatus : snapshots
Elevator --> ElevatorState
Elevator --> Direction
Door --> DoorStateBehaviour: presses as commands, dispatch as a strategy, displays as observers.
classDiagram
class Request {
<<interface>>
+requested_at: float
+apply(sink: RequestSink) str
+label() str
}
class RequestSink {
<<interface>>
+assign_hall_call(request: HallRequest) str
+assign_cabin_call(request: CabinRequest) str
}
class HallRequest {
+floor: int
+direction: Direction
+destination: int
+key() tuple
}
class CabinRequest {
+car_id: str
+floor: int
}
class DispatchStrategy {
<<interface>>
+name: str
+select(cars: List~CarStatus~, request: HallRequest) str
}
class FcfsDispatch
class NearestCarDispatch
class LookDispatch {
+cost(car: CarStatus, request: HallRequest) int
}
class DestinationDispatch
class CarListener {
<<interface>>
+on_car_changed(status: CarStatus) None
}
class Display {
+render() str
+floor_of(car_id: str) int
}
class ElevatorController
Request <|.. HallRequest
Request <|.. CabinRequest
RequestSink <|.. ElevatorController
HallRequest ..> RequestSink : apply
ElevatorController --> DispatchStrategy
DispatchStrategy <|.. FcfsDispatch
DispatchStrategy <|.. NearestCarDispatch
DispatchStrategy <|.. LookDispatch
DispatchStrategy <|.. DestinationDispatch
DestinationDispatch o-- DispatchStrategy : fallback
CarListener <|.. Display
ElevatorController --> "many" CarListener : notifiesDesign patterns applied¶
| Pattern | Where | Why it earns its place |
|---|---|---|
| State | ElevatorState plus the guarded transitions in Elevator.step |
Five statuses, and the same tick means something different in each. The enum-and-table form is the right one here: the transitions are one line each and the whole machine is nine lines you can read in the room. Classes per state would triple the code without adding a decision. |
| Strategy | DispatchStrategy and its four implementations |
"Now use nearest-car instead" is the follow-up you will be asked, and it is a constructor argument. Because strategies read immutable snapshots, they are pure functions and testable without a building. |
| Mediator | ElevatorController |
Landings and cars never reference each other. With N cars and M landings that is N + M edges through the controller instead of N x M between peers, and the dispatch policy lives in one object. |
| Command | HallRequest, CabinRequest behind the Request protocol |
A press becomes an object with a timestamp, so the controller can log it, replay it, and measure how long it waited. submit logs then delegates; the request itself decides which controller method runs. |
| Observer | CarListener and Display |
Indicators subscribe; the controller pushes statuses after each tick, outside its lock. A second display is one more subscriber. |
| Dependency injection | SimulationClock, strategies, cars |
Nothing calls the wall clock, so 90 ticks run in microseconds and every test is exact. |
What was deliberately not used: Singleton for the controller. A building genuinely has one controller, which is exactly why interviewers expect it — but the demo builds four controllers to compare four strategies, and every test builds its own. One instance created in main and injected gives you the same guarantee without the global.
Key flows¶
A hall call from press to answered, with the display updated on every tick.
sequenceDiagram
autonumber
actor P as Passenger
participant HP as Landing panel 7
participant C as ElevatorController
participant S as LookDispatch
participant B as Elevator B
participant D as Display
P->>HP: press up
HP->>C: hall_call(7, UP, to 11)
C->>C: assignment for (7, UP)? none
C->>S: select(snapshots, request)
S-->>C: car B, cost 2, idle
C->>B: add_stop(7, UP)
loop one tick per floor
C->>B: step()
B-->>C: False, doors stayed shut
C-)D: on_car_changed(status)
end
C->>B: step()
B-->>C: True, doors open at 7
C->>B: board() then add_stop(11)
C->>HP: clear the up lamp
Note over C: ServedCall recorded, wait 3 ticksThe car's lifecycle. Every arrow is one branch of step; the maintenance arrows are the emergency stop.
stateDiagram-v2
[*] --> IDLE
IDLE --> DOOR_OPEN : a stop at this floor
IDLE --> MOVING_UP : work above
IDLE --> MOVING_DOWN : work below
MOVING_UP --> DOOR_OPEN : arrived at a stop
MOVING_UP --> MOVING_DOWN : nothing left above
MOVING_UP --> IDLE : no stops left
MOVING_DOWN --> DOOR_OPEN : arrived at a stop
MOVING_DOWN --> MOVING_UP : nothing left below
MOVING_DOWN --> IDLE : no stops left
DOOR_OPEN --> DOOR_OPEN : obstructed, timer restarts
DOOR_OPEN --> MOVING_UP : closed, work above
DOOR_OPEN --> MOVING_DOWN : closed, work below
DOOR_OPEN --> IDLE : closed, nothing to do
IDLE --> MAINTENANCE : emergency stop
MOVING_UP --> MAINTENANCE : emergency stop
MAINTENANCE --> IDLE : return_to_serviceTwo things to say while you draw it. First, DOOR_OPEN --> DOOR_OPEN is the obstruction: the door restarts its dwell timer instead of closing, which is why a car with someone standing in the doorway holds the floor forever and never loses its queue. Second, there is no arrow from MOVING_UP to MOVING_UP: staying in motion is the absence of a transition, so the diagram shows exactly the decisions the code makes.
Implementation¶
Write it in the order the interviewer wants to see: vocabulary, then the presses, then the entities, then the policies, then the car, then the controller that ties them together.
The five car states and three door states are the whole design in one screen; Direction.IDLE is what a car reports when it has nothing to do, and keeping it in the same enum avoids a nullable field.
code/lld/elevator_system/models.py — enums
class Direction(StrEnum):
UP = "up"
DOWN = "down"
IDLE = "idle"
def opposite(self) -> Direction:
if self is Direction.UP:
return Direction.DOWN
if self is Direction.DOWN:
return Direction.UP
return Direction.IDLE
def arrow(self) -> str:
return {Direction.UP: "^", Direction.DOWN: "v", Direction.IDLE: "-"}[self]
class ElevatorState(StrEnum):
"""The five statuses a car can be in. Every tick moves the car through this set."""
IDLE = "idle"
MOVING_UP = "moving_up"
MOVING_DOWN = "moving_down"
DOOR_OPEN = "door_open"
MAINTENANCE = "maintenance"
class DoorState(StrEnum):
CLOSED = "closed"
OPEN = "open"
OBSTRUCTED = "obstructed"
code/lld/elevator_system/models.py — errors
class FloorOutOfRangeError(ValidationError):
"""The requested floor does not exist in this building."""
class NoCarAvailableError(ConflictError):
"""Every car is in maintenance, so the call cannot be assigned to anyone."""
class CarUnavailableError(InvalidStateError):
"""The car is out of service and refuses stops, boarding or movement."""
class DoorObstructedError(InvalidStateError):
"""A door operation is illegal for the door's current state."""
class CapacityExceededError(ConflictError):
"""One more passenger would put the car over its rated load."""
A press is a command object. RequestSink is the two-method view of the controller a request needs, which is how models.py stays free of any import from services.py, and requested_at is what later becomes a wait time.
code/lld/elevator_system/models.py — the presses
class RequestSink(Protocol):
"""The two controller methods a request needs. Keeps models free of service imports."""
def assign_hall_call(self, request: HallRequest) -> str: ...
def assign_cabin_call(self, request: CabinRequest) -> str: ...
class Request(Protocol):
"""Command: a button press as an object the controller can log, queue and replay."""
requested_at: float
def apply(self, sink: RequestSink) -> str: ...
def label(self) -> str: ...
@dataclass(frozen=True, slots=True)
class HallRequest:
"""A landing button: "someone on floor 5 wants to go up"."""
floor: int
direction: Direction
requested_at: float
destination: int | None = None # only a destination-dispatch lobby knows this
def key(self) -> tuple[int, Direction]:
return (self.floor, self.direction)
def apply(self, sink: RequestSink) -> str:
return sink.assign_hall_call(self)
def label(self) -> str:
suffix = "" if self.destination is None else f" to {self.destination}"
return f"hall {self.floor}{self.direction.arrow()}{suffix}"
@dataclass(frozen=True, slots=True)
class CabinRequest:
"""A button inside a car: no dispatch decision to make, the car is already chosen."""
car_id: str
floor: int
requested_at: float
def apply(self, sink: RequestSink) -> str:
return sink.assign_cabin_call(self)
def label(self) -> str:
return f"cabin {self.car_id} to {self.floor}"
The door is a timer, the landing is two lamps, and CarStatus is the immutable snapshot every strategy and every display reads. Nothing here holds a lock, because nothing here is shared until a car or the controller owns it.
code/lld/elevator_system/models.py — entities and snapshots
@dataclass(slots=True)
class Door:
"""A dwell timer, not a motor: ``open`` starts it, ``tick`` runs it down."""
default_hold: int = 2
state: DoorState = DoorState.CLOSED
hold_ticks: int = 0
def is_open(self) -> bool:
return self.state is not DoorState.CLOSED
def open(self) -> None:
self.state = DoorState.OPEN
self.hold_ticks = self.default_hold
def obstruct(self) -> None:
if self.state is DoorState.CLOSED:
raise DoorObstructedError("a closed door cannot be obstructed")
self.state = DoorState.OBSTRUCTED
self.hold_ticks = self.default_hold
def clear_obstruction(self) -> None:
if self.state is not DoorState.OBSTRUCTED:
raise DoorObstructedError("the door is not obstructed")
self.state = DoorState.OPEN
self.hold_ticks = self.default_hold
def tick(self) -> bool:
"""Run the dwell timer for one tick. Returns True on the tick it closes."""
if self.state is DoorState.CLOSED:
return False
if self.state is DoorState.OBSTRUCTED:
self.hold_ticks = self.default_hold # a blocked door restarts its timer
return False
self.hold_ticks -= 1
if self.hold_ticks > 0:
return False
self.state = DoorState.CLOSED
return True
@dataclass(slots=True)
class Floor:
"""A landing with its two hall lamps. Pressing a lit lamp is absorbed here."""
number: int
lit: set[Direction] = field(default_factory=set)
def press(self, direction: Direction) -> bool:
"""Light the lamp. Returns False when it was already lit (a duplicate press)."""
if direction in self.lit:
return False
self.lit.add(direction)
return True
def clear(self, direction: Direction) -> None:
self.lit.discard(direction)
@dataclass(frozen=True, slots=True)
class CarStatus:
"""An immutable snapshot: what a strategy scores and what a display shows."""
car_id: str
floor: int
direction: Direction
state: ElevatorState
door: DoorState
stops: tuple[int, ...]
load: int
capacity: int
def in_service(self) -> bool:
return self.state is not ElevatorState.MAINTENANCE
def is_idle(self) -> bool:
return self.state is ElevatorState.IDLE and not self.stops
def has_room(self) -> bool:
return self.load < self.capacity
def distance_to(self, floor: int) -> int:
return abs(self.floor - floor)
def render(self) -> str:
door = " door" if self.door is not DoorState.CLOSED else ""
return f"{self.car_id}{self.direction.arrow()}{self.floor}{door}"
@dataclass(frozen=True, slots=True)
class ServedCall:
"""One answered hall call: the raw material of the average-wait comparison."""
floor: int
direction: Direction
car_id: str
requested_at: float
served_at: float
def wait(self) -> float:
return self.served_at - self.requested_at
Strategies take snapshots and return a car id, so they never touch a lock and never mutate anything. _eligible is the one rule they all share: a car in maintenance or already full cannot answer.
code/lld/elevator_system/strategies.py — the interface
class DispatchStrategy(Protocol):
"""Scores the cars and names one. Returns None when no car can take the call."""
name: str
def select(self, cars: Sequence[CarStatus], request: HallRequest) -> str | None: ...
def _eligible(cars: Sequence[CarStatus]) -> list[CarStatus]:
return [car for car in cars if car.in_service() and car.has_room()]
The two you write first are the two you then criticise. FCFS ignores geography; nearest-car ignores direction and will happily assign a car that is about to speed past you the other way.
code/lld/elevator_system/strategies.py — FCFS and nearest-car
class FcfsDispatch:
"""First come, first served: the first idle car, else the shortest queue.
The baseline you implement in two minutes and then criticise: it ignores
where the car is and which way it is going, so a call one floor away can
wait for a car at the other end of the shaft.
"""
name = "fcfs"
def select(self, cars: Sequence[CarStatus], request: HallRequest) -> str | None:
eligible = _eligible(cars)
if not eligible:
return None
idle = [car for car in eligible if car.is_idle()]
if idle:
return idle[0].car_id
return min(eligible, key=lambda car: (len(car.stops), car.car_id)).car_id
class NearestCarDispatch:
"""Fewest floors away, idle cars winning ties.
Better than FCFS and still wrong in one obvious way: a car one floor away
that is speeding past in the other direction scores better than an idle car
three floors away, and the passenger watches it go by.
"""
name = "nearest_car"
def select(self, cars: Sequence[CarStatus], request: HallRequest) -> str | None:
eligible = _eligible(cars)
if not eligible:
return None
return min(
eligible,
key=lambda car: (car.distance_to(request.floor), not car.is_idle(), car.car_id),
).car_id
LOOK is the one that earns the interview. Its cost function has three cases, and saying them out loud is the answer: on my way costs the distance, after my turn costs a shaft more, behind me costs two.
code/lld/elevator_system/strategies.py — LOOK and destination dispatch
class LookDispatch:
"""SCAN/LOOK: a car keeps going while stops remain ahead, then turns.
The cost of a call is the number of floors the car must travel before it can
open its doors there, so a car heading your way is cheap, a car that must
finish its run first pays one shaft, and a car that has already passed you
pays two.
"""
name = "look"
def __init__(self, shaft_height: int) -> None:
if shaft_height <= 0:
raise ValueError("shaft_height must be positive")
self._shaft = shaft_height
def cost(self, car: CarStatus, request: HallRequest) -> int:
distance = car.distance_to(request.floor)
if car.is_idle():
return distance
going_up = car.direction is Direction.UP
ahead = request.floor >= car.floor if going_up else request.floor <= car.floor
if ahead and car.direction is request.direction:
return distance # picked up on the way, no turn needed
if ahead:
return distance + self._shaft # served after the car turns at the end
return distance + 2 * self._shaft # the car has already passed this floor
def select(self, cars: Sequence[CarStatus], request: HallRequest) -> str | None:
eligible = _eligible(cars)
if not eligible:
return None
return min(
eligible, key=lambda car: (self.cost(car, request), len(car.stops), car.car_id)
).car_id
class DestinationDispatch:
"""Ask for the destination at the landing, then group passengers by it.
When the lobby panel knows you want floor 9, a car already stopping at 9 and
still able to pick you up costs nothing extra; only when no car matches does
this fall back to the wrapped strategy.
"""
name = "destination"
def __init__(self, fallback: DispatchStrategy) -> None:
self._fallback = fallback
def select(self, cars: Sequence[CarStatus], request: HallRequest) -> str | None:
if request.destination is None:
return self._fallback.select(cars, request)
grouped = [
car
for car in _eligible(cars)
if request.destination in car.stops and self._on_the_way(car, request)
]
if not grouped:
return self._fallback.select(cars, request)
return min(grouped, key=lambda car: (car.distance_to(request.floor), car.car_id)).car_id
@staticmethod
def _on_the_way(car: CarStatus, request: HallRequest) -> bool:
if car.is_idle():
return True
if car.direction is Direction.UP:
return car.floor <= request.floor
return car.floor >= request.floor
The clock is injected and discrete, which is what makes a 90-tick simulation exact and instant.
code/lld/elevator_system/services.py — the simulation clock
class SimulationClock:
"""A discrete clock: one ``tick`` is one second of simulated time.
It satisfies the shared ``Clock`` protocol, so services take it exactly where
they would take ``SystemClock``; nothing in this package sleeps or reads the
wall clock, which is what makes a 400-tick simulation run in milliseconds.
"""
def __init__(self, seconds_per_tick: float = 1.0, start: float = 0.0) -> None:
if seconds_per_tick <= 0:
raise ValueError("seconds_per_tick must be positive")
self._clock = FakeClock(start)
self._seconds = seconds_per_tick
self._ticks = 0
self._lock = threading.Lock()
def now(self) -> float:
return self._clock.now()
def now_dt(self) -> datetime:
return self._clock.now_dt()
@property
def ticks(self) -> int:
with self._lock:
return self._ticks
def tick(self) -> int:
"""Advance one tick and return the new tick number."""
with self._lock:
self._ticks += 1
count = self._ticks
self._clock.advance(self._seconds)
return count
Now the car, which is big enough to deserve its own module. step is the whole state machine: run the door timer, or move one floor, then re-decide. _resume and _stop_here are the LOOK rule, and splitting the stops by service direction is what makes them short. Note that step returns whether the doors opened rather than leaving the controller to infer it — a car standing with its doors open can be given a stop at its own floor and close and reopen inside a single tick, and an edge detector on the status would miss that passenger forever.
code/lld/elevator_system/car.py — the car
class Elevator:
"""One car: its own lock, its own two stop sets, its own state machine.
Stops are split by the direction in which they will be served, which is what
makes LOOK a three-line rule instead of a scheduler: serve everything in the
current direction, then turn.
"""
def __init__(
self,
car_id: str,
floors: int,
capacity: int = 8,
start_floor: int = 0,
door_hold: int = 2,
) -> None:
if floors < 2:
raise ValueError("a shaft needs at least two floors")
self.id = car_id
self.floors = floors
self.capacity = capacity
self.floor = start_floor
self.state = ElevatorState.IDLE
self.direction = Direction.IDLE
self.load = 0
self.travelled = 0 # floors moved so far: the energy half of the strategy bake-off
self.door = Door(default_hold=door_hold)
self._up_stops: set[int] = set()
self._down_stops: set[int] = set()
self._opened = False # set by _open_door, read and reset by step
self._lock = threading.Lock()
# -- commands ------------------------------------------------------------------
def add_stop(self, floor: int, direction: Direction | None = None) -> None:
"""Queue a stop. ``direction`` is the *call* direction for a hall request."""
if not 0 <= floor < self.floors:
raise FloorOutOfRangeError(f"floor {floor} is outside 0..{self.floors - 1}")
with self._lock:
if self.state is ElevatorState.MAINTENANCE:
raise CarUnavailableError(f"car {self.id} is in maintenance")
if direction is Direction.UP:
self._up_stops.add(floor)
elif direction is Direction.DOWN:
self._down_stops.add(floor)
elif floor >= self.floor:
self._up_stops.add(floor)
else:
self._down_stops.add(floor)
def board(self, passengers: int = 1) -> None:
with self._lock:
if not self.door.is_open():
raise CarUnavailableError(f"car {self.id} has its doors closed")
if self.load + passengers > self.capacity:
raise CapacityExceededError(
f"car {self.id} holds {self.capacity}; {self.load} + {passengers} does not fit"
)
self.load += passengers
def leave(self, passengers: int = 1) -> None:
with self._lock:
self.load = max(0, self.load - passengers)
def obstruct_door(self) -> None:
with self._lock:
self.door.obstruct()
def clear_door(self) -> None:
with self._lock:
self.door.clear_obstruction()
def emergency_stop(self) -> list[int]:
"""Drop every stop, open up where we are, go out of service. Returns the dropped floors."""
with self._lock:
dropped = sorted(self._up_stops | self._down_stops)
self._up_stops.clear()
self._down_stops.clear()
self.door.open()
self.direction = Direction.IDLE
self.state = ElevatorState.MAINTENANCE
return dropped
def return_to_service(self) -> None:
with self._lock:
if self.state is not ElevatorState.MAINTENANCE:
raise CarUnavailableError(f"car {self.id} is not in maintenance")
self.door.state = DoorState.CLOSED
self.door.hold_ticks = 0
self.state = ElevatorState.IDLE
# -- the tick ------------------------------------------------------------------
def step(self) -> bool:
"""One tick: run the door timer, or move one floor, then re-decide.
Returns True if the doors opened during this tick. A car standing with
its doors open can be given a stop at its own floor, close and reopen in
the same tick, so the controller cannot detect an arrival by watching the
status change; the car reports the event instead.
"""
with self._lock:
self._opened = False
if self.state is ElevatorState.MAINTENANCE:
return False
if self.state is ElevatorState.DOOR_OPEN:
if self.door.tick():
self._resume()
return self._opened
if self.state is ElevatorState.MOVING_UP:
self.floor += 1
self.travelled += 1
elif self.state is ElevatorState.MOVING_DOWN:
self.floor -= 1
self.travelled += 1
self._resume()
return self._opened
def floors_travelled(self) -> int:
with self._lock:
return self.travelled
def status(self) -> CarStatus:
with self._lock:
return CarStatus(
car_id=self.id,
floor=self.floor,
direction=self.direction,
state=self.state,
door=self.door.state,
stops=tuple(sorted(self._up_stops | self._down_stops)),
load=self.load,
capacity=self.capacity,
)
# -- helpers: called with the lock already held --------------------------------
def _resume(self) -> None:
if self._stop_here():
self._open_door()
return
self.direction = self._next_direction()
self.state = {
Direction.UP: ElevatorState.MOVING_UP,
Direction.DOWN: ElevatorState.MOVING_DOWN,
Direction.IDLE: ElevatorState.IDLE,
}[self.direction]
def _open_door(self) -> None:
self._up_stops.discard(self.floor)
self._down_stops.discard(self.floor)
self.door.open()
self.state = ElevatorState.DOOR_OPEN
self._opened = True
def _stop_here(self) -> bool:
if self.direction is Direction.UP:
return self.floor in self._up_stops or (
self.floor in self._down_stops and not self._work_above()
)
if self.direction is Direction.DOWN:
return self.floor in self._down_stops or (
self.floor in self._up_stops and not self._work_below()
)
return self.floor in self._up_stops or self.floor in self._down_stops
def _next_direction(self) -> Direction:
if self.direction is Direction.UP and self._work_above():
return Direction.UP
if self.direction is Direction.DOWN and self._work_below():
return Direction.DOWN
if self._work_above():
return Direction.UP
if self._work_below():
return Direction.DOWN
return Direction.IDLE
def _work_above(self) -> bool:
return any(f > self.floor for f in self._up_stops | self._down_stops)
def _work_below(self) -> bool:
return any(f < self.floor for f in self._up_stops | self._down_stops)
Displays subscribe and are pushed to; they never poll a car.
code/lld/elevator_system/services.py — displays
class CarListener(Protocol):
"""Observer: anything that wants to be told a car's new status after a tick."""
def on_car_changed(self, status: CarStatus) -> None: ...
class Display:
"""The indicator above the doors: floor number and direction arrow, per car."""
def __init__(self, name: str = "lobby") -> None:
self.name = name
self._lock = threading.Lock()
self._by_car: dict[str, CarStatus] = {}
def on_car_changed(self, status: CarStatus) -> None:
with self._lock:
self._by_car[status.car_id] = status
def floor_of(self, car_id: str) -> int:
with self._lock:
return self._by_car[car_id].floor
def render(self) -> str:
with self._lock:
return " | ".join(self._by_car[car_id].render() for car_id in sorted(self._by_car))
The controller is the mediator. submit is the Command invoker, _assign_locked is the dispatch decision, tick is the simulation, and _on_arrival is the part candidates forget: when the doors open, riders get out, the waiting passenger gets in and presses a floor, and only then is the call recorded as served.
code/lld/elevator_system/services.py — the controller
class ElevatorController:
"""Mediator: the only object that knows both the landings and the cars.
A landing never addresses a car and a car never addresses a landing. The
controller owns the assignment map, the hall lamps, the event log and the
served-call statistics, all under ``_lock``.
"""
def __init__(
self,
cars: Iterable[Elevator],
floors: int,
strategy: DispatchStrategy | None = None,
clock: SimulationClock | None = None,
ids: IdGenerator | None = None,
) -> None:
self._cars = {car.id: car for car in cars}
if not self._cars:
raise ValueError("a controller needs at least one car")
self.floors = floors
self._landings = {number: Floor(number) for number in range(floors)}
self._strategy = strategy or LookDispatch(shaft_height=floors - 1)
self._clock = clock or SimulationClock()
self._ids = ids or SequentialIdGenerator("R")
self._assignments: dict[tuple[int, Direction], AssignedCall] = {}
self._riders: dict[str, dict[int, int]] = {car_id: {} for car_id in self._cars}
self._deferred: list[HallRequest] = []
self._listeners: list[CarListener] = []
self._served: list[ServedCall] = []
self._log: list[str] = []
self._lock = threading.Lock()
# -- the invoker side of Command -------------------------------------------------
def submit(self, request: Request) -> str:
"""Log the press, then let the command choose which controller method runs."""
with self._lock:
self._log.append(f"t{self._clock.ticks} {self._ids.next_id()} {request.label()}")
return request.apply(self)
def hall_call(
self, floor: int, direction: Direction, destination: int | None = None
) -> str:
return self.submit(HallRequest(floor, direction, self._clock.now(), destination))
def cabin_request(self, car_id: str, floor: int) -> str:
return self.submit(CabinRequest(car_id, floor, self._clock.now()))
# -- the receiver side -----------------------------------------------------------
def assign_hall_call(self, request: HallRequest) -> str:
self._check_floor(request.floor)
with self._lock:
return self._assign_locked(request)
def assign_cabin_call(self, request: CabinRequest) -> str:
self._check_floor(request.floor)
with self._lock:
self._car(request.car_id).add_stop(request.floor)
return request.car_id
# -- the tick --------------------------------------------------------------------
def tick(self) -> int:
"""Advance time, step every car, clear the lamps of the calls just answered."""
statuses: list[CarStatus] = []
with self._lock:
self._clock.tick()
deferred, self._deferred = self._deferred, []
for request in deferred:
self._try_assign(request)
for car in self._cars.values():
opened = car.step()
status = car.status()
if opened:
self._on_arrival(status)
statuses.append(status)
self._notify(statuses) # outside the lock: a slow display never stalls the bank
return self._clock.ticks
def run(self, ticks: int) -> None:
for _ in range(ticks):
self.tick()
# -- maintenance -----------------------------------------------------------------
def emergency_stop(self, car_id: str) -> list[str]:
"""Take a car out of service and re-dispatch the hall calls it still owed."""
with self._lock:
car = self._car(car_id)
dropped = car.emergency_stop()
orphans = [a.request for a in self._assignments.values() if a.car_id == car_id]
for request in orphans:
del self._assignments[request.key()]
self._landings[request.floor].clear(request.direction)
self._riders[car_id].clear()
self._log.append(
f"t{self._clock.ticks} emergency stop {car_id}: dropped {dropped}, "
f"re-dispatching {len(orphans)} hall call(s)"
)
return [car for car in map(self._try_assign, orphans) if car is not None]
def return_to_service(self, car_id: str) -> None:
with self._lock:
self._car(car_id).return_to_service()
self._log.append(f"t{self._clock.ticks} {car_id} back in service")
# -- reads -------------------------------------------------------------------------
def subscribe(self, listener: CarListener) -> None:
with self._lock:
self._listeners.append(listener)
statuses = [car.status() for car in self._cars.values()]
for status in statuses:
listener.on_car_changed(status)
def statuses(self) -> list[CarStatus]:
with self._lock:
return [car.status() for car in self._cars.values()]
def waiting_calls(self) -> list[tuple[int, Direction]]:
with self._lock:
return sorted(self._assignments, key=lambda key: (key[0], key[1].value))
def served_calls(self) -> list[ServedCall]:
with self._lock:
return list(self._served)
def average_wait(self) -> float:
served = self.served_calls()
return sum(call.wait() for call in served) / len(served) if served else 0.0
def total_travel(self) -> int:
"""Floors moved by the whole bank: the energy cost a scheduling policy trades against."""
with self._lock:
return sum(car.floors_travelled() for car in self._cars.values())
def event_log(self) -> list[str]:
with self._lock:
return list(self._log)
# -- helpers: called with ``_lock`` already held -------------------------------------
def _assign_locked(self, request: HallRequest) -> str:
landing = self._landings[request.floor]
assigned = self._assignments.get(request.key())
if assigned is not None:
landing.press(request.direction) # already lit: a second press is absorbed here
return assigned.car_id
snapshots = [car.status() for car in self._cars.values()]
car_id = self._strategy.select(snapshots, request)
if car_id is None:
raise NoCarAvailableError(f"no car can answer {request.label()}")
landing.press(request.direction)
self._assignments[request.key()] = AssignedCall(request, car_id)
self._cars[car_id].add_stop(request.floor, request.direction)
self._log.append(f"t{self._clock.ticks} {request.label()} -> {car_id}")
return car_id
def _try_assign(self, request: HallRequest) -> str | None:
"""Re-dispatch after a full car or an emergency stop; defer when nobody can take it."""
try:
return self._assign_locked(request)
except NoCarAvailableError:
self._deferred.append(request)
return None
def _on_arrival(self, status: CarStatus) -> None:
"""Doors just opened: riders get out, waiting passengers get in and press a floor."""
car = self._cars[status.car_id]
leaving = self._riders[car.id].pop(status.floor, 0)
if leaving:
car.leave(leaving)
now = self._clock.now()
for direction in (Direction.UP, Direction.DOWN):
key = (status.floor, direction)
assigned = self._assignments.get(key)
if assigned is None or assigned.car_id != status.car_id:
continue
request = assigned.request
del self._assignments[key]
self._landings[status.floor].clear(direction)
if request.destination is not None:
try:
car.board()
except CapacityExceededError:
self._log.append(f"t{self._clock.ticks} {car.id} is full at {status.floor}")
self._try_assign(request)
continue
car.add_stop(request.destination)
riders = self._riders[car.id]
riders[request.destination] = riders.get(request.destination, 0) + 1
self._served.append(
ServedCall(
floor=status.floor,
direction=direction,
car_id=status.car_id,
requested_at=request.requested_at,
served_at=now,
)
)
self._log.append(
f"t{self._clock.ticks} {status.car_id} opened at {status.floor}, "
f"cleared the {direction} lamp"
)
def _notify(self, statuses: Sequence[CarStatus]) -> None:
for listener in list(self._listeners):
for status in statuses:
listener.on_car_changed(status)
def _car(self, car_id: str) -> Elevator:
try:
return self._cars[car_id]
except KeyError:
raise CarUnavailableError(f"unknown car {car_id}") from None
def _check_floor(self, floor: int) -> None:
if not 0 <= floor < self.floors:
raise FloorOutOfRangeError(f"floor {floor} is outside 0..{self.floors - 1}")
Running python -m lld.elevator_system.demo prints one narrated scenario and then replays a fixed eight-call workload against each strategy:
--- three cars, twelve floors, LOOK dispatch ---
t0 A-0 | B-5 | C-11 waiting: none
hall call 7 up (to 11) -> car B
t1 A-0 | B^5 | C-11 waiting: 7^
t2 A-0 | B^6 | C-11 waiting: 7^
hall call 1 up (to 9) -> car A
t3 A^0 | B^7 door | C-11 waiting: 1^
emergency stop A -> its hall calls move to ['C']
t4 A-0 door | B^7 door | Cv11 waiting: 1^
t5 A-0 door | B^7 | Cv10 waiting: 1^
served 7^ by B after 3 ticks
--- eight calls replayed against each strategy ---
fcfs wait 8.38 ticks 49 floors travelled 8/8 served
nearest_car wait 7.38 ticks 45 floors travelled 8/8 served
look wait 7.00 ticks 32 floors travelled 8/8 served
destination wait 4.62 ticks 28 floors travelled 8/8 served
That table is the reason the clock is injected. FCFS is worst on both measures; nearest-car buys a shorter wait by dragging cars off their runs, which costs floors; LOOK travels 32 floors instead of 45 for a shorter wait than either; destination dispatch wins because it can put two passengers heading to floor 11 in the same car. Say out loud that the ranking is workload-dependent — on a light load with idle cars, nearest-car is hard to beat, and the honest claim is "here is the harness, here are two metrics, run your traffic through it".
Concurrency and edge cases¶
Which lock protects what. There are two levels and one rule.
Elevator._lockguards that car's floor, state, direction, door, load and both stop sets. Every public method takes it; the private helpers (_resume,_stop_here,_next_direction) assume it is already held and never take it again. A call arriving mid-move therefore lands in_up_stopsbetween two ticks, never during one.ElevatorController._lockguards the assignment map, the landing lamps, the deferred queue, the rider counts, the event log and the served-call list. Dispatch reads snapshots and writes the assignment under it, so two threads pressing the same button produce one assignment and one car.
The rule is that the controller lock is always taken before a car lock, never the reverse: a car never calls back into the controller, it only returns values from step and status. That single direction is what makes the ordering trivially safe. An uncontended lock costs about 17 ns (see the latency numbers), so a bank of eight cars pays roughly 8 x 17 ns per tick for locking — nothing next to any real work, which is why per-car granularity is the right trade here rather than one lock over the whole bank.
The duplicate press. _assign_locked checks the assignment map first and returns the existing car; the lamp is idempotent. The concurrency test presses eight buttons five times each from eight threads and asserts that no button was ever answered by two different cars and that no floor is queued in two cars.
SCAN starvation. _next_direction keeps a car going while any stop remains ahead, so a steady stream of upward calls can strand a down call at the bottom. Name it as a known property of LOOK, then offer the fix: age each assignment and let the controller re-dispatch (or force a turnaround) once a call has waited more than a threshold, which is a change to the controller rather than the car.
Capacity overflow. When the doors open and the car is full, the passenger cannot board, so the call is re-dispatched to a car with room; if there is none, it goes on _deferred and is retried at the start of the next tick. That path is tested with a car of capacity 1.
Other edges handled: doors obstructed indefinitely (the dwell timer restarts and the car holds the floor); a stop added at the floor the car is already standing on; an emergency stop that re-homes the hall calls the car owed while dropping the cabin requests of passengers who are getting out; floors outside the building rejected with a validation error; every car in maintenance rejected with NoCarAvailableError.
Common mistake
Modelling the queue as one sorted list of floors. It reads well for thirty seconds and then cannot express "stop at 5 on the way up but not on the way down", which is the entire point of a hall call having a direction. Two sets — one served going up, one going down — make LOOK three lines and make the diagram honest.
Extensibility and follow-ups¶
- Compare policies on real traffic: already built.
replayruns a fixed workload and reports average wait and floors travelled; feed it a seeded random workload for a bigger sample. - Zoning and express cars: a
ZonedDispatchthat filters the snapshots to the cars serving the requested floor's zone and delegates to any inner strategy — the same compositionDestinationDispatchalready uses for its fallback. - Power failure or fire recall: a mode on the controller that cancels all assignments, sends every car to a designated floor, and refuses new calls; the car needs no change because
emergency_stopalready returns the stops it dropped. - Elevator groups across buildings: several controllers behind a facade that routes a call to a group first. This is where the conversation turns into a distributed-systems question — telemetry, per-group health, and a supervisor that reassigns when a group stops answering.
- Real hardware:
Elevatorbecomes an interface with a simulated and a driver-backed implementation; the tick becomes the driver's position callback. Nothing in the controller changes, which is the point of it never having read a car's fields directly.
Interview tip
When you are asked "which scheduling algorithm would you use?", do not pick one. Say "I would make it a strategy and measure": name FCFS, nearest-car, LOOK and destination dispatch, describe the two metrics that matter (wait to pick-up and floors travelled), and point at the seam where a new one plugs in. Then give your default, which is LOOK, and why.
Tests¶
tests/test_elevator_system.py has 18 cases covering the happy path, validation, the state machine, both concurrency invariants and the awkward edges. The two worth walking through in the room are the LOOK sweep and the concurrent presses:
code/lld/elevator_system/tests/test_elevator_system.py — the LOOK sweep
def test_look_finishes_the_upward_sweep_before_it_turns() -> None:
controller = build({"A": 3}, floors=10)
for floor in (6, 8, 1): # the call below arrives while the car is already going up
controller.cabin_request("A", floor)
opened_at: list[int] = []
for _ in range(24):
controller.tick()
status = controller.statuses()[0]
if status.door is DoorState.OPEN and status.floor not in opened_at:
opened_at.append(status.floor)
assert opened_at == [6, 8, 1] # everything above first, then one reversal
code/lld/elevator_system/tests/test_elevator_system.py — concurrent presses
def test_concurrent_presses_assign_every_button_to_exactly_one_car() -> None:
controller = build({"A": 0, "B": 5, "C": 11}, strategy=NearestCarDispatch())
buttons = [(floor, Direction.UP) for floor in range(1, 9)]
presses = [button for button in buttons for _ in range(5)] # 40 presses, 8 buttons
def press(button: tuple[int, Direction]) -> str:
return controller.hall_call(button[0], button[1])
with ThreadPoolExecutor(max_workers=8) as pool:
answers = list(pool.map(press, presses))
cars_per_button: dict[tuple[int, Direction], set[str]] = defaultdict(set)
for button, car_id in zip(presses, answers, strict=True):
cars_per_button[button].add(car_id)
assert all(len(cars) == 1 for cars in cars_per_button.values()) # never split between cars
assert len(controller.waiting_calls()) == len(buttons)
queued = sorted(floor for status in controller.statuses() for floor in status.stops)
assert queued == [floor for floor, _ in buttons] # each floor queued in exactly one car
The rest cover: a hall call served end to end with the passenger boarding and pressing a destination; floors outside the building rejected; the exact six-tick state sequence of one car; a duplicate press producing one served call; an emergency stop re-homing the call it owed; every car in maintenance; an obstructed door holding the floor and then releasing it; a full car handing its passenger back to the bank; boarding over the rated load; three strategies picking three different cars, destination dispatch grouping two passengers bound for the same floor, and the display following every car without polling. Run them with uv run pytest code/lld/elevator_system -q.
45-minute pacing¶
| Minutes | What to do | What to say or write |
|---|---|---|
| 0-5 | Clarify | What is a tick? Do landings know the destination? Capacity, emergency stop, maintenance? Out of scope: fire recall, hardware bus. |
| 5-10 | Entities | Nouns on the board: Controller, Elevator, Door, Floor, HallRequest, CabinRequest, Display. Verbs become methods: hall_call, step, add_stop, select. |
| 10-16 | Diagrams | Draw the five-state machine first, then the class diagram around it. Mark the two locks while you draw. |
| 16-34 | Code | Enums, then HallRequest, then Elevator.step and _resume, then ElevatorController.tick and _assign_locked. Say "two stop sets" when you write them. |
| 34-40 | Concurrency and dispatch | The controller-then-car lock order, the duplicate press, the full car. Then the four strategies and the two metrics you would compare them on. |
| 40-45 | Extensions | Zoning, aging to kill starvation, hardware behind the same interface, groups as the hand-off to a system-design question. |
Related¶
- State — the enum-and-table form used for
ElevatorState - Strategy — the four dispatch policies
- Mediator — the controller, and a smaller version of this same example
- Command — presses as objects with a timestamp
- Concurrency for LLD in Python — lock ordering and granularity
- Mock LLD interview: elevator system — the same problem as a 45-minute transcript