Design a multi-threaded parking lot

01The design, in the six-step framework

1. Requirements and use cases

Many entry and exit gates run at once, each on its own thread. A driver gets a spot of the tightest available size or larger. If two drivers reach the last spot at the same instant, exactly one gets it and the other is told the lot is full, or waits in line if they chose to. Waiting drivers are served in arrival order. A driver leaving frees a spot and wakes the next in line. Leaving twice is harmless.

Out of scope: floors, pricing and the class model of the basic parking lot, which is assumed; this page keeps only what concurrency changes.

Constraints: no spot is ever given to two vehicles; no deadlock under any interleaving; gates working on different sizes do not block each other; a late arrival does not jump a queue of waiters.

2. Core entities and relationships

• FreeSpots: a pool of free spots for one size, with its own lock. The take is atomic.
• ConcurrentLot: holds one pool per size, the active tickets under their own lock, and the FIFO queue of waiters.
• Spot, Vehicle, Ticket: frozen values, safe to share across threads with no lock at all.
• threading.Event: one per waiting driver, so a single driver can be woken.

Spots are immutable: "occupied" is not a flag on the spot but the fact of not being in a free pool. Moving a spot between the pool and a ticket is therefore a single ownership transfer, which removes the most common concurrency bug in this problem.

3. Interfaces and key methods

class FreeSpots:
    """Per-size pool of free spots; one lock per size, never two at once."""
    def try_take(self) -> Spot | None: ...
    def put_back(self, spot: Spot) -> None: ...

class ConcurrentLot:
    def enter(self, vehicle: Vehicle, timeout: float | None = None) -> Ticket:
        """Take the tightest free spot; optionally queue FIFO; raises LotFull."""
    def leave(self, ticket_id: str) -> None:
        """Idempotent: a second call does nothing."""
    def available(self, size: SpotSize) -> int: ...

Invariants: every spot is in exactly one free pool or exactly one ticket; no thread ever holds two pool locks; a thread with waiters ahead of it does not take a spot; leave returns the spot only if it removed the ticket.

4. Design patterns and the choices behind them

Ownership by queue, not a flag per spot

The naive approach scans for a free spot and then marks it, a check-then-act race. A global lock fixes it and makes all gates single file. Scanning with a lock per spot works but holds up on contention and multiplies lock acquisitions. A pool per size turns "find and claim" into one atomic pop: two threads popping the last spot cannot both succeed, because try_take is the critical section. The Python deque pop is safe alone, but the empty check needs the lock, and using a thread-safe structure does not remove the need to think about compound steps.

Deadlock is designed out

A vehicle may need to try several sizes. If a thread held the small lock while requesting the medium lock and another did the reverse, they could deadlock. Here each pool lock is released before the next is tried, so no thread ever holds two pool locks and the circular-wait condition cannot arise. That is stronger than lock ordering, though the ascending loop gives that too. Similarly the ticket lock and wait lock are never held together with a pool lock (see deadlock avoidance).

Explicit FIFO instead of a condition variable

A condition variable with notify gives no ordering, and a late arrival can barge in between the notify and the waiter waking. Fairness needs an explicit queue. Each waiter has an Event and only the head may try; a new arrival checks _has_waiters and joins the back. If the head's vehicle does not fit the freed spot, the head keeps it from the people behind. That strict FIFO is the simple behaviour; a per-size queue is the upgrade to mention.

Why not optimistic

Optimistic concurrency with a version on each spot suits a database row, where a lock across network calls is expensive. In memory the critical section is a few instructions, so a short lock is simpler and retries would only add noise near full capacity.

5. Core flow and code

1. One medium spot remains. Threads A and B call enter for cars at the same moment.
2. Both see no waiters and call _try_park, which tries the pools from the vehicle's size upward.
3. A holds the medium lock first and pops the spot. B then finds the pool empty and moves on to large, which is also empty, so _try_park returns None.
4. B has timeout=30, so it appends its Event to the queue and, as head, retries, then waits.
5. A's driver leaves. leave pops the ticket, puts the spot back, and sets the head's Event.
6. B wakes, is head, and takes the spot. In finally it removes itself and signals the next waiter.
7. A thread with no timeout raises LotFull at step 3 without waiting.

from __future__ import annotations
import collections, itertools, threading, time
from dataclasses import dataclass
from enum import IntEnum

class SpotSize(IntEnum):
    SMALL = 1; MEDIUM = 2; LARGE = 3

class LotFull(Exception):
    pass

@dataclass(frozen=True)
class Vehicle:
    plate: str
    size: SpotSize

@dataclass(frozen=True)
class Spot:
    spot_id: str
    size: SpotSize

@dataclass(frozen=True)
class Ticket:
    ticket_id: int
    plate: str
    spot: Spot

class FreeSpots:
    def __init__(self, spots: list[Spot]) -> None:
        self._free = collections.deque(spots)
        self._lock = threading.Lock()

    def try_take(self) -> Spot | None:
        with self._lock:
            return self._free.popleft() if self._free else None

    def put_back(self, spot: Spot) -> None:
        with self._lock:
            self._free.append(spot)

    def count(self) -> int:
        with self._lock:
            return len(self._free)

class ConcurrentLot:
    def __init__(self, spots: list[Spot]) -> None:
        self._pools = {size: FreeSpots([s for s in spots if s.size == size]) for size in SpotSize}
        self._tickets: dict[int, Ticket] = {}
        self._tickets_lock = threading.Lock()
        self._ids = itertools.count(1)
        self._waiters: collections.deque[threading.Event] = collections.deque()
        self._wait_lock = threading.Lock()

    def _try_park(self, vehicle: Vehicle) -> Ticket | None:
        for size in SpotSize:                       # ascending: the global order of acquisition
            if size < vehicle.size:
                continue
            spot = self._pools[size].try_take()     # each lock held only inside this call
            if spot is not None:
                ticket = Ticket(next(self._ids), vehicle.plate, spot)
                with self._tickets_lock:
                    self._tickets[ticket.ticket_id] = ticket
                return ticket
        return None

    def enter(self, vehicle: Vehicle, timeout: float | None = None) -> Ticket:
        ticket = self._try_park(vehicle) if not self._has_waiters() else None
        if ticket is not None:
            return ticket
        if timeout is None:
            raise LotFull(vehicle.plate)
        deadline = time.monotonic() + timeout
        event = threading.Event()
        with self._wait_lock:
            self._waiters.append(event)
        try:
            while True:
                if self._is_head(event):            # FIFO: only the head may try
                    ticket = self._try_park(vehicle)
                    if ticket is not None:
                        return ticket
                remaining = deadline - time.monotonic()
                if remaining <= 0:
                    raise LotFull(vehicle.plate)
                event.wait(remaining); event.clear()
        finally:
            with self._wait_lock:
                self._waiters.remove(event)
                head = self._waiters[0] if self._waiters else None
            if head is not None:
                head.set()                          # hand the turn to the next in line

    def _has_waiters(self) -> bool:
        with self._wait_lock:
            return bool(self._waiters)

    def _is_head(self, event: threading.Event) -> bool:
        with self._wait_lock:
            return bool(self._waiters) and self._waiters[0] is event

    def leave(self, ticket_id: int) -> None:
        with self._tickets_lock:
            ticket = self._tickets.pop(ticket_id, None)
        if ticket is None:
            return                                   # double leave: harmless
        self._pools[ticket.spot.size].put_back(ticket.spot)
        with self._wait_lock:
            head = self._waiters[0] if self._waiters else None
        if head is not None:
            head.set()

    def available(self, size: SpotSize) -> int:
        return self._pools[size].count()

6. Extensions, concurrency and what interviewers probe

• Lost wakeups: a spot freed just after the head tried and before it waits is safe, because Event stays set until cleared. The head then loops and retries.
• Per-size fairness: a head truck waiting for a large spot blocks cars. Keep a queue per size and route each freed spot to the first waiter it fits.
• Tightest fit under contention: the loop may place a car in a large spot an instant before a medium is freed. That is acceptable; a global optimum would need a global lock.
• Consistency of available: it is a snapshot that may be stale on return. Never use it to decide entry.
• Test first: 100 threads racing for 10 spots, asserting exactly 10 tickets with unique spots; repeat under a barrier so they start together. Then a waiter served in order, a timeout, and a double leave.
• Common mistake: holding a global lock while the driver pays or while waiting, or checking available and then taking in two steps.

02Where this question was reported

• LLD Mastery · Google-tagged (Hard); fine-grained locking, deadlock prevention, fair queuing
• LLD Mastery · Google-tagged; concurrency variant of the classic