Python Programming Questions

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.

Possible Python/library causes

• dict ordering: dict insertion order IS deterministic in CPython 3.7+ (this is a language guarantee, not an implementation quirk), but the actual non-determinism seen in production comes from something upstream feeding the dict in a different order between runs, for example a filesystem glob() returning filenames in a different order, or multiple workers finishing and inserting their results in whatever order they happen to complete, not the dict itself misbehaving.
• set iteration: unordered, causes non-deterministic traversal.
• floating-point reductions (sum/mean) non-associative ordering leads to tiny differences.
• parallelism: thread/process scheduling, race conditions, non-deterministic map ordering.
• use of RNG without fixed seeds (random, numpy.random, torch).
• reliance on OS-level nondeterminism (filesystem glob order, multiprocessing.Queue timing).

Two of these causes, made concrete, verified on CPython 3.12:

Non-associative floating-point reduction, a naive accumulator loop genuinely disagreeing with itself depending on order (built-in sum() on recent CPython uses a more accurate internal algorithm and happens to avoid this specific example, which is exactly why the naive loop below, closer to what a hand-rolled or per-worker partial-sum reduction actually does, is the more honest demonstration):

def manual_sum(values):
    total = 0.0
    for v in values:
        total += v
    return total

vals = [0.1, 0.2, 0.3]
print(manual_sum(vals))
# 0.6000000000000001
print(manual_sum(list(reversed(vals))))
# 0.6

import math
print(math.fsum(vals) == math.fsum(list(reversed(vals))))
# True

The same three floats, summed in two different orders by a plain accumulator, produce two DIFFERENT bit patterns, 0.6000000000000001 versus exactly 0.6, purely from the order of addition (this is exactly what happens when parallel workers each compute a partial sum and the main process combines them in whatever order they finish); math.fsum (a compensated-summation algorithm that tracks and corrects rounding error as it goes) gives the identical, order-independent result either way.

Unseeded RNG producing different output on repeated runs, and the fix:

import random

random.seed(42)
print(random.random())
# 0.6394267984578837
random.seed(42)
print(random.random())
# 0.6394267984578837 -- identical, since the seed was reset

Calling random.seed(42) again before the second random.random() call reproduces the exact same float; omitting the seed (or seeding from the current time, which is random's default) would make this genuinely different on every run, which is exactly the failure mode the fix below addresses.

Concrete fixes

• Sort keys/lists before iteration (e.g., sorted(dict.items())).
• Use deterministic data structures: collections.OrderedDict where explicit order required.
• Fix RNG seeds at pipeline start for all libs (random.seed, np.random.default_rng with seed passed).
• Use stable reduction algorithms: math.fsum or pairwise summation (numpy.sum with dtype and stable algorithm libs).
• Avoid unordered set for grouping; use sorted containers or deterministic grouping keys.
• Control parallelism: fix number of workers, use deterministic partitioning, or run single-threaded for critical steps.
• Pin library versions and document environment (requirements, Docker).

These steps produce reproducible outputs suitable for production ETL runs and debugging.

Approach

I would read the file line by line with bufio.Scanner. Opening errors are fatal, because there is no file to process. Line parse errors are recoverable, meaning a bad line does not stop the whole file, so I return them in a slice and keep the good records. If scanning itself fails, I return the partial data plus the scan error so the caller can decide whether to fail the job.

package main

import (
    "bufio"
    "fmt"
    "os"
    "strconv"
    "strings"
)

type Record struct {
    Name string
    Age  int
}

type ParseIssue struct {
    Line int
    Raw  string
    Err  error
}

func ParseFile(path string) ([]Record, []ParseIssue, error) {
    f, err := os.Open(path)
    if err != nil {
        return nil, nil, err
    }
    defer f.Close()

    scanner := bufio.NewScanner(f)
    scanner.Buffer(make([]byte, 1024), 1024*1024)

    var records []Record
    var issues []ParseIssue
    lineNum := 0

    for scanner.Scan() {
        lineNum++
        line := strings.TrimSpace(scanner.Text())
        if line == "" {
            continue
        }

        parts := strings.Split(line, ",")
        if len(parts) != 2 {
            issues = append(issues, ParseIssue{Line: lineNum, Raw: line, Err: fmt.Errorf("expected name,age")})
            continue
        }

        age, err := strconv.Atoi(strings.TrimSpace(parts[1]))
        if err != nil {
            issues = append(issues, ParseIssue{Line: lineNum, Raw: line, Err: err})
            continue
        }

        records = append(records, Record{
            Name: strings.TrimSpace(parts[0]),
            Age:  age,
        })
    }

    if err := scanner.Err(); err != nil {
        return records, issues, err
    }

    return records, issues, nil
}

Key points

• open failure: return immediately
• parsing failure: collect issue and continue
• scan failure: return partial results and an error

Example: with Ann,32, bad-line, and Bob,41, you get 2 records and 1 issue.

Complexity: O(n) time, O(k) memory for stored records and issues.

Debug & remediation plan:

1. Reproduce locally

• Run the test repeatedly: pytest -k name --maxfail=1 -q --count=100 (pytest-repeat plugin) to reproduce flakiness.
• Run under same env as CI (Python version, env vars).

2. Collect traces

• Add logging with timestamps or use pytest -s; capture stack traces on timeout by increasing timeout to inspect where it hangs.
• Use faulthandler: in test setup call faulthandler.dump_traceback_later and dump on timeout.
• Use pytest --durations and --showlocals to find slow parts.

3. Isolate cause

• Check for test order dependency: run single test and with -k to see isolation.
• Mock external resources (network, DB, filesystem) and use deterministic fixtures.
• Check for race conditions: add sleeps, use thread/async synchronization, run under race detectors or sanitizers if available.

4. Fixes

• Replace real I/O with fixtures/mocks or use ephemeral resources.
• Make async tests use event loop proper awaits and timeouts; use pytest-timeout to fail faster.
• Ensure teardown always cleans up (use tmp_path, monkeypatch).

5. Enforce stability in CI

• Add time budget and stricter assertions; mark inherently flaky tests with @pytest.mark.flaky only after triage.
• Run flaky-test detection in CI: rerun once and fail if consistently flaky.
• Add metrics and alerting if test durations increase.

Result: deterministic, fast tests and fewer CI interruptions.

Worked walkthrough on one concrete flaky test: say test_worker_processes_queue starts a background thread that pushes a result onto a queue.Queue, then the test does time.sleep(0.05) and asserts the queue has an item, assuming 50ms is always enough for the worker thread to finish. On a fast, quiet machine this passes almost every time; on a loaded CI runner the worker thread sometimes has not finished by the time the fixed sleep ends, and the assertion fails, only sometimes, which is exactly what "flaky" means.

1. Reproduce locally: pytest -k test_worker_processes_queue --count=100 (--count comes from the pytest-repeat plugin, which adds that option to run the same test that many times in one pytest invocation) runs the test repeatedly in one process; a genuinely flaky test now fails on some of those runs and passes on others, in the same environment, which rules out "it only fails in CI" as an explanation and confirms the flakiness is reproducible on demand rather than a one-off fluke.
2. Collect traces: pytest --durations=0 --showlocals (--durations=0 prints every test's wall-clock duration sorted slowest-first instead of only the slowest few; --showlocals prints each local variable's value at the point of failure) shows, on a failing run, that the assertion failed with the queue still empty: the worker thread genuinely had not produced a result yet, not that the assertion logic itself was wrong. faulthandler.dump_traceback_later (a standard-library facility you arm with a countdown; if the process is still alive when the countdown expires, it dumps every thread's current stack trace, which is exactly what you need for a hang rather than a fast failure) would show the worker thread still inside its own processing function at the moment the assertion already gave up, confirming a genuine race between the fixed sleep and the thread's actual completion time, not a logic bug in the worker itself.
3. Isolate cause: running the test alone (not the full suite) still reproduces it sometimes, which rules out cross-test state leakage as the cause; the pattern (passes when the machine is quiet, fails more often when it is busy) is the signature of a race condition between a fixed-duration sleep and actual, variable-duration background work, not a logic bug.
4. Fix: replace the fixed time.sleep(0.05) with an actual wait on the real completion signal, queue.get(timeout=5), which blocks only as long as it actually needs to (returning the moment the worker pushes its result, and only failing the test if 5 real seconds pass with nothing produced), removing the race entirely instead of tuning the sleep duration and hoping it is long enough.
5. Enforce in CI: add pytest-timeout (a plugin that fails any individual test that runs longer than a configured limit, rather than letting a hang stall the whole CI job indefinitely) as a backstop for any future test that hangs outright, and only reach for @pytest.mark.flaky (a marker that tells a plugin to automatically retry a failing test before reporting it failed) after a test like this has actually been triaged and understood, never as a first response to red CI, since it hides the exact race just diagnosed instead of fixing it.

A race detector or sanitizer (a class of tool, e.g. a language's -race flag or ThreadSanitizer, that instruments a program at build or run time to flag unsynchronized concurrent access to shared memory as it happens, rather than waiting for it to occasionally produce a wrong answer) is the equivalent tool one level down, in compiled languages with real memory-level data races. Python's GIL rules out that specific class of memory corruption, so the direct equivalent here is exactly the timing/synchronization walkthrough above, not a separate tool to reach for.

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.