C++, Python, pandas, TensorFlow: the working set

What you need to read and write fluently in an inference-optimization interview, with every snippet actually run and the real output pasted underneath.

1. Modern C++ essentials for systems

You are not expected to write template metaprogramming. You are expected to explain who owns what memory, when a copy happens, and why a function signature looks the way it does. Everything below is C++17/20 and compiles clean with -Wall -Wextra.

Copy#include <algorithm>
#include <iostream>
#include <memory>
#include <numeric>
#include <span>
#include <stdexcept>
#include <string>
#include <string_view>
#include <vector>

struct Tracer {  // counts copies vs moves
    static inline int copies = 0, moves = 0;
    std::vector<int> data;
    explicit Tracer(size_t n) : data(n, 1) {}
    Tracer(const Tracer& o) : data(o.data) { ++copies; }
    Tracer(Tracer&& o) noexcept : data(std::move(o.data)) { ++moves; }
};

struct Guard {  // RAII: release happens on every exit path
    std::string name;
    explicit Guard(std::string n) : name(std::move(n)) { std::cout << "  acquire " << name << "\n"; }
    ~Guard() { std::cout << "  release " << name << "\n"; }
    Guard(const Guard&) = delete;
};

void by_value(std::vector<int> v) { v.push_back(99); }
void by_ref(std::vector<int>& v) { v.push_back(99); }
void by_cref(const std::vector<int>& v) { std::cout << "  cref sees " << v.size() << " elems\n"; }

Tracer make() { return Tracer(1000); }  // guaranteed elision: no copy, no move

template <typename T>
T total(std::span<const T> xs) { return std::accumulate(xs.begin(), xs.end(), T{}); }

size_t count_spaces(std::string_view s) { return std::count(s.begin(), s.end(), ' '); }

int main() {
    std::cout << "[value vs reference]\n";
    std::vector<int> v{1, 2, 3};
    by_value(v);  std::cout << "  after by_value: " << v.size() << "\n";
    by_ref(v);    std::cout << "  after by_ref:   " << v.size() << "\n";
    by_cref(v);

    std::cout << "[RAII]\n";
    try {
        Guard g("gpu_buffer");
        throw std::runtime_error("boom");
    } catch (const std::exception& e) { std::cout << "  caught " << e.what() << "\n"; }

    std::cout << "[smart pointers]\n";
    auto u = std::make_unique<Tracer>(10);
    auto u2 = std::move(u);  // ownership transfer; u is now null
    std::cout << "  u is " << (u ? "alive" : "null") << ", u2 owns " << u2->data.size() << " ints\n";
    auto s1 = std::make_shared<Tracer>(10);
    { auto s2 = s1; std::cout << "  use_count inside scope: " << s1.use_count() << "\n"; }
    std::cout << "  use_count after scope:  " << s1.use_count() << "\n";

    std::cout << "[move semantics]\n";
    Tracer a = make();
    std::cout << "  after make():            copies=" << Tracer::copies << " moves=" << Tracer::moves << "\n";
    std::vector<Tracer> vt;
    vt.reserve(4);
    vt.push_back(a);
    std::cout << "  push_back(lvalue):       copies=" << Tracer::copies << " moves=" << Tracer::moves << "\n";
    vt.push_back(std::move(a));
    std::cout << "  push_back(std::move(a)): copies=" << Tracer::copies << " moves=" << Tracer::moves
              << "  (a.data.size() now " << a.data.size() << ")\n";

    std::cout << "[templates, span, string_view]\n";
    std::vector<double> d{1.5, 2.5, 3.0};
    int arr[] = {1, 2, 3, 4};
    std::cout << "  total(vector<double>) = " << total<double>(d) << "\n";
    std::cout << "  total(int[4])         = " << total<int>(arr) << "\n";
    std::cout << "  spaces in literal     = " << count_spaces("a b  c d") << " (no allocation)\n";

    std::cout << "[lambdas]\n";
    int threshold = 2;
    auto by_val = [threshold](int x) { return x > threshold; };
    threshold = 100;  // does not affect by_val
    std::vector<int> xs{5, 1, 4, 2, 3};
    std::cout << "  count(x>2) = " << std::count_if(xs.begin(), xs.end(), by_val) << "\n";
    std::sort(xs.begin(), xs.end(), [](int p, int q) { return p > q; });
    for (int x : xs) std::cout << x << ' ';
    std::cout << "\n";
}

$ g++ -std=c++20 -O2 -Wall -Wextra cpp01_essentials.cpp -o cpp01_essentials
[value vs reference]
  after by_value: 3
  after by_ref:   4
  cref sees 4 elems
[RAII]
  acquire gpu_buffer
  release gpu_buffer
  caught boom
[smart pointers]
  u is null, u2 owns 10 ints
  use_count inside scope: 2
  use_count after scope:  1
[move semantics]
  after make():            copies=0 moves=0
  push_back(lvalue):       copies=1 moves=0
  push_back(std::move(a)): copies=1 moves=1  (a.data.size() now 0)
[templates, span, string_view]
  total(vector<double>) = 7
  total(int[4])         = 10
  spaces in literal     = 4 (no allocation)
[lambdas]
  count(x>2) = 3
5 4 3 2 1 

2. Memory & performance

Stack vs heap. Stack allocation is a pointer bump and is freed automatically on scope exit; the stack is small (typically 8 MB main thread, less in other threads) so large buffers go on the heap. Heap allocation (new, malloc, vector growth) takes a lock-ish allocator path, can fragment, and is the main reason request-path code pre-allocates and reuses buffers (arenas, object pools, reserve()). Cache lines. The CPU moves memory in 64-byte lines (128 on Apple silicon). Touching one byte pulls the whole line, so contiguous, sequentially accessed data is fast because every fetched byte is used and the hardware prefetcher guesses the next line. This single fact explains most of this section.

• AoS vs SoA. Array of Structs interleaves fields; scanning one field drags the others through the cache. Struct of Arrays keeps each field contiguous. GPUs want SoA for the same reason as coalescing (section 4).
• vector vs linked list. A list node is a separate heap allocation: iteration is pointer chasing, each hop a likely cache miss, no prefetching. Even insert-in-the-middle, the list's one advantage, loses for realistic sizes because finding the position is O(n) pointer chasing anyway.
• False sharing. Two threads writing different variables that sit on the same cache line still fight: each write invalidates the other core's copy of the line. Fix with alignas(64) padding or per-thread accumulators merged at the end.
• Alignment/padding. The compiler pads fields to their natural alignment. Ordering fields largest-first often shrinks the struct. SIMD loads and cudaMemcpy fast paths want 16/32/64-byte alignment.

Copy#include <atomic>
#include <chrono>
#include <cstdint>
#include <iostream>
#include <list>
#include <numeric>
#include <random>
#include <thread>
#include <vector>

using Clock = std::chrono::steady_clock;
template <typename F>
double ms(F&& f) {
    auto t0 = Clock::now();
    f();
    return std::chrono::duration<double, std::milli>(Clock::now() - t0).count();
}

struct Padded { char c; double d; char e; };  // compiler inserts padding
struct Packed { double d; char c; char e; };  // same fields reordered: smaller
struct alignas(64) CacheLine { std::atomic<int64_t> v{0}; };

struct ParticleAoS { float x, y, z, vx, vy, vz, mass, charge; };
struct ParticlesSoA { std::vector<float> x; };  // other columns omitted: this query touches only x

int main() {
    std::cout << "sizeof(Padded)=" << sizeof(Padded) << " sizeof(Packed)=" << sizeof(Packed)
              << " sizeof(CacheLine)=" << sizeof(CacheLine) << "\n";

    constexpr int N = 4'000'000;
    std::vector<ParticleAoS> aos(N, ParticleAoS{1, 2, 3, 0, 0, 0, 1, 0});
    ParticlesSoA soa;
    soa.x.assign(N, 1.f);
    float sa = 0, ss = 0;
    double t_aos = ms([&] { for (auto& p : aos) sa += p.x; });
    double t_soa = ms([&] { for (float x : soa.x) ss += x; });
    std::cout << "sum one field, 4M elems   AoS: " << t_aos << " ms   SoA: " << t_soa << " ms   (sums " << sa << " " << ss << ")\n";

    std::mt19937 rng(1);
    std::vector<int> vec(2'000'000);
    for (auto& x : vec) x = rng() % 100;
    std::list<int> lst(vec.begin(), vec.end());
    long sv = 0, sl = 0;
    double t_vec = ms([&] { sv = std::accumulate(vec.begin(), vec.end(), 0L); });
    double t_lst = ms([&] { sl = std::accumulate(lst.begin(), lst.end(), 0L); });
    std::cout << "sum 2M ints   vector: " << t_vec << " ms   list: " << t_lst << " ms   same result: " << (sv == sl) << "\n";

    // Two threads, each incrementing its OWN counter.
    auto run = [](std::atomic<int64_t>& a, std::atomic<int64_t>& b) {
        auto work = [](std::atomic<int64_t>& c) { for (int i = 0; i < 50'000'000; ++i) c.fetch_add(1, std::memory_order_relaxed); };
        return ms([&] { std::thread t1(work, std::ref(a)), t2(work, std::ref(b)); t1.join(); t2.join(); });
    };
    struct Adjacent { std::atomic<int64_t> a{0}, b{0}; } adj;
    CacheLine p1, p2;
    double t_fs = run(adj.a, adj.b);
    double t_ok = run(p1.v, p2.v);
    std::cout << "2 threads, adjacent counters: " << t_fs << " ms   64B-separated: " << t_ok << " ms\n";
}

$ g++ -std=c++20 -O2 -Wall -Wextra -pthread cpp02a_locality.cpp -o cpp02a_locality && ./cpp02a_locality   # on lab pod (x86 Linux)
sizeof(Padded)=24 sizeof(Packed)=16 sizeof(CacheLine)=64
sum one field, 4M elems   AoS: 8.75936 ms   SoA: 3.65783 ms   (sums 4e+06 4e+06)
sum 2M ints   vector: 0.3438 ms   list: 7.71397 ms   same result: 1
2 threads, adjacent counters: 1436.8 ms   64B-separated: 251.041 ms

Undefined behavior and the sanitizers

UB means the compiler is allowed to assume it never happens, so "it worked when I tried it" proves nothing. Common sources: out-of-bounds access, use-after-free, signed integer overflow, shifting by ≥ the width, data races, reading uninitialized memory, dereferencing null, violating strict aliasing. Optimization levels make it visible: -O0 is a debugging build; -O2 is the normal release level; -O3 adds more aggressive vectorization and inlining (not always faster; measure); add -march=native only if you build where you run. For debugging, always add -g. Run tests under -fsanitize=address,undefined (ASan finds memory errors, UBSan finds the arithmetic ones; ~2x slowdown) and, separately, -fsanitize=thread (TSan; cannot be combined with ASan).

Copy#include <cstdio>
#include <vector>

int main() {
    std::vector<int> v(4, 7);
    int sum = 0;
    for (int i = 0; i <= 4; ++i) sum += v[i];  // off-by-one: reads v[4]
    std::printf("sum=%d\n", sum);
}

$ g++ -std=c++20 -g -O1 -fsanitize=address,undefined -fno-omit-frame-pointer cpp02b_asan.cpp -o cpp02b_asan && cpp02b_asan
=================================================================
==81421==ERROR: AddressSanitizer: heap-buffer-overflow on address 0x602000000120 at pc 0x000104e5cb40 bp 0x00016afa1b50 sp 0x00016afa1b48
READ of size 4 at 0x602000000120 thread T0
    #0 0x000104e5cb3c in main cpp02b_asan.cpp:8
    #1 0x000184527da0 in start+0x1b4c (dyld:arm64e+0x1fda0)

0x602000000120 is located 0 bytes after 16-byte region [0x602000000110,0x602000000120)
allocated by thread T0 here:
    #0 0x0001054eff50 in _Znwm+0x74 (libclang_rt.asan_osx_dynamic.dylib:arm64e+0x4ff50)
    #1 0x000104e5ce54 in std::__1::vector<int, std::__1::allocator<int>>::__vallocate[abi:nqe210106](unsigned long) vector.h:574
    #2 0x000104e5cd04 in std::__1::vector<int, std::__1::allocator<int>>::vector[abi:nqe210106](unsigned long, int const&) vector.h:171
    #3 0x000104e5c9a8 in main cpp02b_asan.cpp:6
    #4 0x000184527da0 in start+0x1b4c (dyld:arm64e+0x1fda0)

SUMMARY: AddressSanitizer: heap-buffer-overflow cpp02b_asan.cpp:8 in main
... (shadow-byte dump trimmed)
/bin/sh: line 1: 81421 Abort trap: 6           bin_cpp02b_asan

The loop reads v[4], one past a 16-byte allocation. The program has no visible symptom without the sanitizer. ASan's report names the line (cpp02b_asan.cpp:8), the size (16-byte region) and where it was allocated.

Copy#include <climits>
#include <cstdio>

int main(int argc, char**) {
    int big = INT_MAX - 1 + argc;      // INT_MAX
    int next = big + argc;             // signed overflow: undefined behavior
    std::printf("%d\n", next);
    int shift = 31 + argc;             // 32
    std::printf("%d\n", 1 << shift);   // shift >= width: undefined behavior
}

$ g++ -std=c++20 -g -O1 -fsanitize=undefined cpp02c_ub.cpp -o cpp02c_ub && cpp02c_ub
-2147483648
1
cpp02c_ub.cpp:7:20: runtime error: signed integer overflow: 2147483647 + 1 cannot be represented in type 'int'
SUMMARY: UndefinedBehaviorSanitizer: undefined-behavior cpp02c_ub.cpp:7:20 
cpp02c_ub.cpp:10:27: runtime error: shift exponent 32 is too large for 32-bit type 'int'
SUMMARY: UndefinedBehaviorSanitizer: undefined-behavior cpp02c_ub.cpp:10:27 

UBSan reports both problems with file and column. Note the program also printed a result for each, which is exactly why UB is dangerous.

Copy#include <climits>
#include <cstdio>

// "Obviously" true unless x == INT_MAX... but signed overflow is UB, so -O2 may fold it to true.
__attribute__((noinline)) bool bigger_after_inc(int x) { return x + 1 > x; }

int main() {
    std::printf("INT_MAX+1 > INT_MAX ? %d\n", bigger_after_inc(INT_MAX));
}

$ g++ -std=c++20 -O0 cpp02d_ub_opt.cpp -o cpp02d_ub_opt0 && cpp02d_ub_opt0 && g++ -std=c++20 -O2 cpp02d_ub_opt.cpp -o cpp02d_ub_opt2 && cpp02d_ub_opt2
INT_MAX+1 > INT_MAX ? 0
INT_MAX+1 > INT_MAX ? 1

3. Concurrency

std::thread runs a function on a new OS thread; you must join() (or detach()) before the object dies. Shared mutable data needs one of: a mutex (lock_guard/scoped_lock for RAII unlock), an atomic, or ownership partitioning (each thread owns its slice, merge at the end). A data race (two threads touch the same non-atomic location, at least one writes, no ordering) is undefined behavior even if the output looks right.

Copy#include <atomic>
#include <chrono>
#include <iostream>
#include <mutex>
#include <thread>
#include <vector>

template <typename F>
double run_threads(int n, F f) {
    auto t0 = std::chrono::steady_clock::now();
    std::vector<std::thread> ts;
    for (int i = 0; i < n; ++i) ts.emplace_back(f);
    for (auto& t : ts) t.join();
    return std::chrono::duration<double, std::milli>(std::chrono::steady_clock::now() - t0).count();
}

int main() {
    constexpr int kThreads = 4, kIters = 1'000'000;
    long locked = 0;
    std::mutex m;
    double t_mutex = run_threads(kThreads, [&] {
        for (int i = 0; i < kIters; ++i) { std::lock_guard<std::mutex> g(m); ++locked; }
    });
    std::atomic<long> atom{0};
    double t_atomic = run_threads(kThreads, [&] {
        for (int i = 0; i < kIters; ++i) atom.fetch_add(1, std::memory_order_relaxed);
    });
    // Sharding: each thread accumulates locally, merges once. Usually the real answer.
    std::atomic<long> sharded{0};
    double t_shard = run_threads(kThreads, [&] {
        long local = 0;
        for (int i = 0; i < kIters; ++i) ++local;
        sharded.fetch_add(local);
    });
    std::cout << "expected " << kThreads * kIters << "\n";
    std::cout << "mutex   : " << locked << "  " << t_mutex << " ms\n";
    std::cout << "atomic  : " << atom << "  " << t_atomic << " ms\n";
    std::cout << "sharded : " << sharded << "  " << t_shard << " ms\n";
}

$ g++ -std=c++20 -O2 -Wall -Wextra -pthread cpp03a_mutex_atomic.cpp -o cpp03a_mutex_atomic
expected 4000000
mutex   : 4000000  74.8185 ms
atomic  : 4000000  25.1017 ms
sharded : 4000000  0.048375 ms

Copy#include <iostream>
#include <thread>

int counter = 0;  // plain int shared by two threads: a data race (UB)

int main() {
    auto work = [] { for (int i = 0; i < 100000; ++i) ++counter; };
    std::thread a(work), b(work);
    a.join(); b.join();
    std::cout << "counter=" << counter << " (expected 200000)\n";
}

$ g++ -std=c++20 -O1 -g -fsanitize=thread -pthread cpp03b_race.cpp -o cpp03b_race && cpp03b_race
counter=200000 (expected 200000)
==================
WARNING: ThreadSanitizer: data race (pid=81478)
  Write of size 4 at 0x000104a60000 by thread T2:
    #0 void* std::__1::__thread_proxy[abi:nqe210106]<std::__1::tuple<std::__1::unique_ptr<std::__1::__thread_struct, std::__1::default_delete<std::...

  Previous write of size 4 at 0x000104a60000 by thread T1:
    #0 void* std::__1::__thread_proxy[abi:nqe210106]<std::__1::tuple<std::__1::unique_ptr<std::__1::__thread_struct, std::__1::default_delete<std::...

  Location is global 'counter' at 0x000104a60000 (bin_cpp03b_race+0x100008000)
  [thread-creation stacks trimmed]

SUMMARY: ThreadSanitizer: data race thread.h:168 in void* std::__1::__thread_proxy[abi:nqe210106]<std::__1::tuple<std::__1::unique_ptr<std::__1::__...
ThreadSanitizer: reported 1 warnings
/bin/sh: line 1: 81478 Abort trap: 6           bin_cpp03b_race

Memory order, in five lines

• memory_order_relaxed: the operation itself is atomic, with no ordering guarantees about other memory. Fine for counters.
• release store + acquire load of the same atomic: everything the writer did before the release is visible to a reader that sees the stored value after its acquire. This is the "publish" pattern.
• seq_cst (the default): a single global order of all seq_cst operations. Simplest to reason about, slightly more expensive on weak architectures (ARM, and so Graviton and Apple silicon).
• x86 gives you acquire/release almost for free in hardware, which hides ordering bugs that surface on ARM. Test on both.
• Rule of thumb: use mutexes first; reach for atomics with explicit orders only for a measured hot path, and copy a known-good pattern like the one below.

A lock-free SPSC ring buffer

One producer thread, one consumer thread, fixed power-of-two capacity, no locks and no allocation after construction. This is the classic structure between a network thread and a batching thread. Only the producer writes head_, only the consumer writes tail_, so no compare-and-swap is needed. The release store on head_ publishes the slot write; the consumer's acquire load of head_ is what makes the slot's contents visible.

Copy#include <atomic>
#include <chrono>
#include <cstddef>
#include <iostream>
#include <thread>
#include <vector>

// Single-producer / single-consumer ring buffer. N must be a power of two.
template <typename T, size_t N>
class SpscRing {
    static_assert((N & (N - 1)) == 0, "N must be a power of two");
public:
    bool push(const T& v) {                      // producer thread only
        size_t h = head_.load(std::memory_order_relaxed);
        if (h - tail_.load(std::memory_order_acquire) == N) return false;  // full
        buf_[h & (N - 1)] = v;
        head_.store(h + 1, std::memory_order_release);  // publishes the write above
        return true;
    }
    bool pop(T& out) {                           // consumer thread only
        size_t t = tail_.load(std::memory_order_relaxed);
        if (t == head_.load(std::memory_order_acquire)) return false;      // empty
        out = buf_[t & (N - 1)];
        tail_.store(t + 1, std::memory_order_release);  // hands the slot back
        return true;
    }
private:
    alignas(64) std::atomic<size_t> head_{0};    // separate lines: no false sharing
    alignas(64) std::atomic<size_t> tail_{0};
    alignas(64) T buf_[N];
};

int main() {
    constexpr long kItems = 5'000'000;
    static SpscRing<long, 1024> q;
    auto t0 = std::chrono::steady_clock::now();
    std::thread prod([&] { for (long i = 0; i < kItems; ++i) while (!q.push(i)) {} });
    long expect = 0, bad = 0;
    std::thread cons([&] {
        long v;
        while (expect < kItems) { if (q.pop(v)) { bad += (v != expect); ++expect; } }
    });
    prod.join(); cons.join();
    double s = std::chrono::duration<double>(std::chrono::steady_clock::now() - t0).count();
    std::cout << "items=" << expect << " out_of_order=" << bad << "  " << kItems / s / 1e6 << " M items/s\n";
}

$ g++ -std=c++20 -O2 -Wall -Wextra -pthread cpp03c_spsc.cpp -o cpp03c_spsc && cpp03c_spsc && g++ -std=c++20 -O1 -g -fsanitize=thread -pthread cpp03c_spsc.cpp -o cpp03c_spsc_tsan && cpp03c_spsc_tsan
items=5000000 out_of_order=0  16.8655 M items/s
items=5000000 out_of_order=0  1.54858 M items/s

The first line is the optimized build (~17 M items/s through a 1024-slot ring, zero ordering violations across 5 M items). The second is the same code under ThreadSanitizer: no warnings, and ~11x slower, which is normal for TSan. As a check, changing the head_ store to relaxed made TSan report a data race, and on this ARM machine the consumer even saw 200 stale or out-of-order items out of 5 M.

Condition variables

A condition_variable lets a thread sleep until another thread says "something changed". The pattern is always: hold the mutex, wait(lock, predicate); the other side changes state under the mutex, then notifies. The predicate matters because wakeups can be spurious and because a notify sent before the waiter starts waiting is lost; the predicate re-checks the real condition.

Copy#include <condition_variable>
#include <iostream>
#include <mutex>
#include <queue>
#include <thread>

std::mutex m;
std::condition_variable cv;
std::queue<int> q;
bool done = false;

int main() {
    long sum = 0;
    std::thread consumer([&] {
        std::unique_lock lk(m);
        for (;;) {
            cv.wait(lk, [] { return !q.empty() || done; });  // predicate guards against spurious wakeups
            while (!q.empty()) { sum += q.front(); q.pop(); }
            if (done) return;
        }
    });
    for (int i = 1; i <= 1000; ++i) {
        { std::lock_guard g(m); q.push(i); }
        cv.notify_one();
    }
    { std::lock_guard g(m); done = true; }
    cv.notify_one();
    consumer.join();
    std::cout << "sum=" << sum << " (expected " << 1000 * 1001 / 2 << ")\n";
}

$ g++ -std=c++20 -O2 -Wall -Wextra -pthread cpp03d_condvar.cpp -o cpp03d_condvar
sum=500500 (expected 500500)

4. CUDA C++ primer

Execution model. A kernel is a function run by thousands of threads. Threads are grouped into blocks (up to 1024 threads, run on one SM, can cooperate through shared memory and __syncthreads()); blocks form a grid. Inside a block, threads execute in warps of 32 in lockstep. You compute your global index as blockIdx.x * blockDim.x + threadIdx.x and guard with if (i < n) because the grid is rounded up.

Coalescing. When the 32 threads of a warp read 32 consecutive 4-byte addresses, the hardware serves them with a few wide memory transactions. If they read scattered addresses, each thread can cost a separate transaction. This is the GPU analogue of the cache-line story in section 2.

Vector add

Copy#include <cstdio>
#include <cuda_runtime.h>

#define CHECK(x) do { cudaError_t e = (x); if (e != cudaSuccess) { printf("CUDA error %s at line %d\n", cudaGetErrorString(e), __LINE__); return 1; } } while (0)

__global__ void vec_add(const float* a, const float* b, float* c, int n) {
    int i = blockIdx.x * blockDim.x + threadIdx.x;  // global thread index
    if (i < n) c[i] = a[i] + b[i];                  // guard: grid is rounded up
}

int main() {
    const int n = 1 << 26;                          // 64M floats = 256 MB per array
    const size_t bytes = size_t(n) * sizeof(float);
    float *a, *b, *c;
    CHECK(cudaMallocManaged(&a, bytes)); CHECK(cudaMallocManaged(&b, bytes)); CHECK(cudaMallocManaged(&c, bytes));
    for (int i = 0; i < n; ++i) { a[i] = 1.f; b[i] = float(i % 7); }
    int threads = 256, blocks = (n + threads - 1) / threads;
    vec_add<<<blocks, threads>>>(a, b, c, n);       // warm-up (also migrates pages)
    CHECK(cudaDeviceSynchronize());
    cudaEvent_t t0, t1; cudaEventCreate(&t0); cudaEventCreate(&t1);
    cudaEventRecord(t0);
    for (int r = 0; r < 20; ++r) vec_add<<<blocks, threads>>>(a, b, c, n);
    cudaEventRecord(t1); CHECK(cudaEventSynchronize(t1));
    float ms; cudaEventElapsedTime(&ms, t0, t1); ms /= 20;
    int bad = 0;
    for (int i = 0; i < n; ++i) bad += (c[i] != 1.f + float(i % 7));
    printf("grid=%d blocks x %d threads, errors=%d\n", blocks, threads, bad);
    printf("%.3f ms/iter, %.0f GB/s (3 arrays x 4 B moved per element)\n", ms, 3.0 * bytes / (ms * 1e-3) / 1e9);
    cudaFree(a); cudaFree(b); cudaFree(c);
}

$ nvcc -O3 -arch=sm_120 cu01_vecadd.cu -o cu01_vecadd && ./cu01_vecadd   # on lab pod
grid=262144 blocks x 256 threads, errors=0
0.542 ms/iter, 1485 GB/s (3 arrays x 4 B moved per element)

The kernel does one add per 12 bytes moved, so it is memory-bound: 1485 GB/s is close to what this card's memory can deliver, and no amount of clever arithmetic would change that. Most inference kernels (elementwise ops, layernorm, softmax, decode-time attention) live in this regime, so "how many bytes does this kernel move" is the first question to ask.

Coalesced vs strided access

Copy#include <cstdio>
#include <cuda_runtime.h>

__global__ void copy_coalesced(const float* in, float* out, int n) {
    int i = blockIdx.x * blockDim.x + threadIdx.x;  // neighbouring threads -> neighbouring addresses
    if (i < n) out[i] = in[i];
}
__global__ void copy_strided(const float* in, float* out, int n, int stride) {
    int t = blockIdx.x * blockDim.x + threadIdx.x;
    long i = (long(t) * stride) % n;                // neighbouring threads -> addresses `stride` floats apart
    out[i] = in[i];
}

int main() {
    const int n = 1 << 26;
    float *in, *out;
    cudaMalloc(&in, n * 4L); cudaMalloc(&out, n * 4L);
    cudaMemset(in, 0, n * 4L);
    int threads = 256, blocks = n / threads;
    cudaEvent_t t0, t1; cudaEventCreate(&t0); cudaEventCreate(&t1);
    auto time = [&](auto launch) {
        launch(); cudaDeviceSynchronize();
        cudaEventRecord(t0);
        for (int r = 0; r < 10; ++r) launch();
        cudaEventRecord(t1); cudaEventSynchronize(t1);
        float ms; cudaEventElapsedTime(&ms, t0, t1); return ms / 10;
    };
    float base = time([&] { copy_coalesced<<<blocks, threads>>>(in, out, n); });
    printf("coalesced      : %.3f ms  %.0f GB/s\n", base, 2.0 * n * 4 / (base * 1e-3) / 1e9);
    for (int stride : {2, 8, 32}) {                 // ~stride*4 B between neighbouring threads; at 33 each thread hits its own 128 B line
        float ms = time([&] { copy_strided<<<blocks, threads>>>(in, out, n, stride + 1); });
        printf("stride %2d (odd): %.3f ms  %.0f GB/s useful  (%.1fx slower)\n", stride + 1, ms, 2.0 * n * 4 / (ms * 1e-3) / 1e9, ms / base);
    }
}

$ nvcc -O3 -arch=sm_120 cu02_coalescing.cu -o cu02_coalescing && ./cu02_coalescing   # on lab pod
coalesced      : 0.367 ms  1462 GB/s
stride  3 (odd): 1.793 ms  299 GB/s useful  (4.9x slower)
stride  9 (odd): 4.957 ms  108 GB/s useful  (13.5x slower)
stride 33 (odd): 13.032 ms  41 GB/s useful  (35.5x slower)

Same bytes copied, same arithmetic. Making neighbouring threads read addresses 33 floats apart cost 35x. If you ever see a kernel far below bandwidth in Nsight, check for strided or transposed access first.

Shared-memory reduction

Summing 64 M floats: each block loads its slice, reduces it in shared memory with a tree (half the threads active each step, __syncthreads() between steps), and one thread per block does a single atomicAdd into the result. Adding two elements during the load halves the number of blocks. Production code finishes the tree with warp shuffles (__shfl_down_sync) to skip the last syncs; CUB's DeviceReduce does all this for you.

Copy#include <cstdio>
#include <cuda_runtime.h>

// Each block reduces 2*blockDim.x elements in shared memory, then one atomicAdd per block.
__global__ void reduce_sum(const float* in, float* out, int n) {
    extern __shared__ float s[];                    // per-block scratchpad (on-chip, ~100x lower latency than DRAM)
    int tid = threadIdx.x;
    int i = blockIdx.x * blockDim.x * 2 + tid;
    float v = 0.f;
    if (i < n) v += in[i];
    if (i + blockDim.x < n) v += in[i + blockDim.x];  // first add during load: halves the block count
    s[tid] = v;
    __syncthreads();
    for (int stride = blockDim.x / 2; stride > 0; stride >>= 1) {  // tree reduction
        if (tid < stride) s[tid] += s[tid + stride];
        __syncthreads();                            // every step needs all writes visible
    }
    if (tid == 0) atomicAdd(out, s[0]);
}

int main() {
    const int n = 1 << 26;
    float *in, *out;
    cudaMallocManaged(&in, n * 4L); cudaMallocManaged(&out, 4);
    for (int i = 0; i < n; ++i) in[i] = 1.0f;
    int threads = 256, blocks = (n + threads * 2 - 1) / (threads * 2);
    *out = 0; reduce_sum<<<blocks, threads, threads * sizeof(float)>>>(in, out, n);
    cudaDeviceSynchronize();
    printf("sum = %.0f (expected %d)\n", *out, n);
    cudaEvent_t t0, t1; cudaEventCreate(&t0); cudaEventCreate(&t1);
    cudaEventRecord(t0);
    for (int r = 0; r < 20; ++r) { *out = 0; reduce_sum<<<blocks, threads, threads * sizeof(float)>>>(in, out, n); }
    cudaEventRecord(t1); cudaEventSynchronize(t1);
    float ms; cudaEventElapsedTime(&ms, t0, t1); ms /= 20;
    printf("%.3f ms/iter, %.0f GB/s read\n", ms, n * 4.0 / (ms * 1e-3) / 1e9);
}

$ nvcc -O3 -arch=sm_120 cu03_reduce.cu -o cu03_reduce && ./cu03_reduce   # on lab pod
sum = 67108864 (expected 67108864)
0.235 ms/iter, 1144 GB/s read

~1.1 TB/s read, i.e. the reduction is also bandwidth-limited, at about 77% of the copy rate above. Note that float addition is not associative: the atomicAdd ordering varies run to run, so the low bits of large float sums can differ between runs. Here all inputs were 1.0 so the sum is exact.

Tiled matrix multiply

Each block computes a 32x32 tile of C. It loads one tile of A and one of B into shared memory, every thread reuses those values 32 times, then slides to the next tile. Global-memory traffic drops by roughly the tile width.

Copy#include <cmath>
#include <cstdio>
#include <vector>
#include <cuda_runtime.h>

#define TILE 32

__global__ void matmul_naive(const float* A, const float* B, float* C, int n) {
    int row = blockIdx.y * TILE + threadIdx.y, col = blockIdx.x * TILE + threadIdx.x;
    float acc = 0.f;
    for (int k = 0; k < n; ++k) acc += A[row * n + k] * B[k * n + col];  // every operand from global memory
    C[row * n + col] = acc;
}

__global__ void matmul_tiled(const float* A, const float* B, float* C, int n) {
    __shared__ float As[TILE][TILE], Bs[TILE][TILE];
    int row = blockIdx.y * TILE + threadIdx.y, col = blockIdx.x * TILE + threadIdx.x;
    float acc = 0.f;
    for (int t = 0; t < n; t += TILE) {
        As[threadIdx.y][threadIdx.x] = A[row * n + t + threadIdx.x];     // each global element loaded once per tile...
        Bs[threadIdx.y][threadIdx.x] = B[(t + threadIdx.y) * n + col];
        __syncthreads();
        for (int k = 0; k < TILE; ++k) acc += As[threadIdx.y][k] * Bs[k][threadIdx.x];  // ...and reused TILE times
        __syncthreads();
    }
    C[row * n + col] = acc;
}

int main() {
    const int n = 2048;
    std::vector<float> hA(n * n), hB(n * n);
    for (int i = 0; i < n * n; ++i) { hA[i] = float(i % 13) * 0.1f; hB[i] = float(i % 7) * 0.1f; }
    float *A, *B, *C1, *C2;
    cudaMalloc(&A, n * n * 4L); cudaMalloc(&B, n * n * 4L); cudaMalloc(&C1, n * n * 4L); cudaMalloc(&C2, n * n * 4L);
    cudaMemcpy(A, hA.data(), n * n * 4L, cudaMemcpyHostToDevice);
    cudaMemcpy(B, hB.data(), n * n * 4L, cudaMemcpyHostToDevice);
    dim3 block(TILE, TILE), grid(n / TILE, n / TILE);
    cudaEvent_t t0, t1; cudaEventCreate(&t0); cudaEventCreate(&t1);
    auto time = [&](auto launch) {
        launch(); cudaDeviceSynchronize();
        cudaEventRecord(t0);
        for (int r = 0; r < 5; ++r) launch();
        cudaEventRecord(t1); cudaEventSynchronize(t1);
        float ms; cudaEventElapsedTime(&ms, t0, t1); return ms / 5;
    };
    float tn = time([&] { matmul_naive<<<grid, block>>>(A, B, C1, n); });
    float tt = time([&] { matmul_tiled<<<grid, block>>>(A, B, C2, n); });
    std::vector<float> r1(n * n), r2(n * n);
    cudaMemcpy(r1.data(), C1, n * n * 4L, cudaMemcpyDeviceToHost);
    cudaMemcpy(r2.data(), C2, n * n * 4L, cudaMemcpyDeviceToHost);
    float maxdiff = 0; for (int i = 0; i < n * n; ++i) maxdiff = fmaxf(maxdiff, fabsf(r1[i] - r2[i]));
    double flops = 2.0 * n * n * n;
    printf("naive: %.2f ms  %.0f GFLOP/s\n", tn, flops / (tn * 1e-3) / 1e9);
    printf("tiled: %.2f ms  %.0f GFLOP/s  (%.1fx)\n", tt, flops / (tt * 1e-3) / 1e9, tn / tt);
    printf("max |naive - tiled| = %g\n", maxdiff);
}

$ nvcc -O3 -arch=sm_120 cu04_tiled_matmul.cu -o cu04_tiled_matmul && ./cu04_tiled_matmul   # on lab pod
naive: 2.55 ms  6730 GFLOP/s
tiled: 1.98 ms  8698 GFLOP/s  (1.3x)
max |naive - tiled| = 0

How this connects to Triton and PyTorch

• Triton (the kernel language) lets you write the same thing at block granularity in Python: you load a vector of BLOCK elements with a mask, compute, store. The compiler handles thread mapping, coalescing and shared-memory staging. Section 10 explains why this name collides with Triton Inference Server.
• PyTorch custom ops: write a kernel plus a small C++ wrapper that checks tensors and launches it, then expose it to Python. torch.utils.cpp_extension.load_inline compiles that at import time; for real projects use a setup.py/CMake build and register with torch.library so torch.compile understands the op.

Copyimport torch, triton, triton.language as tl

@triton.jit
def add_kernel(x_ptr, y_ptr, out_ptr, n, BLOCK: tl.constexpr):
    pid = tl.program_id(0)                       # one program = one block of BLOCK elements
    offs = pid * BLOCK + tl.arange(0, BLOCK)
    mask = offs < n
    x = tl.load(x_ptr + offs, mask=mask)
    y = tl.load(y_ptr + offs, mask=mask)
    tl.store(out_ptr + offs, x + y, mask=mask)

def add(x, y):
    out = torch.empty_like(x)
    n = x.numel()
    add_kernel[(triton.cdiv(n, 1024),)](x, y, out, n, BLOCK=1024)
    return out

x = torch.randn(1 << 26, device="cuda"); y = torch.randn_like(x)
print("matches torch:", torch.equal(add(x, y), x + y))
print("triton (kernel language) version:", triton.__version__)
ms = triton.testing.do_bench(lambda: add(x, y))
print(f"triton add: {ms:.3f} ms  {3 * x.numel() * 4 / (ms * 1e-3) / 1e9:.0f} GB/s")

$ python3 cu06_triton_kernel.py   # on lab pod
matches torch: True
triton (kernel language) version: 3.6.0
triton add: 0.547 ms  1471 GB/s

Copyimport time, torch
from torch.utils.cpp_extension import load_inline

cuda_src = r"""
#include <torch/extension.h>
__global__ void saxpy_kernel(const float* x, const float* y, float* out, float a, int64_t n) {
    int64_t i = blockIdx.x * (int64_t)blockDim.x + threadIdx.x;
    if (i < n) out[i] = a * x[i] + y[i];
}
torch::Tensor saxpy(torch::Tensor x, torch::Tensor y, double a) {
    TORCH_CHECK(x.is_cuda() && y.is_cuda(), "tensors must be on the GPU");
    TORCH_CHECK(x.is_contiguous() && y.is_contiguous(), "tensors must be contiguous");
    TORCH_CHECK(x.scalar_type() == torch::kFloat32, "float32 only");
    auto out = torch::empty_like(x);
    int64_t n = x.numel();
    int threads = 256, blocks = (n + threads - 1) / threads;
    saxpy_kernel<<<blocks, threads>>>(x.data_ptr<float>(), y.data_ptr<float>(), out.data_ptr<float>(), (float)a, n);
    return out;
}
"""
cpp_src = "torch::Tensor saxpy(torch::Tensor x, torch::Tensor y, double a);"

t0 = time.time()
ext = load_inline(name="saxpy_ext", cpp_sources=cpp_src, cuda_sources=cuda_src,
                  functions=["saxpy"], extra_cuda_cflags=["-O3"], verbose=False)
print(f"compiled+loaded in {time.time()-t0:.0f}s")

x = torch.randn(1 << 26, device="cuda"); y = torch.randn_like(x)
print("matches torch:", torch.allclose(ext.saxpy(x, y, 2.0), 2.0 * x + y))

def bench(f, reps=20):
    f(); torch.cuda.synchronize()
    s, e = torch.cuda.Event(enable_timing=True), torch.cuda.Event(enable_timing=True)
    s.record()
    for _ in range(reps): f()
    e.record(); torch.cuda.synchronize()
    return s.elapsed_time(e) / reps
print(f"custom kernel : {bench(lambda: ext.saxpy(x, y, 2.0)):.3f} ms")
print(f"torch 2*x + y : {bench(lambda: 2.0 * x + y):.3f} ms  (two kernels + a temporary)")
try:
    ext.saxpy(x.cpu(), y.cpu(), 2.0)
except RuntimeError as e:
    print("CPU tensor ->", str(e).splitlines()[0])

$ python3 cu05_load_inline.py   # on lab pod
compiled+loaded in 0s
matches torch: True
custom kernel : 0.541 ms
torch 2*x + y : 0.908 ms  (two kernels + a temporary)
CPU tensor -> tensors must be on the GPU

5. Python for performance work

The GIL, concretely

CPython has a Global Interpreter Lock: only one thread executes Python bytecode at a time. Consequences for a serving process: (1) threads do not speed up pure-Python CPU work; (2) threads do help when waiting on I/O, because blocking calls (sockets, time.sleep) and most C extensions (numpy, PyTorch ops, tokenizers written in Rust, gRPC) release the GIL; (3) for CPU-bound Python, use processes (or move the work into a native library). Python 3.13 has an experimental free-threaded build, but the standard build and most deployed code still have the GIL.

Copyimport asyncio, time
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor

def cpu_work(n=12_000_000):          # pure-Python loop: holds the GIL
    s = 0
    for i in range(n): s += i * i
    return s

def io_work(_=None):                # stands in for a network call: releases the GIL while waiting
    time.sleep(0.2)

def timed(label, fn):
    t0 = time.perf_counter(); fn(); print(f"{label:<34}{time.perf_counter() - t0:6.2f} s")

async def aio_main(n):
    await asyncio.gather(*[asyncio.sleep(0.2) for _ in range(n)])

if __name__ == "__main__":
    print("CPU-bound, 4 tasks")
    timed("  sequential", lambda: [cpu_work() for _ in range(4)])
    with ThreadPoolExecutor(4) as ex: timed("  4 threads (GIL)", lambda: list(ex.map(lambda _: cpu_work(), range(4))))
    with ProcessPoolExecutor(4) as ex: timed("  4 processes", lambda: list(ex.map(cpu_work, [12_000_000] * 4)))
    print("I/O-bound, 20 tasks of 0.2 s")
    timed("  sequential", lambda: [io_work() for _ in range(20)])
    with ThreadPoolExecutor(20) as ex: timed("  20 threads", lambda: list(ex.map(io_work, range(20))))
    timed("  asyncio.gather", lambda: asyncio.run(aio_main(20)))

$ python3 py05a_gil.py
CPU-bound, 4 tasks
  sequential                        1.07 s
  4 threads (GIL)                   1.02 s
  4 processes                       0.38 s
I/O-bound, 20 tasks of 0.2 s
  sequential                        4.13 s
  20 threads                        0.21 s
  asyncio.gather                    0.20 s

Generators, dataclasses, typing, and the classic gotchas

Copyimport sys
from dataclasses import dataclass, field
from typing import Protocol, Iterator

# ---- generators: lazy, constant memory
def batches(it, size) -> Iterator[list]:
    buf = []
    for x in it:
        buf.append(x)
        if len(buf) == size:
            yield buf; buf = []
    if buf: yield buf

eager = [i for i in range(1_000_000)]
lazy = (i for i in range(1_000_000))
print("list bytes:", sys.getsizeof(eager), " generator bytes:", sys.getsizeof(lazy))
print(list(batches(range(7), 3)))

# ---- dataclasses + typing
@dataclass(frozen=True, slots=True)
class Request:
    model: str
    tokens: int
    tags: tuple[str, ...] = ()

class Scorer(Protocol):              # structural typing: anything with .score() qualifies
    def score(self, r: Request) -> float: ...

class LenScorer:
    def score(self, r: Request) -> float: return r.tokens / 100

def rank(rs: list[Request], s: Scorer) -> list[Request]:
    return sorted(rs, key=s.score, reverse=True)

rs = [Request("a", 50), Request("b", 900), Request("c", 300)]
print([r.model for r in rank(rs, LenScorer())])
try:
    rs[0].tokens = 1
except Exception as e:
    print(type(e).__name__, "-", e)

# ---- gotcha 1: mutable default argument is created once
def add_bad(x, acc=[]):
    acc.append(x); return acc
print(add_bad(1), add_bad(2))
def add_good(x, acc=None):
    acc = [] if acc is None else acc
    acc.append(x); return acc
print(add_good(1), add_good(2))

# ---- gotcha 2: closures capture variables, not values
fs_bad = [lambda: i for i in range(3)]
fs_good = [lambda i=i: i for i in range(3)]
print([f() for f in fs_bad], [f() for f in fs_good])

# ---- gotcha 3: aliasing in nested lists
grid_bad = [[0] * 2] * 2
grid_bad[0][0] = 1
grid_good = [[0] * 2 for _ in range(2)]
grid_good[0][0] = 1
print(grid_bad, grid_good)

# ---- gotcha 4: 'is' vs '=='
a = 1000; b = 1000
print(a == b, [1] == [1], [1] is [1])

$ python3 py05b_basics.py
list bytes: 8448728  generator bytes: 192
[[0, 1, 2], [3, 4, 5], [6]]
['b', 'c', 'a']
FrozenInstanceError - cannot assign to field 'tokens'
[1, 2] [1, 2]
[1] [2]
[2, 2, 2] [0, 1, 2]
[[1, 0], [1, 0]] [[1, 0], [0, 0]]
True True False

• A generator holds 192 bytes regardless of length; the list held 8 MB. Use generators/iterators for streams of requests, lines, batches.
• @dataclass(frozen=True, slots=True) gives immutability, hashability and no per-instance __dict__ (less memory, faster attribute access). Protocol is structural typing: LenScorer never inherited from Scorer.
• Default arguments are evaluated once, at def time, so acc=[] is one shared list. Closures capture the variable, not its value at that moment, so all lambdas saw the final i = 2; bind with a default argument. [[0]*2]*2 repeats a reference to one inner list.

Profiling: find it before you fix it

cProfile is deterministic and shows call counts and where time goes by function (adds overhead, distorts tiny functions). py-spy is a sampling profiler that attaches to a running process with no code changes and low overhead, so it is the right tool against a live server (py-spy dump --pid N for a stack snapshot, py-spy record for a flamegraph). For one-off line timing use time.perf_counter() (monotonic, high resolution; never time.time() for durations) or timeit.

Copyimport cProfile, io, pstats, time, timeit
from contextlib import contextmanager
import numpy as np

def parse(lines):                        # string work
    return [float(l.split(",")[1]) for l in lines]

def normalize(xs):                       # naive Python loop over floats
    m = max(xs); out = []
    for x in xs: out.append(x / m)
    return out

def pipeline(lines):
    return sum(normalize(parse(lines)))

lines = [f"id{i},{i % 997}.5" for i in range(300_000)]
pr = cProfile.Profile(); pr.enable(); pipeline(lines); pr.disable()
s = io.StringIO(); pstats.Stats(pr, stream=s).sort_stats("tottime").print_stats(3)
print("\n".join(l.strip() for l in s.getvalue().splitlines() if l.strip())[:700])

@contextmanager
def span(name):                          # poor man's line timer
    t0 = time.perf_counter(); yield
    print(f"{name}: {(time.perf_counter() - t0) * 1e3:.1f} ms")

xs = np.random.default_rng(0).random(1_000_000)
with span("python loop"): r1 = sum(x * x for x in xs.tolist())
with span("numpy vectorized"): r2 = xs @ xs
print("equal:", np.isclose(r1, r2))
print("timeit xs @ xs: %.3f ms/call" % (timeit.timeit(lambda: xs @ xs, number=100) * 10))

$ python3 py05c_profile.py
600006 function calls in 0.081 seconds
Ordered by: internal time
List reduced from 8 to 3 due to restriction <3>
ncalls  tottime  percall  cumtime  percall filename:lineno(function)
1    0.031    0.031    0.047    0.047 py05c_profile.py:6(parse)
1    0.017    0.017    0.031    0.031 py05c_profile.py:9(normalize)
300000    0.016    0.000    0.016    0.000 {method 'split' of 'str' objects}
python loop: 28.1 ms
numpy vectorized: 0.4 ms
equal: True
timeit xs @ xs: 0.049 ms/call

Vectorizing with numpy was ~50-70x faster for the same sum of squares (the exact ratio varies by run). Below, py-spy sampling a script on the lab pod (200 samples/s for 3 s); each stack ends with its sample count, so the generator expression on line 10 is the hot spot (265 samples), then the enclosing checksum frame itself (214), then tokenize (66):

Copy# py-spy sampling target
import time

def tokenize(n):               # pure-Python string work
    return [str(i) * 3 for i in range(n)]

def checksum(items):           # pure-Python arithmetic
    s = 0
    for it in items:
        s += sum(ord(c) for c in it)
    return s

def main():
    t_end = time.time() + 3
    while time.time() < t_end:
        checksum(tokenize(20000))
main()

$ py-spy record -f raw -r 200 -o spy.txt -- python3 py05e_pyspy.py   # on lab pod; then sum samples per stack, keep the top 4
<module> (py05e_pyspy.py:17)
    main (py05e_pyspy.py:16)
    checksum (py05e_pyspy.py:10)
    <genexpr> (py05e_pyspy.py:10) 265
<module> (py05e_pyspy.py:17)
    main (py05e_pyspy.py:16)
    checksum (py05e_pyspy.py:10) 214
<module> (py05e_pyspy.py:17)
    main (py05e_pyspy.py:16)
    tokenize (py05e_pyspy.py:5) 66
<module> (py05e_pyspy.py:17)
    main (py05e_pyspy.py:16)
    checksum (py05e_pyspy.py:9) 5

Zero-copy: memoryview and the buffer protocol

Slicing bytes/bytearray copies. A memoryview slice is a new window onto the same memory. np.frombuffer wraps existing bytes as an array without copying (read-only if the source is immutable). This matters when handling large tensors/images off a socket or shared-memory segment.

Copyimport numpy as np, time

buf = bytearray(b"hello world, hello zero copy")
mv = memoryview(buf)
head = mv[:5]
buf[0:1] = b"J"                              # mutate the original...
print(bytes(head), "<- slice sees it (no copy)")

big = bytearray(200_000_000)
t0 = time.perf_counter(); c = big[1:]; t1 = time.perf_counter()
m = memoryview(big)[1:]; t2 = time.perf_counter()
print(f"bytearray slice (copy): {(t1-t0)*1e3:.1f} ms   memoryview slice: {(t2-t1)*1e6:.1f} us")

raw = np.arange(6, dtype=np.float32).tobytes()
arr = np.frombuffer(raw, dtype=np.float32)   # view onto the bytes object
print(arr, "writeable:", arr.flags.writeable, " owns data:", arr.flags.owndata)
arr2 = np.frombuffer(bytearray(raw), dtype=np.float32)
arr2[0] = 42; print("writable buffer ->", arr2[:2])

$ python3 py05d_zero_copy.py
b'Jello' <- slice sees it (no copy)
bytearray slice (copy): 53.4 ms   memoryview slice: 8.5 us
[0. 1. 2. 3. 4. 5.] writeable: False  owns data: False
writable buffer -> [42.  1.]

Copying a 200 MB slice took tens of milliseconds (33-53 ms across runs); the memoryview slice took ~10 microseconds.

6. numpy essentials

• Broadcasting: shapes are compared from the right; each pair of dimensions must be equal or one must be 1 (or missing). The size-1 dimension is stretched virtually, no memory copied.
• Strides: an array is a buffer plus (shape, strides in bytes). Transposing, slicing with steps, reshape of contiguous data, and broadcasting just change metadata, so they are views. Boolean masks and fancy (list/array) indexing build a new array: copies.
• dtype: determines memory, speed and overflow. Integer overflow wraps silently; floats lose integer precision beyond 2^24 (fp32) and 2^53 (fp64); fp16 has an 11-bit significand.

Copyimport numpy as np

# ---- broadcasting: align shapes from the right; size-1 (or missing) dims stretch
x = np.arange(6).reshape(2, 3)          # (2,3)
row = np.array([10, 20, 30])            # (3,)   -> (1,3) -> (2,3)
col = np.array([[100], [200]])          # (2,1)  -> (2,3)
print(x + row); print(x + col)
print("(3,) + (2,1) ->", (np.zeros(3) + np.zeros((2, 1))).shape)
try:
    np.zeros((2, 3)) + np.zeros(2)
except ValueError as e:
    print("ValueError:", e)

# ---- the classic: per-row normalization needs keepdims
logits = np.array([[1., 2., 3.], [1., 1., 1.]])
p = np.exp(logits - logits.max(axis=1, keepdims=True))
p /= p.sum(axis=1, keepdims=True)
print(p.round(3), p.sum(axis=1))

# ---- strides, views vs copies
a = np.arange(12, dtype=np.int32).reshape(3, 4)
print("strides (bytes):", a.strides, " transpose:", a.T.strides, " a.T contiguous:", a.T.flags.c_contiguous)
v = a[:, ::2]                           # basic slicing -> view
c = a[[0, 2]]                           # fancy indexing -> copy
print("slice shares memory:", np.shares_memory(a, v), "| fancy index shares:", np.shares_memory(a, c))
v[0, 0] = 99; print("a[0,0] after writing through the view:", a[0, 0])
r = a.reshape(4, 3); rt = a.T.reshape(12)
print("reshape of contiguous is a view:", np.shares_memory(a, r), "| reshape of transpose copies:", not np.shares_memory(a, rt))

# ---- dtype: size, overflow, promotion
print(np.array([1, 2, 3], dtype=np.int8) * 100)         # wraps silently
print(np.float16(2049), np.float32(16777217), np.int64(2**53 + 1) + 0.0)
print((np.ones(3, np.float32) + np.ones(3, np.float64)).dtype, (np.ones(3, np.float16) + 1.5).dtype, (np.arange(3) / 2).dtype)
h = np.zeros(1_000_000, np.float16); f = np.zeros(1_000_000, np.float32)
print("1M elems: fp16", h.nbytes, "B, fp32", f.nbytes, "B")

$ python3 np06_numpy.py
[[10 21 32]
 [13 24 35]]
[[100 101 102]
 [203 204 205]]
(3,) + (2,1) -> (2, 3)
ValueError: operands could not be broadcast together with shapes (2,3) (2,) 
[[0.09  0.245 0.665]
 [0.333 0.333 0.333]] [1. 1.]
strides (bytes): (16, 4)  transpose: (4, 16)  a.T contiguous: False
slice shares memory: True | fancy index shares: False
a[0,0] after writing through the view: 99
reshape of contiguous is a view: True | reshape of transpose copies: True
[100 -56  44]
2.048e+03 1.6777216e+07 9007199254740992.0
float64 float16 float64
1M elems: fp16 2000000 B, fp32 4000000 B

7. pandas essentials, with a latency-log analysis

This page ran pandas 3.0.6. Version matters: pandas 3 enforces copy-on-write and a dedicated string dtype; in 2.x the same trap raises SettingWithCopyWarning instead. Cheat sheet first, on a six-row frame.

Copyimport pandas as pd
pd.set_option("display.width", 100)

df = pd.DataFrame({
    "model": ["bert", "bert", "gpt", "gpt", "gpt", "t5"],
    "node":  ["n1",   "n2",   "n1",  "n2",  "n2",  "n3"],
    "lat_ms": [12.0, 15.0, 80.0, 95.0, 400.0, 30.0],
}, index=list("abcdef"))

print(df.loc["b":"d", ["model", "lat_ms"]])      # label slice: END INCLUSIVE
print(df.iloc[1:3, :2])                          # position slice: end exclusive
print(df[(df.lat_ms > 50) & (df.model == "gpt")]) # & | ~ with parentheses, not and/or

g = df.groupby("model").agg(n=("lat_ms", "size"), mean=("lat_ms", "mean"),
                            p95=("lat_ms", lambda s: s.quantile(0.95)))
print(g)

nodes = pd.DataFrame({"node": ["n1", "n2", "n4"], "gpu": ["L40S", "RTX6000", "H100"]})
for how in ["inner", "left", "outer"]:
    m = df.merge(nodes, on="node", how=how)
    print(f"{how:<5} -> {len(m)} rows; NaN gpu rows: {int(m.gpu.isna().sum())}")

print(df.pivot_table(index="model", columns="node", values="lat_ms", aggfunc="median"))

ts = pd.to_datetime(["2026-01-01 10:05", "2026-01-01 10:55", "2026-01-01 11:10"])
s = pd.Series([1, 2, 3], index=ts)
print(s.resample("h").sum())
print(ts.floor("h").tolist()[0], "|", ts.hour.tolist(), "|", ts.tz)

$ python3 pd07a_basics.py
  model  lat_ms
b  bert    15.0
c   gpt    80.0
d   gpt    95.0
  model node
b  bert   n2
c   gpt   n1
  model node  lat_ms
c   gpt   n1    80.0
d   gpt   n2    95.0
e   gpt   n2   400.0
       n        mean     p95
model                       
bert   2   13.500000   14.85
gpt    3  191.666667  369.50
t5     1   30.000000   30.00
inner -> 5 rows; NaN gpu rows: 0
left  -> 6 rows; NaN gpu rows: 1
outer -> 7 rows; NaN gpu rows: 1
node     n1     n2    n3
model                   
bert   12.0   15.0   NaN
gpt    80.0  247.5   NaN
t5      NaN    NaN  30.0
2026-01-01 10:00:00    3
2026-01-01 11:00:00    3
Freq: h, dtype: int64
2026-01-01 10:00:00 | [10, 10, 11] | None

• .loc is by label and the slice end is inclusive ("b":"d" gave 3 rows); .iloc is by position and end-exclusive (1:3 gave 2). Combine masks with &, |, ~ and parentheses.
• groupby(...).agg(name=(col, func)) is named aggregation. merge: inner keeps matches only (5 rows), left keeps all left rows with NaN for misses (6), outer keeps everything (7). Pass validate="many_to_one" to catch accidental row multiplication from duplicate keys.
• pivot_table is groupby + unstack with an aggfunc. resample("h") needs a DatetimeIndex; .dt.floor("h") gives you a groupable hour column; timestamps are tz-naive by default (ts.tz is None), so be explicit about UTC.

Traps and performance

Copyimport time, warnings
import numpy as np, pandas as pd

print("pandas", pd.__version__)
df = pd.DataFrame({"a": [1, 2, 3, 4], "b": [10, 20, 30, 40]})

# ---- the SettingWithCopy trap (chained assignment)
with warnings.catch_warnings(record=True) as w:
    warnings.simplefilter("always")
    df[df.a > 2]["b"] = -1                  # sets on a temporary, then throws it away
    print("chained assignment changed df:", (df.b == -1).any(),
          "| warning:", w[0].category.__name__ if w else None)
df.loc[df.a > 2, "b"] = -1                  # one .loc call: the correct form
print("single .loc changed df:", df.b.tolist())

# ---- apply vs vectorized vs numpy
n = 500_000
d = pd.DataFrame({"x": np.random.default_rng(0).random(n)})
t0 = time.perf_counter(); r1 = d.x.apply(lambda v: v * 2 + 1); t1 = time.perf_counter()
r2 = d.x * 2 + 1;                                              t2 = time.perf_counter()
print(f"apply: {(t1-t0)*1e3:.0f} ms   vectorized: {(t2-t1)*1e3:.1f} ms   same: {np.allclose(r1, r2)}")

# ---- categorical + memory
s = pd.Series(np.random.default_rng(0).choice(["bert-base", "gpt-large", "t5-xl"], n))
print("object dtype :", s.memory_usage(deep=True) // 1024, "KiB")
print("category     :", s.astype("category").memory_usage(deep=True) // 1024, "KiB")
f64 = pd.Series(np.random.default_rng(0).random(n))
print("float64 vs float32:", f64.memory_usage() // 1024, "vs", f64.astype("float32").memory_usage() // 1024, "KiB")

# ---- percentile gotchas
lat = pd.Series([10, 12, 11, 13, 500])
print("mean:", lat.mean(), " p50:", lat.quantile(.5), " p99:", lat.quantile(.99),
      "| nearest-rank p99:", lat.quantile(.99, interpolation="higher"))

$ python3 pd07b_traps.py
pandas 3.0.6
chained assignment changed df: False | warning: ChainedAssignmentError
single .loc changed df: [10, 20, -1, -1]
apply: 32 ms   vectorized: 0.7 ms   same: True
object dtype : 7649 KiB
category     : 488 KiB
float64 vs float32: 3906 vs 1953 KiB
mean: 109.2  p50: 12.0  p99: 480.52 | nearest-rank p99: 500

• Series.apply(lambda) calls Python once per element: ~30-40 ms vs under 1 ms vectorized for 500 K rows (~40x). Reach for vectorized column ops, np.where, .map with a dict, or groupby.transform before apply.
• category dtype stores each distinct string once plus small integer codes: 7.6 MiB → 0.5 MiB for a 3-valued string column of 500 K rows in this run (an earlier run in an environment without pyarrow, where pandas 3 falls back to Python-object strings, measured ~27 MiB), and groupbys on categoricals are faster. Downcasting float64 → float32 halves memory.
• Percentiles: quantile interpolates linearly by default. For 5 latencies [10,12,11,13,500], "p99" = 480.5, a value that never occurred; interpolation="higher" gave 500. Know which definition your dashboard (Prometheus histograms interpolate within buckets) uses before comparing numbers. And never average percentiles across groups; recompute them from the raw samples or from mergeable sketches/histograms.

Realistic task: p50/p95/p99 per model and hour, outliers, node join

The script generates 240 K synthetic inference requests over 6 hours across 3 models and 6 nodes, with two planted problems: node n5 becomes 3x slower after 03:00, and 0.1% of requests time out (0.9-2 s). Then it answers the questions an on-call engineer would ask. The seed is fixed so the output is reproducible.

Copyimport numpy as np, pandas as pd
pd.set_option("display.width", 110); pd.set_option("display.max_columns", 20)
rng = np.random.default_rng(42)

# ---- 1. synthetic inference-request log: 6 h, 3 models, 6 nodes
N = 240_000
ts = pd.Timestamp("2026-03-01 00:00") + pd.to_timedelta(np.sort(rng.uniform(0, 6 * 3600, N)), unit="s")
log = pd.DataFrame({
    "ts": ts,
    "model": rng.choice(["llmqu-1b", "ranker-v3", "embed-s"], N, p=[.3, .5, .2]),
    "node": rng.choice([f"n{i}" for i in range(1, 7)], N),
    "tokens_in": rng.integers(8, 512, N),
})
base = log.model.map({"llmqu-1b": 30.0, "ranker-v3": 8.0, "embed-s": 4.0})
log["latency_ms"] = base * rng.lognormal(0, 0.25, N) * (1 + log.tokens_in / 1024)
sick = (log.node == "n5") & (log.ts >= "2026-03-01 03:00")        # inject: n5 degrades after 03:00
log.loc[sick, "latency_ms"] *= 3
spike = rng.random(N) < 0.001                                     # inject: 0.1% timeouts
log.loc[spike, "latency_ms"] = rng.uniform(900, 2000, spike.sum())
log["model"] = log.model.astype("category"); log["node"] = log.node.astype("category")
print(log.head(3)); print(f"{len(log):,} rows, {log.memory_usage(deep=True).sum() / 1e6:.1f} MB\n")

# ---- 2. p50/p95/p99 per model and hour
log["hour"] = log.ts.dt.floor("h")
q = (log.groupby(["model", "hour"], observed=True).latency_ms
        .quantile([.5, .95, .99]).unstack().rename(columns=lambda c: f"p{int(c * 100)}").round(1))
print(q.loc["llmqu-1b"]); print()

# ---- 3. outliers: robust z-score (median/MAD) within each model
g = log.groupby("model", observed=True).latency_ms
med = g.transform("median"); mad = (log.latency_ms - med).abs().groupby(log.model, observed=True).transform("median")
log["rz"] = 0.6745 * (log.latency_ms - med) / mad
out = log[log.rz > 10]
print(f"outliers (robust z > 10): {len(out)} ({len(out) / len(log):.2%}); by model:\n{out.model.value_counts().to_string()}\n")

# ---- 4. join node metadata; find the sick node
nodes = pd.DataFrame({"node": [f"n{i}" for i in range(1, 7)],
                      "gpu": ["L40S"] * 3 + ["RTX6000"] * 3, "az": ["a", "a", "b", "b", "c", "c"]})
j = log.merge(nodes, on="node", how="left", validate="many_to_one")
j = j[j.model == "ranker-v3"]
by = j.groupby(["node", j.ts.dt.hour >= 3], observed=True).latency_ms.median().unstack().round(2)
by.columns = ["median before 03:00", "median after 03:00"]
print(by.join(nodes.set_index("node")[["gpu", "az"]]))
print("\nmax p99 across nodes after 03:00 ->", j[j.ts.dt.hour >= 3].groupby("node", observed=True).latency_ms.quantile(.99).idxmax())

$ python3 pd07c_latency_log.py
                             ts      model node  tokens_in  latency_ms
0 2026-03-01 00:00:00.284440814  ranker-v3   n1        253   15.177045
1 2026-03-01 00:00:00.286897792    embed-s   n3        280    5.826168
2 2026-03-01 00:00:00.378636467  ranker-v3   n3         31    7.290388
240,000 rows, 6.2 MB

                      p50    p95    p99
hour                                   
2026-03-01 00:00:00  37.3   58.9   71.1
2026-03-01 01:00:00  37.4   59.1   70.6
2026-03-01 02:00:00  37.8   59.1   72.6
2026-03-01 03:00:00  40.1  130.0  170.2
2026-03-01 04:00:00  40.2  131.4  178.1
2026-03-01 05:00:00  39.9  129.8  172.5

outliers (robust z > 10): 2614 (1.09%); by model:
model
ranker-v3    1302
llmqu-1b      806
embed-s       506

      median before 03:00  median after 03:00      gpu az
node                                                     
n1                   9.97                9.91     L40S  a
n2                  10.03                9.97     L40S  a
n3                   9.90                9.91     L40S  b
n4                   9.98               10.00  RTX6000  b
n5                  10.02               29.93  RTX6000  c
n6                   9.98                9.93  RTX6000  c

max p99 across nodes after 03:00 -> n5

8. TensorFlow essentials

Tensors and eager mode. Immutable n-d arrays with a dtype and (maybe partially unknown) shape; tf.Variable holds mutable state; tf.GradientTape records ops for autodiff. Eager (op-by-op) execution is the default. tf.function traces your Python once per distinct input signature into a graph (a dataflow program TF can optimize, serialize and run without Python). Tracing executes your Python, so Python side effects (print, appending to a list) happen only at trace time. Python scalars/objects are baked into the graph as constants, so a new Python value means a new trace; tensors with a new dtype or shape also trigger a retrace unless you pin an input_signature with None dims. Retracing in a serving loop is a latency killer.

Copyimport os; os.environ["TF_CPP_MIN_LOG_LEVEL"] = "3"
import sys, time, tensorflow as tf
sys.stdout.reconfigure(line_buffering=True)   # keep tf.print and print in order when piped
tf.get_logger().setLevel('ERROR')   # hides the retracing warnings TF prints after the 5th retrace
print("TF", tf.__version__)

x = tf.constant([[1., 2.], [3., 4.]])
print(x.shape, x.dtype, (x @ tf.transpose(x)).numpy().tolist(), tf.reduce_sum(x, axis=0).numpy())
v = tf.Variable(1.0)
with tf.GradientTape() as tape:
    y = v * v * 3
print("d(3v^2)/dv at v=1:", tape.gradient(y, v).numpy())

traces = []
@tf.function
def f(a):
    traces.append(1)                 # Python side effect: runs only while TRACING
    return a * 2

f(tf.constant(1.0)); f(tf.constant(2.0))
print("after 2 calls with same dtype/shape, traces =", len(traces))
f(tf.constant([1.0, 2.0]));  f(tf.constant([1.0, 2.0, 3.0]))
print("two new shapes, traces =", len(traces))
for i in range(3): f(i)              # Python ints are baked into the graph as constants
print("three different Python ints, traces =", len(traces))

traces.clear()
@tf.function(input_signature=[tf.TensorSpec([None], tf.float32)])
def g(a):
    traces.append(1); return a * 2
for n in (1, 2, 3, 4): g(tf.ones([n]))
print("with TensorSpec([None]): 4 shapes, traces =", len(traces))

@tf.function
def counter():
    print("python print (trace time only)")
    tf.print("tf.print (every run)", output_stream=sys.stdout)
counter(); counter()

def gelu_chain(a):                   # elementwise ops: XLA fuses them into one kernel
    for _ in range(6): a = a * tf.sigmoid(1.702 * a) + 0.01
    return a
xla = tf.function(gelu_chain, jit_compile=True)
big = tf.random.normal([4096, 4096])
xla(big)
t0 = time.perf_counter(); [gelu_chain(big) for _ in range(5)]; t1 = time.perf_counter()
[xla(big) for _ in range(5)];                                 t2 = time.perf_counter()
print(f"eager {(t1-t0)*200:.0f} ms/iter, XLA {(t2-t1)*200:.0f} ms/iter (CPU)")
hlo = xla.experimental_get_compiler_ir(big)(stage="optimized_hlo")
print("fusion computations in optimized HLO:", hlo.count("fusion("), "| HLO lines:", len(hlo.splitlines()))

$ python3 tf08a_function.py
TF 2.21.0
(2, 2) <dtype: 'float32'> [[5.0, 11.0], [11.0, 25.0]] [4. 6.]
d(3v^2)/dv at v=1: 6.0
after 2 calls with same dtype/shape, traces = 1
two new shapes, traces = 3
three different Python ints, traces = 6
with TensorSpec([None]): 4 shapes, traces = 1
python print (trace time only)
tf.print (every run)
tf.print (every run)
eager 83 ms/iter, XLA 21 ms/iter (CPU)
fusion computations in optimized HLO: 1 | HLO lines: 59

Keras, tf.data, SavedModel, signatures

tf.data builds input pipelines as a lazy graph of transformations (map, shuffle, batch); .prefetch(AUTOTUNE) overlaps preparing batch N+1 with computing batch N. A SavedModel is a directory with a serialized graph (saved_model.pb), weights (variables/) and one or more signatures: named entry points with fixed input names/dtypes/shapes. serving_default is what TF Serving, and most converters, call. The input name comes from your Keras Input(name=...) or TensorSpec(name=...); name them, because the default (keras_tensor) leaks into your client contract.

Copyimport os; os.environ["TF_CPP_MIN_LOG_LEVEL"] = "3"
import numpy as np, shutil, tensorflow as tf, keras

gpus = tf.config.list_physical_devices("GPU")
for g in gpus: tf.config.experimental.set_memory_growth(g, True)   # must run before GPU init
print("GPUs visible:", len(gpus), "(memory growth set on each; no-op here if 0)")

# ---- tf.data: shuffle -> batch -> prefetch overlaps input prep with compute
rng = np.random.default_rng(0)
X = rng.normal(size=(1024, 8)).astype("float32"); y = (X[:, 0] + X[:, 1] > 0).astype("float32")
ds = tf.data.Dataset.from_tensor_slices((X, y)).shuffle(1024, seed=0).batch(64).prefetch(tf.data.AUTOTUNE)
print("element_spec:", ds.element_spec)

# ---- Keras
model = keras.Sequential([keras.Input((8,), name="features"), keras.layers.Dense(16, activation="relu"), keras.layers.Dense(1, activation="sigmoid")])
model.compile("adam", "binary_crossentropy", metrics=["accuracy"])
hist = model.fit(ds, epochs=5, verbose=0, shuffle=False)
print("loss per epoch:", [round(l, 3) for l in hist.history["loss"]], "| acc:", round(hist.history["accuracy"][-1], 3))

# ---- SavedModel + signatures
shutil.rmtree("/tmp/lw_sm", ignore_errors=True)
model.export("/tmp/lw_sm")                       # Keras 3: writes a serving SavedModel
loaded = tf.saved_model.load("/tmp/lw_sm")
print("signatures:", list(loaded.signatures))
sig = loaded.signatures["serving_default"]
print("inputs :", {k: (v.shape.as_list(), v.dtype.name) for k, v in sig.structured_input_signature[1].items()})
print("outputs:", {k: v.shape.as_list() for k, v in sig.structured_outputs.items()})
out = sig(**{list(sig.structured_input_signature[1])[0]: tf.constant(X[:2])})
print("matches model.predict:", np.allclose(list(out.values())[0].numpy(), model.predict(X[:2], verbose=0), atol=1e-6))
print(sorted(os.listdir("/tmp/lw_sm")))

$ python3 tf08b_keras_serving.py
GPUs visible: 0 (memory growth set on each; no-op here if 0)
element_spec: (TensorSpec(shape=(None, 8), dtype=tf.float32, name=None), TensorSpec(shape=(None,), dtype=tf.float32, name=None))
loss per epoch: [0.782, 0.719, 0.669, 0.628, 0.594] | acc: 0.652
Saved artifact at '/tmp/lw_sm'. The following endpoints are available:

* Endpoint 'serve'
  args_0 (POSITIONAL_ONLY): TensorSpec(shape=(None, 8), dtype=tf.float32, name='features')
Output Type:
  TensorSpec(shape=(None, 1), dtype=tf.float32, name=None)
Captures:
signatures: ['serve', 'serving_default']
inputs : {'features': ([None, 8], 'float32')}
outputs: {'output_0': [None, 1]}
matches model.predict: True
['assets', 'fingerprint.pb', 'saved_model.pb', 'variables']

The captured-resource lines are the model's variables. The check matches model.predict confirms the exported signature computes the same function. GPU memory growth: by default TF reserves nearly all GPU memory at startup, which breaks co-tenancy (two TF processes, or TF next to anything else). set_memory_growth(gpu, True) before GPU initialization makes it allocate on demand; the alternative is a hard cap via set_logical_device_configuration(... memory_limit=MB). The output above shows 0 GPUs, so the call was a no-op here: TF on GPU was not run.

TF Serving vs Triton's TensorFlow backend

TF to ONNX with tf2onnx

Outline: (1) export a SavedModel or build a tf.function with an explicit input_signature (use None for the batch dimension, or the ONNX input will be fixed-batch); (2) run tf2onnx choosing an opset; (3) validate numerically against TF on real inputs, and with several batch sizes; (4) inspect ops for anything exotic that your target runtime lacks. Both entry points were run:

Copyimport os; os.environ["TF_CPP_MIN_LOG_LEVEL"] = "3"
import numpy as np, tensorflow as tf, keras, tf2onnx, onnxruntime as ort, onnx
tf.get_logger().setLevel("ERROR")

model = keras.Sequential([keras.Input((8,), name="features"), keras.layers.Dense(16, activation="relu"), keras.layers.Dense(1, activation="sigmoid")])

# Wrap in a tf.function with a dynamic batch dim; this is the portable entry point.
@tf.function(input_signature=[tf.TensorSpec([None, 8], tf.float32, name="features")])
def serve(features): return {"score": model(features)}

onnx_model, _ = tf2onnx.convert.from_function(serve, input_signature=serve.input_signature, opset=17,
                                              output_path="/tmp/lw_model.onnx")
g = onnx.load("/tmp/lw_model.onnx").graph
print("inputs :", [(i.name, [d.dim_param or d.dim_value for d in i.type.tensor_type.shape.dim]) for i in g.input])
print("outputs:", [o.name for o in g.output], "| ops:", sorted({n.op_type for n in g.node}))

X = np.random.default_rng(1).normal(size=(5, 8)).astype("float32")
sess = ort.InferenceSession("/tmp/lw_model.onnx", providers=["CPUExecutionProvider"])
ref = serve(tf.constant(X))["score"].numpy()
got = sess.run(None, {"features": X})[0]
print("max |TF - ORT| =", float(np.abs(ref - got).max()))
print("batch 1 and batch 64 both run:", sess.run(None, {"features": X[:1]})[0].shape, sess.run(None, {"features": np.tile(X, (13, 1))[:64]})[0].shape)

$ python3 tf08c_tf2onnx.py
inputs : [('features', ['unk__6', 8])]
outputs: ['score'] | ops: ['MatMul', 'Relu', 'Sigmoid']
max |TF - ORT| = 5.960464477539063e-08
batch 1 and batch 64 both run: (1, 1) (64, 1)

The dynamic batch dimension came through (unk__6 is the symbolic batch), the three ops are as expected, TF and ONNX Runtime agree to 6e-8 (float32 rounding), and batch sizes 1 and 64 both work. The command-line form on the SavedModel exported above:

Copypython -m tf2onnx.convert --saved-model /tmp/lw_sm --opset 17 --output model.onnx

$ python -m tf2onnx.convert --saved-model /tmp/lw_sm --opset 17 --output /tmp/lw_cli.onnx   # last 4 log lines
2026-10-16 18:04:34,691 - INFO - Successfully converted TensorFlow model /tmp/lw_sm to ONNX
2026-10-16 18:04:34,691 - INFO - Model inputs: ['features']
2026-10-16 18:04:34,691 - INFO - Model outputs: ['output_0']
2026-10-16 18:04:34,691 - INFO - ONNX model is saved at /tmp/lw_cli.onnx

9. PyTorch essentials, side by side with TF, and a short JAX comparison

Copyimport torch, torch.nn as nn
from torch.utils.data import Dataset, DataLoader

# ---- autograd: dynamic graph recorded as you run
w = torch.tensor(2.0, requires_grad=True)
loss = (w * 3 - 1) ** 2
loss.backward()
print("d/dw (3w-1)^2 at w=2:", w.grad.item(), "(analytic 6*(3w-1) = 30)")
with torch.no_grad():                    # serving: no graph, less memory
    print("requires_grad inside no_grad:", (w * 2).requires_grad)

# ---- nn.Module + state_dict
class Net(nn.Module):
    def __init__(self):
        super().__init__()
        self.fc1, self.fc2 = nn.Linear(8, 16), nn.Linear(16, 1)
    def forward(self, x): return torch.sigmoid(self.fc2(torch.relu(self.fc1(x))))

net = Net().cuda()
print("state_dict keys:", list(net.state_dict().keys()))
torch.save(net.state_dict(), "/tmp/lw_net.pt")
net2 = Net().cuda(); net2.load_state_dict(torch.load("/tmp/lw_net.pt"))
x = torch.randn(4, 8, device="cuda")
print("reloaded weights give same output:", torch.equal(net(x), net2(x)))

# ---- Dataset / DataLoader
class Toy(Dataset):
    def __len__(self): return 1000
    def __getitem__(self, i): 
        g = torch.Generator().manual_seed(i)
        v = torch.randn(8, generator=g); return v, (v[0] + v[1] > 0).float()
dl = DataLoader(Toy(), batch_size=64, shuffle=True, num_workers=2, pin_memory=True)
opt = torch.optim.Adam(net.parameters(), 1e-2)
for epoch in range(3):
    tot = 0.0
    for xb, yb in dl:
        xb, yb = xb.cuda(non_blocking=True), yb.cuda(non_blocking=True)
        loss = nn.functional.binary_cross_entropy(net(xb).squeeze(1), yb)
        opt.zero_grad(); loss.backward(); opt.step(); tot += loss.item() * len(xb)
    print(f"epoch {epoch}: loss {tot / 1000:.3f}")
net.eval()   # inference: dropout/batchnorm switch behaviour; a separate switch from no_grad
print("training flag after eval():", net.training)

$ python3 pt09a_basics.py   # on lab pod
d/dw (3w-1)^2 at w=2: 30.0 (analytic 6*(3w-1) = 30)
requires_grad inside no_grad: False
state_dict keys: ['fc1.weight', 'fc1.bias', 'fc2.weight', 'fc2.bias']
reloaded weights give same output: True
epoch 0: loss 0.615
epoch 1: loss 0.406
epoch 2: loss 0.228
training flag after eval(): False

• torch.no_grad() (or inference_mode()) stops graph recording and saves memory; model.eval() switches dropout/batch-norm behavior. They are independent switches and serving needs both.
• state_dict is an ordered map of names to tensors: weights only, so loading needs the class definition. Save that, not the pickled model object.
• In training loops the order is: zero_grad → forward → backward → step. DataLoader workers overlap CPU data prep with GPU compute; pin_memory + non_blocking=True overlaps the host-to-device copy.

torch.compile, CUDA graphs, profiler

A 16-layer MLP at batch 8 is deliberately tiny: each kernel runs for a few microseconds, so the GPU is idle waiting for the CPU to launch the next one. This is the launch-bound regime that low-latency decode lives in.

Copyimport time, warnings, torch, torch.nn as nn
warnings.filterwarnings('ignore')
from torch.profiler import profile, ProfilerActivity

dev = "cuda"
mlp = nn.Sequential(*[m for _ in range(8) for m in (nn.Linear(512, 512), nn.GELU())]).to(dev).eval()
x = torch.randn(8, 512, device=dev)

def bench(f, reps=200):
    for _ in range(10): f()
    torch.cuda.synchronize(); t0 = time.perf_counter()
    for _ in range(reps): f()
    torch.cuda.synchronize(); return (time.perf_counter() - t0) / reps * 1e6

with torch.no_grad():
    eager_us = bench(lambda: mlp(x))
    t0 = time.perf_counter(); cmlp = torch.compile(mlp); cmlp(x); torch.cuda.synchronize()
    print(f"first torch.compile call (trace+codegen): {time.perf_counter() - t0:.1f} s")
    comp_us = bench(lambda: cmlp(x))

    # CUDA graph: record the whole launch sequence once, replay with one launch
    static = x.clone()
    s = torch.cuda.Stream(); s.wait_stream(torch.cuda.current_stream())
    with torch.cuda.stream(s):
        for _ in range(3): mlp(static)                  # warm-up on a side stream before capture
    torch.cuda.current_stream().wait_stream(s)
    g = torch.cuda.CUDAGraph()
    with torch.cuda.graph(g): out = mlp(static)
    def replay(): static.copy_(x); g.replay()
    graph_us = bench(replay)

    print(f"eager {eager_us:.0f} us | torch.compile {comp_us:.0f} us | CUDA graph replay {graph_us:.0f} us   (batch 8, 16 tiny layers)")
    print("graph output equals eager:", torch.allclose(out, mlp(x), atol=1e-5))

    with profile(activities=[ProfilerActivity.CPU, ProfilerActivity.CUDA]) as prof:
        for _ in range(20): mlp(x)
        torch.cuda.synchronize()
top = sorted(prof.key_averages(), key=lambda e: e.self_device_time_total, reverse=True)[:3]
for e in top: print(f"{e.key[:46]:<48}{e.count:>4} calls {e.self_device_time_total / e.count:>7.1f} us GPU each")

$ python3 pt09b_compile_profile.py   # on lab pod
first torch.compile call (trace+codegen): 3.0 s
eager 252 us | torch.compile 207 us | CUDA graph replay 45 us   (batch 8, 16 tiny layers)
graph output equals eager: True
aten::addmm                                      160 calls     3.9 us GPU each
void gemmSN_TN_kernel<float, 128, 16, 2, 4, 8,   160 calls     3.9 us GPU each
aten::gelu                                       160 calls     1.1 us GPU each

JAX in one screen

JAX is "numpy + autodiff + XLA" built on pure functions: no hidden state, no in-place mutation. That purity is what lets jit, grad and vmap compose.

Copyimport warnings; warnings.filterwarnings("ignore")
import jax, jax.numpy as jnp

@jax.jit
def f(x):
    print("tracing f for shape", x.shape)       # side effect: runs at trace time only (purity!)
    return jnp.tanh(x) * 2

f(jnp.ones(3)); f(jnp.zeros(3)); f(jnp.ones(5))   # 3rd call has a new shape -> retrace

loss = lambda w, x: ((w * x - 1.0) ** 2).sum()
print("grad:", jax.grad(loss)(2.0, jnp.array([1., 2.])))      # d/dw sum((wx-1)^2) = sum(2x(wx-1)) = 2*1*1 + 2*2*3
xs = jnp.arange(6.).reshape(3, 2)
print("vmap over batch:", jax.vmap(lambda row: row @ row)(xs))  # per-example fn, batched automatically

key = jax.random.key(0)                             # no global RNG state: keys are passed explicitly
k1, k2 = jax.random.split(key)
print("same key -> same draw:", bool(jax.random.normal(k1) == jax.random.normal(k1)), "| split key differs:", bool(jax.random.normal(k1) != jax.random.normal(k2)))
x = jnp.arange(3.)
y = x.at[0].set(9.)                                # arrays are immutable: .at[].set returns a new array
print(x, y)

$ python3 jx09_jax.py
tracing f for shape (3,)
tracing f for shape (5,)
grad: 14.0
vmap over batch: [ 1. 13. 41.]
same key -> same draw: True | split key differs: True
[0. 1. 2.] [9. 1. 2.]

The print inside the jitted function fired twice for three calls: shape (3,) traced once and reused, shape (5,) retraced; same rule as tf.function. vmap turned a per-row function into a batched one without rewriting it. Random numbers use explicit keys instead of global state, and arrays are immutable (x.at[0].set(9.) returns a new array).

10. Common libraries map

Copyimport warnings; warnings.filterwarnings("ignore")
import importlib, torch
from transformers import LlamaConfig, AutoModelForCausalLM

cfg = LlamaConfig(hidden_size=2048, intermediate_size=8192, num_hidden_layers=16, num_attention_heads=32,
                  num_key_value_heads=8, vocab_size=128256)
with torch.device("meta"):                      # no memory allocated: shapes only
    m = AutoModelForCausalLM.from_config(cfg)
n = sum(p.numel() for p in m.parameters())
kv = 2 * cfg.num_hidden_layers * cfg.num_key_value_heads * (cfg.hidden_size // cfg.num_attention_heads) * 2
print(f"params: {n / 1e9:.2f} B -> {n * 2 / 2**30:.1f} GiB in bf16")
print(f"KV cache: {kv / 1024:.0f} KiB/token (GQA, bf16) -> {kv * 4096 / 2**20:.0f} MiB per 4k-token sequence")

# 'triton' is two unrelated things
for name in ("triton", "tritonserver", "tritonclient"):
    try:
        mod = importlib.import_module(name)
        print(f"import {name:<13} OK   {getattr(mod, '__doc__', '') and mod.__doc__.strip().splitlines()[0][:60]!r}" if name != "triton" else f"import triton        OK   (GPU kernel language, v{mod.__version__}; has triton.jit)")
    except ImportError as e:
        print(f"import {name:<13} FAILS ({e.name} not installed)")

$ python3 lib10e_hf_triton_names.py   # on lab pod
params: 1.50 B -> 2.8 GiB in bf16
KV cache: 32 KiB/token (GQA, bf16) -> 128 MiB per 4k-token sequence
import triton        OK   (GPU kernel language, v3.6.0; has triton.jit)
import tritonserver  OK   'Triton Inference Server In-Process Python API'
import tritonclient  FAILS (tritonclient not installed)

That script also shows the sizing arithmetic an inference engineer does constantly. A 1.5 B-parameter model is ~2.8 GiB in bf16. With grouped-query attention (8 KV heads, head dim 64, 16 layers) the KV cache costs 2 (K and V) x 16 x 8 x 64 x 2 bytes = 32 KiB per token, i.e. 128 MiB for one 4k-token sequence, so 16 GiB of cache holds ~128 such sequences. Concurrency is capped by KV memory, not weights.

A tour in code

Copyimport warnings; warnings.filterwarnings("ignore")
import numpy as np, pandas as pd, polars as pl, scipy, scipy.stats as st
import sklearn, xgboost as xgb, prometheus_client as prom
from sklearn.ensemble import GradientBoostingRegressor
from sklearn.model_selection import train_test_split

rng = np.random.default_rng(0)
lat = rng.lognormal(3.0, 0.5, 20_000)
sigma, _, scale = st.lognorm.fit(lat, floc=0)
print(f"scipy lognorm fit (true mu=3.0, sigma=0.5): mu={np.log(scale):.2f} sigma={sigma:.2f}")

df = pd.DataFrame({"m": rng.choice(list("ab"), 20_000), "lat": lat})
print("pandas :", df.groupby("m").lat.quantile(.99).round(1).to_dict())
print("polars :", pl.from_pandas(df).group_by("m").agg(pl.col("lat").quantile(.99)).sort("m").with_columns(pl.col("lat").round(1)).rows())

X = rng.normal(size=(5000, 6)); y = X[:, 0] * 3 + np.sin(X[:, 1]) + rng.normal(0, .1, 5000)
Xtr, Xte, ytr, yte = train_test_split(X, y, random_state=0)
sk = GradientBoostingRegressor(n_estimators=100).fit(Xtr, ytr)
xg = xgb.XGBRegressor(n_estimators=100, tree_method="hist").fit(Xtr, ytr)
print(f"sklearn GBR R2={sk.score(Xte, yte):.3f}   xgboost R2={xg.score(Xte, yte):.3f}")

reg = prom.CollectorRegistry()
h = prom.Histogram("request_latency_seconds", "latency", ["model"], buckets=(.01, .05, .1, .5), registry=reg)
for v in (.004, .03, .2): h.labels("ranker").observe(v)
print("\n".join(l for l in prom.generate_latest(reg).decode().splitlines() if "bucket" in l or "_count" in l))

$ python3 lib10a_tour.py
scipy lognorm fit (true mu=3.0, sigma=0.5): mu=3.00 sigma=0.50
pandas : {'a': 64.0, 'b': 63.5}
polars : [('a', 64.0), ('b', 63.4)]
sklearn GBR R2=0.998   xgboost R2=0.996
request_latency_seconds_bucket{le="0.01",model="ranker"} 1.0
request_latency_seconds_bucket{le="0.05",model="ranker"} 2.0
request_latency_seconds_bucket{le="0.1",model="ranker"} 2.0
request_latency_seconds_bucket{le="0.5",model="ranker"} 3.0
request_latency_seconds_bucket{le="+Inf",model="ranker"} 3.0
request_latency_seconds_count{model="ranker"} 3.0

• The lognormal fit recovered the true parameters (mu=3.0, sigma=0.5). Heavy-tailed latency is usually lognormal-ish, which is why means mislead.
• pandas and polars disagree on p99 for group b (63.5 vs 63.4) because their default quantile methods differ (linear interpolation vs nearest). Same data, different "p99". Pin the method when comparing tools.
• Histogram buckets (0.01, 0.05, 0.1, 0.5) are cumulative: le="0.05" includes everything ≤ 0.05 s. A Prometheus percentile estimate can only be as precise as the bucket edges around it.

Copyimport warnings; warnings.filterwarnings('ignore')
from fastapi import FastAPI
from fastapi.testclient import TestClient
from fastapi.responses import PlainTextResponse
from pydantic import BaseModel
from prometheus_client import Counter, CollectorRegistry, generate_latest

reg = CollectorRegistry()
REQS = Counter("infer_requests_total", "requests", ["status"], registry=reg)
app = FastAPI()

class Req(BaseModel):
    prompt: str
    max_tokens: int = 16

@app.post("/v1/generate")
async def generate(r: Req):                # async def: runs on the event loop, so never block in here
    REQS.labels("ok").inc()
    return {"echo": r.prompt[:10], "n": r.max_tokens}

@app.get("/metrics", response_class=PlainTextResponse)
def metrics(): return generate_latest(reg).decode()

c = TestClient(app)
print(c.post("/v1/generate", json={"prompt": "hello world, long text"}).json())
bad = c.post("/v1/generate", json={"prompt": 5, "max_tokens": "x"})
print(bad.status_code, [e["loc"] for e in bad.json()["detail"]])
print([l for l in c.get("/metrics").text.splitlines() if l.startswith("infer_requests_total{")])

$ python3 lib10b_fastapi.py
{'echo': 'hello worl', 'n': 16}
422 [['body', 'prompt'], ['body', 'max_tokens']]
['infer_requests_total{status="ok"} 1.0']

Pydantic rejected the bad body with a 422 and told us which fields failed; the counter incremented only for the valid request. Not run: that a blocking call inside async def stalls other requests is the standard asyncio behavior shown in section 5 (time.sleep vs asyncio.sleep), but it was not demonstrated against FastAPI itself. Use plain def handlers (FastAPI runs them in a thread pool) or await truly async clients.

protobuf + gRPC: define the schema once, generate stubs, and the bytes on the wire are field-number tagged binary.

Copy// infer.proto
syntax = "proto3";
message Request  { string model = 1; repeated float features = 2; }
message Response { repeated float scores = 1; }
service Inference { rpc Predict(Request) returns (Response); }

Copypython -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. infer.proto

$ python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. infer.proto && ls
__pycache__
infer_pb2_grpc.py
infer_pb2.py
infer.proto

Copyimport json, infer_pb2 as pb

r = pb.Request(model="ranker", features=[0.5, 1.5])
wire = r.SerializeToString()
print("wire bytes:", len(wire), wire.hex())            # tag 0x0a=field1/len-delimited, 0x12=field2 packed floats
print("roundtrip :", list(pb.Request.FromString(wire).features))
print("same payload as JSON:", len(json.dumps({"model": "ranker", "features": [0.5, 1.5]}, separators=(",", ":"))), "bytes")
print("service stub class generated:", hasattr(__import__("infer_pb2_grpc"), "InferenceStub"))

$ PYTHONPATH=. python lib10d_proto.py
wire bytes: 18 0a0672616e6b657212080000003f0000c03f
roundtrip : [0.5, 1.5]
same payload as JSON: 39 bytes
service stub class generated: True

18 bytes vs 39 for the JSON of the same payload: no field names on the wire (just tag numbers 0x0a, 0x12) and floats packed as 4 raw bytes each. Field numbers are the contract; never reuse or renumber them.

11. Interview prep: coding questions in Python and C++

How these usually go: you get a problem, you are expected to ask about edge cases and constraints, state the approach and complexity before coding, write something correct, then test it by hand. For infra roles the follow-ups are about concurrency, memory and what happens at scale. Every solution below was compiled/run (C++ with -Wall -Wextra; the concurrent ones also under ThreadSanitizer, the rest under ASan+UBSan, all clean) and its real output is shown. Try each before opening it.

1. LRU cache (O(1) get and put)

Idea: a hash map for lookup plus an ordering structure you can reorder in O(1). Python: OrderedDict (a hash map plus linked list). C++: std::list + unordered_map of iterators; list::splice relinks a node without invalidating iterators. Complexity: O(1) time per op, O(capacity) space. Follow-ups: thread safety (a mutex around both structures, or sharding by key hash because even get mutates order), TTL, byte-size-based eviction (KV/embedding caches).

Copyfrom collections import OrderedDict

class LRUCache:
    def __init__(self, capacity: int):
        self.cap, self.d = capacity, OrderedDict()
    def get(self, k):
        if k not in self.d: return -1
        self.d.move_to_end(k)                 # most recently used goes to the back
        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)   # evict the front (least recent)

c = LRUCache(2)
c.put("a", 1); c.put("b", 2); print(c.get("a"))   # a becomes most recent
c.put("c", 3)                                      # evicts b
print(c.get("b"), c.get("a"), c.get("c"))

$ python3 iv_lru.py
1
-1 1 3

Copy#include <iostream>
#include <list>
#include <unordered_map>

class LRUCache {
    size_t cap_;
    std::list<std::pair<int, int>> items_;  // front = most recent
    std::unordered_map<int, std::list<std::pair<int, int>>::iterator> idx_;
public:
    explicit LRUCache(size_t cap) : cap_(cap) {}
    int get(int k) {
        auto it = idx_.find(k);
        if (it == idx_.end()) return -1;
        items_.splice(items_.begin(), items_, it->second);  // O(1) move, iterators stay valid
        return it->second->second;
    }
    void put(int k, int v) {
        if (auto it = idx_.find(k); it != idx_.end()) {
            it->second->second = v;
            items_.splice(items_.begin(), items_, it->second);
            return;
        }
        items_.emplace_front(k, v);
        idx_[k] = items_.begin();
        if (items_.size() > cap_) { idx_.erase(items_.back().first); items_.pop_back(); }
    }
};

int main() {
    LRUCache c(2);
    c.put(1, 1); c.put(2, 2); std::cout << c.get(1) << "\n";
    c.put(3, 3);
    std::cout << c.get(2) << " " << c.get(1) << " " << c.get(3) << "\n";
}

$ g++ -std=c++20 -O2 -Wall -Wextra iv_lru.cpp -o iv_lru
1
-1 1 3

2. Top-k and streaming percentiles

Top-k largest of a stream: keep a min-heap of size k; a new item replaces the root only if larger. O(n log k) time, O(k) space (better than sorting all, O(n log n), when k is small). Streaming percentile without storing every sample: a log-bucketed histogram (the idea behind DDSketch/HDR histograms). Bucket i covers (gamma^(i-1), gamma^i]; returning the bucket's midpoint gives bounded relative error (1% here). O(1) per add, memory depends on the dynamic range, not the count. Mergeable by adding bucket counts, which is why distributed systems use sketches instead of averaging per-host percentiles. Follow-up: exact median of a stream = two heaps (max-heap lower half, min-heap upper half), O(log n) per add.

Copyimport heapq, math, random
import numpy as np

class TopK:                                   # k largest of a stream: O(log k) per item, O(k) memory
    def __init__(self, k): self.k, self.h = k, []
    def add(self, x):
        if len(self.h) < self.k: heapq.heappush(self.h, x)
        elif x > self.h[0]: heapq.heapreplace(self.h, x)   # beats the smallest of the current top-k
    def result(self): return sorted(self.h, reverse=True)

class LogHistogram:                           # streaming quantiles with bounded RELATIVE error (DDSketch idea)
    def __init__(self, rel_err=0.01):
        self.gamma = (1 + rel_err) / (1 - rel_err); self.lg = math.log(self.gamma)
        self.counts, self.n = {}, 0
    def add(self, x):                         # x > 0
        i = math.ceil(math.log(x) / self.lg)
        self.counts[i] = self.counts.get(i, 0) + 1; self.n += 1
    def quantile(self, q):
        target, seen = q * (self.n - 1), 0
        for i in sorted(self.counts):
            seen += self.counts[i]
            if seen > target: return 2 * self.gamma ** i / (self.gamma + 1)

random.seed(0)
xs = [random.lognormvariate(3, 0.8) for _ in range(200_000)]
t, h = TopK(3), LogHistogram(0.01)
for x in xs: t.add(x); h.add(x)
print("top3 matches sorted():", t.result() == sorted(xs, reverse=True)[:3])
for q in (.5, .95, .99):
    exact = float(np.quantile(xs, q)); est = h.quantile(q)
    print(f"p{int(q*100)}: exact {exact:7.2f}  sketch {est:7.2f}  rel.err {abs(est-exact)/exact:.4f}  (buckets used: {len(h.counts)})")

$ python3 iv_topk.py
top3 matches sorted(): True
p50: exact   20.07  sketch   19.89  rel.err 0.0089  (buckets used: 321)
p95: exact   74.98  sketch   74.45  rel.err 0.0072  (buckets used: 321)
p99: exact  128.33  sketch  127.75  rel.err 0.0045  (buckets used: 321)

Copy#include <algorithm>
#include <cmath>
#include <functional>
#include <iostream>
#include <map>
#include <queue>
#include <random>
#include <vector>

class TopK {  // k largest of a stream
    size_t k_;
    std::priority_queue<double, std::vector<double>, std::greater<>> h_;  // min-heap
public:
    explicit TopK(size_t k) : k_(k) {}
    void add(double x) {
        if (h_.size() < k_) h_.push(x);
        else if (x > h_.top()) { h_.pop(); h_.push(x); }
    }
    std::vector<double> result() { std::vector<double> r; auto h = h_; while (!h.empty()) { r.push_back(h.top()); h.pop(); } std::reverse(r.begin(), r.end()); return r; }
};

class LogHistogram {  // bounded relative error quantiles
    double gamma_, lg_; std::map<int, long> counts_; long n_ = 0;
public:
    explicit LogHistogram(double rel = 0.01) : gamma_((1 + rel) / (1 - rel)), lg_(std::log(gamma_)) {}
    void add(double x) { ++counts_[(int)std::ceil(std::log(x) / lg_)]; ++n_; }
    double quantile(double q) const {
        double target = q * (n_ - 1); long seen = 0;
        for (auto& [i, c] : counts_) { seen += c; if (seen > target) return 2 * std::pow(gamma_, i) / (gamma_ + 1); }
        return NAN;
    }
    size_t buckets() const { return counts_.size(); }
};

int main() {
    std::mt19937 rng(0); std::lognormal_distribution<double> d(3, 0.8);
    std::vector<double> xs(200000); for (auto& x : xs) x = d(rng);
    TopK t(3); LogHistogram h(0.01);
    for (double x : xs) { t.add(x); h.add(x); }
    auto sorted = xs; std::sort(sorted.begin(), sorted.end());
    auto top = t.result();
    std::cout << "top3 matches sort: " << (top[0] == sorted[sorted.size()-1] && top[2] == sorted[sorted.size()-3]) << "\n";
    for (double q : {0.5, 0.95, 0.99}) {
        double exact = sorted[(size_t)(q * (sorted.size() - 1))], est = h.quantile(q);
        std::printf("p%d: exact %7.2f  sketch %7.2f  rel.err %.4f  (buckets: %zu)\n", (int)(q * 100), exact, est, std::fabs(est - exact) / exact, h.buckets());
    }
}

$ g++ -std=c++20 -O2 -Wall -Wextra iv_topk.cpp -o iv_topk
top3 matches sort: 1
p50: exact   20.07  sketch   19.89  rel.err 0.0093  (buckets: 321)
p95: exact   74.97  sketch   74.45  rel.err 0.0069  (buckets: 321)
p99: exact  129.19  sketch  130.34  rel.err 0.0089  (buckets: 321)

3. Thread-safe bounded queue

Idea: one mutex, two condition variables (not_full, not_empty), waits with predicates, and a close() that wakes everyone so consumers can drain and exit (the usual forgotten requirement; without it shutdown deadlocks). Returns None/nullopt when closed and empty. Complexity: O(1) per op; blocked threads cost nothing. Follow-ups: multiple producers/consumers (this handles them), timeouts (wait_for), lock-free alternatives (section 3's SPSC ring if there is exactly one of each), backpressure vs dropping, priority.

Copyimport threading
from collections import deque

class BoundedQueue:
    def __init__(self, cap):
        self.q, self.cap, self.closed = deque(), cap, False
        self.lock = threading.Lock()
        self.not_full, self.not_empty = threading.Condition(self.lock), threading.Condition(self.lock)  # one lock, two conditions
    def put(self, x):
        with self.lock:
            self.not_full.wait_for(lambda: len(self.q) < self.cap or self.closed)
            if self.closed: raise RuntimeError("closed")
            self.q.append(x); self.not_empty.notify()
    def get(self):                              # returns None once closed AND drained
        with self.lock:
            self.not_empty.wait_for(lambda: self.q or self.closed)
            if not self.q: return None
            x = self.q.popleft(); self.not_full.notify(); return x
    def close(self):
        with self.lock:
            self.closed = True; self.not_empty.notify_all(); self.not_full.notify_all()

q, total, lock = BoundedQueue(4), [0], threading.Lock()
def consume():
    while (x := q.get()) is not None:
        with lock: total[0] += x
ws = [threading.Thread(target=consume) for _ in range(3)]
for w in ws: w.start()
for i in range(1, 1001): q.put(i)               # blocks whenever 4 items are waiting
q.close()
for w in ws: w.join()
print("sum =", total[0], "(expected 500500)")

$ python3 iv_queue.py
sum = 500500 (expected 500500)

Copy#include <atomic>
#include <condition_variable>
#include <deque>
#include <iostream>
#include <mutex>
#include <optional>
#include <thread>
#include <vector>

template <typename T>
class BoundedQueue {
    std::deque<T> q_; size_t cap_; bool closed_ = false;
    std::mutex m_; std::condition_variable not_full_, not_empty_;
public:
    explicit BoundedQueue(size_t cap) : cap_(cap) {}
    bool put(T x) {                                  // false if closed
        std::unique_lock lk(m_);
        not_full_.wait(lk, [&] { return q_.size() < cap_ || closed_; });
        if (closed_) return false;
        q_.push_back(std::move(x)); not_empty_.notify_one(); return true;
    }
    std::optional<T> get() {                         // nullopt once closed and drained
        std::unique_lock lk(m_);
        not_empty_.wait(lk, [&] { return !q_.empty() || closed_; });
        if (q_.empty()) return std::nullopt;
        T x = std::move(q_.front()); q_.pop_front(); not_full_.notify_one(); return x;
    }
    void close() { { std::lock_guard g(m_); closed_ = true; } not_empty_.notify_all(); not_full_.notify_all(); }
};

int main() {
    BoundedQueue<int> q(4);
    std::atomic<long> total{0};
    std::vector<std::thread> ws;
    for (int i = 0; i < 3; ++i) ws.emplace_back([&] { while (auto x = q.get()) total += *x; });
    for (int i = 1; i <= 1000; ++i) q.put(i);
    q.close();
    for (auto& w : ws) w.join();
    std::cout << "sum = " << total << " (expected 500500)\n";
}

$ g++ -std=c++20 -O2 -Wall -Wextra -pthread iv_queue.cpp -o iv_queue
sum = 500500 (expected 500500)

4. Token-bucket rate limiter

Idea: a bucket holds up to burst tokens and refills at rate per second. Do not run a timer: on each request compute tokens lazily from the elapsed time since the last call, clamp to burst, then spend. O(1) time and space per limiter. Inject the clock so it is testable (the output uses a fake clock). Follow-ups: per-key limiters (hash map plus eviction of idle keys), thread safety (mutex, or an atomic holding "next allowed time" in the GCRA variant), distributed limiting (Redis Lua script, or local buckets with a shared budget), token-based limits for LLMs (cost = tokens).

Copyimport time

class TokenBucket:
    def __init__(self, rate, burst, clock=time.monotonic):
        self.rate, self.burst, self.clock = rate, burst, clock
        self.tokens, self.last = float(burst), clock()
    def allow(self, cost=1):
        now = self.clock()
        self.tokens = min(self.burst, self.tokens + (now - self.last) * self.rate)   # lazy refill: no timer thread
        self.last = now
        if self.tokens >= cost:
            self.tokens -= cost; return True
        return False

t = [0.0]                                       # fake clock makes the test deterministic
b = TokenBucket(rate=2, burst=3, clock=lambda: t[0])
print("burst  :", [b.allow() for _ in range(5)])
t[0] = 1.0; print("t=1.0s :", [b.allow() for _ in range(3)], "(2 tokens/s refilled)")
t[0] = 100.0; print("t=100s :", [b.allow() for _ in range(5)], "(capped at burst=3)")

$ python3 iv_bucket.py
burst  : [True, True, True, False, False]
t=1.0s : [True, True, False] (2 tokens/s refilled)
t=100s : [True, True, True, False, False] (capped at burst=3)

Copy#include <algorithm>
#include <functional>
#include <iostream>

class TokenBucket {
    double rate_, burst_, tokens_, last_; std::function<double()> now_;
public:
    TokenBucket(double rate, double burst, std::function<double()> now) : rate_(rate), burst_(burst), tokens_(burst), last_(now()), now_(std::move(now)) {}
    bool allow(double cost = 1) {
        double t = now_();
        tokens_ = std::min(burst_, tokens_ + (t - last_) * rate_);
        last_ = t;
        if (tokens_ < cost) return false;
        tokens_ -= cost; return true;
    }
};

int main() {
    double t = 0;
    TokenBucket b(2, 3, [&] { return t; });
    auto run = [&](int n) { for (int i = 0; i < n; ++i) std::cout << (b.allow() ? "true " : "false "); std::cout << "\n"; };
    std::cout << "burst  : "; run(5);
    t = 1.0;   std::cout << "t=1.0s : "; run(3);
    t = 100.0; std::cout << "t=100s : "; run(5);
}

$ g++ -std=c++20 -O2 -Wall -Wextra iv_bucket.cpp -o iv_bucket
burst  : true true true false false 
t=1.0s : true true false 
t=100s : true true true false false 

5. Merge intervals

Idea: sort by start; sweep, extending the last merged interval when the next one starts at or before its end. Complexity: O(n log n) time (the sort), O(n) output space. Edge cases to say out loud: empty input, touching intervals ([1,4],[4,5] merge if endpoints are inclusive), one interval containing another, unsorted input. If the input arrives pre-sorted or from k sorted streams, it is O(n) or O(n log k) with a heap.

Copydef merge(iv):
    out = []
    for s, e in sorted(iv):                       # O(n log n) sort dominates
        if out and s <= out[-1][1]: out[-1][1] = max(out[-1][1], e)
        else: out.append([s, e])
    return out

print(merge([(8, 10), (1, 3), (2, 6), (15, 18), (17, 20)]))
print(merge([(1, 4), (4, 5)]), merge([]), merge([(1, 10), (2, 3)]))

$ python3 iv_intervals.py
[[1, 6], [8, 10], [15, 20]]
[[1, 5]] [] [[1, 10]]

Copy#include <algorithm>
#include <iostream>
#include <vector>

using Iv = std::pair<int, int>;
std::vector<Iv> merge(std::vector<Iv> iv) {      // by value: we sort our own copy
    std::sort(iv.begin(), iv.end());
    std::vector<Iv> out;
    for (auto [s, e] : iv) {
        if (!out.empty() && s <= out.back().second) out.back().second = std::max(out.back().second, e);
        else out.push_back({s, e});
    }
    return out;
}
void show(const std::vector<Iv>& v) { for (auto [s, e] : v) std::cout << "[" << s << "," << e << "] "; std::cout << "\n"; }

int main() {
    show(merge({{8, 10}, {1, 3}, {2, 6}, {15, 18}, {17, 20}}));
    show(merge({{1, 4}, {4, 5}})); show(merge({})); show(merge({{1, 10}, {2, 3}}));
}

$ g++ -std=c++20 -O2 -Wall -Wextra iv_intervals.cpp -o iv_intervals
[1,6] [8,10] [15,20] 
[1,5] 

[1,10] 

6. Numerically stable softmax

Idea: softmax is invariant to subtracting a constant, so subtract the max first; then every exponent is ≤ 0 and cannot overflow. The naive version returned NaN for logits around 1000 (exp overflows to inf, inf/inf = NaN). With axis handling it is the standard numpy one-liner. Complexity: O(n) time, three passes over the data (max, exp+sum, normalize). Follow-up that matters here: online softmax computes the max and the sum in a single pass by rescaling the running sum when a new max appears (s = s*exp(m_old - m_new) + exp(x - m_new)), which is the trick that makes FlashAttention possible, avoiding materializing the full attention matrix. Also mention: in fp16/bf16, accumulate in fp32; log_softmax via log-sum-exp avoids taking the log of a tiny probability.

Copyimport numpy as np
import warnings; warnings.simplefilter("ignore")

def softmax_naive(x): e = np.exp(x); return e / e.sum()
def softmax(x, axis=-1):
    z = x - x.max(axis=axis, keepdims=True)    # shift so the max is 0: exp <= 1, no overflow; result unchanged
    e = np.exp(z); return e / e.sum(axis=axis, keepdims=True)

x = np.array([1000., 1001., 1002.])
print("naive :", softmax_naive(x))
print("stable:", softmax(x).round(4), softmax(x).sum())
print("2-D rows sum to 1:", softmax(np.random.default_rng(0).normal(size=(3, 5))).sum(axis=1))
print("matches shift-free on small input:", np.allclose(softmax(np.array([1., 2., 3.])), softmax_naive(np.array([1., 2., 3.]))))

$ python3 iv_softmax.py
naive : [nan nan nan]
stable: [0.09   0.2447 0.6652] 0.9999999999999999
2-D rows sum to 1: [1. 1. 1.]
matches shift-free on small input: True

Copy#include <algorithm>
#include <cmath>
#include <iostream>
#include <vector>

std::vector<float> softmax_naive(const std::vector<float>& x) {
    std::vector<float> e(x.size()); float s = 0;
    for (size_t i = 0; i < x.size(); ++i) { e[i] = std::exp(x[i]); s += e[i]; }
    for (auto& v : e) v /= s; return e;
}
std::vector<float> softmax(const std::vector<float>& x) {
    float m = *std::max_element(x.begin(), x.end());   // pass 1: max   (online-softmax fuses passes 1+2)
    std::vector<float> e(x.size()); float s = 0;
    for (size_t i = 0; i < x.size(); ++i) { e[i] = std::exp(x[i] - m); s += e[i]; }   // pass 2: exp + sum
    for (auto& v : e) v /= s; return e;                 // pass 3: normalize
}
void show(const char* n, const std::vector<float>& v) { std::cout << n; for (float f : v) std::cout << " " << f; std::cout << "\n"; }

int main() {
    std::vector<float> x{1000, 1001, 1002};
    show("naive :", softmax_naive(x));
    show("stable:", softmax(x));
}

$ g++ -std=c++20 -O2 -Wall -Wextra iv_softmax.cpp -o iv_softmax
naive : nan nan nan
stable: 0.0900306 0.244728 0.665241

7. Dynamic batcher with timeout

Idea: the core of every inference server. Requests arrive one at a time; the GPU is far more efficient on a batch. A worker thread blocks for the first request (that starts a timer), then collects more until the batch is full or the timeout expires, then calls the model once and fulfils each caller's future. The two knobs trade throughput against latency: max_batch bounds memory and tail latency per call; max_wait bounds the delay added to a lone request (the output shows a lone request waiting the full 50 ms). Run the model outside the lock, and propagate exceptions to every future in the batch so callers never hang. Complexity: O(batch) per flush. Follow-ups: bounded queue and load shedding, per-request deadlines, padding/bucketing variable-length inputs, continuous batching for LLMs (join and leave the batch at every decode step instead of waiting for whole batches), and how Triton's dynamic_batching { max_queue_delay_microseconds } maps onto max_wait.

Copyimport queue, threading, time
from concurrent.futures import Future

class DynamicBatcher:
    """Flush when max_batch items are waiting OR the oldest has waited max_wait seconds."""
    def __init__(self, fn, max_batch=4, max_wait=0.05):
        self.fn, self.max_batch, self.max_wait = fn, max_batch, max_wait
        self.q = queue.Queue()
        threading.Thread(target=self._loop, daemon=True).start()
    def submit(self, x) -> Future:
        f = Future(); self.q.put((x, f)); return f
    def _loop(self):
        while True:
            batch = [self.q.get()]                          # block until the first item: starts the timer
            deadline = time.monotonic() + self.max_wait
            while len(batch) < self.max_batch:
                try: batch.append(self.q.get(timeout=max(0, deadline - time.monotonic())))
                except queue.Empty: break
            try:
                outs = self.fn([x for x, _ in batch])       # one model call for the whole batch
                for (_, f), o in zip(batch, outs): f.set_result(o)
            except Exception as e:
                for _, f in batch: f.set_exception(e)       # one bad batch must not hang callers

sizes = []
def model(xs): sizes.append(len(xs)); return [x * 2 for x in xs]

b = DynamicBatcher(model, max_batch=4, max_wait=0.05)
t0 = time.monotonic()
futs = [b.submit(i) for i in range(10)]
print("results:", [f.result() for f in futs])
print("batch sizes:", sizes, f"| total {time.monotonic() - t0:.2f}s (last partial batch waited for the timeout)")
t1 = time.monotonic(); print("lone request ->", b.submit(21).result(), f"after {time.monotonic() - t1:.2f}s (the max_wait latency cost)")

$ python3 iv_batcher.py
results: [0, 2, 4, 6, 8, 10, 12, 14, 16, 18]
batch sizes: [4, 4, 2] | total 0.06s (last partial batch waited for the timeout)
lone request -> 42 after 0.05s (the max_wait latency cost)

Copy#include <chrono>
#include <condition_variable>
#include <deque>
#include <functional>
#include <future>
#include <iostream>
#include <mutex>
#include <thread>
#include <vector>

using namespace std::chrono;

class DynamicBatcher {
    using Item = std::pair<int, std::promise<int>>;
    std::function<std::vector<int>(const std::vector<int>&)> fn_;
    size_t max_batch_; milliseconds max_wait_;
    std::mutex m_; std::condition_variable cv_; std::deque<Item> q_; bool stop_ = false;
    std::thread worker_;
    void loop() {
        for (;;) {
            std::vector<Item> batch;
            {
                std::unique_lock lk(m_);
                cv_.wait(lk, [&] { return !q_.empty() || stop_; });
                if (q_.empty()) return;                       // stopping and drained
                auto deadline = steady_clock::now() + max_wait_;  // timer starts at the first item
                cv_.wait_until(lk, deadline, [&] { return q_.size() >= max_batch_ || stop_; });
                while (!q_.empty() && batch.size() < max_batch_) { batch.push_back(std::move(q_.front())); q_.pop_front(); }
            }
            std::vector<int> in; for (auto& [x, p] : batch) in.push_back(x);
            auto out = fn_(in);                               // model call runs outside the lock
            for (size_t i = 0; i < batch.size(); ++i) batch[i].second.set_value(out[i]);
        }
    }
public:
    DynamicBatcher(decltype(fn_) fn, size_t max_batch, milliseconds max_wait)
        : fn_(std::move(fn)), max_batch_(max_batch), max_wait_(max_wait), worker_([this] { loop(); }) {}
    ~DynamicBatcher() { { std::lock_guard g(m_); stop_ = true; } cv_.notify_all(); worker_.join(); }
    std::future<int> submit(int x) {
        std::promise<int> p; auto f = p.get_future();
        { std::lock_guard g(m_); q_.emplace_back(x, std::move(p)); }
        cv_.notify_one(); return f;
    }
};

int main() {
    std::vector<size_t> sizes;
    DynamicBatcher b([&](const std::vector<int>& xs) { sizes.push_back(xs.size()); std::vector<int> o; for (int x : xs) o.push_back(x * 2); return o; }, 4, 50ms);
    auto t0 = steady_clock::now();
    std::vector<std::future<int>> fs;
    for (int i = 0; i < 10; ++i) fs.push_back(b.submit(i));
    std::cout << "results:"; for (auto& f : fs) std::cout << " " << f.get();
    std::cout << "\nbatch sizes:"; for (auto s : sizes) std::cout << " " << s;
    std::cout << "  | total " << duration<double>(steady_clock::now() - t0).count() << "s\n";
}

$ g++ -std=c++20 -O2 -Wall -Wextra -pthread iv_batcher.cpp -o iv_batcher
results: 0 2 4 6 8 10 12 14 16 18
batch sizes: 4 4 2  | total 0.0600383s

8. Debounce

Idea: collapse a burst of calls into one call that fires after a quiet period, with the last arguments. Every call resets the deadline. The output shows 5 rapid calls (0 to 4, 30 ms apart) producing a single call with the last value, then a later call firing on its own. (Throttle is the sibling: at most one call per interval, the first or the last one.) Complexity: O(1) per call. Follow-ups: leading-edge vs trailing-edge firing, a maximum wait so a never-ending burst still fires, cancellation/flush on shutdown (the C++ destructor joins the worker; pending work is dropped), and using asyncio (loop.call_later + cancel) instead of threads.

Copyimport threading, time

def debounce(wait):
    def deco(fn):
        timer, lock = [None], threading.Lock()
        def wrapper(*a, **kw):
            with lock:
                if timer[0]: timer[0].cancel()              # every call pushes the deadline out
                timer[0] = threading.Timer(wait, fn, a, kw); timer[0].start()
        return wrapper
    return deco

calls = []
@debounce(0.1)
def save(x): calls.append((x, round(time.monotonic() - t0, 1)))

t0 = time.monotonic()
for i in range(5): save(i); time.sleep(0.03)      # burst of 5 calls, 30 ms apart
time.sleep(0.3)
save("later"); time.sleep(0.3)
print(calls)                                          # only the LAST call of each burst fires, after the quiet period

$ python3 iv_debounce.py
[(4, 0.3), ('later', 0.6)]

Copy#include <chrono>
#include <condition_variable>
#include <functional>
#include <iostream>
#include <mutex>
#include <optional>
#include <thread>

using namespace std::chrono;

class Debouncer {
    std::function<void(int)> fn_; milliseconds wait_;
    std::mutex m_; std::condition_variable cv_;
    std::optional<int> pending_; steady_clock::time_point deadline_; bool stop_ = false;
    std::thread worker_;
    void loop() {
        std::unique_lock lk(m_);
        while (!stop_) {
            if (!pending_) { cv_.wait(lk); continue; }
            if (cv_.wait_until(lk, deadline_) == std::cv_status::timeout && pending_ && steady_clock::now() >= deadline_) {
                int v = *pending_; pending_.reset();
                lk.unlock(); fn_(v); lk.lock();               // never call user code under the lock
            }                                                 // else: woken by a new call -> loop re-reads the deadline
        }
    }
public:
    Debouncer(std::function<void(int)> fn, milliseconds wait) : fn_(std::move(fn)), wait_(wait), worker_([this] { loop(); }) {}
    ~Debouncer() { { std::lock_guard g(m_); stop_ = true; } cv_.notify_all(); worker_.join(); }
    void call(int v) { { std::lock_guard g(m_); pending_ = v; deadline_ = steady_clock::now() + wait_; } cv_.notify_all(); }
};

int main() {
    std::vector<int> fired;
    {
        Debouncer d([&](int v) { fired.push_back(v); }, 100ms);
        for (int i = 0; i < 5; ++i) { d.call(i); std::this_thread::sleep_for(30ms); }
        std::this_thread::sleep_for(300ms);
        d.call(99); std::this_thread::sleep_for(300ms);
    }
    std::cout << "fired:"; for (int v : fired) std::cout << " " << v; std::cout << "\n";
}

$ g++ -std=c++20 -O2 -Wall -Wextra -pthread iv_debounce.cpp -o iv_debounce
fired: 4 99