Fork–Join Pattern #

package main

import (
	"fmt"
	"sync"
)

// sumChunk sums one slice segment and sends the result to the channel
func sumChunk(nums []int, resultCh chan<- int, wg *sync.WaitGroup) {
	defer wg.Done()
	sum := 0
	for _, n := range nums {
		sum += n
	}
	resultCh <- sum
}

func parallelSum(numbers []int, numWorkers int) int {
	chunkSize := (len(numbers) + numWorkers - 1) / numWorkers
	resultCh := make(chan int, numWorkers) // buffered: one slot per sub-task
	var wg sync.WaitGroup

	// FORK: split the array and run each part on a separate goroutine
	for i := 0; i < len(numbers); i += chunkSize {
		end := i + chunkSize
		if end > len(numbers) {
			end = len(numbers)
		}
		wg.Add(1)
		go sumChunk(numbers[i:end], resultCh, &wg)
	}

	// Close the channel after all goroutines finish
	go func() {
		wg.Wait()  // JOIN: wait for all sub-tasks
		close(resultCh)
	}()

	// Aggregation: collect all partial results
	total := 0
	for partial := range resultCh {
		total += partial
	}
	return total
}

func main() {
	numbers := make([]int, 1_000_000)
	for i := range numbers {
		numbers[i] = i + 1
	}

	result := parallelSum(numbers, 8) // use 8 parallel sub-tasks
	fmt.Printf("Total: %d\n", result)
}

package main

import (
	"fmt"
	"sync"
)

const threshold = 500 // below this, do it sequentially

// parallelMergeSort sorts a slice recursively with Fork–Join
func parallelMergeSort(data []int, wg *sync.WaitGroup) {
	if wg != nil {
		defer wg.Done()
	}

	n := len(data)
	if n <= 1 {
		return
	}

	// If the size is small, finish sequentially (avoid goroutine overhead)
	if n <= threshold {
		sequentialSort(data)
		return
	}

	mid := n / 2
	left := make([]int, mid)
	right := make([]int, n-mid)
	copy(left, data[:mid])
	copy(right, data[mid:])

	var childWG sync.WaitGroup
	childWG.Add(2)

	// FORK: two sub-tasks run in parallel
	go parallelMergeSort(left, &childWG)
	go parallelMergeSort(right, &childWG)

	// JOIN: wait for both sub-tasks to finish
	childWG.Wait()

	// Combine the results
	merge(data, left, right)
}

func merge(dst, left, right []int) {
	i, j, k := 0, 0, 0
	for i < len(left) && j < len(right) {
		if left[i] <= right[j] {
			dst[k] = left[i]; i++
		} else {
			dst[k] = right[j]; j++
		}
		k++
	}
	for i < len(left) { dst[k] = left[i]; i++; k++ }
	for j < len(right) { dst[k] = right[j]; j++; k++ }
}

func sequentialSort(data []int) {
	// simple insertion sort for small chunks
	for i := 1; i < len(data); i++ {
		key := data[i]
		j := i - 1
		for j >= 0 && data[j] > key {
			data[j+1] = data[j]
			j--
		}
		data[j+1] = key
	}
}

func main() {
	data := []int{9, 3, 7, 1, 5, 8, 2, 6, 4, 0}
	parallelMergeSort(data, nil)
	fmt.Println("Sorted:", data)
}

package main

import (
	"fmt"
	"sync"
)

type PartialResult struct {
	Value int
	Err   error
}

// processChunk processes one data segment and can produce an error
func processChunk(id int, data []int, resultCh chan<- PartialResult, wg *sync.WaitGroup) {
	defer wg.Done()

	// Simulated error on a specific chunk
	if id == 2 {
		resultCh <- PartialResult{Err: fmt.Errorf("chunk %d failed to process: corrupted data", id)}
		return
	}

	sum := 0
	for _, v := range data {
		sum += v
	}
	resultCh <- PartialResult{Value: sum}
}

func forkJoinWithErrors(data []int, numChunks int) (int, []error) {
	chunkSize := (len(data) + numChunks - 1) / numChunks
	resultCh := make(chan PartialResult, numChunks)
	var wg sync.WaitGroup

	// FORK
	for i := 0; i < numChunks; i++ {
		start := i * chunkSize
		if start >= len(data) {
			break
		}
		end := start + chunkSize
		if end > len(data) {
			end = len(data)
		}
		wg.Add(1)
		go processChunk(i, data[start:end], resultCh, &wg)
	}

	// JOIN
	go func() {
		wg.Wait()
		close(resultCh)
	}()

	// Aggregate results and collect all errors
	total := 0
	var errs []error
	for result := range resultCh {
		if result.Err != nil {
			errs = append(errs, result.Err)
			continue
		}
		total += result.Value
	}

	return total, errs
}

func main() {
	data := make([]int, 100)
	for i := range data {
		data[i] = i + 1
	}

	total, errs := forkJoinWithErrors(data, 5)
	if len(errs) > 0 {
		fmt.Println("Some chunks failed:")
		for _, e := range errs {
			fmt.Println(" -", e)
		}
	}
	fmt.Printf("Total from successful chunks: %d\n", total)
}

package main

import (
	"fmt"
	"sync"
	"time"
)

// forkJoinBounded runs parallel tasks but limits concurrency
func forkJoinBounded(tasks []func() int, maxConcurrent int) []int {
	results := make([]int, len(tasks))
	sem := make(chan struct{}, maxConcurrent) // semaphore: max N active goroutines
	var wg sync.WaitGroup
	var mu sync.Mutex

	// FORK with a concurrency limit
	for i, task := range tasks {
		wg.Add(1)
		go func(idx int, t func() int) {
			defer wg.Done()
			sem <- struct{}{}        // take a slot — blocks if full
			defer func() { <-sem }() // release the slot when done

			result := t()
			mu.Lock()
			results[idx] = result
			mu.Unlock()
		}(i, task)
	}

	// JOIN
	wg.Wait()
	return results
}

func main() {
	tasks := make([]func() int, 20)
	for i := range tasks {
		id := i
		tasks[i] = func() int {
			time.Sleep(100 * time.Millisecond)
			return id * id // square of id
		}
	}

	// Only 4 tasks run simultaneously, even though there are 20 tasks
	results := forkJoinBounded(tasks, 4)
	fmt.Println("Results (20 tasks, max 4 concurrent):", results[:5], "...")
}

// ANTI-PATTERN: too fine-grained — one goroutine per element
// the goroutine overhead (stack, scheduler) is far more expensive than the computation
func tooFineGrained(data []int) int {
	resultCh := make(chan int, len(data))
	var wg sync.WaitGroup
	for _, v := range data { // 1 million elements = 1 million goroutines ✗
		wg.Add(1)
		go func(n int) {
			defer wg.Done()
			resultCh <- n // goroutine overhead >> cost of the addition
		}(v)
	}
	go func() { wg.Wait(); close(resultCh) }()
	total := 0
	for v := range resultCh { total += v }
	return total
}

// ANTI-PATTERN: too coarse-grained — only one goroutine
// no parallelism at all
func tooCoarseGrained(data []int) int {
	resultCh := make(chan int, 1)
	go func() { // only 1 goroutine — no better than sequential ✗
		sum := 0
		for _, v := range data { sum += v }
		resultCh <- sum
	}()
	return <-resultCh
}

// CORRECT: appropriate granularity — number of chunks = number of CPU cores
func wellGrained(data []int) int {
	numWorkers := runtime.NumCPU() // ✓ match the core count
	chunkSize := (len(data) + numWorkers - 1) / numWorkers
	resultCh := make(chan int, numWorkers)
	var wg sync.WaitGroup

	for i := 0; i < len(data); i += chunkSize {
		end := i + chunkSize
		if end > len(data) { end = len(data) }
		wg.Add(1)
		go func(chunk []int) {
			defer wg.Done()
			sum := 0
			for _, v := range chunk { sum += v }
			resultCh <- sum
		}(data[i:end])
	}

	go func() { wg.Wait(); close(resultCh) }()
	total := 0
	for partial := range resultCh { total += partial }
	return total
}

// ✗ Fork without join — goroutine leak, results not collected
func noJoin(data []int) {
	for _, chunk := range splitData(data) {
		go processChunk(chunk) // ✗ no WaitGroup, no join
	}
	// the function returns before sub-tasks finish — results are lost
}

// ✓ Always have an explicit join before using the results
func withJoin(data []int) int {
	resultCh := make(chan int, len(data))
	var wg sync.WaitGroup
	for _, chunk := range splitData(data) {
		wg.Add(1)
		go func(c []int) {
			defer wg.Done()
			resultCh <- process(c) // ✓ result sent to the channel
		}(chunk)
	}
	go func() { wg.Wait(); close(resultCh) }() // ✓ explicit join
	total := 0
	for r := range resultCh { total += r }
	return total
}

// ✗ Shared mutable state without synchronization — data race
var sharedTotal int
func racyFork(data []int) {
	var wg sync.WaitGroup
	for _, chunk := range splitData(data) {
		wg.Add(1)
		go func(c []int) {
			defer wg.Done()
			for _, v := range c {
				sharedTotal += v // ✗ race condition — multiple goroutines write simultaneously
			}
		}(chunk)
	}
	wg.Wait()
}

// ✓ Each sub-task produces a partial result, combined at the end
func safeFork(data []int) int {
	resultCh := make(chan int, 8)
	var wg sync.WaitGroup
	for _, chunk := range splitData(data) {
		wg.Add(1)
		go func(c []int) {
			defer wg.Done()
			local := 0
			for _, v := range c { local += v } // ✓ local variable, no race
			resultCh <- local
		}(chunk)
	}
	go func() { wg.Wait(); close(resultCh) }()
	total := 0
	for r := range resultCh { total += r } // ✓ combine here, not in the goroutine
	return total
}

// ✗ No recursion threshold — goroutines for size-1 data
func noThreshold(data []int) int {
	if len(data) <= 1 { // ✗ should have a much larger threshold
		return data[0]
	}
	// ... recursive fork
	return 0
}

// ✓ A threshold prevents goroutine overhead for small data
func withThreshold(data []int) int {
	if len(data) <= 1000 { // ✓ do it sequentially for small chunks
		sum := 0
		for _, v := range data { sum += v }
		return sum
	}
	// ... recursive fork only for large data
	return 0
}