Fork-Join Model

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

"""
Dependency Analysis Examples:
Identifying which operations can execute in parallel.
"""
 
# Example 1: Loop with NO dependencies - PARALLELIZABLE
def independent_iterations():
    """Each iteration operates on independent data."""
    results = [0] * 1000
    
    # Each iteration writes to a different index
    # No iteration reads another's output
    # All iterations can execute in parallel
    for i in range(1000):
        results[i] = expensive_compute(i)  # ✓ Parallel
    
    return results
 
 
# Example 2: Loop with TRUE dependency - SEQUENTIAL
def dependent_iterations():
    """Each iteration depends on the previous result."""
    result = 1
    
    # Each iteration reads 'result' written by previous
    # True dependency: read-after-write
    for i in range(1, 1000):
        result = result * i  # ✗ Sequential
    
    return result
 
 
# Example 3: Loop with REDUCIBLE dependency - PARTIALLY PARALLELIZABLE
def reducible_pattern():
    """Reduction: combine results associatively."""
    data = [1, 2, 3, 4, 5, 6, 7, 8]
    
    # Naive sequential sum
    total = 0
    for x in data:
        total += x  # Seems sequential due to 'total' dependency
    
    # But addition is associative: (a+b)+c = a+(b+c)
    # Can decompose into parallel partial sums, then combine:
    # Partition: [1,2,3,4], [5,6,7,8]
    # Parallel:  sum1=10,   sum2=26
    # Combine:   10 + 26 = 36
    # This is the REDUCTION pattern
 
 
# Example 4: Map pattern - EMBARRASSINGLY PARALLEL
def map_pattern(data):
    """Apply independent function to each element."""
    
    # Each call is independent - perfect parallelism
    results = [transform(x) for x in data]  # ✓ Parallel
    
    return results
 
 
# Example 5: Prefix sum - SEEMINGLY SEQUENTIAL, ACTUALLY PARALLEL
def prefix_sum_analysis():
    """
    Prefix sum: result[i] = sum(data[0:i+1])
    
    Naive approach seems sequential:
        result[0] = data[0]
        result[i] = result[i-1] + data[i]  # Depends on previous!
    
    But there's a parallel algorithm (Blelloch scan):
        O(n) work, O(log n) span
    
    Key insight: operations can be restructured to expose parallelism.
    """
    pass
 
 
# Example 6: Dependency Graph Analysis
def analyze_dependencies():
    """
    Consider these operations:
    
    A: x = input1 * 2
    B: y = input2 + 3  
    C: z = x + y
    D: w = input3 - 1
    E: result = z * w
    
    Dependency graph:
    
        A ──┐
            ├──► C ──┐
        B ──┘       ├──► E
        D ─────────┘
    
    Parallel schedule:
    - Level 1: A, B, D (all independent)
    - Level 2: C (depends on A, B)
    - Level 3: E (depends on C, D)
    
    Speedup: 5 sequential operations → 3 parallel steps
    """
    pass

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

import java.util.concurrent.*;
import java.util.stream.*;
 
/**
 * Comparison of data parallelism and task parallelism approaches.
 */
public class ParallelismComparison {
    
    // ==========================================
    // DATA PARALLELISM: Same operation on partitioned data
    // ==========================================
    
    /**
     * Data-parallel array processing.
     * Each element processed identically; data is partitioned.
     */
    static long dataParallelSum(int[] data) {
        // Fork-Join style: divide array, sum partitions, combine
        return IntStream.of(data)
            .parallel()  // Automatic data partitioning
            .mapToLong(x -> x)  // Same operation on each element
            .sum();  // Reduction
    }
    
    /**
     * Data-parallel image processing.
     * Each pixel processed by same kernel; image partitioned into tiles.
     */
    static void dataParallelImageProcess(int[][] image, int[][] output) {
        int height = image.length;
        int width = image[0].length;
        
        IntStream.range(0, height).parallel().forEach(y -> {
            for (int x = 0; x < width; x++) {
                // Same convolution operation applied to each pixel
                output[y][x] = applyConvolution(image, x, y);
            }
        });
    }
    
    // ==========================================
    // TASK PARALLELISM: Different operations concurrently
    // ==========================================
    
    /**
     * Task-parallel pipeline.
     * Different stages process concurrently on different data.
     */
    static void taskParallelPipeline(Queue<RawData> input, 
                                      Queue<Result> output) {
        ExecutorService executor = Executors.newFixedThreadPool(3);
        
        BlockingQueue<DecodedData> decoded = new LinkedBlockingQueue<>();
        BlockingQueue<ProcessedData> processed = new LinkedBlockingQueue<>();
        
        // Task 1: Decode (runs concurrently)
        executor.submit(() -> {
            while (true) {
                RawData raw = input.poll();
                if (raw == null) break;
                decoded.put(decode(raw));  // Different function
            }
        });
        
        // Task 2: Process (runs concurrently)
        executor.submit(() -> {
            while (true) {
                DecodedData d = decoded.take();
                processed.put(process(d));  // Different function
            }
        });
        
        // Task 3: Encode (runs concurrently)
        executor.submit(() -> {
            while (true) {
                ProcessedData p = processed.take();
                output.add(encode(p));  // Different function
            }
        });
        
        executor.shutdown();
    }
    
    /**
     * Task-parallel independent computations.
     * Different analyses run on same data concurrently.
     */
    static AnalysisResult taskParallelAnalysis(Data data) 
            throws Exception {
        ExecutorService executor = Executors.newFixedThreadPool(4);
        
        // Four different analyses on same data
        Future<StatisticalResult> statsF = 
            executor.submit(() -> computeStatistics(data));
        Future<TrendResult> trendF = 
            executor.submit(() -> detectTrends(data));
        Future<AnomalyResult> anomalyF = 
            executor.submit(() -> findAnomalies(data));
        Future<ForecastResult> forecastF = 
            executor.submit(() -> generateForecast(data));
        
        // Wait for all (join)
        return new AnalysisResult(
            statsF.get(),
            trendF.get(),
            anomalyF.get(),
            forecastF.get()
        );
    }
    
    // ==========================================
    // HYBRID: Both data and task parallelism
    // ==========================================
    
    /**
     * Hybrid parallelism: parallel pipeline with data-parallel stages.
     */
    static void hybridParallelism(int[][][] videoFrames) {
        // Task parallelism: different stages
        // Data parallelism within each stage: parallel over frames/pixels
        
        ForkJoinPool pool = ForkJoinPool.commonPool();
        
        pool.submit(() -> {
            Arrays.stream(videoFrames)
                .parallel()  // Data parallel over frames
                .forEach(frame -> {
                    // Each frame processed by many workers
                    IntStream.range(0, frame.length)
                        .parallel()  // Data parallel over rows
                        .forEach(y -> processRow(frame[y]));
                });
        });
    }
}

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

import java.util.concurrent.*;
 
/**
 * Demonstrates granularity selection and sequential thresholds.
 */
public class GranularityExample extends RecursiveTask<Long> {
    
    // Sequential threshold - tune based on workload
    // Start with ~1000-10000 elements, benchmark to optimize
    private static final int SEQUENTIAL_THRESHOLD = 5000;
    
    private final long[] array;
    private final int lo;
    private final int hi;
    
    public GranularityExample(long[] array, int lo, int hi) {
        this.array = array;
        this.lo = lo;
        this.hi = hi;
    }
    
    @Override
    protected Long compute() {
        int size = hi - lo;
        
        // COARSE CHECK: Is task small enough to do sequentially?
        if (size <= SEQUENTIAL_THRESHOLD) {
            // Sequential execution - no more forking
            return sequentialSum(array, lo, hi);
        }
        
        // DECOMPOSITION: Split into two halves
        int mid = (lo + hi) / 2;
        
        // Fork left half
        GranularityExample left = new GranularityExample(array, lo, mid);
        left.fork();
        
        // Compute right half directly (fork-once pattern)
        GranularityExample right = new GranularityExample(array, mid, hi);
        Long rightResult = right.compute();
        
        // Join left and combine results
        Long leftResult = left.join();
        
        return leftResult + rightResult;
    }
    
    /**
     * Sequential computation for small chunks.
     * Optimized for cache locality - sequential access pattern.
     */
    private long sequentialSum(long[] arr, int start, int end) {
        long sum = 0;
        for (int i = start; i < end; i++) {
            sum += arr[i];
            
            // If per-element work is expensive, threshold can be lower
            // If per-element work is trivial, threshold should be higher
        }
        return sum;
    }
    
    // ==========================================
    // ADAPTIVE GRANULARITY
    // ==========================================
    
    /**
     * Adaptive approach: consider available parallelism and work remaining.
     */
    static class AdaptiveTask extends RecursiveTask<Long> {
        private static final int MIN_GRANULARITY = 1000;
        
        private final long[] array;
        private final int lo, hi;
        private final int surplusTasks;
        
        AdaptiveTask(long[] array, int lo, int hi, int surplusTasks) {
            this.array = array;
            this.lo = lo;
            this.hi = hi;
            this.surplusTasks = surplusTasks;
        }
        
        @Override
        protected Long compute() {
            int size = hi - lo;
            
            // Stop splitting when:
            // 1. Task is small enough, OR
            // 2. We have enough tasks queued (surplus)
            if (size <= MIN_GRANULARITY || surplusTasks <= 0) {
                return sequentialSum(array, lo, hi);
            }
            
            int mid = (lo + hi) / 2;
            
            // Reduce surplus as we create more tasks
            AdaptiveTask left = new AdaptiveTask(
                array, lo, mid, surplusTasks - 1);
            AdaptiveTask right = new AdaptiveTask(
                array, mid, hi, surplusTasks - 1);
            
            left.fork();
            Long rightResult = right.compute();
            Long leftResult = left.join();
            
            return leftResult + rightResult;
        }
        
        private long sequentialSum(long[] arr, int start, int end) {
            long sum = 0;
            for (int i = start; i < end; i++) sum += arr[i];
            return sum;
        }
    }
    
    // ==========================================
    // BENCHMARKING GRANULARITY
    // ==========================================
    
    public static void benchmarkThresholds() {
        long[] data = new long[10_000_000];
        for (int i = 0; i < data.length; i++) data[i] = i;
        
        int[] thresholds = {100, 1000, 5000, 10000, 50000, 100000};
        ForkJoinPool pool = ForkJoinPool.commonPool();
        
        System.out.println("Threshold | Time (ms) | Tasks Created");
        System.out.println("----------+-----------+--------------");
        
        for (int threshold : thresholds) {
            // Reset task counter...
            long start = System.nanoTime();
            
            // Would need to modify SEQUENTIAL_THRESHOLD dynamically
            // or create threshold-specific subclass
            Long result = pool.invoke(
                new GranularityExample(data, 0, data.length));
            
            long elapsed = (System.nanoTime() - start) / 1_000_000;
            
            // Optimal threshold minimizes time
            System.out.printf("%9d | %9d | (measure)%n", threshold, elapsed);
        }
    }
}

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

import java.util.concurrent.*;
import java.util.concurrent.atomic.*;
 
/**
 * Load balancing strategies for parallel decomposition.
 */
public class LoadBalancing {
    
    // ==========================================
    // PROBLEM: Non-uniform work distribution
    // ==========================================
    
    /**
     * Example: Processing items where work varies dramatically.
     * Item 0 might take 1ms, item 999 might take 1000ms.
     */
    static void nonUniformWorkProblem() {
        int[] items = new int[1000];
        
        // Work per item: items[i] takes i milliseconds
        // Total work: 0+1+2+...+999 = ~500k ms = 500 seconds
        // If we split evenly: Worker 1 handles [0-499], Worker 2 handles [500-999]
        // Worker 1 work: 0+1+...+499 = 124,750 ms
        // Worker 2 work: 500+501+...+999 = 374,750 ms
        // Worker 1 finishes 3x earlier and idles!
    }
    
    // ==========================================
    // STATIC CYCLIC DISTRIBUTION
    // ==========================================
    
    /**
     * Cyclic distribution: Round-robin assignment.
     * Worker 0: items 0,4,8,...  Worker 1: items 1,5,9,... etc.
     * Balances if variation is regular.
     */
    static void cyclicDistribution(int[] items, int numWorkers) {
        ExecutorService executor = Executors.newFixedThreadPool(numWorkers);
        
        for (int worker = 0; worker < numWorkers; worker++) {
            final int workerId = worker;
            
            executor.submit(() -> {
                // Process items workerId, workerId+numWorkers, ...
                for (int i = workerId; i < items.length; i += numWorkers) {
                    processItem(items[i]);
                }
            });
        }
        
        executor.shutdown();
    }
    
    // ==========================================
    // DYNAMIC CHUNKED DISTRIBUTION
    // ==========================================
    
    /**
     * Dynamic chunked: Workers grab chunks from shared counter.
     * Good balance but has synchronization overhead.
     */
    static void dynamicChunked(int[] items, int numWorkers, int chunkSize) {
        AtomicInteger nextChunk = new AtomicInteger(0);
        ExecutorService executor = Executors.newFixedThreadPool(numWorkers);
        
        for (int worker = 0; worker < numWorkers; worker++) {
            executor.submit(() -> {
                while (true) {
                    // Atomically grab next chunk
                    int start = nextChunk.getAndAdd(chunkSize);
                    if (start >= items.length) break;
                    
                    int end = Math.min(start + chunkSize, items.length);
                    for (int i = start; i < end; i++) {
                        processItem(items[i]);
                    }
                }
            });
        }
        
        executor.shutdown();
    }
    
    // ==========================================
    // WORK-STEALING: Fork-Join's Built-in Solution
    // ==========================================
    
    /**
     * Work-stealing automatically balances load.
     * Each worker has a deque; idle workers steal from busy workers.
     */
    static class WorkStealingSum extends RecursiveTask<Long> {
        private static final int THRESHOLD = 1000;
        private final long[] array;
        private final int lo, hi;
        
        WorkStealingSum(long[] array, int lo, int hi) {
            this.array = array;
            this.lo = lo;
            this.hi = hi;
        }
        
        @Override
        protected Long compute() {
            if (hi - lo <= THRESHOLD) {
                return sequentialCompute();
            }
            
            int mid = (lo + hi) / 2;
            WorkStealingSum left = new WorkStealingSum(array, lo, mid);
            WorkStealingSum right = new WorkStealingSum(array, mid, hi);
            
            // Fork creates task in local deque
            left.fork();
            
            // While 'left' might sit in deque, other workers can STEAL it
            // This is the work-stealing magic
            Long rightResult = right.compute();
            Long leftResult = left.join();
            
            return leftResult + rightResult;
        }
        
        private long sequentialCompute() {
            // Variable work simulation
            long sum = 0;
            for (int i = lo; i < hi; i++) {
                sum += expensiveOp(array[i]);  // Time varies per value
            }
            return sum;
        }
        
        private long expensiveOp(long value) {
            // Simulate variable work: more work for larger values
            long result = value;
            for (int i = 0; i < value % 100; i++) {
                result = result * 31 + i;
            }
            return result;
        }
    }
    
    // ==========================================
    // GUIDED SELF-SCHEDULING
    // ==========================================
    
    /**
     * Guided scheduling: chunk size decreases as work progresses.
     * Large chunks initially (low overhead), small chunks at end (good balance).
     */
    static void guidedScheduling(int[] items, int numWorkers) {
        AtomicInteger remaining = new AtomicInteger(items.length);
        AtomicInteger currentIndex = new AtomicInteger(0);
        ExecutorService executor = Executors.newFixedThreadPool(numWorkers);
        
        for (int worker = 0; worker < numWorkers; worker++) {
            executor.submit(() -> {
                while (true) {
                    int rem = remaining.get();
                    if (rem <= 0) break;
                    
                    // Chunk size proportional to remaining work divided by workers
                    // Minimum chunk size of 1
                    int chunkSize = Math.max(1, rem / (numWorkers * 2));
                    
                    int start = currentIndex.getAndAdd(chunkSize);
                    int actualRemaining = remaining.addAndGet(-chunkSize);
                    
                    if (start >= items.length) break;
                    
                    int end = Math.min(start + chunkSize, items.length);
                    for (int i = start; i < end; i++) {
                        processItem(items[i]);
                    }
                }
            });
        }
        
        executor.shutdown();
    }
    
    private static void processItem(int item) {
        // Simulate work proportional to item value
        try { Thread.sleep(item % 10); } 
        catch (InterruptedException e) {}
    }
}

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

import java.util.concurrent.*;
import java.util.stream.*;
import java.util.*;
 
/**
 * Implementation examples of common decomposition patterns.
 */
public class DecompositionPatterns {
    
    // ==========================================
    // PATTERN 1: MAP (Embarrassingly Parallel)
    // ==========================================
    
    /**
     * Map pattern: independent transformation of each element.
     */
    static double[] parallelMap(double[] input, DoubleUnaryOperator f) {
        return Arrays.stream(input)
            .parallel()
            .map(f)
            .toArray();
    }
    
    // Fork-Join version for illustration
    static class MapTask extends RecursiveAction {
        private static final int THRESHOLD = 10000;
        private final double[] input, output;
        private final int lo, hi;
        private final DoubleUnaryOperator f;
        
        MapTask(double[] input, double[] output, int lo, int hi, 
                DoubleUnaryOperator f) {
            this.input = input; this.output = output;
            this.lo = lo; this.hi = hi; this.f = f;
        }
        
        @Override
        protected void compute() {
            if (hi - lo <= THRESHOLD) {
                for (int i = lo; i < hi; i++) {
                    output[i] = f.applyAsDouble(input[i]);
                }
            } else {
                int mid = (lo + hi) / 2;
                invokeAll(
                    new MapTask(input, output, lo, mid, f),
                    new MapTask(input, output, mid, hi, f)
                );
            }
        }
    }
    
    // ==========================================
    // PATTERN 2: REDUCE
    // ==========================================
    
    /**
     * Reduce pattern: combine elements with associative operation.
     */
    static class ReduceTask extends RecursiveTask<Long> {
        private static final int THRESHOLD = 10000;
        private final long[] array;
        private final int lo, hi;
        private final LongBinaryOperator combine;
        private final long identity;
        
        ReduceTask(long[] array, int lo, int hi, 
                   long identity, LongBinaryOperator combine) {
            this.array = array;
            this.lo = lo; this.hi = hi;
            this.identity = identity;
            this.combine = combine;
        }
        
        @Override
        protected Long compute() {
            if (hi - lo <= THRESHOLD) {
                long result = identity;
                for (int i = lo; i < hi; i++) {
                    result = combine.applyAsLong(result, array[i]);
                }
                return result;
            }
            
            int mid = (lo + hi) / 2;
            ReduceTask left = new ReduceTask(array, lo, mid, identity, combine);
            ReduceTask right = new ReduceTask(array, mid, hi, identity, combine);
            
            left.fork();
            Long rightResult = right.compute();
            Long leftResult = left.join();
            
            return combine.applyAsLong(leftResult, rightResult);
        }
    }
    
    static long parallelSum(long[] array) {
        return ForkJoinPool.commonPool().invoke(
            new ReduceTask(array, 0, array.length, 0L, Long::sum));
    }
    
    static long parallelMax(long[] array) {
        return ForkJoinPool.commonPool().invoke(
            new ReduceTask(array, 0, array.length, Long.MIN_VALUE, Long::max));
    }
    
    // ==========================================
    // PATTERN 3: DIVIDE AND CONQUER
    // ==========================================
    
    /**
     * Divide-and-conquer: recursive subdivision into independent subproblems.
     * Classic example: parallel merge sort.
     */
    static class ParallelMergeSort extends RecursiveAction {
        private static final int THRESHOLD = 4096;
        private final int[] array;
        private final int lo, hi;
        private final int[] buffer;
        
        ParallelMergeSort(int[] array, int lo, int hi, int[] buffer) {
            this.array = array;
            this.lo = lo; this.hi = hi;
            this.buffer = buffer;
        }
        
        @Override
        protected void compute() {
            if (hi - lo <= THRESHOLD) {
                Arrays.sort(array, lo, hi);  // Sequential fallback
                return;
            }
            
            int mid = (lo + hi) / 2;
            
            // DIVIDE: Create subproblems
            ParallelMergeSort left = 
                new ParallelMergeSort(array, lo, mid, buffer);
            ParallelMergeSort right = 
                new ParallelMergeSort(array, mid, hi, buffer);
            
            // CONQUER: Solve subproblems in parallel
            invokeAll(left, right);
            
            // COMBINE: Merge sorted halves
            merge(array, lo, mid, hi, buffer);
        }
        
        private void merge(int[] arr, int lo, int mid, int hi, int[] buf) {
            System.arraycopy(arr, lo, buf, lo, hi - lo);
            int i = lo, j = mid, k = lo;
            while (i < mid && j < hi) {
                arr[k++] = (buf[i] <= buf[j]) ? buf[i++] : buf[j++];
            }
            while (i < mid) arr[k++] = buf[i++];
            while (j < hi) arr[k++] = buf[j++];
        }
    }
    
    // ==========================================
    // PATTERN 4: STENCIL
    // ==========================================
    
    /**
     * Stencil pattern: each output depends on neighborhood of inputs.
     * Exercise: avoid data races at tile boundaries.
     */
    static void parallelStencil(double[][] input, double[][] output, 
                                 double[][] kernel) {
        int height = input.length;
        int width = input[0].length;
        int kRadius = kernel.length / 2;
        
        // Process rows in parallel (each row independent if reading input)
        IntStream.range(kRadius, height - kRadius).parallel().forEach(y -> {
            for (int x = kRadius; x < width - kRadius; x++) {
                double sum = 0;
                for (int ky = -kRadius; ky <= kRadius; ky++) {
                    for (int kx = -kRadius; kx <= kRadius; kx++) {
                        sum += input[y + ky][x + kx] * 
                               kernel[ky + kRadius][kx + kRadius];
                    }
                }
                output[y][x] = sum;
            }
        });
    }
}