Python Programming Questions

Go vocabulary, translated for a Python reader: a goroutine is Go's lightweight, concurrently-running function, similar in spirit to a Python thread but far cheaper to start and typically used in much greater numbers in real Go code; a channel is a typed, thread-safe queue you send values into and receive values out of, roughly like a queue.Queue shared between Python threads, except the compiler enforces the type of what flows through it; a WaitGroup is a counter that lets the caller block until every worker goroutine has signaled it is done, similar to calling .join() on a list of Python Thread objects; ctx (short for context) carries a shared cancellation signal through the call tree, and ctx.Done() returns a channel that closes the moment that signal fires, so any goroutine can cheaply check "has someone asked everything to stop?" without polling a shared boolean, comparable to checking a Python threading.Event.

Structure

I’d use a bounded worker pool. A feeder sends files with an index into a jobs channel. Each worker validates one file, sends {index, result, err} to a results channel, and watches ctx.Done() so context cancellation, meaning a cooperative stop signal, is fast.

Ordering
A single collector keeps nextIndex and a map of out-of-order results. If result 7 arrives before 6, store it until 6 is ready, then flush in submission order.

Fatal parse error
If a worker sees an unrecoverable parse error, it sends the error and calls cancel(). That stops the feeder, makes workers exit on ctx.Done(), and prevents new work from starting.

Shutdown

• close jobs after feeding stops
• WaitGroup waits for workers
• close results after workers finish
• collector drains until closed or canceled

Worked example: files 1, 2, 3, 4 arrive. If file 3 has a fatal parse error, 1 and 2 can still be emitted, 4 is never started, and the summary reports the error immediately.

Shape of the code (a sketch of the structure, not a full compiled program):

type job struct {
    index int
    path  string
}
type result struct {
    index  int
    output string
    err    error
}

func run(ctx context.Context, paths []string, numWorkers int) []result {
    ctx, cancel := context.WithCancel(ctx)
    defer cancel()

    jobs := make(chan job)
    results := make(chan result)
    var wg sync.WaitGroup

    for w := 0; w < numWorkers; w++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for j := range jobs {
                out, err := validate(j.path)
                if err != nil && isFatal(err) {
                    cancel()
                }
                results <- result{index: j.index, output: out, err: err}
            }
        }()
    }

    go func() {
        for i, p := range paths {
            select {
            case jobs <- job{index: i, path: p}:
            case <-ctx.Done():
                close(jobs)
                return
            }
        }
        close(jobs)
    }()

    go func() {
        wg.Wait()
        close(results)
    }()

    return collectInOrder(results)
}

The jobs and results lines are the channel declarations; the three go func() { ... }() blocks are the goroutine launches, one pool of workers, one feeder, one closer. select { case jobs <- job{...}: ... case <-ctx.Done(): ... } is how the feeder stays responsive to cancellation even while trying to send a job that a worker isn't ready to receive yet, instead of blocking forever on a full channel after a fatal error.

This gives parallelism, ordered output, and fast failure without deadlocks.

Approach

I would stream the file line by line so memory stays bounded, meaning it stays roughly constant no matter how large the file is. For each line, trim whitespace, skip blanks, try to parse the fields, and update counters. Malformed lines go into a small issue count or log so the job can continue.

from dataclasses import dataclass


@dataclass
class Report:
    total: int = 0
    overdue: int = 0
    days_overdue_sum: int = 0
    malformed: int = 0


def process_tickets(path: str) -> Report:
    report = Report()

    with open(path, 'r', encoding='utf-8') as f:
        for line_no, line in enumerate(f, start=1):
            line = line.strip()
            if not line:
                continue

            parts = line.split(',')
            if len(parts) != 3:
                report.malformed += 1
                continue

            try:
                status = parts[1].strip().lower()
                days = int(parts[2].strip())
            except ValueError:
                report.malformed += 1
                continue

            if not parts[0].strip():
                report.malformed += 1
                continue

            report.total += 1
            if status == 'overdue':
                report.overdue += 1
                report.days_overdue_sum += days

    return report

Worked example: with A1,overdue,3, a blank line, and B2,open,0, the report counts 2 valid records, 1 overdue ticket, and 3 total overdue days.

Tradeoffs

• Fast and memory-safe for huge files
• One pass only, so you should add any needed counters up front
• If operators need examples of bad lines, log the first few rather than storing every bad line

Complexity: O(n) time and O(1) memory for the counters.

Tests and benchmarks:

• Correctness tests: unit tests comparing NumPy reference outputs vs Numba outputs on varied inputs including edge cases (NaN, infinities, zero-length arrays, non-contiguous/strided arrays). Use property-based tests (the hypothesis library: instead of writing individual test inputs by hand, you describe a rule the output must always satisfy, and the library generates many random inputs trying to break that rule) for broader coverage.
• Numerical accuracy: assert tolerances (rtol/atol, relative and absolute tolerance) for floating ops, using np.testing.assert_allclose rather than exact equality, since compiled and interpreted floating-point code can differ in the last few bits even when both are correct.
• Performance benchmarks: microbenchmarks measuring time and compilation overhead (first call vs steady-state) using asv (airspeed velocity, a benchmarking tool that tracks a function's performance across commits, so a regression shows up tied to a specific commit rather than just "it got slower at some point") or pytest-benchmark.

A minimal correctness check, concretely (verified on CPython 3.12; the pattern below is the standard, documented Numba usage, whether or not numba happens to be installed in the environment reading this):

import numpy as np
from numba import njit

def f_numpy(x):
    return np.sqrt(x ** 2 + 1)

@njit
def f_numba(x):
    out = np.empty_like(x)
    for i in range(x.shape[0]):
        out[i] = (x[i] ** 2 + 1) ** 0.5
    return out

x = np.array([0.0, 1.0, 2.0, 3.0])
np.testing.assert_allclose(f_numpy(x), f_numba(x))
# passes silently if the two implementations agree within tolerance;
# assert_allclose raises AssertionError with the offending values if they don't

This is the shape every correctness test in this answer means: compute the same thing both ways, on real (including edge-case) input, and assert they agree, rather than eyeballing the code and assuming they do.

Gotchas & how to test:

• Compilation overhead: measure both cold (first call, which includes the one-time cost of compiling the function for that argument type) and hot runs (every call after, which reuse the already-compiled machine code); prefer caching compiled functions when appropriate.
• Unsupported Python features: ensure no reliance on Python objects in hot loops; test for object-mode fallback by enabling nopython=True in tests and catching TypingError. Numba's default @njit mode (short for nopython=True) compiles the whole function straight to machine code with no Python-object fallback; if some part of the function cannot be compiled that way, @njit raises TypingError immediately rather than silently degrading, which is exactly what you want in tests: a raised TypingError tells you precisely which line Numba could not compile, instead of the function quietly running slow through the interpreter (the older, more permissive default mode's "object-mode fallback").
• ABI/dtype differences: test with different dtypes and memory layouts (C/F contiguous, non-contiguous views). ABI (application binary interface) is the compiled-code calling convention two pieces of native code, here a NumPy build and a Numba-compiled function, must agree on to call each other correctly; a mismatch is rare but shows up as crashes or garbage values rather than a clean error. C-contiguous means an array's elements sit one after another in memory in row-major order (the default for a freshly created array); F-contiguous (Fortran-contiguous) means the same but column-major; a strided or non-contiguous view (e.g. arr[::2]) is neither, and a Numba function compiled assuming one layout can behave incorrectly or need an implicit, costly copy when handed the other, which is exactly why testing with a strided view as input matters, not just a fresh contiguous array.
• Thread-safety: test with multithreaded calls if using parallel=True; check for race conditions.

Automation: integrate benchmarks in CI (nightly) and gate correctness on PRs; document performance expectations and maintain regression alerts.

Experiment design
Compare three functions in a tight loop: (A) raise/catch exception on error, (B) return error code and check, (C) pre-validated path (no error). Use timeit and large N.

Code:

import timeit

def raise_path(n):
    for i in range(n):
        try:
            if i%100==0: raise ValueError
        except ValueError:
            pass

def errorcode_path(n):
    for i in range(n):
        ok = True            # the normal, no-error case
        if i%100==0: ok = False   # the simulated error case
        if not ok: pass

def prevalidated_path(n):
    for i in range(n):
        pass                 # no error branch at all: the baseline

n=1000000
print(timeit.timeit(lambda: raise_path(n), number=3))
print(timeit.timeit(lambda: errorcode_path(n), number=3))
print(timeit.timeit(lambda: prevalidated_path(n), number=3))

A correction to the code above (a genuine bug, not just a style choice): as originally sketched, errorcode_path set ok = False unconditionally at the top of every iteration and never set it back to True on the normal path, so if not ok: was True on every single iteration, not just the 1-in-100 simulated errors. That does not model "check an error code" at all, it just runs the pass branch every time. The fix is the one shown above: ok = True by default (the common, no-error case), flipped to False only on the simulated error, so errorcode_path and raise_path are actually testing the same 1% error frequency against each other. The third function, prevalidated_path, is also added here: it has no error branch whatsoever, and exists specifically as the baseline "cost of the loop itself, with no error handling of any kind" that the question's three-way comparison (raise/catch, return-code, pre-validated) asks for; the original sketch defined only the first two.

Interpretation

• Run this at varying simulated error frequencies (change i%100==0 to i%1==0 for 100% errors, or remove the if entirely for 0%) rather than trusting a single frequency; the shape of the gap between raise_path and errorcode_path as a function of error frequency is the actual finding, a single absolute number from one machine is not portable or reproducible elsewhere and should not be reported as "the" result. prevalidated_path's time is the floor: the cost of iterating n times with no error-handling machinery of any kind, useful as the baseline both other paths are measured against.
• The mechanism, not a specific number, is what to lead with: constructing and raising a real exception involves allocating an exception object, populating a traceback, and unwinding the stack to find a matching except, real, nonzero work that a plain boolean check never does; at 0% error frequency, the try block itself still costs something (setting up the exception-handling frame) even though nothing is ever raised, so raise_path at 0% errors is a fair comparison of that baseline try-frame overhead against errorcode_path's baseline if overhead. As error frequency rises toward 100%, the actual raise/unwind cost starts to dominate raise_path's total time in a way it never does for errorcode_path, which does the same constant amount of work (ok = True, one comparison) whether or not that iteration's error flag ends up True.
• For production ETL, prefer error-code or pre-validation for expected, frequent errors, reserve exceptions for truly exceptional control flow, situations that are rare enough that even a real per-raise cost barely matters in aggregate, and where the cleaner control-flow and forced handling (you cannot silently ignore a raised exception the way an unchecked error code can be ignored) outweigh that cost.
• Profile memory and CPU (not just the timeit total) to ensure GC or traceback construction isn't dominating the comparison for reasons unrelated to the actual branch being tested, for example, a test harness that also does unrelated allocation inside the timed loop.

Approach: implement LRU with a dict for storage and a doubly-linked list for order; use threading.RLock (a lock that the SAME thread can safely acquire again without deadlocking itself, unlike a plain threading.Lock, in case the cache logic ever needs to re-enter the lock while already holding it) for concurrency. Decorator returns wrapper that locks around lookups and updates, minimizing lock hold time.

Implementation:

import threading
from functools import wraps

def lru_cache(maxsize=128):
    def deco(func):
        cache = {}
        head = tail = None
        lock = threading.RLock()
        class Node:
            __slots__=('key','val','prev','next')
            def __init__(self,k,v):
                self.key=k;self.val=v;self.prev=self.next=None
        def _move_to_front(node):
            nonlocal head, tail
            if node is head:
                return  # already the most-recently-used entry, nothing to do
            # unlink node from wherever it currently sits
            if node.prev:
                node.prev.next = node.next
            if node.next:
                node.next.prev = node.prev
            if node is tail:
                tail = node.prev
            # relink it at the head (the most-recently-used end)
            node.prev = None
            node.next = head
            if head:
                head.prev = node
            head = node
            if tail is None:
                tail = node
        @wraps(func)
        def wrapper(*args, **kwargs):
            nonlocal head, tail   # wrapper reassigns both below on eviction; without
                                   # this declaration LEGB makes them locals instead
            key=(args,tuple(sorted(kwargs.items())))
            with lock:
                node=cache.get(key)
                if node:
                    _move_to_front(node); return node.val
            val=func(*args, **kwargs)
            with lock:
                if key in cache: return cache[key].val
                node=Node(key,val); cache[key]=node; _move_to_front(node)
                if len(cache)>maxsize:
                    # cache is over capacity: evict the true tail, the
                    # least-recently-used entry, from both the linked list
                    # and the dict
                    lru_node = tail
                    tail = lru_node.prev
                    if tail:
                        tail.next = None
                    else:
                        head = None
                    del cache[lru_node.key]
            return val
        return wrapper
    return deco

Worked example: verified on CPython 3.12, calling the decorated function through a full eviction cycle so the policy can actually be watched working, not just taken on faith.

calls = []

@lru_cache(maxsize=2)
def square(n):
    calls.append(n)
    return n * n

print(square(1))   # 1  (miss: cache empty, computed and cached; order (MRU->LRU): [1])
print(square(2))   # 4  (miss: computed and cached; order: [2, 1])
print(square(1))   # 1  (hit: served from cache, 1 moves back to the front; order: [1, 2])
print(square(3))   # 9  (miss: cache was full at {1, 2}; since 1 was just reused, 2 is now
                    # the least-recently-used entry and gets evicted to make room for 3; order: [3, 1])
print(square(2))   # 4  (miss again: 2 was evicted in the previous step, so this recomputes
                    # instead of hitting the cache; order: [2, 3])
print(calls)        # [1, 2, 3, 2]  -- 2 appears twice: proof it was actually evicted and
                     # had to be recomputed, not just a claim

Complexity: O(1) average get/set.

Contention: a single global lock, one shared threading.RLock protecting the whole cache, serializes cache access: only one thread at a time can even check whether something is cached, regardless of which key it wants, which becomes a bottleneck under heavy concurrent traffic. Two ways to reduce that: a read-mostly optimistic check (read the dict for a hit without taking the lock first, since a plain dict read is safe to race on for a snapshot lookup, and only take the lock to confirm the hit and update the linked-list ordering, so the common cache-hit path spends less time holding the lock), or shard locks (split one cache into several smaller caches, each with its own separate lock, and route each key to one shard by hashing it, e.g. shard = hash(key) % num_shards; two threads reading keys that land in different shards no longer contend for the same lock at all, at the cost of maxsize now being enforced per shard rather than globally).