[Embedded System Lab] Chapter 6 - Systolic Array Theory
"Systolic Array Theory"
[Objective]
HW Architecture의 연산 Mechanism을 SW(Python) Simulation을 통해 분석하고 이해하기
[Theory]
▣ 1D Systolic Array
> Algorithm Intensity
· High Operational Intensity : Deep Learning의 핵심인 GEMM 연산은 연산 횟수(Ops)가 데이터 양에 비해 압도적으로 많음
· 이론적으로 GEMM은 Compute-Bound에 해당
→ But, 실제 HW에서는 메모리 대역폭의 한계로 인해 'Memory-Bound' 구간에 머물며 최대 성능을 낼 수 없음
※ Memory-Bound (메모리 대역폭 제한)
· HW의 성능이 메모리 대역폭(Bandwidth)의 한계로 인해 제한되는 상태
· 연산 장치는 매우 빠르지만, 데이터를 메모리에서 가져오는 속도가 느려 ALU(연산 Unit)가 데이터를 기다리며 노는 시간 발생
→ 데이터 전송에 소모되는 에너지와 지연 시간이 실제 연산 비용보다 월등히 크기 때문에 발생
→ Von Neumann Bottleneck
※ Compute-Bound (연산 중심적)
· HW의 최대 성능이 Processor의 연산 속도(FLOPS)에 의해 제한되는 상태
· 데이터 공급 속도가 충분히 빨라서, ALU가 쉬지 않고 돌아갈 때 발생
→ GEMM은 이론적으로 이 구간에 해당
> 데이터 이동 비용의 비대칭성 (Energy & Latency)
· Data Movement Cost : 데이터 전송에 소모되는 에너지가 연산에 사용되는 에너지에 비해 월등히 큼
· Von Neumann Bottleneck : CPU와 메모리가 분리된 구조에서는 매 연산마다 발생하는 Bus Traffic이 에너지 효율과 성능의 근본적인 한계로 작용
> HW적 설계 목표 - Data Locality 최적화
· Temporal & Spatial Locality : 데이터가 한 번 ALU로 들어왔을 때, 최대한 활용하기 위해 Reuse 할 수 있도록 설계
> 병렬 연산의 기초 - 1D Systolic Array (GEMV 가속)
▼
▼
▼
▼
· Single-Dimensional Flow : 데이터를 한 줄로 세워진 PE(연산기)에 통과시키는 구조
· GEMV : Vector x의 요소들이 Pipeline을 타고 한 칸씩 이동하며 행렬 A의 각 행과 연산됨
→ 각 PE는 연산 후 결과를 바로 메모리에 Write하지 않고, 다음 Clock에 옆 PE로 데이터를 밀어줌
→ 이 과정에서 데이터 x는 메모리에서 한 번만 읽히고, 모든 PE를 거치며 Reuse
** HW Insight : Memory Access를 Local Data Passing으로 대체하여 Bandwidth 절약 가능
▣ 2D Systolic Array
> Concept - From Line(Vector) to Mesh
· 1D 구조를 통해 Vector 연산을 진행했다면, 2D Mesh는 행렬 A와 B의 모든 원소를 평면상에서 교차시켜 결과 행렬 C를 생성
※ PE Grid : 각 PE는 결과 행렬의 부분합을 계산하고 저장하는 전용 공간이 됨
> Mechanism - 데이터의 교차와 동기화
· Horizontal Streaming (행렬 A) : 행렬 A의 i번째 행은 왼쪽에서 오른쪽으로 흐르며, 해당 행의 모든 PE에 값을 전달
· Vector Streaming (행렬 B) : 행렬 B의 j번째 열은 위에서 아래로 흐르며, 해당 열의 모든 PE에 값을 전달
· "Skewed" Input : 데이터가 이동하는 시간이 필요하기 때문에 모든 PE가 동시에 연산 시작 불가
→ 데이터가 정확한 타이밍에 특정 PE에 만나도록 입력을 계단식(Skewed)으로 넣어줌(Temporal Alignment)
> 2차원 데이터의 Reuse
· Weight/Input Reuse : 메모리에서 행렬 A를 한 번 읽으면 N개의 열 PE가 공유하고, B를 한 번 읽으면 M개의 행 PE가 공유
· Local Accumulation : 부분합을 외부 메모리에 매번 Write하지 않고, PE 내부 Register에 보관하며 연산을 누적
▼
▼
▼
▼
▣ Adder Tree
> 분산 연산, 집중 집계 (Map-Reduce)
· Broadcasting : 입력 데이터를 모든 ALU에 동시에 뿌려줌
· Parallel Multiplication : 수천 개의 곱셈이 단 한 Clock에 병렬적으로 발생
· Tree-Based Reduction : 연산기들 사이에 물리적인 Binary Tree 망을 구축하여, 발생한 모든 곱셈 결과를 즉시 수렴시킴
> HW적 특징 (Interconnect Perspective)
· Parallel Throughput : Pipeline Register를 Tree 각 층에 배치하여, 매 Clock마다 새로운 Vector 내적 결과를 연산할 수 있는 연산기 생성 가능
· Low Latency : ALU가 확장되어도 결과 도출까지 걸리는 시간 적음
▼
▼
▼
▼
▣ Systolic Array vs Adder Tree
[Experiment]
▣ import
1 2 3 | import numpy as np import time from IPython.display import clear_output, display | cs |
1 : 행렬 생성, 행렬 곱셈, 배열 초기화에 사용할 "numpy" Module import
2 : 시간 지연을 설정하기 위한 "time" Module import
3 : 이전 출력 화면을 지우고 새로운 결과를 출력하여 Simulation 결과를 애니메이션처럼 보이게 하기 위해 import
▣ Systolic Array Simulator
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 | class SystolicArraySimulator: def __init__(self, size): self.size = size # Internal state of PEs: {'acc': accumulator, 'a': current_a, 'b': current_b} self.pe_grid = [[{'acc': 0, 'a': 0, 'b': 0} for _ in range(size)] for _ in range(size)] def _format_input_buffers(self, A, B): """ Skew logic: A[r, k] enters row r at cycle r + k B[k, c] enters col c at cycle c + k """ # Total cycles needed for all data to pass through: ~3 * size total_time = 3 * self.size # a_stream[row][time], b_stream[col][time] a_stream = np.full((self.size, total_time), None, dtype=object) b_stream = np.full((self.size, total_time), None, dtype=object) for r in range(self.size): for c in range(self.size): # A streams in from the left: row 'r' at time 'r + c' a_stream[r, r + c] = A[r, c] # B streams in from the top: col 'c' at time 'c + r' b_stream[c, c + r] = B[r, c] return a_stream, b_stream def _format_horizontal_matrices(self, A, B, Res): """Helper to string-ify matrices and place them side-by-side.""" a_lines = str(A).split('\n') b_lines = str(B).split('\n') r_lines = str(Res).split('\n') # Determine height for padding (in case of weird formatting) max_h = max(len(a_lines), len(b_lines), len(r_lines)) # Pad lines to equal height a_lines += [""] * (max_h - len(a_lines)) b_lines += [""] * (max_h - len(b_lines)) r_lines += [""] * (max_h - len(r_lines)) output = [] output.append(f"{'Matrix A':<20} {'Matrix B':<20} {'Expected (A@B)':<20}") output.append("-" * 80) for i in range(max_h): row = f"{a_lines[i]:<20} {b_lines[i]:<20} = {r_lines[i]:<20}" output.append(row) return "\n".join(output) def run(self, A, B): a_stream, b_stream = self._format_input_buffers(A, B) total_cycles = a_stream.shape[1] expected = np.matmul(A, B) # Generate the static side-by-side string once matrix_header = self._format_horizontal_matrices(A, B, expected) # --- STATIC DISPLAY SECTION --- # This part stays at the top static_header = f""" {'='*80} SYSTOLIC ARRAY GEMM SIMULATION: {self.size}x{self.size} {'='*80} {matrix_header} {'='*80} CYCLE LEVEL SIMULATION {'='*80} """ for t in range(total_cycles): # 1. Generate the animation frame display_str = self._generate_display(t, a_stream, b_stream) # 2. Refresh the whole output but rewrite static info + new frame clear_output(wait=True) print(static_header) print(display_str) # 3. Logic: Update Accumulators for r in range(self.size): for c in range(self.size): pe = self.pe_grid[r][c] if pe['a'] is not None and pe['b'] is not None: pe['acc'] += pe['a'] * pe['b'] # 4. Shift Data (Movement) # Iterate backwards to avoid overwriting data in the same cycle for r in range(self.size - 1, -1, -1): for c in range(self.size - 1, -1, -1): # Update A (Horizontal flow) if c == 0: self.pe_grid[r][c]['a'] = a_stream[r, t] if t < a_stream.shape[1] else None else: self.pe_grid[r][c]['a'] = self.pe_grid[r][c-1]['a'] # Update B (Vertical flow) if r == 0: self.pe_grid[r][c]['b'] = b_stream[c, t] if t < b_stream.shape[1] else None else: self.pe_grid[r][c]['b'] = self.pe_grid[r-1][c]['b'] time.sleep(2) def _generate_display(self, t, a_stream, b_stream): lines = [] lines.append(f"Cycle: {t}") lines.append("-" * 80) vis_depth = self.size left_margin_width = (vis_depth * 4) + 5 margin_str = " " * left_margin_width # TOP SECTION (B Streaming) for d in range(vis_depth, 0, -1): row_str = margin_str for c in range(self.size): idx = t + d - 1 val = b_stream[c, idx] if idx < b_stream.shape[1] else None val_str = f"{val:>3}" if val is not None else " ." row_str += f" {val_str} " lines.append(row_str) lines.append(margin_str + " ￬ " * self.size) # MIDDLE SECTION (A Streaming + PEs) for r in range(self.size): left_str = "" for d in range(vis_depth, 0, -1): idx = t + d - 1 val = a_stream[r, idx] if idx < a_stream.shape[1] else None val_str = f"{val:>3}" if val is not None else " ." left_str += f"{val_str} " left_str += " -> " line_top = left_str line_bot = " " * len(left_str) for c in range(self.size): pe = self.pe_grid[r][c] a_v = pe['a'] if pe['a'] is not None else 0 b_v = pe['b'] if pe['b'] is not None else 0 acc_v = pe['acc'] line_top += f"[{a_v:>2}|{b_v:>2}] " line_bot += f" Σ {acc_v:<3} " lines.append(line_top) lines.append(line_bot) lines.append("") return "\n".join(lines) | cs |
> def __init__
① 입력 받은 Size로 크기가 (size) x (size)인 PE 배열을 생성하고 {acc, a, b}를 초기화
→ acc : Accumulator / a : 현재 PE에 들어와 있는 A 행렬 값 / b : 현재 PE에 들어와 있는 B 행렬 값
> def _format_input_buffers
· Systolic Array는 각 PE에서 올바른 행렬 원소들이 정확한 Cycle에 동시에 만나도록 데이터가 순차적으로 들어가야 함
→ Data Skewing을 구현하여 A와 B 행렬의 값들이 Systolic Array에 언제 들어갈지 미리 정리
① 전체 Simulation Cycle 수를 3 x (size)로 잡고, 각 Cycle마다 어떤 데이터가 입력될지 저장하는 타임라인 배열 생성
→ 빈 공간은 None으로 처리
② 반복문을 돌면서 행렬 A, B의 원소들이 진입할 정확한 타이밍을 계산 → Data Skewing 발생
→ a_stream[r, r+c] = A[r, c] : A의 [r, c] 원소는 (r + c) Cycle에 Row r에 입력 (A[0, 0] : 0 Cycle에 입력 / A[1, 0] : 1 Cycle에 입력)
→ b_stream[c, c+r] = B[r, c] : B의 [r, c] 원소는 (c + r) Cycle에 Col c에 입력 (B[0, 0] : 0 Cycle에 입력 / B[0, 1] : 1 Cycle에 입력)
> def _format_horizontal_matrices
· A, B, 정답 행렬을 옆으로 나란히 출력하기 위한 Format 정의
① A, B, 정답 행렬을 문자열로 바꾼 후 줄 단위로 분리
② 3개의 행렬 중 가장 Line 수가 많은 것을 기준으로 높이를 맞춰 출력(Line이 부족한 경우 빈 Line을 추가)
> def run
① "_format_input_buffers"를 통해 A, B가 각 Cycle에 어떻게 입력될지 미리 계산
② 전체 Cycle 수는 a_stream의 Column 개수를 가져오고 "numpy"의 "matmul"을 이용하여 정답 행렬을 계산
③ A, B, 정답 행렬을 나란히 보여주는 문자열 생성하여 출력
④ "_generate_display"를 통해 현재 Cycle t에서 화면에 보여줄 문자열 생성
⑤ "clear_output"을 통해 이전 출력 화면을 지워 매 Cycle마다 새로운 화면이 나타나도록 하여 애니메이션처럼 출력
⑥ 고정으로 출력 될 문자열과 현재 Cycle의 PE를 출력
⑦ 최종 결과 값이 배열 내의 PE에 고정되어 계산되는 Output-Stationary 방식의 MAC 연산 진행
→ 현재 PE에 A, B 값이 둘 다 존재하는 경우 : 곱해서 누적 / 둘 중 하나라도 존재하지 않는 경우 : 곱셈 연산 진행 X
⑧ Data Shift 진행
· A 행렬
→ 현재 PE가 가장 왼쪽 열(0번째 Col)이면 외부에서 새로운 A 값을 받아오고, 가장 왼쪽 열이 아니면 왼쪽 PE의 A 값을 받아옴
· B 행렬
→ 현재 PE가 가장 위쪽 행(0번째 Row)이면 외부에서 새로운 B 값을 받아오고, 가장 위쪽 행이 아니면 위쪽 PE의 B 값을 받아옴
** 반복문을 역순으로 진행하는 이유
→ 정방향으로 반복문을 진행하면 앞쪽 데이터가 뒤쪽을 덮어써버리기 때문에 데이터가 제대로 Shift 되지 않음
⑨ 각 Cycle마다 2초 동안 지연시켜 데이터가 Shift하는 Simulation을 관찰
> def _generate_display
① 현재 Cycle 번호와 구분선을 출력
② 입력 Stream 깊이를 Array 크기만큼 설정 후, A 입력 Stream 때문에 필요한 여백 크기를 계산하여 공백 문자열 생성
③ 각 Column에 대해 B 입력 값을 확인하여 현재 Cycle 기준으로 앞으로 들어올 B 값을 보여주기 위한 Index를 구함
④ Index로 해당 Column, 해당 시간의 B 값을 가져오고 범위를 벗어나면 None으로 처리
→ 값이 있는 경우 : 오른쪽 정렬로 출력 문자열에 추가 / 값이 없는 경우 : " . "으로 표시해서 출력 문자열에 추가
⑤ A 입력 Stream을 여러 칸으로 보여주고 현재 Cycle 기준으로 표시할 A 입력 시간 Index를 설정
⑥ Index로 해당 Row, 해당 시간의 A 값을 가져오고 범위를 벗어나면 None으로 처리하여 출력 문자열에 추가
⑦ PE의 현재 A, B 값과 누적합을 출력하기 위해 "pe"를 통해 현재 PE 안의 A, B, 누적합을 갖고옴
→ A, B는 값이 없는 경우 0을 사용
▣ Adder Tree Simulator
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 | class AdderTreeSimulator: def __init__(self, size): self.size = size self.stages = int(np.log2(size)) # Pipeline State: # stage 0: Multiplier outputs (size N) # stage 1..S: Adder outputs (size N/2, N/4...) # stored as dictionaries {'val': value, 'row_id': row_index} to track data flow self.pipeline = [] current_w = size for _ in range(self.stages + 1): self.pipeline.append([None] * current_w) current_w //= 2 self.results = [] # Stores final calculated results self.weights = None # Will hold Vector V def _format_horizontal_matrices(self, A, V, Res): """Helper to display A, V, and Expected Result side-by-side statically.""" a_lines = str(A).split('\n') v_lines = str(V).split('\n') r_lines = str(Res).split('\n') max_h = max(len(a_lines), len(v_lines), len(r_lines)) # Pad with empty lines a_lines += [""] * (max_h - len(a_lines)) v_lines += [""] * (max_h - len(v_lines)) r_lines += [""] * (max_h - len(r_lines)) output = [] output.append(f"{'Matrix A':<25} {'Vector V':<15} {'Expected Result':<15}") output.append("-" * 80) for i in range(max_h): # Formatted columns row = f"{a_lines[i]:<25} * {v_lines[i]:<15} = {r_lines[i]:<15}" output.append(row) return "\n".join(output) def _generate_tree_display(self, t, n_rows): lines = [] # --- CONSTANTS FOR LAYOUT --- BLOCK_WIDTH = 9 # Width of [aa|bb] GAP_WIDTH = 4 # Gap between blocks at level 0 STRIDE = BLOCK_WIDTH + GAP_WIDTH TREE_RIGHT_MARGIN = 65 # Column index where the tree visualization ends # --- DRAW TREE STAGES --- # We iterate through stages to build string lines # Stage 0: Multipliers line_str = "" vals = self.pipeline[0] # Calculate indent/spacing for Stage 0 current_stride = STRIDE current_indent = 0 # Build Stage 0 string line_str += " " * current_indent for i, node in enumerate(vals): if node is not None: # [A|V] format a_val = node['a_val'] v_val = node['v_val'] txt = f"[{a_val:>2}|{v_val:>2}]" else: txt = f"[{'.':>2}|{'.':>2}]" line_str += f"{txt:^{BLOCK_WIDTH}}" if i < len(vals) - 1: line_str += " " * GAP_WIDTH lines.append(f"{line_str:<{TREE_RIGHT_MARGIN}}") # Add to output lines # Subsequent Adder Stages for s in range(1, self.stages + 1): prev_stride = current_stride current_stride = prev_stride * 2 # Connector lines (\ /) conn_line = "" # The center of the previous block is at: indent + (width/2) + index*prev_stride # We want slashes connecting parents to children # Simplified approach for fixed text layout: # Shift indent by half the previous stride roughly prev_indent = current_indent current_indent = prev_indent + (prev_stride // 2) conn_line = " " * current_indent data_line = " " * current_indent node_count = len(self.pipeline[s]) for i in range(node_count): # Connector # A rough visual approx for the tree branches conn_txt = "\\ /" conn_line += f"{conn_txt:^{BLOCK_WIDTH}}" # Data Node node = self.pipeline[s][i] if node is not None: val = node['val'] txt = f"[{val:>3}]" else: txt = f"[{'.':>3}]" data_line += f"{txt:^{BLOCK_WIDTH}}" if i < node_count - 1: # Spacing between nodes at this level # Calculate gap based on stride difference gap_len = current_stride - BLOCK_WIDTH conn_line += " " * gap_len data_line += " " * gap_len lines.append(f"{conn_line:<{TREE_RIGHT_MARGIN}}") lines.append(f"{data_line:<{TREE_RIGHT_MARGIN}}") # --- RIGHT COLUMN: RESULT VECTOR --- # We need to append the result vector to the right of these lines. # The result vector might be taller than the tree, or shorter. final_output = [] max_visual_rows = max(len(lines), n_rows) # Header for the dynamic section final_output.append(f"Cycle: {t:<5}") final_output.append("-" * 80) for r in range(max_visual_rows): # Left side: Tree (or empty if below tree) left_part = lines[r] if r < len(lines) else " " * TREE_RIGHT_MARGIN # Right side: Result Vector element # We display the vector building up. if r < n_rows: if r < len(self.results): res_val = self.results[r] right_part = f"| {res_val:>4} |" else: right_part = "| . |" else: right_part = "" final_output.append(f"{left_part}{right_part}") return "\n".join(final_output) def run(self, A, V): self.weights = V.flatten() n_rows, n_cols = A.shape total_cycles = n_rows + self.stages + 2 expected = np.dot(A, V) # Static Header Generation static_header = f""" {'='*80} ADDER TREE GEMV SIMULATION: {self.size} Inputs {'='*80} {self._format_horizontal_matrices(A, V, expected)} {'='*80} CYCLE LEVEL SIMULATION {'='*80} """ self.results = [] # Clear results for t in range(total_cycles + 2): # +2 for flush # 1. GENERATE DISPLAY (Before state update to show initial state or current latch) # However, for consistency with flow, usually we calculate state then show. # But to show inputs *entering*, we often setup inputs then show. # We will calculate the "Next State" logic but display the "Current State" # Actually, let's just render the current pipeline state. display_str = self._generate_tree_display(t, n_rows) clear_output(wait=True) print(static_header) print(display_str) # 2. UPDATE LOGIC (Flow from Bottom to Top to simulate registers) # A. Final Result Collection (Output of last stage) last_stage_idx = self.stages if self.pipeline[last_stage_idx][0] is not None: # We have a valid result from the tree root res_node = self.pipeline[last_stage_idx][0] # Append to results list self.results.append(res_node['val']) # B. Adder Stages (Propagate Data) # Move from s-1 to s for s in range(self.stages, 0, -1): input_stage = self.pipeline[s-1] # We need pairs of data to produce output new_stage_data = [] for i in range(0, len(input_stage), 2): left = input_stage[i] right = input_stage[i+1] if left is not None and right is not None: # Perform Addition new_sum = left['val'] + right['val'] # Inherit row_id (should be same) new_stage_data.append({'val': new_sum, 'row_id': left['row_id']}) else: new_stage_data.append(None) self.pipeline[s] = new_stage_data # C. Multiplier Stage (Input Feed) # Load A[row] and V if t < n_rows: # Stream in new row current_row_vals = A[t] new_mult_stage = [] for i in range(self.size): val_a = current_row_vals[i] val_v = self.weights[i] prod = val_a * val_v new_mult_stage.append({ 'val': prod, 'a_val': val_a, 'v_val': val_v, 'row_id': t }) self.pipeline[0] = new_mult_stage else: # No more input, fill with bubbles self.pipeline[0] = [None] * self.size time.sleep(1.5) | cs |
> def __init__
① 입력 개수를 나타내는 size를 입력 받아 Adder Tree의 단계 수(log2n)를 계산 (Ex. Size = 4 : 4 → 2 → 1)
② Pipeline의 각 Stage마다 현재 어떤 값이 들어있는지 저장할 List를 생성 후 현재 Stage의 Node 개수를 저장할 변수 생성
③ Pipeline의 Stage 개수만큼 반복하여 현재 Stage에 해당하는 공간 생성 (∵ 0번째 단계도 포함해야 하므로 Stages + 1 만큼 반복)
④ 초기에는 아무 값도 없으므로 None으로 할당
⑤ Adder Tree는 2개를 더해서 하나로 만들기 때문에 다음 Stage는 Node의 개수가 절반
→ 병렬 처리를 통한 연산 지연 시간 단축을 하는 Logic
> def _format_horizontal_matrices
· A, B, 정답 행렬을 옆으로 나란히 출력하기 위한 Format 정의
① A, B, 정답 행렬을 문자열로 바꾼 후 줄 단위로 분리
② 3개의 행렬 중 가장 Line 수가 많은 것을 기준으로 높이를 맞춰 출력(Line이 부족한 경우 빈 Line을 추가)
> def _generate_tree_display
① 먼저 Tree Layout을 정의하고 Stage 0(Multiplier)의 각 Multiplier 결과를 하나씩 확인
→ 해당 위치에 유효한 데이터가 있는 경우 : Node 안에서 A, V 값을 꺼냄 / 데이터가 없는 경우 : " . "을 이용해 빈 상태를 표시
② 마지막 Node가 아닌 경우 Node와 Node 사이에 공백 추가
③ Stage 0 ~ 마지막 Stage(Adder Stage)를 반복하며 Format 정의 진행
④ Multiplier Stage처럼 Adder 결과를 하나씩 확인
→ 해당 위치에 유효한 데이터가 있는 경우 : Adder 결과 출력 / 데이터가 없는 경우 : " . "을 이용해 빈 상태를 표시
⑤ 마지막 Node가 아닌 경우 다음 Node와의 간격을 추가
⑥ (왼쪽 : Adder Tree / 오른쪽 : 계산이 완료되어 저장된 결과 벡터) 출력을 위해 Tree와 결과 벡터의 Line 수 중 더 큰 값을 선택
→ 출력이 깨지지 않도록 Formatting을 정의
⑦ "left_part"를 이용해 왼쪽에는 Tree 그림 출력을 담당 & 결과 벡터의 Row 범위 안에 속하면 오른쪽 결과 칸에 출력
(이미 계산 완료된 결과가 있으면 출력 / 아직 계산되지 않은 행은 " . "으로 표시)
> def run
① 벡터 V를 1차원 배열로 바꿔 저장 & 행렬 A의 행과 열 개수를 구한 후 전체 Simulation Cycle 수 계산
→ +2 : Pipeline 내부에 남아있는 데이터들이 최종 Stage까지 도달하기 위한 여유 Cycle
② Simulation 결과와 비교하기 위한 정답 행렬을 "numpy"의 "matmul"로 계산 후 Formatting 정의
③ Cycle을 진행하면서 현재 Pipeline 상태를 문자열로 생성하고 이전 출력 화면을 지운 후 현재 Cycle의 Tree 상태를 출력
→ +2 : Pipeline 내부에 남아있는 데이터들이 최종 Stage까지 도달하기 위한 여유 Cycle
④ 마지막 Stage의 Index를 구한 다음 Adder Tree의 최종 합인 마지막 Stage의 Root Node에 결과 확인
→ 결과가 있으면 최종 결과 Node를 결과 벡터에 추가
⑤ Adder Stage를 마지막부터 Stage 1까지 거꾸로 Update 진행
(∵ 정방향으로 Update 시 새로 계산된 값이 같은 Cycle에 다음 Stage로 넘어갈 수 있음)
⑥ Adder Stage의 왼쪽 입력과 오른쪽 입력을 가져와서 두 입력이 모두 존재하면 덧셈 수행(하나라도 None인 경우 계산 진행 X)
⑦ 아직 넣을 Row가 남아있는 경우 현재 Row의 i번째 값과 벡터 v의 i번째 값을 가져와서 A, V 값을 곱한 결과를 Dictionary로 저장
⑧ Dictionary 값을 Multiplier Stage(Stage 0)에 할당하여 새로운 값 갱신 (만약 더이상 넣을 Row가 없는 경우 Stage 0을 비움)
⑨ 각 Cycle마다 1.5초 동안 지연시켜 Simulation을 관찰
▣ Simulator Setting
1 2 3 4 5 6 7 8 9 10 11 | np.random.seed(42) # Small 4x4 A4 = np.random.randint(1, 5, (4, 4)) B4 = np.random.randint(1, 5, (4, 4)) V4 = np.random.randint(1, 5, (4, 1)) # Large 8x8 A8 = np.random.randint(1, 5, (8, 8)) B8 = np.random.randint(1, 5, (8, 8)) V8 = np.random.randint(1, 5, (8, 1)) | cs |
1 : "numpy"의 랜덤 생성 기준 값을 42로 고정하여 매번 같은 랜덤 행렬과 벡터 생성 → 반복해도 동일한 출력 결과
4 ~ 5 : 1이상 5미만의 정수 중에서 랜덤 값을 뽑아 4x4 행렬 생성
6 : 1이상 5미만의 정수 중에서 랜덤 값을 뽑아 4x1 벡터 생성
9 ~ 10 : 1이상 5미만의 정수 중에서 랜덤 값을 뽑아 8x8 행렬 생성
11 : 1이상 5미만의 정수 중에서 랜덤 값을 뽑아 8x1 벡터 생성
▣ Implementation
1 2 3 4 5 6 7 8 9 10 11 | # --- Task 1: 4x4 Systolic Array --- sim_sys_4 = SystolicArraySimulator(size=4) sim_sys_4.run(A4, B4) # --- Task 2: 8x8 Systolic Array --- sim_sys_8 = SystolicArraySimulator(size=8) sim_sys_8.run(A8, B8) # --- Task 3: 4x4 Adder Tree --- sim_tree_4 = AdderTreeSimulator(size=4) sim_tree_4.run(A4, V4) | cs |
▣ Result
<4x4 Systolic Array (Cycle 0)>
· 0번째 행 또는 0번째 열이 아닌 경우 Data Skewing을 위해 " . "으로 대체되어 Cycle 대기· 모든 PE와 Accumulator가 0으로 초기화
· 사용하는 A, B 행렬과 예상 결과 행렬이 상단에 위치
· 사용하는 A, B 행렬과 예상 결과 행렬이 상단에 위치
<4x4 Systolic Array (Cycle 6)>
· A Stream과 B Stream이 제일 마지막 1만 남았고 나머지는 모두 입력되어 비어있는 것을 " . "로 표시한 것을 확인 가능· PE Grid의 (0, 0), (0, 1), (1, 0)은 계산이 완료되어 PE의 A, B 값 상태는 0 & 누적합은 예상 결과 행렬과 동일
<4x4 Adder Tree (Cycle 4)>
> Cycle 1· Stage 0 : (3 x 4) | (4 x 2) | (1 x 2) | (3 x 1)
> Cycle 2
· Stage 0 : (3 x 4) | (4 x 2) | (1 x 2) | (1 x 1)
· Stage 0 : (3 x 4) | (4 x 2) | (1 x 2) | (1 x 1)
· Stage 1 : {(3 x 4) + (4 x 2)} | {(1 x 2) + (3 x 1)} = 20 | 5
> Cycle 3
· Stage 0 : (3 x 4) | (2 x 2) | (3 x 2) | (3 x 1)
· Stage 1 : {(3 x 4) + (4 x 2)} | {(1 x 2) + (1 x 1)} = 20 | 3
· Stage 2 : [{(3 x 4) + (4 x 2)} + {(1 x 2) + (3 x 1)}] = 25
> Cycle 4
· Stage 0 : (3 x 4) | (2 x 2) | (3 x 2) | (3 x 1)
· Stage 1 : {(3 x 4) + (2 x 2)} | {(3 x 2) + (3 x 1)} = 16 | 9
· Stage 2 : [{(3 x 4) + (4 x 2)} + {(1 x 2) + (1 x 1)}] = 23
· 결과 벡터 : 25
∴ Pipeline이 잘 수행되는 것을 확인 가능
[Discussion]
> Systolic Array
· A와 B, 두 행렬을 입력으로 받아 GEMM(AxB)을 수행
· 각 PE가 부분합을 내부에 저장하면서 A 데이터는 오른쪽, B 데이터는 아래쪽으로 전달
· 각 PE에서 계산된 부분합을 즉시 메모리에 기록하지 않고 내부 Register에 유지하는 Output-Stationary 방식 사용
→ 같은 데이터가 여러 PE에서 Reuse되며, 제한된 메모리 대역폭 환경에서 메모리 접근 횟수↓
∴ GEMM 또는 Convolution처럼 데이터 재사용이 중요한 연산에 적합
> Adder Tree
· 행렬 A와 Vector V를 입력으로 받아 GEMV(AxV)을 수행
· Stage 0(Multiplier Stage)에서 A와 V를 병렬로 곱셈
· Stage 1(Adder Stage)부터 인접한 두 값을 더하여 하나로 줄이는 Reduction 진행
→ 여러 곱셈 결과를 빠르게 하나의 Dot Product로 합치는 Parallel Reduction 구조
∴ GEMV 또는 Dot Product처럼 많은 곱셈 결과를 빠르게 합산해야 하는 연산에 적합
[Reference]
· Systolic_Array_6 (Embedded System Lab : Chapter6) - William J. Song
댓글
댓글 쓰기