GPU共享内存机制：CUDA编程模型中的性能关键

# 共享内存容量配置查询(Python模拟)# NVIDIA GPU中共享内存与L1缓存共享SRAMshared_mem_kb = 48   # 默认48KB共享内存l1_cache_kb = 16     # 默认16KB L1缓存total_sram = shared_mem_kb + l1_cache_kbprint(f"片上SRAM总量: {total_sram}KB")print(f"可配置: 48KB共享+16KB缓存 或 32KB+32KB")

# 共享内存声明与生命周期(Python模拟CUDA概念)# 静态声明: __shared__ float tile[16][16]# 动态声明: extern __shared__ float dynamic_tile[] BLOCK_SIZE = 16# 模拟线程块内共享内存分配tile_a = [[0.0] * BLOCK_SIZE for _ in range(BLOCK_SIZE)]tile_b = [[0.0] * BLOCK_SIZE for _ in range(BLOCK_SIZE)]# 块内线程协作加载数据tid = 0  # 线程IDtile_a[tid // BLOCK_SIZE][tid % BLOCK_SIZE] = 1.0# __syncthreads() 同步后才能读取print(f"共享内存tile大小: {BLOCK_SIZE}x{BLOCK_SIZE} = {BLOCK_SIZE*BLOCK_SIZE}个元素")

# Bank conflict检测与padding优化(Python模拟)BLOCK_SIZE = 16bank_count = 32 # 无padding: 列访问可能冲突for col in range(BLOCK_SIZE):    bank = (col * BLOCK_SIZE) % bank_count    # 同一列的所有行映射到同一bank # Padding优化: 增加一列消除冲突PADDED_SIZE = BLOCK_SIZE + 1  # 17列for col in range(BLOCK_SIZE):    bank = (col * PADDED_SIZE) % bank_count    # padding后列访问分散到不同bankprint(f"无padding列宽: {BLOCK_SIZE}, padding后: {PADDED_SIZE}")print("Padding使列访问分散到不同bank消除冲突")

# 双缓冲技术模拟(Python模拟CUDA概念)BUFFER_COUNT = 2TILE_SIZE = 16# 分配两个缓冲区buffers = [[[0.0]*TILE_SIZE for _ in range(TILE_SIZE)]            for _ in range(BUFFER_COUNT)] current_buf = 0for step in range(3):    next_buf = 1 - current_buf    # 计算当前缓冲区数据(同时预加载下一批到另一缓冲区)    print(f"Step {step}: 计算buffer[{current_buf}], 预加载buffer[{next_buf}]")    current_buf = next_buf  # 交换缓冲区print("双缓冲使计算与数据加载重叠隐藏同步开销")

# Tile化矩阵乘法的全局内存访问减少量计算TILE = 16N = 1024  # 矩阵维度 # 无tiling: 每个输出元素需读取N个A元素和N个B元素access_without_tiling = 2 * N # 有tiling: 每个输出元素需读取N/TILE次共享内存加载# 每次加载TILE个元素由TILE*TILE个线程复用access_with_tiling = 2 * N / TILE  # 每线程的全局内存访问 reduction = access_without_tiling / access_with_tilingprint(f"矩阵维度N={N}, TILE={TILE}")print(f"无tiling每元素全局访问: {access_without_tiling}")print(f"有tiling每元素全局访问: {access_with_tiling}")print(f"全局内存访问减少: {reduction}倍")

import json  class SharedMemSimulator:    """GPU共享内存tiling矩阵乘法模拟器"""     def __init__(self, tile_size, bank_count=32):        self.tile_size = tile_size          # tile边长        self.bank_count = bank_count        # 共享内存bank数        # 模拟共享内存(两个tile缓冲区)        self.tile_a = [[0.0] * tile_size for _ in range(tile_size)]        self.tile_b = [[0.0] * tile_size for _ in range(tile_size)]     def check_bank_conflict(self, access_pattern, padded_width=None):        """检测bank conflict"""        width = padded_width or self.tile_size        bank_access = {}        for thread_id, offset in enumerate(access_pattern):            addr = offset            bank = (addr // 4) % self.bank_count  # 每个bank 4字节            if bank not in bank_access:                bank_access[bank] = []            bank_access[bank].append(thread_id)        max_conflict = max(len(v) for v in bank_access.values())        return max_conflict, bank_access     def tiled_matmul(self, A, B):        """模拟tile化矩阵乘法"""        N = len(A)        TILE = self.tile_size        C = [[0.0] * N for _ in range(N)]         global_access_naive = 0        global_access_tiled = 0         for i in range(0, N, TILE):            for j in range(0, N, TILE):                # 加载tile到共享内存                for ti in range(TILE):                    for tj in range(TILE):                        if i + ti < N and j + tj < N:                            self.tile_a[ti][tj] = A[i + ti][j + tj]                            self.tile_b[ti][tj] = B[i + ti][j + tj]                            global_access_tiled += 2                # __syncthreads() 模拟                # 从共享内存计算部分结果                for ki in range(0, N, TILE):                    for ti in range(TILE):                        for tj in range(TILE):                            for tk in range(TILE):                                if (i+ti < N and j+tj < N and                                        ki+tk < N):                                    C[i+ti][j+tj] += (self.tile_a[ti][tk] *                                                      self.tile_b[tk][tj])         # 朴素方法全局访问次数        for i in range(N):            for j in range(N):                global_access_naive += 2 * N         return C, global_access_naive, global_access_tiled     def compute_access_reduction(self, matrix_size):        """计算tiling带来的全局内存访问减少倍数"""        # 朴素: 每个输出元素读2N次全局内存        naive = 2 * matrix_size        # Tiling: 每个输出元素读2N/TILE次(被TILE*TILE线程复用)        tiled = 2 * matrix_size / self.tile_size        return round(naive / tiled, 1)  # 运行模拟sim = SharedMemSimulator(tile_size=16)print("=== 共享内存Bank Conflict检测 ===")# 列访问模式(无padding)col_access = [col * 16 for col in range(16)]conflict_no_pad, _ = sim.check_bank_conflict(col_access)print(f"  无padding列访问最大冲突数: {conflict_no_pad}")# 列访问模式(有padding, 17列)col_access_padded = [col * 17 for col in range(16)]conflict_pad, _ = sim.check_bank_conflict(col_access_padded, 17)print(f"  有padding(17列)列访问最大冲突数: {conflict_pad}") print("\n=== Tile化矩阵乘法模拟 ===")N = 32A = [[float(i * N + j) for j in range(N)] for i in range(N)]B = [[1.0 if i == j else 0.0 for j in range(N)] for i in range(N)]C, naive_access, tiled_access = sim.tiled_matmul(A, B)print(f"  矩阵大小: {N}x{N}, TILE: {sim.tile_size}")print(f"  朴素方法全局访问: {naive_access}")print(f"  Tiling方法全局访问: {tiled_access}")print(f"  访问减少倍数: {round(naive_access/tiled_access, 1)}x") print("\n=== 不同TILE大小的访问减少 ===")for tile in [8, 16, 32]:    s = SharedMemSimulator(tile_size=tile)    reduction = s.compute_access_reduction(1024)    print(f"  TILE={tile}: 全局访问减少{reduction}x")