PyTorch Engineering Masterclass — Chapter 1 of 16

Welcome to Chapter 1 of the PyTorch Engineering Masterclass. PyTorch 2.x is the primary software foundation for modern artificial intelligence, driving state-of-the-art research and massive production deployments across Meta, OpenAI, Hugging Face, and autonomous vehicle stacks. This chapter bridges the gap between basic syntax and systems-level ML engineering: you will learn reproducible environment configuration, hardware execution architectures (NVIDIA CUDA, Apple Silicon MPS, and CPU), PyTorch 2.x compiler internals, tensor memory layouts, stride mathematics, and asynchronous accelerator synchronization.

2. Prerequisites & Core Learning Objectives

Before proceeding, ensure you have Python 3.10+ installed and a fundamental grasp of linear algebra (vectors, matrices, and dot products). By the conclusion of this chapter, you will be able to:

• Configure deterministic, reproducible PyTorch 2.x development environments across Linux, Windows, and macOS.
• Differentiate eager interpreter execution from Ahead-of-Time graph capture (torch.compile).
• Engineer robust, device-agnostic fallback utilities supporting NVIDIA CUDA, Apple Silicon MPS, and CPU.
• Diagnose tensor memory layouts, storage offsets, and multidimensional stride calculations.
• Avoid the “non-contiguous tensor view” trap by mastering view(), reshape(), and contiguous().
• Apply broadcasting rules to eliminate redundant memory allocations in arithmetic operations.
• Benchmark accelerator throughput accurately using asynchronous execution barriers.

3. PyTorch 2.x Architecture: Eager Dispatch vs. TorchInductor Compilation

To understand why PyTorch revolutionized deep learning, we must examine its dual execution architecture. In traditional frameworks (like TensorFlow 1.x), models were defined as rigid, static abstract graphs that required compilation before any data flowed through them. This made inspection and debugging with native Python tools nearly impossible.

PyTorch pioneered Eager Execution (Define-by-Run). Every mathematical line of PyTorch code immediately allocates memory and executes an operator kernel via the ATen (A Tensors) C++ library. You can place standard Python breakpoints (breakpoint() or pdb), inspect intermediate tensor shapes, and write dynamic Python control flows (such as if branches or recursive loops) inside neural networks.

However, eager execution introduces an architectural cost: memory bandwidth saturation. In eager mode, an expression such as y = torch.relu(torch.matmul(x, w) + b) launches three separate GPU kernels. Each kernel writes its intermediate output to high-bandwidth device memory (VRAM) and immediately reads it back for the next operation. This round-trip data movement frequently bottlenecks high-speed accelerators.

With PyTorch 2.x, the framework introduces torch.compile(). It operates via three core subsystems:

1. TorchDynamo: Safely intercepts Python frame evaluation using the PEP 523 API, extracting pure PyTorch computation graphs while gracefully handling dynamic Python code through graph breaks.
2. AOTAutograd: Captures not just the forward computational graph, but generates the exact joint forward-and-backward automatic differentiation graph prior to execution.
3. TorchInductor: The default compiler backend that compiles graphs into fused, high-performance GPU kernels using OpenAI Triton, or vectorized C++ OpenMP code for CPUs. Fused operations execute entirely within fast on-chip registers and SRAM, eliminating intermediate VRAM roundtrips.

4. System Installation, Virtual Isolation & Environment Verification

Never install machine learning dependencies into your global operating system Python environment. Doing so invites dependency collisions between conflicting versions of CUDA runtimes, NumPy, and system libraries. Always create an isolated virtual environment.

4.1 Creating an Isolated Environment

You can use standard Python venv, modern high-speed package managers like uv, or Conda:

# Option 1: Standard Python venv
python -m venv .venv

# Activate on Linux / macOS:
source .venv/bin/activate

# Activate on Windows (PowerShell):
.venv\Scripts\Activate.ps1

# Option 2: Ultra-fast installation via uv (Recommended for modern workflows)
uv venv .venv
source .venv/bin/activate  # or .venv\Scriptsctivate on Windows

4.2 Installing PyTorch for Your Target Hardware

PyTorch builds are tied directly to hardware acceleration runtimes. Consult the official distribution matrix below:

# ==============================================================================
# TARGET 1: NVIDIA GPU (Linux / Windows with CUDA 12.x or 11.8)
# Check your installed driver capability first with: nvidia-smi
# ==============================================================================
# Pip (CUDA 12.4 wheel):
pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu124

# Conda (NVIDIA channel):
conda install pytorch torchvision torchaudio pytorch-cuda=12.4 -c pytorch -c nvidia

# ==============================================================================
# TARGET 2: Apple Silicon (macOS M1, M2, M3, M4 - Metal Performance Shaders)
# ==============================================================================
# Apple Silicon binaries include native MPS acceleration out-of-the-box:
pip install torch torchvision torchaudio

# ==============================================================================
# TARGET 3: CPU-Only Development (Lightweight testing, CI/CD runners, Ryzen/Intel)
# ==============================================================================
pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cpu

5. Hardware Engineering: The Production Device Manager Utility

Writing hardcoded strings such as tensor.cuda() or tensor.to("cuda:0") is a major anti-pattern. If your code is deployed to an Apple Silicon development workstation, a cloud CPU instance, or a multi-GPU server, it will immediately crash with a RuntimeError.

Furthermore, machine learning engineers must distinguish between Discrete VRAM and Unified Memory (UMA):

• NVIDIA CUDA GPUs: The GPU maintains physically separate high-speed VRAM (GDDR6X or HBM3, delivering up to 3 TB/s bandwidth). However, data must be transferred from Host RAM over the PCIe bus (32–64 GB/s), creating a transfer latency bottleneck. Calling .to("cuda") copies bytes across the physical PCIe bus.
• Apple Silicon MPS: The CPU, GPU, and Neural Engine share a single physical pool of high-bandwidth unified memory. Calling .to("mps") registers the tensor with the Apple Metal driver without a physical PCIe bus hop, but memory is subject to macOS operating system allocation limits (typically 75% of total system RAM).

Below is a production-grade, reusable device management module featuring hardware inspection, memory diagnostics, and automatic fallback:

import logging
import platform
import sys
from typing import Optional, Tuple
import torch

logging.basicConfig(level=logging.INFO, format="[%(asctime)s] %(levelname)s: %(message)s")
logger = logging.getLogger("DeviceManager")

class DeviceManager:
    """
    Production-grade hardware accelerator discovery, memory inspection,
    and device allocation utility for PyTorch 2.x.
    """

    def __init__(self, preferred_device: Optional[str] = None):
        self.device = self._resolve_device(preferred_device)
        self._log_system_profile()

    def _resolve_device(self, preferred: Optional[str]) -> torch.device:
        if preferred is not None:
            try:
                target = torch.device(preferred)
                logger.info(f"Using explicitly requested device: {target}")
                return target
            except Exception as e:
                logger.warning(f"Preferred device '{preferred}' unavailable ({e}). Falling back to autodiscovery.")

        # 1. Check NVIDIA / AMD ROCm CUDA
        if torch.cuda.is_available():
            device_idx = torch.cuda.current_device()
            device_name = torch.cuda.get_device_name(device_idx)
            logger.info(f"Hardware Accelerator Discovered: NVIDIA CUDA [{device_name}] (Device {device_idx})")
            return torch.device(f"cuda:{device_idx}")

        # 2. Check Apple Silicon Metal Performance Shaders (MPS)
        if hasattr(torch.backends, "mps") and torch.backends.mps.is_available():
            if not torch.backends.mps.is_built():
                logger.warning("MPS is available on this system, but PyTorch was not compiled with MPS support.")
            else:
                logger.info("Hardware Accelerator Discovered: Apple Silicon Metal (MPS)")
                return torch.device("mps")

        # 3. Fallback to Host CPU
        logger.info("No hardware accelerator detected. Falling back to Host CPU.")
        return torch.device("cpu")

    def _log_system_profile(self) -> None:
        logger.info(f"Operating System: {platform.system()} {platform.release()} ({platform.machine()})")
        logger.info(f"Python Runtime: {sys.version.split()[0]} | PyTorch Version: {torch.__version__}")
        
        if self.device.type == "cuda":
            idx = self.device.index or 0
            props = torch.cuda.get_device_properties(idx)
            total_vram_gb = props.total_memory / (1024 ** 3)
            cuda_cap = f"{props.major}.{props.minor}"
            logger.info(f"CUDA Compute Capability: {cuda_cap} | Total VRAM: {total_vram_gb:.2f} GB | Multi-Processors: {props.multi_processor_count}")
        elif self.device.type == "mps":
            logger.info("Apple Unified Memory Architecture (UMA) active. Memory dynamically shared with macOS.")
        else:
            logger.info("CPU Execution active. Utilizing host system RAM and thread pool.")

    def allocate(self, shape: Tuple[int, ...], dtype: torch.dtype = torch.float32) -> torch.Tensor:
        """Allocates a tensor directly on the active hardware target."""
        return torch.zeros(shape, dtype=dtype, device=self.device)

    def to_device(self, tensor: torch.Tensor, non_blocking: bool = False) -> torch.Tensor:
        """Safely transfers an existing tensor to the managed hardware accelerator."""
        if tensor.device == self.device:
            return tensor
        return tensor.to(self.device, non_blocking=non_blocking)

    def synchronize(self) -> None:
        """Blocks host thread until all queued asynchronous accelerator operations complete."""
        if self.device.type == "cuda":
            torch.cuda.synchronize(self.device)
        elif self.device.type == "mps":
            torch.mps.synchronize()

# Self-contained verification execution:
if __name__ == "__main__":
    hw = DeviceManager()
    sample_tensor = hw.allocate((1000, 1000))
    print(f"Allocated Tensor: Shape={sample_tensor.shape}, Device={sample_tensor.device}, DType={sample_tensor.dtype}")

6. Tensor Fundamentals Deep Dive

6.1 What is a Tensor?

Mathematically, a tensor is a multidimensional array of numerical values characterized by its order or rank. In computing, a PyTorch tensor encapsulates three critical architectural layers:

1. Storage: A contiguous 1D array of untyped raw bytes in physical RAM or accelerator memory.
2. Metadata (Tensor View): Size (shape), strides, data type (dtype), storage offset, and target compute device.
3. Autograd History: Pointers to computational graph nodes (grad_fn) that enable backward automatic differentiation.

6.2 Tensor Data Types and Precision Engineering

Every tensor is bound to a specific floating-point or integer bit-width. Choosing the correct precision is one of the highest-leverage engineering decisions in modern AI:

import torch

# Precision inspection and casting:
x_fp32 = torch.tensor([1.0, 2.5, 3.8], dtype=torch.float32)
x_bf16 = x_fp32.to(dtype=torch.bfloat16)

print(f"FP32 Element Size: {x_fp32.element_size()} bytes | Total Memory: {x_fp32.nelement() * x_fp32.element_size()} bytes")
print(f"BF16 Element Size: {x_bf16.element_size()} bytes | Total Memory: {x_bf16.nelement() * x_bf16.element_size()} bytes")

6.3 Tensor Creation Methods and Deterministic Seeding

PyTorch offers multiple constructor functions with distinct performance and memory implications:

• torch.empty(*size): Allocates uninitialized memory. Blazingly fast because it skips zeroing memory, but contains garbage values. Ideal only when immediately overwritten (e.g., as an output buffer).
• torch.zeros(*size) and torch.ones(*size): Allocates and clears memory. Safe and deterministic.
• torch.arange(start, end, step) and torch.linspace(start, end, steps): 1D series generators.
• torch.randn(*size): Samples from standard normal distribution $\mathcal{N}(0, 1)$.

For research reproducibility, always explicitly seed the pseudo-random number generators across all devices:

import torch

def set_seed(seed: int = 42) -> None:
    torch.manual_seed(seed)
    if torch.cuda.is_available():
        torch.cuda.manual_seed(seed)
        torch.cuda.manual_seed_all(seed)  # Multi-GPU setups
        # Guarantee deterministic algorithms (may incur slight performance penalty):
        torch.backends.cudnn.deterministic = True
        torch.backends.cudnn.benchmark = False

set_seed(42)
rand_sample = torch.randn(2, 3)
print("Deterministic Random Sample:\n", rand_sample)

6.4 Indexing, Slicing & Ellipsis Notation

PyTorch implements full NumPy-compatible indexing semantics, including strided slicing, boolean masking, integer array indexing, and multidimensional ellipsis expansion:

import torch

# 4D Computer Vision Batch: [Batch Size, Channels, Height, Width]
batch = torch.arange(2 * 3 * 4 * 4).reshape(2, 3, 4, 4)

# 1. Extract first sample, all channels, bottom-right 2x2 spatial patch:
patch = batch[0, :, 2:, 2:]
print("Bottom-Right Spatial Patch Shape:", patch.shape)  # torch.Size([3, 2, 2])

# 2. Ellipsis notation: Select last pixel for all leading dimensions:
corners = batch[..., -1, -1]
print("Corners across batch and channels:", corners.shape)  # torch.Size([2, 3])

# 3. Boolean Masking: Zero out negative activations
activations = torch.randn(3, 3)
mask = activations > 0
positive_values = activations[mask]  # Flattens to 1D vector
print(f"Masked activations count: {positive_values.numel()} of 9")

6.5 Memory Strides, Storage Offsets & The Contiguity Trap

Why does tensor.view() occasionally raise a crash with: RuntimeError: view size is not compatible with input tensor's size and stride?

To answer this, examine how multidimensional tensors exist in physical hardware memory. Computer RAM is a linear, 1-dimensional array of bytes. A 2D matrix $M \in \mathbb{R}^{R \imes C}$ is stored sequentially in row-major order. The formula mapping 2D indices $(r, c)$ to the linear memory address is:

linear_offset = storage_offset + (r × stride[0]) + (c × stride[1])

When you create a tensor with shape (2, 3), its stride is (3, 1): jumping 1 row skips 3 elements in memory; jumping 1 column skips 1 element. When you transpose the matrix (t = x.t()), PyTorch does not move or copy any bytes in memory. It simply swaps the stride metadata to (1, 3)!

Because jumping columns now requires jumping 3 elements in memory, the tensor is no longer C-contiguous. Calling .view() fails because view() requires contiguous physical memory. To resolve this, call .contiguous(), which allocates a new memory buffer and re-orders the bytes sequentially, or use .reshape(), which automatically calls .contiguous() if necessary.

import torch

# Create a contiguous 2x3 tensor
x = torch.tensor([[1, 2, 3], [4, 5, 6]], dtype=torch.int32)
print(f"Original x: Shape={x.shape}, Strides={x.stride()}, Contiguous={x.is_contiguous()}")

# Transpose creates a non-contiguous view
x_t = x.t()
print(f"Transposed x_t: Shape={x_t.shape}, Strides={x_t.stride()}, Contiguous={x_t.is_contiguous()}")

try:
    # Fails with RuntimeError!
    invalid_view = x_t.view(6)
except RuntimeError as e:
    print(f"Expected View Error: {e}")

# Fix: Call contiguous() to re-allocate buffer in physical order
x_t_contiguous = x_t.contiguous()
valid_view = x_t_contiguous.view(6)
print("Successfully reshaped after .contiguous():", valid_view)

6.6 Broadcasting Rules & Zero-Copy Dimension Expansion

Broadcasting allows element-wise operations between tensors of differing shapes without allocating duplicate memory buffers for the smaller tensor. PyTorch follows the NumPy broadcasting algorithm:

The Broadcasting Algorithm:

1. Right-to-Left Alignment: Align shape tuples starting from the trailing (rightmost) dimension.
2. Prepend Singleton Dimensions: If one tensor has fewer dimensions, prepend 1s to its shape until lengths match.
3. Compatibility Test: For every dimension pair, the two sizes must be equal, or one of them must be 1. If not, broadcasting fails with a shape mismatch error.
4. Zero-Copy Stride Assignment: Any dimension with size 1 expanded to size $N$ is assigned a stride of 0. The hardware reads the exact same memory address $N$ times without allocating extra memory!

import torch

# Tensor A: 3x1 column matrix (Shape: [3, 1])
a = torch.tensor([[10.0], [20.0], [30.0]])

# Tensor B: 1D vector of length 4 (Shape: [4] -> Prepend 1 -> [1, 4])
b = torch.tensor([1.0, 2.0, 3.0, 4.0])

# Broadcasted addition: Output shape [3, 4]
c = a + b
print("Broadcasted Sum (Shape [3, 4]):\n", c)

6.7 Matrix Multiplication ($A \imes B$) vs. Element-Wise Multiplication

A frequent error among beginners is confusing element-wise multiplication (A * B) with matrix multiplication (A @ B or torch.matmul(A, B)):

• A * B (Hadamard Product): Performs element-wise multiplication. Both tensors must have identical shapes or be broadcastable.
• torch.matmul(A, B) or A @ B: Performs linear algebraic matrix multiplication. For 2D matrices, the inner dimensions must match: $(M \imes K) \imes (K \imes N)
  ightarrow (M \imes N)$.

import torch

# Batch Matrix Multiplication (BMM) for Transformer Attention Heads:
batch_size, num_heads, seq_len, head_dim = 16, 8, 128, 64

queries = torch.randn(batch_size, num_heads, seq_len, head_dim)
keys = torch.randn(batch_size, num_heads, seq_len, head_dim)

# Transpose last two dimensions of keys for dot product: [B, H, head_dim, seq_len]
keys_transposed = keys.transpose(-2, -1)

# Attention Scores: [16, 8, 128, 64] @ [16, 8, 64, 128] -> [16, 8, 128, 128]
attention_scores = queries @ keys_transposed
print("Batch Matrix Product Shape:", attention_scores.shape)

6.8 Zero-Copy NumPy Interoperability

PyTorch provides bidirectional interoperability with NumPy via torch.from_numpy() and tensor.numpy(). In CPU memory, this operation is zero-copy: both objects point to the exact same physical byte address (data_ptr()). Mutating values in one framework instantly alters values in the other!

import numpy as np
import torch

# Allocate NumPy array
np_arr = np.array([10.0, 20.0, 30.0], dtype=np.float32)

# Zero-copy conversion to PyTorch
torch_tensor = torch.from_numpy(np_arr)

# Verify identical physical memory pointers:
print(f"NumPy Pointer:   {np_arr.__array_interface__['data'][0]}")
print(f"PyTorch Pointer: {torch_tensor.data_ptr()}")

# In-place mutation reflects instantly across both:
np_arr[0] = 999.0
print("PyTorch tensor reflects NumPy mutation:", torch_tensor[0].item())  # 999.0

# CRITICAL GPU / MPS CAVEAT:
# Tensors residing on an accelerator CANNOT directly convert to NumPy.
# You must explicitly transfer them to Host CPU first:
# np_copy = gpu_tensor.cpu().numpy()

6.9 Asynchronous Accelerator Execution & Accurate Benchmarking

When you call an operation on a GPU (e.g., z = x @ y), the PyTorch CPU thread simply places the execution request onto the device’s CUDA or Metal work queue and returns immediately. The GPU executes the math asynchronously in the background.

If you benchmark an accelerator using standard Python time.perf_counter() without synchronization, you are merely timing how many microseconds it took to enqueue the kernel—not how long it took to calculate the result! Always invoke a synchronization barrier:

import time
import torch

device = torch.device("cuda" if torch.cuda.is_available() else ("mps" if torch.backends.mps.is_available() else "cpu"))

# Allocate matrices on accelerator
size = 4096
a = torch.randn(size, size, device=device)
b = torch.randn(size, size, device=device)

# Warm-up iterations (initialize driver kernels, allocate caches)
for _ in range(5):
    _ = a @ b

# Synchronize before timing
if device.type == "cuda":
    torch.cuda.synchronize()
elif device.type == "mps":
    torch.mps.synchronize()

start_time = time.perf_counter()

# Perform target workload
c = a @ b

# Mandatory synchronization barrier: Wait for GPU execution to complete!
if device.type == "cuda":
    torch.cuda.synchronize()
elif device.type == "mps":
    torch.mps.synchronize()

elapsed_ms = (time.perf_counter() - start_time) * 1000
# Matrix multiplication floating-point operations: 2 * N^3 ops
flops = 2 * (size ** 3)
tflops = (flops / (elapsed_ms / 1000)) / 1e12

print(f"Device: {device} | Elapsed Time: {elapsed_ms:.2f} ms | Compute Throughput: {tflops:.2f} TFLOPS")

7. Five Hands-on Engineering Laboratories

Lab 1: Tensor Anatomy & Dimensionality Inspection

Objective: Construct and introspect tensors across dimensions 0 to 4, verifying element counts, data types, and byte allocations.

import torch

def inspect_tensor(name: str, t: torch.Tensor) -> None:
    print(f"--- {name} ---")
    print(f"Shape: {list(t.shape)} | Rank: {t.ndim} | Numel: {t.numel()}")
    print(f"DType: {t.dtype} | Bytes/Elem: {t.element_size()} | Total Bytes: {t.numel() * t.element_size()}")
    print(f"Device: {t.device} | Strides: {t.stride()} | Contiguous: {t.is_contiguous()}\n")

scalar = torch.tensor(3.14159)
vector = torch.linspace(0, 10, 5)
matrix = torch.eye(3, 4)
tensor_3d = torch.zeros(3, 32, 32)
tensor_4d = torch.randn(16, 3, 224, 224)

inspect_tensor("Scalar", scalar)
inspect_tensor("Vector", vector)
inspect_tensor("Matrix", matrix)
inspect_tensor("3D Tensor", tensor_3d)
inspect_tensor("4D Batch", tensor_4d)

Lab 2: Multi-Axis Slicing, Boolean Masking & Broadcast Arithmetic

Objective: Normalize a tabular batch of feature embeddings using broadcast mean and standard deviation.

import torch

# Simulated feature matrix: 100 samples, 5 features
torch.manual_seed(42)
features = torch.randn(100, 5) * 5.0 + 20.0

# Compute mean and standard deviation along sample dimension (dim=0)
mean = features.mean(dim=0, keepdim=True)  # Shape: [1, 5]
std = features.std(dim=0, keepdim=True)    # Shape: [1, 5]

# Broadcast subtraction and division: [100, 5] - [1, 5] -> [100, 5]
normalized = (features - mean) / std

print("Original Mean Feature 0:", features[:, 0].mean().item())
print("Normalized Mean Feature 0 (Target ~0.0):", normalized[:, 0].mean().item())
print("Normalized Std Feature 0 (Target ~1.0):", normalized[:, 0].std().item())

Lab 3: Transpose, Permute & The Contiguous Buffer Trap

Objective: Transform a Computer Vision NCHW tensor to channels-last NHWC for optimized inference, safely handling non-contiguous strides.

import torch

# Standard PyTorch image batch: [Batch=4, Channels=3, Height=32, Width=32]
img_nchw = torch.randn(4, 3, 32, 32)
print("NCHW Strides:", img_nchw.stride())  # (3072, 1024, 32, 1)

# Permute dimensions to Channels-Last: [4, 32, 32, 3]
img_nhwc = img_nchw.permute(0, 2, 3, 1)
print("NHWC Strides:", img_nhwc.stride())  # (3072, 32, 1, 1024) - Non-contiguous!
print("Is NHWC Contiguous?", img_nhwc.is_contiguous())

# Enforce physical contiguity before passing to low-level C++ / TensorRT runtimes:
img_nhwc_contiguous = img_nhwc.contiguous()
print("Enforced Contiguous Strides:", img_nhwc_contiguous.stride())  # (3072, 96, 3, 1)
print("Is Enforced Contiguous?", img_nhwc_contiguous.is_contiguous())

Lab 4: Linear Algebra, Batch Matrix Multiplication & Reductions

Objective: Implement multi-head self-attention projection math using batch matrix multiplication and softmax reductions.

import torch

batch_size = 4
seq_len = 8
d_model = 16

# Simulated Token Embeddings
x = torch.randn(batch_size, seq_len, d_model)
# Projection Weights
w_q = torch.randn(d_model, d_model)
w_k = torch.randn(d_model, d_model)

# Compute Projections: [4, 8, 16] @ [16, 16] -> [4, 8, 16]
q = x @ w_q
k = x @ w_k

# Compute Raw Attention Affinities: Q @ K^T -> [4, 8, 16] @ [4, 16, 8] -> [4, 8, 8]
scores = q @ k.transpose(-2, -1) / (d_model ** 0.5)

# Softmax Reduction along last dimension:
attn_weights = torch.softmax(scores, dim=-1)

# Verify Softmax sums to 1.0 along trailing dimension
sums = attn_weights.sum(dim=-1)
print("Softmax Row Sums Check (Should be all 1.0):\n", sums[0])

Lab 5: Asynchronous Accelerator Profiling & Synchronization

Objective: Measure real accelerator execution vs. asynchronous queuing overhead.

import time
import torch

device = torch.device("cuda" if torch.cuda.is_available() else ("mps" if torch.backends.mps.is_available() else "cpu"))
print(f"Running Lab 5 on Target Device: {device}")

dim = 6000
x = torch.randn(dim, dim, device=device)
y = torch.randn(dim, dim, device=device)

# 1. Flawed naive benchmark (No synchronization):
start_naive = time.perf_counter()
z_naive = x @ y
elapsed_naive_ms = (time.perf_counter() - start_naive) * 1000
print(f"Naive Timing (Misleading Queue Time): {elapsed_naive_ms:.4f} ms")

# 2. Accurate synchronized benchmark:
if device.type == "cuda":
    torch.cuda.synchronize()
elif device.type == "mps":
    torch.mps.synchronize()

start_sync = time.perf_counter()
z_sync = x @ y

if device.type == "cuda":
    torch.cuda.synchronize()
elif device.type == "mps":
    torch.mps.synchronize()

elapsed_sync_ms = (time.perf_counter() - start_sync) * 1000
print(f"Synchronized Timing (Actual Compute Time): {elapsed_sync_ms:.2f} ms")

8. Capstone Mini-Project: Hardware Computation Benchmarking Utility

In production MLOps, engineers regularly need to evaluate inference and training throughput across differing cloud instances (AWS g5, Apple M-series, Google Cloud TPUs/GPUs). Below is a complete, self-contained Python CLI benchmarking suite:

import argparse
import logging
import platform
import sys
import time
from typing import Dict, Any, List
import torch

logging.basicConfig(level=logging.INFO, format="%(message)s")
logger = logging.getLogger("BenchmarkSuite")

class HardwareBenchmark:
    """
    Production Tensor Computation and Hardware Performance Evaluation Suite.
    """

    def __init__(self, device_str: str = "auto"):
        self.device = self._init_device(device_str)
        self.system_info = self._collect_system_info()

    def _init_device(self, req: str) -> torch.device:
        if req != "auto":
            return torch.device(req)
        if torch.cuda.is_available():
            return torch.device("cuda:0")
        if hasattr(torch.backends, "mps") and torch.backends.mps.is_available():
            return torch.device("mps")
        return torch.device("cpu")

    def _collect_system_info(self) -> Dict[str, Any]:
        info = {
            "os": f"{platform.system()} {platform.release()}",
            "arch": platform.machine(),
            "python": sys.version.split()[0],
            "pytorch": torch.__version__,
            "device": str(self.device)
        }
        if self.device.type == "cuda":
            info["device_name"] = torch.cuda.get_device_name(self.device)
            info["vram_total_gb"] = f"{torch.cuda.get_device_properties(self.device).total_memory / (1024**3):.2f} GB"
        elif self.device.type == "mps":
            info["device_name"] = "Apple Silicon Metal Performance Shaders"
        else:
            info["device_name"] = platform.processor() or "Host CPU"
        return info

    def _sync(self) -> None:
        if self.device.type == "cuda":
            torch.cuda.synchronize(self.device)
        elif self.device.type == "mps":
            torch.mps.synchronize()

    def run_benchmark(self, matrix_size: int = 4096, warmup: int = 5, iterations: int = 20) -> Dict[str, Any]:
        logger.info(f"Allocating [{matrix_size}x{matrix_size}] FP32 tensors on {self.device}...")
        a = torch.randn(matrix_size, matrix_size, dtype=torch.float32, device=self.device)
        b = torch.randn(matrix_size, matrix_size, dtype=torch.float32, device=self.device)

        # Warm-up phase
        for _ in range(warmup):
            _ = a @ b
        self._sync()

        # Timed Matmul Benchmark
        logger.info(f"Running {iterations} matrix multiplication trials...")
        times: List[float] = []
        for _ in range(iterations):
            self._sync()
            t0 = time.perf_counter()
            _ = a @ b
            self._sync()
            times.append(time.perf_counter() - t0)

        avg_sec = sum(times) / len(times)
        min_sec = min(times)
        
        # 2 * N^3 floating point operations per square matrix multiply
        flops = 2.0 * (matrix_size ** 3)
        avg_tflops = (flops / avg_sec) / 1e12
        peak_tflops = (flops / min_sec) / 1e12

        # Element-wise Addition Benchmark
        t0 = time.perf_counter()
        for _ in range(iterations):
            _ = a + b
        self._sync()
        add_sec = (time.perf_counter() - t0) / iterations

        return {
            "matrix_size": matrix_size,
            "avg_matmul_ms": avg_sec * 1000,
            "min_matmul_ms": min_sec * 1000,
            "avg_tflops": avg_tflops,
            "peak_tflops": peak_tflops,
            "avg_add_ms": add_sec * 1000
        }

    def print_report(self, results: Dict[str, Any]) -> None:
        print("\n" + "="*70)
        print("          PYTORCH HARDWARE BENCHMARK REPORT          ")
        print("="*70)
        print(f"Platform:       {self.system_info['os']} ({self.system_info['arch']})")
        print(f"PyTorch Ver:    {self.system_info['pytorch']} (Python {self.system_info['python']})")
        print(f"Compute Target: {self.system_info['device']} - {self.system_info['device_name']}")
        if "vram_total_gb" in self.system_info:
            print(f"Hardware VRAM:  {self.system_info['vram_total_gb']}")
        print("-"*70)
        print(f"Workload:       Square Matmul [{results['matrix_size']} x {results['matrix_size']}] FP32")
        print(f"Avg Latency:    {results['avg_matmul_ms']:.2f} ms")
        print(f"Peak Latency:   {results['min_matmul_ms']:.2f} ms")
        print(f"Avg Compute:    {results['avg_tflops']:.2f} TFLOPS")
        print(f"Peak Compute:   {results['peak_tflops']:.2f} TFLOPS")
        print(f"Elementwise:    {results['avg_add_ms']:.2f} ms")
        print("="*70 + "\n")

if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="PyTorch Hardware Benchmark")
    parser.add_argument("--size", type=int, default=4096, help="Matrix dimension size")
    parser.add_argument("--device", type=str, default="auto", help="Compute target (auto, cpu, cuda, mps)")
    args = parser.parse_args()

    bench = HardwareBenchmark(device_str=args.device)
    res = bench.run_benchmark(matrix_size=args.size)
    bench.print_report(res)

9. Ten Practical Engineering Exercises with Solutions

import torch
numel = 64 * 3 * 256 * 256  # 12,582,912 elements
bytes_per_elem = 4          # float32 = 4 bytes
total_bytes = numel * bytes_per_elem
mb = total_bytes / (1024 ** 2)
print(f"Calculated: {mb:.2f} MB")

# Verification:
t = torch.empty(64, 3, 256, 256, dtype=torch.float32)
actual_mb = (t.numel() * t.element_size()) / (1024 ** 2)
print(f"PyTorch Verification: {actual_mb:.2f} MB")  # 48.00 MB

import torch
x = torch.randn(5, 5)
print("Original:", x.is_contiguous())  # True

# Contiguous transformation (slicing row dimension):
row_slice = x[1:3, :]
print("Row Slice:", row_slice.is_contiguous())  # True

# Non-contiguous transformation (transposing or slicing columns):
col_slice = x[:, 1:3]
print("Column Slice:", col_slice.is_contiguous())  # False (skips elements)
transposed = x.t()
print("Transposed:", transposed.is_contiguous())  # False

import torch

# Define target device once at module load:
TARGET_DEVICE = torch.device("cuda" if torch.cuda.is_available() else ("mps" if torch.backends.mps.is_available() else "cpu"))

def safe_to_accelerator(t: torch.Tensor) -> torch.Tensor:
    return t.to(TARGET_DEVICE)

sample = torch.ones(3)
migrated = safe_to_accelerator(sample)
print("Migrated device:", migrated.device)

import torch
b_matrices = torch.randn(10, 4, 8)
# Swap dimension 1 (Rows) and dimension 2 (Cols), leaving dimension 0 untouched:
b_transposed = b_matrices.permute(0, 2, 1)
print("Original Shape:", b_matrices.shape)      # [10, 4, 8]
print("Transposed Shape:", b_transposed.shape)  # [10, 8, 4]

import torch
img = torch.randn(3, 32, 32)
flat_view = img.view(-1)  # -1 infers length automatically: 3072
print("Flat view length:", flat_view.shape)
print("Shares storage:", img.data_ptr() == flat_view.data_ptr())  # True (Zero Copy)

# If non-contiguous:
img_noncontig = img.permute(1, 2, 0)
# flat_view_err = img_noncontig.view(-1) -> FAILS!
flat_safe = img_noncontig.reshape(-1)  # Safe: clones only when necessary

import torch
x = torch.tensor([1.0, 2.0])
y = torch.tensor([10.0, 20.0])

out = x.add(y)  # Out-of-place: Allocates new tensor
print("x after out-of-place add:", x)  # [1.0, 2.0]

x.add_(y)  # In-place (trailing underscore): Modifies memory of x directly!
print("x after in-place add_:", x)    # [11.0, 22.0]

import torch
u = torch.tensor([1.0, 2.0, 3.0, 4.0])
v = torch.tensor([2.0, 0.5, -1.0, 3.0])

# Element-wise multiply followed by reduction sum:
dot_product = (u * v).sum()
print("Manual Dot Product:", dot_product.item())  # (2.0 + 1.0 - 3.0 + 12.0) = 12.0
print("Verification with torch.dot:", torch.dot(u, v).item())

import torch
t = torch.empty(2, 3, 4)
print("PyTorch Stride:", t.stride())  # (12, 4, 1)

import torch
mat = torch.arange(16).reshape(4, 4)
sub = mat[2:, 2:]
print("Submatrix:\n", sub)
print("Submatrix Storage Offset:", sub.storage_offset())  # 10

10. Ten Technical Interview Questions with In-Depth Answers

11. Final Knowledge Assessment

Test your systems-level PyTorch understanding with these five core review questions:

1. Why does broadcasting a dimension from size 1 to size $N$ assign a stride of 0?
2. What specific line of Python code ensures that a multi-GPU training script will not raise an unhandled exception when executed on an Apple Silicon Mac?
3. Why does calling .contiguous() on an already-contiguous tensor incur zero performance penalty?
4. What is the memory size difference between an integer tensor holding token IDs in torch.int64 vs torch.int32 over a sequence of 4,096 tokens across a batch of 32?
5. What does TorchDynamo do when it encounters an unsupported dynamic third-party library call inside a compiled model function?

12. Official References & Further Reading

• Official PyTorch Installation Matrix
• PyTorch Core Tensor API Documentation
• PyTorch 2.x torch.compile In-Depth Tutorial
• Apple Developer Metal Performance Shaders (MPS) for PyTorch
• NVIDIA CUDA Developer Documentation