KernelScope is a CUDA Compiler Explorer that transforms CUDA source code into an interactive visualization of GPU execution. This document explains every component of the backend in detail.
backend/src/kernelscope/
├── api/ # FastAPI REST API
│ ├── app.py # App factory, CORS, static file serving
│ ├── routes.py # All HTTP endpoints
│ └── schemas.py # Pydantic request/response models
│
├── ptx/ # PTX Assembly Parser
│ ├── parser.py # PtxParser class with regex-based parsing
│ └── models.py # PtxInstruction, PtxKernel, PtxModule
│
├── simulator/ # Execution Simulator
│ ├── engine.py # PtxSimulator, memory analysis, cache
│ └── models.py # Thread, Warp, Block, TraceEvent
│
├── sass/ # SASS Machine Code Parser
│ └── parser.py # Parse cuobjdump output
│
├── occupancy/ # SM Occupancy Calculator
│ └── calculator.py # GPU specs database, occupancy formulas
│
├── latency/ # Instruction Timing Model
│ └── model.py # Per-instruction latency and throughput
│
└── hints/ # Performance Analyzer
└── analyzer.py # Automated issue detection
Location: ptx/parser.py, ptx/models.py
Purpose: Parse NVIDIA PTX (Parallel Thread Execution) assembly into a structured representation suitable for simulation and visualization.
PTX is NVIDIA's intermediate representation for GPU code. When you compile CUDA:
CUDA Source (.cu) → nvcc → PTX Assembly (.ptx) → cubin/SASS (machine code)
PTX is architecture-independent, allowing the same PTX to run on different GPU generations (with JIT compilation to native SASS at load time).
.version 7.5 // PTX version
.target sm_80 // Target architecture
.address_size 64 // 64-bit pointers
.visible .entry vector_add( // Kernel entry point
.param .u64 param_a, // Parameters (device pointers)
.param .u64 param_b,
.param .u64 param_c,
.param .u32 param_n
)
{
.reg .pred %p<2>; // Predicate registers (for branching)
.reg .f32 %f<4>; // 32-bit float registers
.reg .b32 %r<6>; // 32-bit general registers
.reg .b64 %rd<8>; // 64-bit registers
.loc 1 2 0 // Source location: file 1, line 2, column 0
ld.param.u64 %rd1, [param_a]; // Load parameter
mov.u32 %r2, %tid.x; // Get thread ID
mad.lo.s32 %r5, %r3, %r4, %r2; // Multiply-add: r5 = r3*r4 + r2
setp.ge.s32 %p1, %r5, %r1; // Set predicate: p1 = (r5 >= r1)
@%p1 bra $L__exit; // Conditional branch if p1 is true
ld.global.f32 %f1, [%rd5]; // Global memory load
add.f32 %f3, %f1, %f2; // Floating-point add
st.global.f32 [%rd7], %f3; // Global memory store
$L__exit:
ret; // Return
}
The parser uses compiled regex patterns for efficiency:
class PtxParser:
# Module-level directives
VERSION_PATTERN = re.compile(r"\.version\s+(\d+\.\d+)")
TARGET_PATTERN = re.compile(r"\.target\s+(\w+)")
ADDRESS_SIZE_PATTERN = re.compile(r"\.address_size\s+(\d+)")
# Kernel structure
ENTRY_PATTERN = re.compile(r"\.visible\s+\.entry\s+(\w+)\s*\(")
PARAM_PATTERN = re.compile(r"\.param\s+\.(\w+)\s+(\w+)")
REG_PATTERN = re.compile(r"\.reg\s+\.(\w+)\s+%(\w+)<(\d+)>")
# Source mapping (CUDA line numbers)
LOC_PATTERN = re.compile(r"\.loc\s+(\d+)\s+(\d+)\s+(\d+)")
# Instructions with optional predicate
INSTRUCTION_PATTERN = re.compile(
r"^(?:@(!?)%(\w+)\s+)?" # Optional predicate: @%p1 or @!%p1
r"(\w+)" # Opcode: add, ld, bra, etc.
r"((?:\.\w+)*)" # Modifiers: .f32, .global, .rn
r"\s*(.*?)" # Operands: %r1, %r2, %r3
r"(?:\s*;.*)?$" # Optional trailing comment
)Instructions are categorized by type for analysis:
class PtxOpType(Enum):
ARITHMETIC = "arithmetic" # add, sub, mul, mad, fma, div, sqrt, sin, cos
MEMORY = "memory" # ld, st, atom, red, prefetch
CONTROL = "control" # bra, call, ret, exit, setp
SYNC = "sync" # bar, barrier, membar, fence
SPECIAL = "special" # Other instructions
DIRECTIVE = "directive" # .reg, .param, .loc
LABEL = "label" # $L__exit:
ARITHMETIC_OPS = frozenset({
"add", "sub", "mul", "mad", "div", "rem", "abs", "neg",
"min", "max", "fma", "sqrt", "rsqrt", "sin", "cos",
"lg2", "ex2", "rcp", "cvt", "cvta",
"and", "or", "xor", "not", "shl", "shr"
})
MEMORY_OPS = frozenset({"ld", "st", "ldu", "prefetch", "atom", "red", "mov"})
CONTROL_OPS = frozenset({"bra", "brx", "call", "ret", "exit", "setp", "selp", "set"})
SYNC_OPS = frozenset({"bar", "barrier", "membar", "fence"})@dataclass(frozen=True, slots=True)
class PtxInstruction:
line: int # Line number in PTX
opcode: str # "add", "ld", "bra", etc.
op_type: PtxOpType # Classification
modifiers: tuple[str, ...] = () # ("f32", "global", "rn")
operands: tuple[str, ...] = () # ("%r1", "%r2", "[%rd5]")
predicate: str | None = None # "@%p1" or "@!%p1"
source_line: int | None = None # Original CUDA source line
@dataclass(slots=True)
class PtxKernel:
name: str # "vector_add"
params: list[PtxParam] # Kernel parameters
registers: dict[str, int] # {"f32": 4, "b32": 6, "b64": 8}
instructions: list[PtxInstruction]
source_map: dict[int, int] # PTX line → CUDA line
def _build_source_to_ptx_map(self) -> dict[int, list[int]]:
"""Reverse mapping: CUDA line → [PTX lines]"""
# Used for bidirectional source highlighting
@dataclass(slots=True)
class PtxModule:
version: str # "7.5"
target: str # "sm_80"
address_size: int # 64
kernels: list[PtxKernel]The .loc directive maps PTX back to CUDA source:
.loc 1 5 0 // File 1, Line 5, Column 0
ld.global.f32 %f1, [%rd5];
This enables the UI to highlight corresponding CUDA lines when hovering over PTX.
Location: simulator/engine.py, simulator/models.py
Purpose: Step through PTX execution, tracking thread states, memory accesses, and performance characteristics.
NVIDIA GPUs execute threads in groups of 32 called warps. All threads in a warp execute the same instruction simultaneously (SIMT - Single Instruction, Multiple Threads).
Grid (entire kernel launch)
└── Block (thread block, runs on one SM)
└── Warp (32 threads, execute in lockstep)
└── Thread (individual execution context)
@dataclass(slots=True)
class Thread:
tid: int # Thread ID within block
active: bool = True # Is thread currently executing?
registers: dict[str, int | float] # %r1, %f2, etc.
predicates: dict[str, bool] # %p1, %p2, etc.
def __post_init__(self) -> None:
# Initialize special registers
self.registers["%tid.x"] = self.tid % 32
self.registers["%tid.y"] = 0
self.registers["%tid.z"] = 0@dataclass(slots=True)
class Warp:
warp_id: int
block_id: int
threads: list[Thread] # Always 32 threads
pc: int = 0 # Program counter (PTX line number)
active_mask: int = 0xFFFFFFFF # 32-bit mask of active threads
at_barrier: bool = False # Waiting at __syncthreads()?
# SIMT stack for handling divergence
# Each entry: (program_counter, active_mask, reconvergence_point)
simt_stack: list[tuple[int, int, int]] = field(default_factory=list)
def __post_init__(self) -> None:
if not self.threads:
base_tid = self.warp_id * 32
self.threads = [Thread(tid=base_tid + i) for i in range(32)]
def get_active_threads(self) -> list[Thread]:
"""Return threads where corresponding bit in active_mask is 1"""
return [t for i, t in enumerate(self.threads)
if (self.active_mask >> i) & 1]The active_mask is a 32-bit integer where each bit represents one thread:
active_mask = 0xFFFFFFFF # All 32 threads active
= 0b11111111111111111111111111111111
active_mask = 0x0000FFFF # Only first 16 threads active
= 0b00000000000000001111111111111111
When a conditional branch causes divergence, some threads take the branch while others don't. The warp must execute both paths serially:
Before: active_mask = 0xFFFFFFFF (all 32 active)
if (threadIdx.x < 16):
path_a() # active_mask = 0x0000FFFF (threads 0-15)
else:
path_b() # active_mask = 0xFFFF0000 (threads 16-31)
After reconvergence: active_mask = 0xFFFFFFFF
@dataclass(slots=True)
class Block:
block_id: int
warps: list[Warp] # Number depends on block_dim
shared_memory: bytearray = field(
default_factory=lambda: bytearray(48 * 1024) # 48KB
)
barrier_count: int = 0 # For __syncthreads()class PtxSimulator:
def __init__(self, config: SimulationConfig | None = None) -> None:
self.config = config or SimulationConfig()
self.blocks: list[Block] = []
self.trace: list[TraceEvent] = []
self.instructions: list[dict[str, Any]] = []
self.pc_to_idx: dict[int, int] = {} # PTX line → instruction index
# Register tracking
self.live_registers: set[str] = set()
self.max_registers: int = 255
# Cycle counting
self.current_cycle: int = 0
self.pending_memory_ops: dict[tuple[int, int], int] = {} # (block, warp) → completion_cycle
# Cache simulation
self.l1_cache = SimpleCache(L1_CACHE_LINES, L1_CACHE_LINE_SIZE)
self.l2_cache = SimpleCache(L2_CACHE_LINES, L2_CACHE_LINE_SIZE)
def load_kernel(self, parsed_ptx: dict, kernel_name: str | None = None) -> None:
"""Load parsed PTX and initialize simulation state"""
kernel = parsed_ptx["kernels"][0] # Use first kernel
# Filter to executable instructions (skip most directives)
self.instructions = [
inst for inst in kernel["instructions"]
if inst["type"] not in ("directive",) or inst["opcode"] == ".loc"
]
# Build PC lookup table
self.pc_to_idx = {inst["line"]: i for i, inst in enumerate(self.instructions)}
# Create thread blocks based on grid dimensions
self.blocks = [
Block(
block_id=bid,
warps=[
Warp(warp_id=wid, block_id=bid)
for wid in range(self.config.warps_per_block)
],
)
for bid in range(self.config.grid_dim[0])
]
# Set initial PC for all warps
if self.instructions:
first_pc = self.instructions[0]["line"]
for block in self.blocks:
for warp in block.warps:
warp.pc = first_pc
def step(self) -> list[TraceEvent]:
"""Execute one instruction per active warp"""
events: list[TraceEvent] = []
for block in self.blocks:
for warp in block.warps:
# Skip warps waiting at barrier
if warp.at_barrier:
continue
# Skip warps that have finished
if warp.pc not in self.pc_to_idx:
continue
idx = self.pc_to_idx[warp.pc]
inst = self.instructions[idx]
# Execute instruction and generate trace event
event = self._execute_instruction(block, warp, inst)
if event:
events.append(event)
self.trace.append(event)
# Advance PC
if idx + 1 < len(self.instructions):
warp.pc = self.instructions[idx + 1]["line"]
else:
warp.pc = -1 # Mark warp as finished
return events
def run(self, max_steps: int = 1000) -> list[TraceEvent]:
"""Run until completion or max_steps"""
for _ in range(max_steps):
self.step()
# Check if all warps finished
all_finished = all(
warp.pc < 0 or warp.pc not in self.pc_to_idx
for block in self.blocks
for warp in block.warps
if not warp.at_barrier
)
if all_finished:
break
return self.traceEvery instruction execution generates a TraceEvent:
class TraceEventType(Enum):
WARP_EXECUTE = "warp_execute" # Normal instruction
MEMORY_ACCESS = "memory_access" # Load/store
DIVERGE = "diverge" # Branch divergence
RECONVERGE = "reconverge" # Threads rejoining
BARRIER_WAIT = "barrier_wait" # Hit __syncthreads()
BARRIER_RELEASE = "barrier_release" # All warps reached barrier
@dataclass(frozen=True, slots=True)
class TraceEvent:
event_type: TraceEventType
block_id: int
warp_id: int
pc: int # PTX line number
instruction: str # "add.f32 %f1, %f2, %f3"
active_mask: int # Which threads executed
source_line: int | None = None # Original CUDA line
memory_pattern: MemoryAccessPattern | None # Bank conflicts, coalescing
register_pressure: RegisterPressure | None # Live register count
divergence_info: DivergenceInfo | None # Branch divergence details
schedule_info: WarpScheduleInfo | None # Stall reasons
cache_info: CacheAccessInfo | None # L1/L2 hit/miss
@property
def active_threads(self) -> int:
"""Count of active threads (popcount of active_mask)"""
return bin(self.active_mask).count("1")Location: simulator/engine.py
# Shared Memory
SHARED_MEM_BANKS = 32 # GPUs have 32 shared memory banks
SHARED_MEM_BANK_WIDTH = 4 # Each bank is 4 bytes wide
# Global Memory
GLOBAL_CACHE_LINE_SIZE = 128 # 128-byte cache lines
GLOBAL_SECTOR_SIZE = 32 # 32-byte sectors within a cache lineShared memory is divided into 32 banks. If multiple threads access different addresses in the same bank, accesses are serialized (a bank conflict).
Bank assignment formula:
bank = (address / 4) % 32
Example addresses and their banks:
Address 0 → Bank 0
Address 4 → Bank 1
Address 128 → Bank 0 (conflict with Address 0!)
Address 132 → Bank 1 (conflict with Address 4!)
Exception: If all threads access the same address in a bank, the hardware broadcasts the value (no conflict).
def analyze_shared_memory_access(
addresses: list[int], # Address per thread
active_mask: int # Which threads are active
) -> MemoryAccessPattern:
banks: dict[int, list[tuple[int, int]]] = {} # bank → [(thread_id, address)]
for tid in range(32):
if not (active_mask >> tid) & 1:
continue
addr = addresses[tid] if tid < len(addresses) else tid * 4
bank = (addr // SHARED_MEM_BANK_WIDTH) % SHARED_MEM_BANKS
if bank not in banks:
banks[bank] = []
banks[bank].append((tid, addr))
# Analyze each bank
bank_accesses = []
max_conflict = 1
for bank_id in range(SHARED_MEM_BANKS):
if bank_id not in banks:
continue
accesses = banks[bank_id]
thread_ids = tuple(t for t, _ in accesses)
addrs = tuple(a for _, a in accesses)
# Check for broadcast (all same address)
is_broadcast = len(set(addrs)) == 1 and len(addrs) > 1
# Count conflict degree (excluding broadcasts)
if not is_broadcast and len(thread_ids) > 1:
max_conflict = max(max_conflict, len(thread_ids))
bank_accesses.append(BankAccess(
bank_id=bank_id,
thread_ids=thread_ids,
addresses=addrs,
is_broadcast=is_broadcast,
))
efficiency = 1.0 / max_conflict if max_conflict > 0 else 1.0
return MemoryAccessPattern(
memory_space=MemorySpace.SHARED,
bank_accesses=tuple(bank_accesses),
conflict_degree=max_conflict, # 1 = no conflict, 32 = worst case
is_coalesced=True, # Always coalesced for shared
transactions=max_conflict,
efficiency=efficiency, # 1.0 = perfect, 0.03125 = 32-way conflict
)# Perfect access: each thread hits a different bank
addresses = [i * 4 for i in range(32)] # [0, 4, 8, 12, ..., 124]
# Thread 0 → Bank 0, Thread 1 → Bank 1, ..., Thread 31 → Bank 31
# Result: conflict_degree=1, efficiency=1.0
# 32-way conflict: all threads hit bank 0
addresses = [i * 128 for i in range(32)] # [0, 128, 256, 384, ...]
# All addresses map to Bank 0
# Result: conflict_degree=32, efficiency=0.03125Global memory accesses are most efficient when threads access consecutive addresses that fit in one 128-byte cache line.
def analyze_global_memory_access(
addresses: list[int],
active_mask: int,
access_size: int = 4 # Bytes per thread (typically 4)
) -> MemoryAccessPattern:
cache_lines: set[int] = set()
sectors: dict[int, list[int]] = {}
for tid in range(32):
if not (active_mask >> tid) & 1:
continue
addr = addresses[tid] if tid < len(addresses) else tid * 4
# Which cache line does this address fall into?
cache_line = addr // GLOBAL_CACHE_LINE_SIZE
cache_lines.add(cache_line)
# Track sectors (32-byte chunks within cache line)
sector = (addr % GLOBAL_CACHE_LINE_SIZE) // GLOBAL_SECTOR_SIZE
if sector not in sectors:
sectors[sector] = []
sectors[sector].append(tid)
transactions = len(cache_lines) # Each unique cache line = 1 transaction
active_threads = bin(active_mask).count("1")
is_coalesced = transactions <= 1
# Efficiency: how close to ideal?
ideal_transactions = (active_threads * access_size + GLOBAL_CACHE_LINE_SIZE - 1) // GLOBAL_CACHE_LINE_SIZE
efficiency = ideal_transactions / transactions if transactions > 0 else 1.0
efficiency = min(1.0, efficiency)
return MemoryAccessPattern(
memory_space=MemorySpace.GLOBAL,
bank_accesses=tuple(...),
conflict_degree=transactions,
is_coalesced=is_coalesced,
transactions=transactions,
efficiency=efficiency,
)# Perfect coalescing: consecutive 4-byte accesses
addresses = [i * 4 for i in range(32)] # 128 bytes total, 1 cache line
# Result: transactions=1, is_coalesced=True, efficiency=1.0
# Strided access: each thread accesses a different cache line
addresses = [i * 128 for i in range(32)] # 32 cache lines!
# Result: transactions=32, is_coalesced=False, efficiency=0.03125
# Spanning two cache lines
addresses = [64 + i * 4 for i in range(32)] # Bytes 64-191
# First 16 threads: cache line 0 (bytes 64-127)
# Last 16 threads: cache line 1 (bytes 128-191)
# Result: transactions=2, efficiency=0.5Location: simulator/engine.py
# Cache parameters (modeled after Ampere/A100)
L1_CACHE_SIZE = 128 * 1024 # 128 KB per SM
L1_CACHE_LINE_SIZE = 128 # 128-byte cache lines
L1_CACHE_LINES = 1024 # 128KB / 128B = 1024 lines
L1_HIT_LATENCY = 28 # ~28 cycles
L2_CACHE_SIZE = 6 * 1024 * 1024 # 6 MB total (shared across all SMs)
L2_CACHE_LINE_SIZE = 128
L2_CACHE_LINES = 49152 # 6MB / 128B
L2_HIT_LATENCY = 150 # ~150 cycles
DRAM_LATENCY = 400 # ~400 cycles for memory missclass SimpleCache:
"""Direct-mapped cache with LRU tracking"""
def __init__(self, num_lines: int, line_size: int) -> None:
self.num_lines = num_lines
self.line_size = line_size
self.tags: dict[int, int] = {} # set_index → cache_line_tag
self.access_order: list[int] = [] # LRU tracking
self.hits = 0
self.misses = 0
def access(self, address: int) -> bool:
"""Returns True if cache hit, False if miss"""
cache_line = address // self.line_size
set_index = cache_line % self.num_lines # Direct-mapped
# Check for hit
if set_index in self.tags and self.tags[set_index] == cache_line:
self.hits += 1
# Update LRU
if set_index in self.access_order:
self.access_order.remove(set_index)
self.access_order.append(set_index)
return True
# Cache miss - install new line
self.misses += 1
self.tags[set_index] = cache_line
if set_index in self.access_order:
self.access_order.remove(set_index)
self.access_order.append(set_index)
return False
def get_hit_rate(self) -> float:
total = self.hits + self.misses
return self.hits / total if total > 0 else 0.0def _execute_instruction(self, block, warp, inst):
# ... for global memory operations ...
if mem_space == MemorySpace.GLOBAL:
representative_addr = addresses[0] # Use first thread's address
# Check L1 cache
l1_hit = self.l1_cache.access(representative_addr)
l2_hit = False
if l1_hit:
latency = L1_HIT_LATENCY # 28 cycles
else:
# L1 miss, check L2
l2_hit = self.l2_cache.access(representative_addr)
if l2_hit:
latency = L2_HIT_LATENCY # 150 cycles
else:
latency = DRAM_LATENCY # 400 cycles
cache_info = CacheAccessInfo(
l1_hit=l1_hit,
l2_hit=l2_hit,
cache_line=representative_addr // L1_CACHE_LINE_SIZE,
access_size=4 * active_count,
latency_cycles=latency,
)Location: simulator/engine.py, simulator/models.py
class WarpStallReason(Enum):
NONE = "none" # Ready to execute
MEMORY_THROTTLE = "memory_throttle" # Too many pending memory ops
MEMORY_DEPENDENCY = "memory_dependency" # Waiting for load to complete
EXECUTION_DEPENDENCY = "execution_dependency" # RAW hazard
BARRIER = "barrier" # Waiting at __syncthreads()
NOT_SELECTED = "not_selected" # Ready but scheduler picked another
DIVERGENCE = "divergence" # Handling divergent execution
MATH_PIPE = "math_pipe" # Waiting for SFU (sin, cos, sqrt)def _determine_stall_reason(
self, inst_type: str, opcode: str, warp: Warp, block: Block
) -> WarpStallReason:
warp_key = (block.block_id, warp.warp_id)
# Check if waiting at barrier
if warp.at_barrier:
return WarpStallReason.BARRIER
# Check for pending memory operations
if warp_key in self.pending_memory_ops:
if self.current_cycle < self.pending_memory_ops[warp_key]:
return WarpStallReason.MEMORY_DEPENDENCY
else:
# Memory op completed, clear it
del self.pending_memory_ops[warp_key]
# Check for memory throttling (too many ops in flight)
if inst_type == "memory":
if len(self.pending_memory_ops) >= 4:
return WarpStallReason.MEMORY_THROTTLE
# Check for expensive math operations
if opcode in ("div", "sqrt", "rsqrt", "rcp", "sin", "cos", "ex2", "lg2"):
return WarpStallReason.MATH_PIPE
return WarpStallReason.NONE@dataclass(frozen=True, slots=True)
class WarpScheduleInfo:
cycle: int # Current simulation cycle
stall_reason: WarpStallReason
ready_warps: int # How many warps could execute
active_warps: int # Total warps not finished
issued_this_cycle: bool # Did this warp issue an instruction?Location: sass/parser.py
Purpose: Parse the output of cuobjdump --dump-sass to extract actual machine code instructions.
- PTX: Architecture-independent intermediate representation
- SASS: Native GPU machine code, architecture-specific
SASS example (from cuobjdump):
Function : vector_add
/*0000*/ MOV R1, c[0x0][0x28] ;
/*0010*/ S2R R0, SR_CTAID.X ;
/*0020*/ S2R R2, SR_TID.X ;
/*0030*/ IMAD R0, R0, c[0x0][0x0], R2 ;
/*0040*/ @P0 EXIT ;
/*0050*/ ISETP.GE.AND P0, PT, R0, c[0x0][0x168], PT ;
/*0060*/ IMAD.WIDE R2, R0, 0x4, c[0x0][0x160] ;
/*0070*/ LDG.E R2, [R2.64] ;
/*0080*/ IMAD.WIDE R4, R0, 0x4, c[0x0][0x158] ;
/*0090*/ LDG.E R0, [R4.64] ;
/*00a0*/ FADD R0, R2, R0 ;
/*00b0*/ STG.E [R2.64], R0 ;
/*00c0*/ EXIT ;
KERNEL_HEADER_PATTERN = re.compile(r"Function\s*:\s*(\w+)")
INSTRUCTION_PATTERN = re.compile(
r"/\*([0-9a-fA-F]+)\*/\s+" # Address: /*0030*/
r"(@[!]?\w+)?\s*" # Optional predicate: @P0
r"(\S+)\s*" # Opcode: IMAD.WIDE
r"(.*?);\s*$" # Operands and semicolon
)
def parse_sass(sass_output: str) -> dict:
kernels: list[SassKernel] = []
current_kernel: SassKernel | None = None
for line in sass_output.split("\n"):
line = line.strip()
# Check for kernel header
kernel_match = KERNEL_HEADER_PATTERN.match(line)
if kernel_match:
if current_kernel:
kernels.append(current_kernel)
current_kernel = SassKernel(name=kernel_match.group(1))
continue
# Parse instruction
if current_kernel:
inst_match = INSTRUCTION_PATTERN.match(line)
if inst_match:
address = inst_match.group(1) # "0030"
predicate = inst_match.group(2) # "@P0" or None
opcode_full = inst_match.group(3) # "IMAD.WIDE"
operands_str = inst_match.group(4) # "R0, R0, c[0x0][0x0], R2"
opcode_parts = opcode_full.split(".")
opcode = opcode_parts[0] # "IMAD"
modifiers = tuple(opcode_parts[1:]) # ("WIDE",)
operands = tuple(op.strip() for op in operands_str.split(","))
current_kernel.instructions.append(SassInstruction(
address=address,
opcode=opcode,
operands=operands,
predicate=predicate,
modifiers=modifiers,
raw=line,
))
if current_kernel:
kernels.append(current_kernel)
return {"kernels": [k.to_dict() for k in kernels]}Location: occupancy/calculator.py
Purpose: Calculate SM (Streaming Multiprocessor) occupancy based on kernel resource usage.
Occupancy = Active Warps / Maximum Warps per SM
Higher occupancy generally helps hide memory latency, but isn't always necessary for optimal performance.
GPU_SPECS: dict[str, GpuSpecs] = {
"sm_75": GpuSpecs(
name="Turing (T4)",
compute_capability="7.5",
max_threads_per_sm=1024, # 32 warps
max_warps_per_sm=32,
max_blocks_per_sm=16,
max_threads_per_block=1024,
registers_per_sm=65536, # 64K 32-bit registers
max_registers_per_thread=255,
shared_memory_per_sm=65536, # 64KB
max_shared_memory_per_block=49152, # 48KB
),
"sm_80": GpuSpecs(
name="Ampere (A100)",
compute_capability="8.0",
max_threads_per_sm=2048, # 64 warps
max_warps_per_sm=64,
max_blocks_per_sm=32,
registers_per_sm=65536,
shared_memory_per_sm=167936, # ~164KB configurable
max_shared_memory_per_block=163840, # ~160KB
),
"sm_86": GpuSpecs(
name="Ampere (RTX 3090)",
compute_capability="8.6",
max_threads_per_sm=1536, # 48 warps
max_warps_per_sm=48,
max_blocks_per_sm=16,
...
),
"sm_89": GpuSpecs(
name="Ada (RTX 4090)",
compute_capability="8.9",
max_threads_per_sm=1536,
max_warps_per_sm=48,
max_blocks_per_sm=24,
...
),
"sm_90": GpuSpecs(
name="Hopper (H100)",
compute_capability="9.0",
max_threads_per_sm=2048,
max_warps_per_sm=64,
max_blocks_per_sm=32,
shared_memory_per_sm=232448, # ~227KB
...
),
}Three resources limit occupancy:
- Registers: Each thread uses registers, and the SM has limited register file
- Shared Memory: Each block uses shared memory
- Block Limit: Maximum concurrent blocks per SM
def calculate_occupancy(
arch: str,
threads_per_block: int,
registers_per_thread: int,
shared_memory_per_block: int = 0,
grid_dim: tuple[int, int, int] = (1, 1, 1),
) -> OccupancyResult:
specs = get_gpu_specs(arch)
warps_per_block = (threads_per_block + 31) // 32
# 1. Limit from registers
if registers_per_thread > 0:
# Registers are allocated in groups of 256
regs_per_warp = registers_per_thread * 32
regs_per_warp = ((regs_per_warp + 255) // 256) * 256
warps_by_regs = specs.registers_per_sm // regs_per_warp
else:
warps_by_regs = specs.max_warps_per_sm
# 2. Limit from shared memory
if shared_memory_per_block > 0:
# Shared memory allocated in 256-byte chunks
smem_per_block = ((shared_memory_per_block + 255) // 256) * 256
blocks_by_smem = specs.shared_memory_per_sm // smem_per_block
warps_by_smem = blocks_by_smem * warps_per_block
else:
warps_by_smem = specs.max_warps_per_sm
# 3. Limit from block size
blocks_by_threads = specs.max_threads_per_sm // threads_per_block
blocks_by_limit = specs.max_blocks_per_sm
max_blocks = min(blocks_by_threads, blocks_by_limit)
warps_by_block_size = max_blocks * warps_per_block
# Take minimum of all limits
active_warps = min(warps_by_regs, warps_by_smem, warps_by_block_size)
active_warps = min(active_warps, specs.max_warps_per_sm)
# Determine limiting factor
if active_warps == warps_by_regs:
limiting = OccupancyLimiter.REGISTERS
elif active_warps == warps_by_smem:
limiting = OccupancyLimiter.SHARED_MEMORY
else:
limiting = OccupancyLimiter.BLOCK_SIZE
occupancy_percent = (active_warps / specs.max_warps_per_sm) * 100
return OccupancyResult(
active_warps_per_sm=active_warps,
max_warps_per_sm=specs.max_warps_per_sm,
occupancy_percent=occupancy_percent,
limiting_factor=limiting,
...
)Architecture: sm_80 (A100)
- Max warps per SM: 64
- Registers per SM: 65536
Kernel:
- 256 threads per block (8 warps)
- 32 registers per thread
Calculation:
- Regs per warp: 32 * 32 = 1024, rounded to 1024
- Warps by regs: 65536 / 1024 = 64 warps
- Warps by block size: min(2048/256, 32) * 8 = 8 * 8 = 64 warps
- Active warps: min(64, 64) = 64
- Occupancy: 64 / 64 = 100%
Location: latency/model.py
Purpose: Provide per-instruction timing information for each GPU architecture.
@dataclass(frozen=True, slots=True)
class InstructionTiming:
latency_cycles: int # Cycles until result is available
throughput_per_sm: float # Operations per cycle per SM
pipe: str # Which execution unit
notes: str = "" # Human-readable description
INSTRUCTION_TIMINGS["sm_80"] = {
# Integer ALU (4 cycles latency, 64 ops/cycle throughput)
"add.s32": InstructionTiming(4, 64, "INT", "32-bit integer add"),
"add.s64": InstructionTiming(4, 32, "INT", "64-bit integer add"),
"mul.lo.s32": InstructionTiming(4, 64, "INT", "32-bit multiply low"),
# Floating-point (4 cycles latency)
"add.f32": InstructionTiming(4, 64, "FP32", "32-bit float add"),
"mul.f32": InstructionTiming(4, 64, "FP32", "32-bit float multiply"),
"fma.rn.f32": InstructionTiming(4, 64, "FP32", "Fused multiply-add"),
# Double precision (8 cycles, half throughput)
"add.f64": InstructionTiming(8, 32, "FP64", "64-bit float add"),
"mul.f64": InstructionTiming(8, 32, "FP64", "64-bit float multiply"),
# Special Function Unit (transcendentals)
"div.approx.f32": InstructionTiming(16, 16, "SFU", "Fast approximate divide"),
"div.rn.f32": InstructionTiming(36, 2, "SFU", "IEEE-compliant divide"),
"sqrt.approx.f32": InstructionTiming(8, 16, "SFU", "Fast approximate sqrt"),
"sin.approx.f32": InstructionTiming(8, 16, "SFU", "Fast sine"),
"cos.approx.f32": InstructionTiming(8, 16, "SFU", "Fast cosine"),
"rsqrt.approx.f32": InstructionTiming(8, 16, "SFU", "Fast reciprocal sqrt"),
# Memory operations (highly variable)
"ld.global": InstructionTiming(200, 0, "MEM", "Global memory ~200 cycles"),
"st.global": InstructionTiming(200, 0, "MEM", "Global memory store"),
"ld.shared": InstructionTiming(23, 32, "LDST", "Shared memory ~23 cycles"),
"st.shared": InstructionTiming(23, 32, "LDST", "Shared memory store"),
"ld.const": InstructionTiming(4, 64, "LDST", "Constant memory (cached)"),
# Atomics
"atom.global.add": InstructionTiming(200, 0, "MEM", "Global atomic add"),
"atom.shared.add": InstructionTiming(23, 8, "LDST", "Shared atomic add"),
# Warp shuffle
"shfl.sync": InstructionTiming(4, 64, "SHFL", "Warp shuffle"),
# Control flow
"bra": InstructionTiming(4, 64, "CTRL", "Unconditional branch"),
"bar.sync": InstructionTiming(20, 0, "CTRL", "Block-level barrier"),
...
}# RTX 4090 (Ada) has 2x FP32 throughput
INSTRUCTION_TIMINGS["sm_89"] = {
**INSTRUCTION_TIMINGS["sm_80"],
"add.f32": InstructionTiming(4, 128, "FP32", "32-bit float (2x throughput)"),
"mul.f32": InstructionTiming(4, 128, "FP32", "32-bit float (2x throughput)"),
"fma.rn.f32": InstructionTiming(4, 128, "FP32", "FMA (2x throughput)"),
}
# H100 (Hopper) has faster shared memory
INSTRUCTION_TIMINGS["sm_90"] = {
**INSTRUCTION_TIMINGS["sm_89"],
"ld.shared": InstructionTiming(20, 64, "LDST", "Shared memory (faster)"),
}PIPE_COLORS = {
"INT": "#4ade80", # Green
"FP32": "#60a5fa", # Blue
"FP64": "#a78bfa", # Purple
"SFU": "#f472b6", # Pink
"MEM": "#fb923c", # Orange
"LDST": "#fbbf24", # Yellow
"CTRL": "#94a3b8", # Gray
"SHFL": "#22d3d8", # Cyan
"UNKNOWN": "#6b7280", # Dark gray
}def annotate_trace_with_latency(
trace: list[dict],
arch: str = "sm_80"
) -> list[dict]:
"""Add timing info to each trace event"""
annotated = []
cumulative_cycles = 0
for event in trace:
instruction = event.get("instruction", "")
opcode = instruction.split()[0] if instruction else ""
timing = get_instruction_latency(opcode, arch)
annotated_event = {
**event,
"timing": timing.to_dict(),
"pipe": timing.pipe,
"pipe_color": PIPE_COLORS.get(timing.pipe, PIPE_COLORS["UNKNOWN"]),
"cumulative_cycles": cumulative_cycles,
}
cumulative_cycles += timing.latency_cycles
annotated.append(annotated_event)
return annotatedLocation: hints/analyzer.py
Purpose: Automatically detect performance issues and provide optimization suggestions.
class HintSeverity(Enum):
INFO = "info" # Minor suggestion
WARNING = "warning" # Notable performance impact
CRITICAL = "critical" # Severe performance problem
class HintCategory(Enum):
MEMORY = "memory" # Bank conflicts, coalescing
DIVERGENCE = "divergence" # Branch divergence
OCCUPANCY = "occupancy" # Low SM utilization
REGISTERS = "registers" # High register pressuredef analyze_trace_hints(
trace: list[dict],
occupancy: dict | None = None,
) -> list[dict]:
hints: list[PerformanceHint] = []
# Count issues across entire trace
bank_conflict_events = 0
uncoalesced_events = 0
divergent_events = 0
high_pressure_events = 0
for idx, event in enumerate(trace):
# Analyze memory accesses
if event.get("type") == "memory_access":
pattern = event.get("memory_pattern", {})
if pattern.get("memory_space") == "shared":
conflict = pattern.get("conflict_degree", 1)
if conflict > 1:
bank_conflict_events += 1
# Individual event hint for severe conflicts
if conflict >= 8:
hints.append(PerformanceHint(
category=HintCategory.MEMORY,
severity=HintSeverity.CRITICAL,
message=f"{conflict}-way bank conflict",
detail="Severe bank conflict causing serialized access",
event_index=idx,
source_line=event.get("source_line"),
))
elif conflict > 1:
hints.append(PerformanceHint(
category=HintCategory.MEMORY,
severity=HintSeverity.WARNING,
message=f"{conflict}-way bank conflict",
detail="Bank conflict reducing throughput",
event_index=idx,
source_line=event.get("source_line"),
))
elif pattern.get("memory_space") == "global":
efficiency = pattern.get("efficiency", 1.0)
if efficiency < 0.25:
hints.append(PerformanceHint(
category=HintCategory.MEMORY,
severity=HintSeverity.CRITICAL,
message=f"Poor memory efficiency ({int(efficiency*100)}%)",
detail="Scattered access causing many transactions",
event_index=idx,
))
elif efficiency < 0.5:
uncoalesced_events += 1
# Track divergence
if event.get("type") == "diverge":
divergent_events += 1
div_info = event.get("divergence_info", {})
if div_info.get("divergence_ratio", 0) > 0.4:
hints.append(PerformanceHint(
category=HintCategory.DIVERGENCE,
severity=HintSeverity.WARNING,
message=f"High divergence ({int(div_info['divergence_ratio']*100)}%)",
detail="Significant warp divergence at this branch",
event_index=idx,
))
# Track register pressure
pressure = event.get("register_pressure", {})
if pressure.get("pressure_ratio", 0) > 0.7:
high_pressure_events += 1
# Summary hints
if bank_conflict_events > 5:
hints.append(PerformanceHint(
category=HintCategory.MEMORY,
severity=HintSeverity.WARNING,
message=f"Shared memory bank conflicts ({bank_conflict_events} events)",
detail="Consider padding arrays or restructuring access patterns",
))
if divergent_events > 3:
hints.append(PerformanceHint(
category=HintCategory.DIVERGENCE,
severity=HintSeverity.WARNING,
message=f"Branch divergence detected ({divergent_events} events)",
detail="Consider restructuring conditionals or using predication",
))
# Occupancy hints
if occupancy:
occ_percent = occupancy.get("occupancy_percent", 100)
limiting = occupancy.get("limiting_factor", "")
if occ_percent < 25:
hints.append(PerformanceHint(
category=HintCategory.OCCUPANCY,
severity=HintSeverity.CRITICAL,
message=f"Very low occupancy ({occ_percent}%)",
detail=f"Limited by {limiting}. Consider reducing resource usage.",
))
elif occ_percent < 50:
hints.append(PerformanceHint(
category=HintCategory.OCCUPANCY,
severity=HintSeverity.WARNING,
message=f"Low occupancy ({occ_percent}%)",
detail=f"Limited by {limiting}.",
))
return [h.to_dict() for h in hints]Location: api/app.py, api/routes.py, api/schemas.py
def create_app() -> FastAPI:
app = FastAPI(
title="KernelScope API",
description="CUDA Compiler Explorer - Compile, Parse, and Simulate",
version="0.1.0",
openapi_tags=[
{"name": "Compilation", "description": "Compile CUDA to PTX/SASS"},
{"name": "Library", "description": "Sample kernel library"},
{"name": "System", "description": "Health checks"},
],
)
# CORS for frontend
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["*"],
allow_headers=["*"],
)
app.include_router(router, prefix="/api")
return app@router.post("/compile")
async def compile_cuda(request: CompileRequest) -> CompileResponse:
"""
Complete CUDA → Analysis pipeline:
1. Send CUDA to Modal backend for nvcc compilation
2. Parse PTX locally
3. Optionally parse SASS
4. Simulate execution
5. Calculate occupancy
6. Analyze for hints
7. Annotate with latency info
"""
source = request.source.strip()
sim_config = _get_sim_config(request.config)
# Call Modal GPU backend
async with httpx.AsyncClient(timeout=60.0) as client:
response = await client.post(
f"{MODAL_BACKEND_URL}/compile",
json={
"source": source,
"arch": request.arch,
"include_sass": request.include_sass,
},
)
if response.status_code == 200:
data = response.json()
if data.get("success") and data.get("ptx"):
ptx = data["ptx"]
sass = data.get("sass")
# Parse PTX locally
parsed = parse_ptx(ptx)
# Parse SASS if available
parsed_sass = None
if sass:
parsed_sass = parse_sass(sass)
# Simulate execution
simulation = simulate_kernel(parsed, sim_config)
simulation["trace"] = simulation["trace"][:500] # Limit trace size
# Add latency annotations
simulation["trace"] = annotate_trace_with_latency(
simulation["trace"], request.arch
)
# Calculate occupancy
occupancy = _calculate_occupancy(parsed, request.arch, sim_config)
# Analyze for performance hints
hints = analyze_trace_hints(simulation["trace"], occupancy)
return CompileResponse(
success=True,
ptx=ptx,
sass=sass,
parsed=parsed,
parsed_sass=parsed_sass,
simulation=simulation,
occupancy=occupancy,
hints=hints,
warnings=data.get("warnings"),
)
else:
return CompileResponse(
success=False,
errors=data.get("errors", ["Compilation failed"]),
)
# Fallback for offline testing
return CompileResponse(
success=False,
errors=["Modal backend unavailable"],
)class CompileRequest(BaseModel):
source: str # CUDA source code
arch: str = "sm_80" # Target: sm_75, sm_80, sm_86, sm_89, sm_90
include_sass: bool = False # Include machine code
config: dict | None = None # {grid_dim: [1,1,1], block_dim: [64,1,1]}
class CompileResponse(BaseModel):
success: bool
ptx: str | None = None # PTX assembly
sass: str | None = None # SASS machine code
parsed: dict | None = None # Parsed PTX structure
parsed_sass: dict | None = None # Parsed SASS structure
simulation: dict | None = None # Execution trace
occupancy: dict | None = None # SM occupancy analysis
hints: list[dict] | None = None # Performance suggestions
errors: list[str] | None = None # Compilation errors
warnings: list[str] | None = None # Compilation warnings┌─────────────────────────────────────────────────────────────────────────────┐
│ Frontend (React) │
│ │ │
│ POST /api/compile │
│ {source, arch, config} │
└─────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────┐
│ routes.py │
│ │
│ 1. Forward CUDA to Modal backend │
│ │
│ ┌─────────────────────────────────────────────┐ │
│ │ Modal GPU Backend │ │
│ │ - nvcc compilation │ │
│ │ - cuobjdump for SASS │ │
│ │ Returns: PTX, SASS, warnings, errors │ │
│ └─────────────────────────────────────────────┘ │
│ │ │
│ 2. Parse PTX locally ◄─────────────┘ │
│ parse_ptx() → PtxModule (kernels, instructions, source_map) │
│ │ │
│ 3. Parse SASS (optional) │ │
│ parse_sass() → SassKernel (instructions) │
│ │ │
│ 4. Simulate execution │ │
│ simulate_kernel() → trace: List[TraceEvent] │
│ - Memory access patterns (bank conflicts, coalescing) │
│ - Register pressure tracking │
│ - Divergence detection │
│ - Cache simulation (L1/L2 hits) │
│ - Warp scheduling state │
│ │ │
│ 5. Annotate with latency │ │
│ annotate_trace_with_latency() → timing, pipe, cumulative_cycles │
│ │ │
│ 6. Calculate occupancy │ │
│ calculate_occupancy() → active_warps, limiting_factor │
│ │ │
│ 7. Analyze hints │ │
│ analyze_trace_hints() → warnings, suggestions │
│ │ │
│ 8. Return CompileResponse ─────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────────┘
│
▼
┌─────────────────────────────────────────────────────────────────────────────┐
│ Frontend (React) │
│ │
│ - Display PTX in editor with syntax highlighting │
│ - Display SASS in editor │
│ - Step through trace, visualize thread states │
│ - Show bank conflicts, coalescing efficiency │
│ - Show occupancy chart │
│ - Display performance hints │
└─────────────────────────────────────────────────────────────────────────────┘
A warp is a group of 32 threads that execute in lockstep. All threads in a warp execute the same instruction simultaneously (SIMT - Single Instruction, Multiple Threads).
Warp 0: threads 0-31 execute instruction N
Warp 1: threads 32-63 execute instruction M (may be different)
Warp 2: threads 64-95 execute instruction P
A 32-bit integer indicating which threads in a warp are currently active:
0xFFFFFFFF = all 32 threads active
0x0000FFFF = only threads 0-15 active
0x00000001 = only thread 0 activeThreads become inactive due to:
- Thread ID >= block size
- Divergent branches (some threads took different path)
- Thread exited early
Shared memory is organized into 32 banks. Accessing different addresses in the same bank causes serialization:
Bank 0: addresses 0, 128, 256, 384, ...
Bank 1: addresses 4, 132, 260, 388, ...
Bank 2: addresses 8, 136, 264, 392, ...
...
Bank 31: addresses 124, 252, 380, 508, ...
No conflict: Each thread accesses different bank Broadcast: All threads access same address (hardware broadcasts) N-way conflict: N threads access different addresses in same bank → N serial accesses
Global memory is accessed in 128-byte cache lines. Coalescing combines multiple thread accesses into fewer memory transactions:
Coalesced: 32 threads access consecutive 4-byte values = 1 transaction Uncoalesced: 32 threads access scattered addresses = up to 32 transactions
When threads in a warp take different paths at a branch, the warp must execute both paths serially:
if (threadIdx.x < 16) {
// Path A: threads 0-15 active, 16-31 masked
a();
} else {
// Path B: threads 16-31 active, 0-15 masked
b();
}
// Reconvergence: all threads active againThe ratio of active warps to the maximum warps a SM can support:
Occupancy = Active Warps / Max Warps per SM
A100 (sm_80): max 64 warps
If kernel runs 32 warps: 32/64 = 50% occupancy
Higher occupancy helps hide memory latency but isn't always required for optimal performance.
Each thread uses registers (up to 255 per thread). High register usage limits occupancy:
A100: 65536 registers per SM
If kernel uses 128 registers per thread:
- 128 regs × 32 threads/warp = 4096 registers/warp
- 65536 / 4096 = 16 warps maximum
- Occupancy limited to 16/64 = 25%