Skip to content

LRU/LFU/TTL Cache β€” Implementation

Single-file Python implementation (lru_cache.py, stdlib only, 3.10+) with a unittest suite (test_lru_cache.py).


▢️ How to Run

cd python-low-level-design/lru-cache
python3 lru_cache.py                  # deterministic demo (manual clock for TTL)
python3 -m unittest test_lru_cache    # 24 tests incl. model-based and concurrency tests

πŸ—ΊοΈ Map of the Code

Piece Responsibility
_Node key, value, expires_at, freq, prev/next (__slots__)
_DList Doubly linked list with head/tail sentinels; push_front, remove, back all O(1)
BaseCache Template method: owns the key -> node map, capacity, TTL, stats. Policy plugs in through four hooks
LRUCache One _DList, front = most recently used
LFUCache freq -> _DList buckets + min_freq; ties broken by recency
ThreadSafeCache One Lock around any cache; get_or_load with single-flight
StripedCache N ThreadSafeCache shards, hash(key) % N
CacheStats hits, misses, evictions (capacity), expirations (TTL)
ManualClock Injectable clock for deterministic TTL tests
Operation LRU LFU
get / put / delete O(1) O(1)
… with a TTL set + O(log n) heap push + O(log n) heap push
purge_expired O(k log n) for k expired same
evict after a delete emptied the min bucket β€” O(F) once, F = distinct frequencies

πŸ”‘ Key Design Decisions

1. Hashmap + doubly linked list, with sentinels

The map answers "where is key k?" in O(1); the list answers "who is least recent?" in O(1) and lets you unlink any node in O(1) because each node knows its neighbours. A singly linked list can't unlink in O(1) (you need the predecessor). The node stores its key so eviction from the tail can delete the map entry.

Head and tail sentinel nodes mean push_front and remove never test for None β€” the head/tail special cases are where hand-written LRUs usually break in interviews.

Why not just OrderedDict? In production Python, do use it (or functools.lru_cache):

class OrderedDictLRU:
    def __init__(self, cap): self.cap, self.d = cap, OrderedDict()
    def get(self, k):
        if k not in self.d: return None
        self.d.move_to_end(k); return self.d[k]
    def put(self, k, v):
        if k in self.d: self.d.move_to_end(k)
        self.d[k] = v
        if len(self.d) > self.cap: self.d.popitem(last=False)

CPython's OrderedDict is a hashmap plus a doubly linked list internally. Interviewers ask for the hand-built version to see the pointer work, and because LFU and TTL need node-level control (per-frequency lists, expiry metadata) that OrderedDict doesn't give you. Say both.

2. One template, policies as hooks

BaseCache.get/put/delete do the map, TTL and stats work once. A policy implements four hooks: _insert (new node), _touch (hit or overwrite), _unlink (evict/expire/delete), _victim (who goes when full). LRU is 12 lines; LFU is 40. FIFO would be LRU with an empty _touch.

This replaces the old EvictionStrategy design, which treated TTL as an eviction policy β€” a TTL-only cache that was full with nothing expired raised ValueError on put. TTL is orthogonal to the eviction policy: any cache can have expiring entries.

3. O(1) LFU

_buckets: {1: [e, d], 2: [c], 3: [a]}     each a DLL, front = most recent
_min_freq: 1
  • get/overwrite: move the node from bucket f to the front of bucket f+1. If that emptied bucket f and f was the minimum, the new minimum is f+1 β€” the node just moved there. O(1).
  • insert: new node goes to bucket 1, min_freq = 1. O(1).
  • evict: back (least recent) of _buckets[min_freq]. O(1). Tie-breaking by recency matters: without it, eviction among equal frequencies is arbitrary and tests become non-deterministic.
  • delete/expire: if it empties the min bucket, the next minimum is unknown. Mark min_freq = None and recompute with min(_buckets) only if an eviction happens before the next insert (which resets it to 1). In practice that is rare.

For strict O(1) on every path, replace the freq -> list dict with a doubly linked list of frequency buckets in ascending order (Shah, Mitra & Matani's O(1) LFU): the first bucket is always the minimum. More pointer code for a case that rarely matters β€” mention it, don't write it unless asked.

Classic LFU weakness: an entry that was hot an hour ago keeps a high count forever. Production variants decay counts (Redis's 8-bit logarithmic counter is decremented once per lfu-decay-time minutes idle) or use W-TinyLFU (Caffeine).

4. TTL: lazy expiry plus a heap

  • Lazy: get checks expires_at and treats an expired entry as a miss (and removes it).
  • Active: each TTL'd put pushes (expires_at, seq, node) onto a min-heap. purge_expired() pops while the top is due. Overwrites and deletes leave stale heap entries; a pop is honoured only if map[key] is node and the expiry still matches. The heap is rebuilt when it exceeds 2 Γ— size + 32, so it stays O(n).
  • Before evicting a live entry, a full cache purges expired ones first: expired entries are free space.

Redis does the same split: lazy expiry on access plus a periodic job that samples keys with a TTL and deletes the expired ones. A timer-per-key would cost a thread or a timer-wheel slot per entry. Time comes from an injected clock (default time.monotonic β€” wall-clock jumps must not expire things); tests use ManualClock instead of sleep.

5. Thread-safety and lock granularity

  • get is a write. It moves a node (LRU) or changes its frequency (LFU). A read-write lock therefore gives no read parallelism β€” every call needs exclusive access. This is the most common wrong answer.
  • ThreadSafeCache: one plain Lock per cache, held for the whole operation, so map, list and heap change together. Not RLock: nothing re-enters.
  • StripedCache (lock striping): N independent shards, each with its own lock and capacity / N slots. Unrelated keys no longer contend. Costs: eviction is per shard (approximate global LRU), a hot shard evicts while others have room, and whole-cache operations (size, clear, resize) touch every shard. Pick N β‰ˆ 2–4Γ— core count, a power of two, and use a well-mixed hash.
  • CPython caveat: under the GIL, pure-Python cache operations are serialised anyway, so striping mainly cuts lock convoying. On free-threaded 3.13t+, or in Java/Go/C++, it scales throughput. Caffeine goes further: reads go to lock-free ring buffers and are replayed into the LRU in batches under a lock that is only tryLocked, so readers never block on recency updates.
  • Single-flight get_or_load: on a miss, the first caller registers an in-flight marker under the lock, releases the lock, runs the loader, then publishes. Concurrent callers for the same key wait on that marker. Ten threads missing on one key cause one database read, not ten (cache-stampede protection). Errors go to all waiters and are not cached.

Known gap to say out loud: if another thread puts a newer value while a load is in flight, the loader's older result overwrites it. Fix with a per-key version checked before publishing, or by invalidating the in-flight entry on put.


🧩 Where to Extend

Ask Change
FIFO / MRU / random New BaseCache subclass, 4 hooks
Size-aware eviction (bytes, not count) Track weight per node; evict in a loop until total_weight + new <= max
Eviction listener / write-back Call a callback in _remove with (key, value, reason)
Scan resistance Segmented LRU / 2Q: new entries enter a probation list, a second hit promotes them
Refresh-ahead In get, if expires_at - now < threshold, trigger an async get_or_load
Distributed See the HLD: consistent hashing across nodes, each node one of these

πŸ“„ Full Source

"""
LRU / LFU Cache with TTL - Low Level Design
-------------------------------------------
    BaseCache (ABC, template method)
      owns: key -> node map, capacity, TTL (lazy + heap-driven purge), stats
      hooks: _insert / _touch / _unlink / _victim   <- the eviction policy
    LRUCache   hashmap + one doubly linked list             get/put O(1)
    LFUCache   hashmap + one DLL per frequency + min_freq   get/put O(1)
    ThreadSafeCache   one lock around any BaseCache, plus single-flight get_or_load
    StripedCache      N independently locked shards, hash(key) % N

Complexities (n = entries):
    get / put / delete           O(1)  (+ O(log n) heap push when a TTL is set)
    purge_expired                O(k log n) for k expired entries
    LFU delete/expire of the     O(F) to recompute min_freq, F = distinct
      last key at min_freq         frequencies; deferred until the next eviction

Not thread-safe on their own. Note that get() MUTATES (moves a node / bumps a
frequency), so a read-write lock buys nothing for LRU/LFU: every call needs
exclusive access. Scale with striping, not RW locks.
"""

from __future__ import annotations

import heapq
import itertools
import threading
import time
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Callable, Generic, Hashable, Iterator, Optional, TypeVar

K = TypeVar("K", bound=Hashable)
V = TypeVar("V")
Clock = Callable[[], float]

_MISSING = object()


# --- Doubly linked list with sentinels -----------------------------------

class _Node(Generic[K, V]):
    __slots__ = ("key", "value", "expires_at", "freq", "prev", "next")

    def __init__(self, key, value, expires_at: Optional[float]):
        self.key = key
        self.value = value
        self.expires_at = expires_at
        self.freq = 0
        self.prev: Optional[_Node] = None
        self.next: Optional[_Node] = None


class _DList:
    """head <-> n1 <-> ... <-> tail. Front = most recent. Sentinels remove
    every head/tail None check, which is where hand-rolled DLLs go wrong."""

    __slots__ = ("_head", "_tail", "_len")

    def __init__(self):
        self._head = _Node(None, None, None)
        self._tail = _Node(None, None, None)
        self._head.next = self._tail
        self._tail.prev = self._head
        self._len = 0

    def push_front(self, node: _Node) -> None:
        node.prev = self._head
        node.next = self._head.next
        self._head.next.prev = node
        self._head.next = node
        self._len += 1

    def remove(self, node: _Node) -> None:
        node.prev.next = node.next
        node.next.prev = node.prev
        node.prev = node.next = None
        self._len -= 1

    def back(self) -> Optional[_Node]:
        return None if self._len == 0 else self._tail.prev

    def __len__(self) -> int:
        return self._len

    def __iter__(self) -> Iterator[_Node]:   # front (MRU) to back (LRU)
        node = self._head.next
        while node is not self._tail:
            yield node
            node = node.next


# --- Stats ----------------------------------------------------------------

@dataclass
class CacheStats:
    hits: int = 0
    misses: int = 0
    evictions: int = 0      # live entries pushed out by capacity
    expirations: int = 0    # entries removed because their TTL passed

    @property
    def hit_rate(self) -> float:
        total = self.hits + self.misses
        return self.hits / total if total else 0.0

    def __add__(self, other: CacheStats) -> CacheStats:
        return CacheStats(self.hits + other.hits, self.misses + other.misses,
                          self.evictions + other.evictions,
                          self.expirations + other.expirations)


# --- Base cache (template method) -----------------------------------------

class BaseCache(ABC, Generic[K, V]):
    def __init__(self, capacity: int, default_ttl: Optional[float] = None,
                 clock: Clock = time.monotonic):
        if capacity <= 0:
            raise ValueError("capacity must be positive")
        if default_ttl is not None and default_ttl <= 0:
            raise ValueError("default_ttl must be positive")
        self._capacity = capacity
        self._default_ttl = default_ttl
        self._clock = clock
        self._map: dict[K, _Node] = {}
        # (expires_at, seq, node). Entries go stale when a key is rewritten
        # or deleted; we skip those on pop and compact when the heap gets big.
        self._expiry_heap: list[tuple[float, int, _Node]] = []
        self._seq = itertools.count()
        self._stats = CacheStats()

    # ---- public API ----
    def get(self, key: K, default: Optional[V] = None) -> Optional[V]:
        """Value for key, or `default` if absent or expired. Counts as a use."""
        node = self._map.get(key)
        if node is None:
            self._stats.misses += 1
            return default
        if self._is_expired(node):
            self._remove(node)
            self._stats.expirations += 1
            self._stats.misses += 1
            return default
        self._touch(node)
        self._stats.hits += 1
        return node.value

    def put(self, key: K, value: V, ttl: Optional[float] = None) -> None:
        """Insert or overwrite. ttl (seconds) overrides default_ttl for this key.
        Overwriting counts as a use and resets the TTL."""
        if ttl is not None and ttl <= 0:
            raise ValueError("ttl must be positive")
        ttl = ttl if ttl is not None else self._default_ttl
        expires_at = self._clock() + ttl if ttl is not None else None

        node = self._map.get(key)
        if node is not None:
            node.value = value
            node.expires_at = expires_at
            self._schedule(node)
            self._touch(node)
            return

        if len(self._map) >= self._capacity:
            # Reclaim dead entries before evicting a live one.
            if self.purge_expired() == 0:
                victim = self._victim()
                self._remove(victim)
                self._stats.evictions += 1
        node = _Node(key, value, expires_at)
        self._map[key] = node
        self._insert(node)
        self._schedule(node)

    def delete(self, key: K) -> bool:
        node = self._map.get(key)
        if node is None:
            return False
        self._remove(node)
        return True

    def purge_expired(self) -> int:
        """Drop every expired entry. Amortised O(log n) per removed entry."""
        now = self._clock()
        removed = 0
        heap = self._expiry_heap
        while heap and heap[0][0] <= now:
            expires_at, _, node = heapq.heappop(heap)
            if self._map.get(node.key) is node and node.expires_at == expires_at:
                self._remove(node)
                self._stats.expirations += 1
                removed += 1
        return removed

    def __contains__(self, key: K) -> bool:
        """Present and not expired. Does NOT count as a use."""
        node = self._map.get(key)
        return node is not None and not self._is_expired(node)

    def __len__(self) -> int:
        """Entries held, including expired ones not yet purged."""
        return len(self._map)

    @property
    def capacity(self) -> int:
        return self._capacity

    @property
    def stats(self) -> CacheStats:
        s = self._stats
        return CacheStats(s.hits, s.misses, s.evictions, s.expirations)

    # ---- internals ----
    def _is_expired(self, node: _Node) -> bool:
        return node.expires_at is not None and node.expires_at <= self._clock()

    def _schedule(self, node: _Node) -> None:
        if node.expires_at is None:
            return
        heapq.heappush(self._expiry_heap, (node.expires_at, next(self._seq), node))
        # Rewrites leave stale heap entries; keep the heap O(live entries).
        if len(self._expiry_heap) > 2 * len(self._map) + 32:
            self._expiry_heap = [(n.expires_at, next(self._seq), n)
                                 for n in self._map.values() if n.expires_at is not None]
            heapq.heapify(self._expiry_heap)

    def _remove(self, node: _Node) -> None:
        del self._map[node.key]
        self._unlink(node)

    # ---- policy hooks ----
    @abstractmethod
    def _insert(self, node: _Node) -> None:
        """A new node entered the cache."""

    @abstractmethod
    def _touch(self, node: _Node) -> None:
        """An existing node was read or overwritten."""

    @abstractmethod
    def _unlink(self, node: _Node) -> None:
        """Forget a node (eviction, expiry or delete)."""

    @abstractmethod
    def _victim(self) -> _Node:
        """The node to evict when full. Cache is non-empty."""

    def _check_invariants(self) -> None:   # for tests
        assert len(self._map) <= self._capacity


class LRUCache(BaseCache[K, V]):
    """Evicts the least recently used entry. Front of the list = MRU."""

    def __init__(self, capacity: int, default_ttl: Optional[float] = None,
                 clock: Clock = time.monotonic):
        super().__init__(capacity, default_ttl, clock)
        self._order = _DList()

    def _insert(self, node: _Node) -> None:
        self._order.push_front(node)

    def _touch(self, node: _Node) -> None:
        self._order.remove(node)
        self._order.push_front(node)

    def _unlink(self, node: _Node) -> None:
        self._order.remove(node)

    def _victim(self) -> _Node:
        return self._order.back()

    def keys_mru_to_lru(self) -> list[K]:
        return [n.key for n in self._order]

    def _check_invariants(self) -> None:
        super()._check_invariants()
        listed = [n.key for n in self._order]
        assert len(listed) == len(self._order) == len(self._map)
        assert set(listed) == set(self._map)


class LFUCache(BaseCache[K, V]):
    """Evicts the least frequently used entry; ties go to the least recently
    used among them. New entries start at freq 1.

    _buckets[f] is a DLL of nodes used exactly f times (front = most recent).
    _min_freq is the smallest non-empty f, or None when it must be recomputed.
    """

    def __init__(self, capacity: int, default_ttl: Optional[float] = None,
                 clock: Clock = time.monotonic):
        super().__init__(capacity, default_ttl, clock)
        self._buckets: dict[int, _DList] = {}
        self._min_freq: Optional[int] = None

    def _bucket(self, freq: int) -> _DList:
        b = self._buckets.get(freq)
        if b is None:
            b = self._buckets[freq] = _DList()
        return b

    def _detach(self, node: _Node) -> bool:
        """Remove from its bucket; True if that emptied the min bucket."""
        b = self._buckets[node.freq]
        b.remove(node)
        if len(b) == 0:
            del self._buckets[node.freq]
            return node.freq == self._min_freq
        return False

    def _insert(self, node: _Node) -> None:
        node.freq = 1
        self._bucket(1).push_front(node)
        self._min_freq = 1                       # nothing can be below 1

    def _touch(self, node: _Node) -> None:
        emptied_min = self._detach(node)
        node.freq += 1
        self._bucket(node.freq).push_front(node)
        if emptied_min:
            self._min_freq = node.freq           # it moved exactly one bucket up

    def _unlink(self, node: _Node) -> None:
        if self._detach(node):
            self._min_freq = None                # recompute lazily in _victim

    def _victim(self) -> _Node:
        if self._min_freq is None:
            self._min_freq = min(self._buckets)  # O(F); rare: only after a
        return self._buckets[self._min_freq].back()  # delete/expire of the min

    def frequency(self, key: K) -> int:
        node = self._map.get(key)
        return node.freq if node else 0

    def _check_invariants(self) -> None:
        super()._check_invariants()
        assert sum(len(b) for b in self._buckets.values()) == len(self._map)
        for f, b in self._buckets.items():
            assert len(b) > 0 and all(n.freq == f for n in b)
        if self._buckets and self._min_freq is not None:
            assert self._min_freq == min(self._buckets)


# --- Concurrency ----------------------------------------------------------

class _Flight:
    __slots__ = ("done", "value", "error")

    def __init__(self):
        self.done = threading.Event()
        self.value = None
        self.error: Optional[BaseException] = None


class ThreadSafeCache(Generic[K, V]):
    """One mutex around a BaseCache. A plain Lock, not an RLock: nothing
    re-enters, and not a read-write lock: get() mutates recency/frequency."""

    def __init__(self, cache: BaseCache[K, V]):
        self._cache = cache
        self._lock = threading.Lock()
        self._inflight: dict[K, _Flight] = {}

    def get(self, key: K, default: Optional[V] = None) -> Optional[V]:
        with self._lock:
            return self._cache.get(key, default)

    def put(self, key: K, value: V, ttl: Optional[float] = None) -> None:
        with self._lock:
            self._cache.put(key, value, ttl)

    def delete(self, key: K) -> bool:
        with self._lock:
            return self._cache.delete(key)

    def purge_expired(self) -> int:
        with self._lock:
            return self._cache.purge_expired()

    def get_or_load(self, key: K, loader: Callable[[K], V],
                    ttl: Optional[float] = None) -> V:
        """Cache-aside with single-flight: on a miss, exactly one caller runs
        loader(key) (outside the lock); concurrent callers for the same key
        wait for its result instead of stampeding the backing store."""
        with self._lock:
            value = self._cache.get(key, _MISSING)
            if value is not _MISSING:
                return value
            flight = self._inflight.get(key)
            leader = flight is None
            if leader:
                flight = self._inflight[key] = _Flight()
        if not leader:
            flight.done.wait()
            if flight.error is not None:
                raise flight.error
            return flight.value
        try:
            flight.value = loader(key)
        except BaseException as e:
            flight.error = e
            raise
        finally:
            with self._lock:
                if flight.error is None:
                    self._cache.put(key, flight.value, ttl)
                del self._inflight[key]
            flight.done.set()
        return flight.value

    def __contains__(self, key: K) -> bool:
        with self._lock:
            return key in self._cache

    def __len__(self) -> int:
        with self._lock:
            return len(self._cache)

    @property
    def stats(self) -> CacheStats:
        with self._lock:
            return self._cache.stats

    def _check_invariants(self) -> None:
        with self._lock:
            self._cache._check_invariants()


class StripedCache(Generic[K, V]):
    """N shards, each its own cache + lock; a key lives in shard hash(key) % N.

    Contention drops roughly N-fold because unrelated keys no longer share a
    lock. The price: eviction is per shard (approximate global LRU/LFU), and
    capacity is split evenly, so a hot shard evicts while others have room.
    Under CPython's GIL pure-Python work is still serialised; striping pays
    off with free-threaded builds or when the critical section releases the
    GIL. In Java/Go/C++ this is the standard design (cf. Guava/Caffeine
    segments, ConcurrentHashMap's per-bin locking).
    """

    def __init__(self, capacity: int, stripes: int,
                 factory: Callable[[int], BaseCache[K, V]]):
        if stripes <= 0 or capacity < stripes:
            raise ValueError("need 1 <= stripes <= capacity")
        per_shard = -(-capacity // stripes)          # ceil
        self._shards = [ThreadSafeCache(factory(per_shard)) for _ in range(stripes)]

    def _shard(self, key: K) -> ThreadSafeCache[K, V]:
        return self._shards[hash(key) % len(self._shards)]

    def get(self, key: K, default: Optional[V] = None) -> Optional[V]:
        return self._shard(key).get(key, default)

    def put(self, key: K, value: V, ttl: Optional[float] = None) -> None:
        self._shard(key).put(key, value, ttl)

    def delete(self, key: K) -> bool:
        return self._shard(key).delete(key)

    def get_or_load(self, key: K, loader: Callable[[K], V],
                    ttl: Optional[float] = None) -> V:
        return self._shard(key).get_or_load(key, loader, ttl)

    def __contains__(self, key: K) -> bool:
        return key in self._shard(key)

    def __len__(self) -> int:
        return sum(len(s) for s in self._shards)

    @property
    def stats(self) -> CacheStats:
        total = CacheStats()
        for s in self._shards:
            total = total + s.stats
        return total

    def _check_invariants(self) -> None:
        for s in self._shards:
            s._check_invariants()


# --- Demo -----------------------------------------------------------------

class ManualClock:
    """Deterministic clock for demos and tests."""

    def __init__(self, start: float = 0.0):
        self.now = start

    def __call__(self) -> float:
        return self.now

    def advance(self, seconds: float) -> None:
        self.now += seconds


def demo() -> None:
    print("=== LRU ===")
    lru: LRUCache[str, int] = LRUCache(3)
    for k, v in (("a", 1), ("b", 2), ("c", 3)):
        lru.put(k, v)
    lru.get("a")                    # a becomes MRU; b is now LRU
    lru.put("d", 4)                 # evicts b
    print("  order MRU->LRU:", lru.keys_mru_to_lru(), "| b present:", "b" in lru)

    print("\n=== LFU (ties broken by recency) ===")
    lfu: LFUCache[str, int] = LFUCache(3)
    for k, v in (("a", 1), ("b", 2), ("c", 3)):
        lfu.put(k, v)
    lfu.get("a"); lfu.get("a"); lfu.get("c")
    lfu.put("d", 4)                 # b has freq 1 and is the only one -> evicted
    lfu.put("e", 5)                 # d (freq 1) evicted, not c (freq 2)
    print("  freqs:", {k: lfu.frequency(k) for k in "abcde"})

    print("\n=== TTL (manual clock) ===")
    clock = ManualClock()
    ttl_cache: LRUCache[str, str] = LRUCache(2, default_ttl=10, clock=clock)
    ttl_cache.put("session", "alice")
    ttl_cache.put("token", "xyz", ttl=2)
    clock.advance(3)
    print("  after 3s: token =", ttl_cache.get("token"), "| session =", ttl_cache.get("session"))
    ttl_cache.put("x", "1")
    clock.advance(8)                # session (t=10) expires; x (t=13) does not
    ttl_cache.put("y", "2")         # full: purges expired 'session' instead of evicting x
    print("  after 11s: keys =", ttl_cache.keys_mru_to_lru(), "|", ttl_cache.stats)

    print("\n=== Thread-safe + single-flight loader ===")
    calls = []
    cache: ThreadSafeCache[int, str] = ThreadSafeCache(LRUCache(100))

    def slow_load(key: int) -> str:
        calls.append(key)
        time.sleep(0.05)
        return f"row-{key}"

    threads = [threading.Thread(target=cache.get_or_load, args=(7, slow_load))
               for _ in range(10)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    print(f"  10 concurrent misses for key 7 -> loader ran {len(calls)} time(s);",
          "value:", cache.get(7))

    striped: StripedCache[int, int] = StripedCache(64, 8, LRUCache)
    for i in range(200):
        striped.put(i, i * i)
    print(f"  striped: 8 shards x 8 slots, inserted 200 -> holds {len(striped)}")


if __name__ == "__main__":
    demo()