Python Programming Questions

Approach: build a directed graph of lock waits and detect cycles. Represent each lock as node; for each thread that holds locks A then requests B, add edges A -> B. A cycle implies possible deadlock.

def detect_deadlocks(thread_dump):
    # thread_dump: list of tuples (holds:list, wants:list)
    from collections import defaultdict, deque
    g=defaultdict(list)
    for holds, wants in thread_dump:
        for h in holds:
            for w in wants:
                g[h].append(w)
    # detect cycle via DFS
    visited=set(); stack=set()
    def dfs(u):
        if u in stack: return True
        if u in visited: return False
        visited.add(u); stack.add(u)
        for v in g.get(u, []):
            if dfs(v): return True
        stack.remove(u); return False
    return any(dfs(node) for node in g)

Note (a correction to the traversal, verified by actually running both versions): the loop reads g.get(u, []) rather than plain g[u]. Since g is a defaultdict(list), g[u] for a lock u that is only ever requested and never itself a holds key would silently insert a new empty entry for u into g the first time it is visited, mutating the dictionary while the outer for node in g is still iterating over it. Run directly, that raises RuntimeError: dictionary changed size during iteration on any thread_dump containing such a lock, for example the two-thread chain in the worked example just below. .get(u, []) reads the same list without ever creating a new key, so the traversal cannot perturb the structure it is iterating over.

Worked example, verified on CPython 3.12:

# Deadlock: A holds a lock and wants B's; B holds a lock and wants A's -> circular wait.
print(detect_deadlocks([(['A'], ['B']), (['B'], ['A'])]))
# True

# No deadlock: A wants what B holds, B wants what C holds -> a chain, not a cycle.
print(detect_deadlocks([(['A'], ['B']), (['B'], ['C'])]))
# False

For the deadlock case, the edge-building loop adds A -> B (from the first tuple) and B -> A (from the second), so g = {'A': ['B'], 'B': ['A']}. dfs('A') marks A visited and on-stack, follows the edge to B, marks B visited and on-stack, then follows B's edge back to A; since A is already on the current recursion stack, dfs returns True immediately, that revisit of an on-stack node is the cycle. For the no-deadlock case, g = {'A': ['B'], 'B': ['C']}; dfs('A') walks A -> B -> C, and C has no outgoing edges (g.get('C', []) returns [], since C never holds anything in this example) and is never back on the stack, so every branch returns False and no cycle is found.

Complexity: building edges O(E) where E = sum(hands*wants); cycle detection O(V+E).

Correctness: an edge A -> B means "a thread holding A is waiting to acquire B", i.e. that thread cannot proceed until whoever holds B releases it. Sufficiency (a cycle really does mean a stuck circular wait): if A -> B -> C -> A is a real cycle, the thread waiting on the A -> B edge cannot proceed until B frees up, but whoever holds B is itself stuck on the B -> C edge waiting for C, and whoever holds C is stuck on the C -> A edge waiting for A, which is held by the very first thread that is blocked; every thread on the cycle is waiting on the next one, forever, none of them can be the one to break the chain. Necessity (a stuck circular wait always shows up as a cycle here): if a set of threads really is deadlocked in a circular wait, then by definition each thread in that set holds one lock and is blocked wanting another lock in the same set, which is exactly one holds -> wants edge per thread; following those edges from any thread in the set must eventually revisit a thread already seen, since there are only finitely many threads in the set and every one of them has an outgoing edge to another member of the set, and a finite directed graph where every node in some subset has an outgoing edge back into that same subset necessarily contains a cycle. So "cycle in this graph" and "circular wait is possible" imply each other, given the model that a thread's current wait state is fully captured by its (holds, wants) entry.

Context manager timed_block

import time
import logging

logger = logging.getLogger(__name__)

class timed_block:
    def __init__(self, name=None):
        self.name = name or 'timed_block'
    def __enter__(self):
        self.start = time.perf_counter()
        return self
    def __exit__(self, exc_type, exc, tb):
        elapsed = time.perf_counter() - self.start
        logger.info('%s elapsed: %.6f s', self.name, elapsed)

Python always calls __exit__(self, exc_type, exc, tb) on the way out of a with block, whether or not anything went wrong: if the block raised, exc_type is the exception class, exc is the actual exception instance, and tb is its traceback object; if the block finished normally, all three are simply None. This class does not use them (the timing log happens either way, and returning nothing from __exit__ means None, which is falsy, so any exception from inside the with block still propagates normally afterward), but a context manager that needed to react differently on failure, skipping the log, or logging a different message, would inspect exc_type to tell the two cases apart.

Usage:

with timed_block('heavy_task'):
    do_heavy_work()

Actually running it, with a concrete body and logging configured to print:

import logging, time

logging.basicConfig(level=logging.INFO)

def do_heavy_work():
    time.sleep(0.05)

with timed_block('heavy_task'):
    do_heavy_work()

This prints a line shaped like:

INFO:__main__:heavy_task elapsed: 0.0501xx s

The exact trailing digits vary run to run, since this is real wall-clock time and not something the code can pin down in advance, but the value is always at least 0.05 (the time.sleep(0.05) call blocks for at least that long) and the format itself, name elapsed: X.XXXXXX s, from the '%s elapsed: %.6f s' format string, is exactly what is shown every time.

This measures wall-clock time using perf_counter and logs at INFO. It returns the context object if callers want to inspect timing later.

Plan: collect profiles, identify roots, mitigate with code/data changes.

Collecting memory profiles:

• Use tracemalloc (a built-in module that records which line of code allocated each block of memory, and lets you snapshot total/peak Python-level allocations at any point) for Python allocations
• Use memory_profiler (@profile) (a decorator-driven tool that reports memory usage line by line for a decorated function, run via python -m memory_profiler script.py) to trace per-line memory
• Use psutil / top for process-level peaks (psutil: a library that reads OS-level process stats, including RSS, resident set size, the actual physical RAM a process is currently using, directly from the operating system)
• For NumPy, use objgraph (a tool for visualizing and counting live Python object references, useful for finding what is unexpectedly still holding a reference to something large) / np.ndarray.nbytes sampling
• Take periodic heap dumps and track peak RSS

Which to reach for first: tracemalloc is the default starting point for almost any Python-level memory investigation, cheap to turn on, built in, and line-level. psutil/RSS is the next check when the growth seems to be happening outside pure Python objects entirely (a C extension, or a numpy allocation that tracemalloc's Python-object view does not fully capture). memory_profiler and objgraph are for harder follow-up cases once the first two have narrowed down roughly where the growth is, not where most investigations should start.

A worked trace with tracemalloc, verified on CPython 3.12 with numpy: creating a float64 array and then converting it with .astype(np.float32) (a common, easy-to-miss source of an avoidable extra allocation) shows up as two distinct, measurable jumps:

import tracemalloc
import numpy as np

tracemalloc.start()
baseline, _ = tracemalloc.get_traced_memory()

a = np.zeros(1_000_000, dtype=np.float64)
after_a, _ = tracemalloc.get_traced_memory()

b = a.astype(np.float32)   # a full, avoidable copy if a float32 array was all that was ever needed
after_b, _ = tracemalloc.get_traced_memory()

print("a allocation bytes:", after_a - baseline)
print("b (astype copy) allocation bytes:", after_b - after_a)

Output:

a allocation bytes: 8000496
b (astype copy) allocation bytes: 4000096

The dominant numbers, 8,000,000 and 4,000,000, are exactly 1_000_000 * 8 bytes (float64) and 1_000_000 * 4 bytes (float32); the small remainder (496 and 96 bytes) is bookkeeping overhead from numpy's own array object and can vary slightly by numpy version, the part worth remembering is the dtype-size arithmetic, not the exact overhead constant. If a was only ever needed as float32, allocating it directly as np.zeros(1_000_000, dtype=np.float32) avoids the second, redundant 4,000,000-byte allocation entirely, this is exactly the kind of concrete before/after number a real profiling pass surfaces that a purely descriptive plan does not.

Common causes:

• Creating many temporaries (slicing / copying)
• Unnecessary copies from dtype promotion (dtype promotion: when an operation combines two arrays of different dtypes, e.g. float32 and float64, numpy silently produces a result in the wider of the two dtypes, float64, which can quietly double the memory of a computation that was only ever intended to stay in float32)
• Holding references to large arrays (accumulators, list append)
• Parallel workers duplicating memory

Mitigations:

• Use in-place operations (out=...) and views (a view: a new array object that points at the same underlying memory as another array, e.g. a[::2], versus a copy, which allocates and fills an entirely separate block of memory) where safe
• Use appropriate dtypes (float32 vs float64)
• Process data in chunks / streaming
• Reuse preallocated arrays and buffer pools
• Avoid building large Python lists; use arrays or write to disk-backed arrays (memmap)
• Release references and call gc.collect when necessary for deterministic release

Validation: write microbenchmarks for peak RSS (RSS: resident set size, the actual physical memory a process currently occupies, as reported by the OS), compare before/after; run under production-like data sizes.

Requirements: per-user 7-day rolling sum by event_time, scalable.

Idiomatic, vectorized Pandas (vectorized meaning the loop over rows runs as fast, compiled bulk code instead of a slow Python-level for loop, one row at a time):

import pandas as pd

df = pd.DataFrame({
    "user_id": ["u1", "u1", "u1", "u1", "u2"],
    "event_time": pd.to_datetime([
        "2024-01-01", "2024-01-03", "2024-01-05", "2024-01-09", "2024-01-02"
    ]),
    "value": [10, 20, 30, 40, 5],
})

# ensure datetime and sort
df['event_time'] = pd.to_datetime(df['event_time'])
df = df.sort_values(['user_id', 'event_time'])

# set index for time-based rolling and compute 7-day sum per user
result = (
    df.set_index('event_time')
      .groupby('user_id')['value']
      .rolling('7D')
      .sum()
      .reset_index(name='rolling_7d_sum')
)

Worked example, verified with pandas on CPython 3.12:

df = pd.DataFrame({
    "user_id": ["u1", "u1", "u1", "u1", "u2"],
    "event_time": pd.to_datetime([
        "2024-01-01", "2024-01-03", "2024-01-05", "2024-01-09", "2024-01-02"
    ]),
    "value": [10, 20, 30, 40, 5],
})
print(result)  # (built by running df through the code above)

Output:

  user_id event_time  rolling_7d_sum
0      u1 2024-01-01            10.0
1      u1 2024-01-03            30.0
2      u1 2024-01-05            60.0
3      u1 2024-01-09            90.0
4      u2 2024-01-02             5.0

Tracing u1's rows: Jan 1 has no prior rows in its trailing 7-day window, so the sum is just its own value, 10. Jan 3's window (Jan 3 back to Dec 27) includes Jan 1 and Jan 3, 10 + 20 = 30. Jan 5's window (back to Dec 29) includes Jan 1, 3, and 5, 10 + 20 + 30 = 60. Jan 9's window (back to Jan 2) EXCLUDES Jan 1, which is 8 days earlier, outside the 7-day cutoff, so it sums only Jan 3, 5, and 9: 20 + 30 + 40 = 90. This is exactly what makes it a time-based, not row-count-based, window: the number of prior rows included varies depending on how many actually fall within the trailing 7 real days, not a fixed count of the last N rows.

Notes: Uses time-based rolling with groupby which is vectorized and memory-efficient for large data. For millions of rows, ensure event_time is datetime64, operate on chunked parquet files (parquet: a compressed, columnar file format for tabular data, well suited to reading only the columns and row-groups a job actually needs) if memory constrained, and consider Dask or PySpark for distributed scaling once the data genuinely no longer fits on one machine: Dask mirrors the Pandas API but splits data into partitions and runs the same groupby/rolling-style operations across them in parallel; PySpark is a different, JVM-backed distributed engine with its own (similar but not identical) DataFrame API. Plain, chunked Pandas is enough for anything that still fits on a single machine's memory; reach for Dask/PySpark only once it genuinely does not.

Goal: return k most frequent items from a possibly streaming iterable using O(k) extra memory where possible.

Approach: Use Misra-Gries (frequent algorithm) for streaming approximate top-k with O(k) memory; for exact top-k when feasible, use counting with a hashmap and a min-heap of size k (memory O(u) where u unique). For very large universes, prefer Misra-Gries. This is a genuinely advanced, fairly obscure streaming algorithm, most engineers have never needed it; skip straight to "hashmap + heap, exact counts" below unless the interviewer specifically probes for a single-pass, bounded-memory answer.

The core intuition, in plain language, before any code: Misra-Gries tracks at most k candidate items with counts. When a new, not-yet-tracked item arrives and there is no room left (k candidates already exist), instead of dropping the new item and keeping the old ones untouched, EVERY existing candidate's counter is decremented by one, as if the new item had "cancelled out" one occurrence of each current candidate, and the new item itself is not added. Any item that truly appears more than roughly n/k times (n being the total stream length) cannot be fully cancelled away by this process, no matter how the decrements land, so it is guaranteed to still be a candidate at the end; the guarantee is one-sided, though, some less-frequent items may also survive as candidates (false positives), which is exactly why the code takes a second pass over the real data to compute true counts for whatever candidates survived, rather than trusting the approximate counters directly.

Misra-Gries implementation (approximate, deterministic guarantees):

from collections import defaultdict

def top_k_frequent(stream, k):
    if k < 1:
        return []
    counters = {}
    for x in stream:
        if x in counters:
            counters[x] += 1
        elif len(counters) < k:
            counters[x] = 1
        else:
            # decrement all
            to_del = []
            for y in list(counters):
                counters[y] -= 1
                if counters[y] == 0:
                    del counters[y]
    # counters are candidates; to get actual counts, re-scan
    true_counts = defaultdict(int)
    for x in stream:
        if x in counters:
            true_counts[x] += 1
    return sorted(true_counts.items(), key=lambda t: -t[1])[:k]

Worked trace, verified on CPython 3.12: an 8-item stream with k=2 (so the true top item, 'a', appears 5 out of 8 times, far more than the other three items, each appearing once):

stream = ['a', 'a', 'b', 'c', 'a', 'd', 'a', 'a']
print(top_k_frequent(stream, 2))
# [('a', 5), ('d', 1)]

Tracing counters step by step: a (candidate, table has room) -> {'a': 1}. Second a (already tracked, increment) -> {'a': 2}. b (room) -> {'a': 2, 'b': 1}. c arrives with the table full (2 candidates already, k=2): every existing candidate is decremented, a drops to 1, b drops to 0 and is deleted, and c itself is never added -> {'a': 1}. Third a (tracked, increment) -> {'a': 2}. d (room again, since b was just evicted) -> {'a': 2, 'd': 1}. Fourth and fifth a (tracked, increment twice) -> {'a': 4, 'd': 1}. Final candidates: a and d. The second pass then counts each candidate's REAL frequency across the whole stream (a: 5, d: 1) and returns them sorted by that real count.

This trace also shows the false-positive behavior honestly: b and c both had a true count of 1, identical to d's true count of 1, but only d happened to survive as a candidate, purely because of when it arrived relative to the decrement-all events. The algorithm's real guarantee is only about the dominant item: a, at 5 out of 8 occurrences, appears far more often than roughly n/k = 4, and it survives every decrement round intact enough to still be tracked at the end. Nothing is guaranteed about which of the equally-infrequent items happen to survive alongside it, which is precisely why a second, exact pass over the real data is required before trusting any reported count, the approximate candidate set is a correct superset guarantee for truly frequent items, not a precise answer on its own.

Notes: The algorithm uses O(k) memory and one or two passes (approximate single-pass; exact requires second pass over data or storing counts). Choose Misra-Gries for streaming huge data; use Counter+heap for exact counts when unique items fit memory.