Classic software engineering interview questions — the kind that test concurrency, data structures, and system thinking — solved in Python and in U. Toggle between them to see what the pit of success looks like when the compiler enforces what Python leaves to discipline.
Implement a cache with a maximum size that evicts the least-recently-used entry when full. It must be safe under concurrent access.
import threading
from collections import OrderedDict
class LRUCache:
def __init__(self, capacity: int):
self.capacity = capacity
self.cache = OrderedDict()
self.lock = threading.Lock()
def get(self, key: str):
with self.lock:
if key not in self.cache:
return None
self.cache.move_to_end(key)
return self.cache[key]
def put(self, key: str, value):
with self.lock:
if key in self.cache:
self.cache.move_to_end(key)
self.cache[key] = value
else:
if len(self.cache) >= self.capacity:
self.cache.popitem(last=False)
self.cache[key] = value
# Usage
cache = LRUCache(100)
# Thread A Thread B
cache.put("x", 1) # cache.put("y", 2)
val = cache.get("x") # val = cache.get("x")
# Problems:
# - Lock is manual — forget it and you get a data race
# - Nothing prevents cache.cache[k] = v bypassing the lock
# - OrderedDict is not thread-safe on its own
# - The lock is pessimistic — all readers block each other
# - A new developer can add a method without the lock
# and it compiles, tests pass, prod breaks
d LRUCache capacity: I -M entries: {S: Tree} +G +M = {} order: [S] +G +M = [] f get(key: S) -> Tree +N key_idx = t.order.find(key) key_idx == none ? r => none // Touch: move to end (MVCC — atomic) LRUCache << { order: t.order.without(key_idx).with(key) } r => t.entries[key] f put(key: S, val: Tree) -> none t.order.find(key) != none ? ( LRUCache << { entries: t.entries.set(key, val), order: t.order.without(t.order.find(key)).with(key) } r => none ) t.order.len >= t.capacity ? ( oldest = t.order[1] LRUCache << { entries: t.entries.remove(oldest).set(key, val), order: t.order.without(1).with(key) } r => none ) LRUCache << { entries: t.entries.set(key, val), order: t.order.with(key) }
<< is MVCC — readers never block, writers CAS-retry automatically. You cannot forget the lock because there is no lock — direct assignment to +G +M fields is a compile error. A new developer adding a method MUST use << or the compiler rejects the code.
BFS from a seed URL, discovering all links on the same domain. Start synchronous, then make it concurrent.
from concurrent.futures import ThreadPoolExecutor
from urllib.parse import urlparse, urldefrag
def crawl(seed_url, get_urls, max_workers=10):
domain = urlparse(seed_url).netloc
visited = set()
visited.add(seed_url)
queue = [seed_url]
results = []
def process(url):
links = get_urls(url)
return [urldefrag(l)[0] for l in links
if urlparse(urldefrag(l)[0]).netloc == domain]
with ThreadPoolExecutor(max_workers=max_workers) as pool:
while queue:
futures = {pool.submit(process, u): u
for u in queue}
queue = []
for future in futures:
for link in future.result():
if link not in visited:
visited.add(link)
queue.append(link)
results.append(link)
return results
# Problems:
# - visited set is shared mutable state with no protection
# - ThreadPoolExecutor hides the GIL limitation
# - get_urls could throw — no error boundary visible
# - Nothing in the type system says "this is concurrent"
d Crawl visited: {S} +G +M = {} results: [S] +G +M = [] f same_domain(seed: S, link: S) -> L URL.parse(seed).host == URL.parse(link).host ? r => true r => false a f process(url: S) -> [S] links = a HTTP.get(url) r => links f crawl(seed: S) -> [S] queue: [S] +M = [seed] Crawl << { visited: Crawl.visited.add(seed) } queue.len > 0 ? ( tasks = queue.map(url => a process(url)) queue = [] tasks.on(task => ( links = task links.on(link => ( clean = URL.defrag(link) same_domain(seed, clean) == true ? ( Crawl.visited.has(clean) == false ? ( << ( Crawl << { visited: Crawl.visited.add(clean), results: Crawl.results.with(clean) } ) queue = queue.with(clean) )))))))) r => Crawl.results
a before a call makes it async — U's fiber scheduler handles concurrency on a single thread with epoll. The visited set uses << (MVCC) so concurrent adds never race. HTTP.get is declared +A +E in the stdlib — the compiler knows it's async and effectful.
Transfer money between two accounts atomically. If one fails, neither should change.
import threading
class Account:
def __init__(self, balance):
self.balance = balance
self.lock = threading.Lock()
def transfer(src, dst, amount):
# Must lock in consistent order to avoid deadlock
first, second = sorted([src, dst], key=id)
with first.lock:
with second.lock:
if src.balance < amount:
raise ValueError("Insufficient funds")
src.balance -= amount
dst.balance += amount
# Problems:
# - Two locks, manually ordered by id() to avoid deadlock
# - If you forget the ordering, deadlock under contention
# - Nothing prevents src.balance = 0 outside the lock
# - The ValueError is untyped — caller doesn't know what fails
# - A log() call inside the lock holds it longer
# - If process crashes between writes, money vanishes
# - None of this is visible in the function signature
d AccSrc bal: Q +G +M = 0.0 d AccDst bal: Q +G +M = 0.0 f transfer(amount: Q) -> none ! InsufficientFunds rate = get_fee_rate() // +D — before the block << ( fee = compute_fee(AccSrc.bal, rate) // -E -D ✓ AccSrc.bal < amount + fee ? x Rollback("insufficient") AccSrc << { bal: AccSrc.bal - amount - fee } AccDst << { bal: AccDst.bal + amount } ) // atomic commit or retry log("transferred") // +E — after the block
<< ( ... ) is an optimistic transaction — all patches buffer, all versions check, all-or-nothing. The compiler enforces only pure, deterministic code inside: try log() inside the block and the compiler rejects it (+E). The error type (! InsufficientFunds) is in the signature.
Given a vocabulary, tokenize input text using greedy matching. Handle unknown characters. Ensure round-trip.
def tokenize(text: str, vocab: dict) -> list:
tokens = []
key = ""
for ch in text:
key += ch
if key in vocab:
tokens.append(vocab[key])
key = ""
elif not any(k.startswith(key) for k in vocab):
tokens.append(vocab.get("UNK", -1))
key = ""
if key:
tokens.append(vocab.get("UNK", -1))
return tokens
def detokenize(tokens: list, vocab: dict) -> str:
rev = {v: k for k, v in vocab.items()}
return "".join(rev.get(t, "?") for t in tokens)
# Problems:
# - any(k.startswith(key)...) is O(n) per char — O(n*m) total
# - vocab is dict[str,int] — nothing stops {1: "a"}
# - UNK sentinel is stringly-typed
# - Return type is list (of what?)
d Token UNK: I +G = -1 f tokenize(text: S, vocab: {S: I}) -> [I] tokens: [I] +M = [] key: S +M = "" [1..text.len].on(ii => ( key = key + text[ii] vocab.has(key) ? ( tokens.push(vocab[key]) key = "" none ) ! ( has_prefix = vocab.keys.any(kk => kk.starts_with(key)) has_prefix == false ? ( tokens.push(Token.UNK) key = "" ) none ) )) key.len > 0 ? tokens.push(Token.UNK) r => tokens f detokenize(tokens: [I], vocab: {S: I}) -> S rev: {I: S} = vocab.invert() r => tokens.map(tt => rev.has(tt) ? rev[tt] ! "?").join("")
{S: I} not untyped dict. vocab is -M — the function cannot mutate it. Token.UNK is a +G constant, not a magic string. The two +M annotations mark exactly which state changes in the loop.
Convert periodic call-stack samples into a timeline of start/end events for a profiler visualization.
def convert_to_trace(samples):
events = []
prev_stack = []
for sample in samples:
ts, stack = sample["ts"], sample["stack"]
common = 0
while (common < len(prev_stack)
and common < len(stack)
and prev_stack[common] == stack[common]):
common += 1
# End departed frames (innermost first)
for i in range(len(prev_stack) - 1, common - 1, -1):
events.append({"kind": "end", "ts": ts,
"name": prev_stack[i]})
# Start new frames
for i in range(common, len(stack)):
events.append({"kind": "start", "ts": ts,
"name": stack[i]})
prev_stack = stack
return events
# Problems:
# - sample is untyped dict — no guarantee "ts" exists
# - events is list of dicts — no schema, keys can be misspelled
# - Nothing distinguishes recursive calls at the same depth
d Sample ts: N -M stack: [S] -M d Event kind: S -M ts: N -M name: S -M f convert_to_trace(samples: [Sample]) -> [Event] events: [Event] +M = [] prev: [S] +M = [] samples.on(sample => ( common: I +M = 0 common < prev.len ? common < sample.stack.len ? ( prev[common + 1] == sample.stack[common + 1] ? ( common = common + 1 none )) // End departed frames (innermost first) [prev.len..common + 1].on(ii => ( events.push(Event({ kind: "end", ts: sample.ts, name: prev[ii] })) )) // Start new frames [common + 1..sample.stack.len].on(ii => ( events.push(Event({ kind: "start", ts: sample.ts, name: sample.stack[ii] })) )) prev = sample.stack none )) r => events
Sample and Event are typed — no misspelled keys. Fields are -M (records, not mutable bags). samples is a -M parameter — the function can't corrupt it. The three +M accumulators are visible. Every other binding is frozen.
The pattern across all five.
Every Python solution requires the developer to remember the right thing: remember the lock, remember the lock ordering, remember not to mutate shared state, remember to type-check the dict keys. Every U solution makes the wrong thing unwritable: direct assignment to MVCC state is a compile error, +E calls inside transactions are rejected, -M parameters can't be mutated.
The Python solutions are 24–38 lines each. The U solutions are 14–28 lines. But the line count isn't the point — what's missing is. No import threading. No Lock(). No with self.lock:. No lock ordering. No ThreadPoolExecutor. No GIL.
For an LLM generating code, this is the difference between “hope the model remembers the lock” and “the compiler catches it if it doesn't.” The pit of success means the model that writes the obvious code writes the correct code.