Go: Go Goroutines و WaitGroups

// Serial processing: 100k entries × 200ms/entry = 5.5 hours
func processLogs(logs []LogEntry) []Result {
    results := make([]Result, 0, len(logs))
    for _, log := range logs {
        result := processSingleLog(log)  // 200ms each, including network IO
        results = append(results, result)
    }
    return results
}

// log_processor.go
package main

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

type LogEntry struct {
    ID      int
    Message string
    Level   string
}

type Result struct {
    ID     int
    Status string
}

func processSingleLog(log LogEntry) Result {
    time.Sleep(200 * time.Millisecond)
    return Result{ID: log.ID, Status: "processed"}
}

func processLogsConcurrent(logs []LogEntry, workerCount int) []Result {
    results := make([]Result, 0, len(logs))
    var mu sync.Mutex
    var wg sync.WaitGroup

    // Use channels for task distribution
    jobs := make(chan LogEntry, len(logs))
    for _, log := range logs {
        jobs <- log
    }
    close(jobs)

    // Start workerCount workers
    for w := 0; w < workerCount; w++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            for log := range jobs {
                result := processSingleLog(log)
                mu.Lock()
                results = append(results, result)
                mu.Unlock()
            }
        }(w)
    }

    wg.Wait()
    return results
}

func main() {
    // Simulate 100 log entries (demo; can scale to 100k)
    logs := make([]LogEntry, 100)
    for i := 0; i < 100; i++ {
        logs[i] = LogEntry{ID: i, Message: fmt.Sprintf("log-%d", i), Level: "INFO"}
    }

    start := time.Now()
    results := processLogsConcurrent(logs, 10)
    elapsed := time.Since(start)

    fmt.Printf("Processed %d logs in %v\n", len(results), elapsed)
    fmt.Printf("Current goroutine count: %d\n", runtime.NumGoroutine())
}

Processed 100 logs in 2.05s
Current goroutine count: 1

package main

import (
    "fmt"
    "time"
)

func printNumbers() {
    for i := 1; i <= 5; i++ {
        time.Sleep(100 * time.Millisecond)
        fmt.Printf("%d ", i)
    }
}

func printLetters() {
    for _, c := range "ABCDE" {
        time.Sleep(150 * time.Millisecond)
        fmt.Printf("%c ", c)
    }
}

func main() {
    go printNumbers()  // Execute concurrently in a goroutine
    go printLetters()
    time.Sleep(1 * time.Second)
    fmt.Println()
}

1 A 2 B 3 C 4 D 5 E

package main

import (
    "fmt"
    "time"
)

func main() {
    go func() {
        fmt.Println("Anonymous goroutine running")
    }()

    go func(msg string) {
        fmt.Println("Anonymous goroutine with parameter:", msg)
    }("hello")

    time.Sleep(100 * time.Millisecond)
    fmt.Println("main done")
}

package main

import (
    "fmt"
    "sync"
)

func worker(id int, wg *sync.WaitGroup) {
    defer wg.Done()  // 3. Tell WaitGroup this worker is done
    fmt.Printf("Worker %d started\n", id)
    // Simulate work...
    fmt.Printf("Worker %d done\n", id)
}

func main() {
    var wg sync.WaitGroup

    for i := 1; i <= 3; i++ {
        wg.Add(1)   // 1. Increment counter
        go worker(i, &wg)
    }

    wg.Wait()  // 2. Block until all workers finish
    fmt.Println("All workers done!")
}

package main

import (
    "fmt"
    "sync"
)

func main() {
    var wg sync.WaitGroup

    // ❌ Wrong: Closure captures loop variable
    for i := 1; i <= 3; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            fmt.Printf("Wrong i=%d\n", i)  // All print 3 or 4
        }()
    }
    wg.Wait()

    // ✅ Correct: Pass parameter (creates a copy)
    for i := 1; i <= 3; i++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            fmt.Printf("Correct i=%d\n", id)
        }(i)
    }
    wg.Wait()
}

package main

import (
    "fmt"
    "runtime"
)

func main() {
    // View current GOMAXPROCS
    fmt.Printf("GOMAXPROCS: %d\n", runtime.GOMAXPROCS(0))
    fmt.Printf("CPU cores: %d\n", runtime.NumCPU())
    fmt.Printf("Current goroutine count: %d\n", runtime.NumGoroutine())
}

package main

import (
    "fmt"
    "runtime"
)

func main() {
    fmt.Printf("Default GOMAXPROCS: %d\n", runtime.GOMAXPROCS(0))

    // CPU-intensive: GOMAXPROCS = NumCPU
    // IO-intensive: GOMAXPROCS = NumCPU * 2~10
    runtime.GOMAXPROCS(4)  // Set to 4
    fmt.Printf("After setting GOMAXPROCS: %d\n", runtime.GOMAXPROCS(0))
}

package main

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

func cpuIntensiveTask() {
    sum := 0
    for i := 0; i < 10000000; i++ {
        sum += i
    }
}

func benchmarkGOMAXPROCS(n int) time.Duration {
    runtime.GOMAXPROCS(n)
    var wg sync.WaitGroup
    start := time.Now()

    for i := 0; i < 8; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            cpuIntensiveTask()
        }()
    }

    wg.Wait()
    return time.Since(start)
}

func main() {
    for _, p := range []int{1, 2, 4} {
        elapsed := benchmarkGOMAXPROCS(p)
        fmt.Printf("GOMAXPROCS=%d: %v\n", p, elapsed)
    }
}

package main

import (
    "fmt"
    "runtime"
    "time"
)

// Leaky goroutine: reads from a channel that is never closed
func leakyGoroutine() {
    ch := make(chan int)
    go func() {
        <-ch  // Blocks forever—no one writes to ch
    }()
}

func main() {
    for i := 0; i < 10; i++ {
        leakyGoroutine()
    }

    time.Sleep(100 * time.Millisecond)
    fmt.Printf("Goroutine count after leak: %d\n", runtime.NumGoroutine())
    // Output: Goroutine count after leak: 11 (10 leaked + 1 main)
}

package main

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

func say(msg string) {
    fmt.Println(msg)
}

func main() {
    var wg sync.WaitGroup

    // Style 1: Named function
    wg.Add(1)
    go func() {
        defer wg.Done()
        say("Style 1: Named function")
    }()

    // Style 2: Anonymous function
    wg.Add(1)
    go func() {
        defer wg.Done()
        fmt.Println("Style 2: Anonymous function")
    }()

    // Style 3: Anonymous function with parameter (recommended—avoids closure trap)
    wg.Add(1)
    go func(msg string) {
        defer wg.Done()
        fmt.Println(msg)
    }("Style 3: With parameter")

    // Style 4: Function as value
    wg.Add(1)
    fn := func() {
        defer wg.Done()
        fmt.Println("Style 4: Function variable")
    }
    go fn()

    wg.Wait()
}

package main

import (
    "fmt"
    "runtime"
    "sync"
)

func main() {
    var memStats runtime.MemStats
    var wg sync.WaitGroup
    num := 100000  // 100k goroutines

    runtime.GC()
    runtime.ReadMemStats(&memStats)
    before := memStats.HeapAlloc

    for i := 0; i < num; i++ {
        wg.Add(1)
        go func(n int) {
            defer wg.Done()
            _ = n
        }(i)
    }
    wg.Wait()

    runtime.ReadMemStats(&memStats)
    after := memStats.HeapAlloc
    perGoroutine := float64(after-before) / float64(num)

    fmt.Printf("Started %d goroutines\n", num)
    fmt.Printf("Memory increase: %.2f MB\n", float64(after-before)/1024/1024)
    fmt.Printf("Per goroutine: ~%.2f KB\n", perGoroutine/1024)
    fmt.Printf("Total goroutine count: %d\n", runtime.NumGoroutine())
}

Started 100000 goroutines
Memory increase: 21.45 MB
Per goroutine: ~0.22 KB
Total goroutine count: 1

package main

import (
    "fmt"
    "runtime"
    "time"
)

func safeWorker(done chan struct{}) {
    <-done  // Wait for exit signal
}

func main() {
    done := make(chan struct{})

    // Start 5 workers
    for i := 0; i < 5; i++ {
        go safeWorker(done)
    }

    fmt.Printf("Goroutine count after start: %d\n", runtime.NumGoroutine())

    // Send exit signal
    close(done)
    time.Sleep(10 * time.Millisecond)

    fmt.Printf("Goroutine count after close: %d\n", runtime.NumGoroutine())
}

// log_processor.go
package main

import (
    "fmt"
    "math/rand"
    "runtime"
    "sync"
    "time"
)

type LogLevel int

const (
    Info LogLevel = iota
    Warn
    Error
    Debug
)

type LogLine struct {
    Timestamp time.Time
    Level     LogLevel
    Message   string
    Source    string
}

type ProcessedLog struct {
    Original    LogLine
    Severity    string
    Alert       bool
    ProcessedAt time.Time
}

func (l LogLine) Process() ProcessedLog {
    // Simulate processing time (IO wait)
    time.Sleep(time.Duration(50+rand.Intn(50)) * time.Millisecond)

    severity := "low"
    alert := false
    switch l.Level {
    case Error:
        severity = "critical"
        alert = true
    case Warn:
        severity = "medium"
    }

    return ProcessedLog{
        Original:    l,
        Severity:    severity,
        Alert:       alert,
        ProcessedAt: time.Now(),
    }
}

// Serial processing
func processSerial(logs []LogLine) []ProcessedLog {
    results := make([]ProcessedLog, 0, len(logs))
    for _, log := range logs {
        results = append(results, log.Process())
    }
    return results
}

// Concurrent processing: worker pool pattern
func processConcurrent(logs []LogLine, workers int) []ProcessedLog {
    jobs := make(chan LogLine, len(logs))
    results := make(chan ProcessedLog, len(logs))

    // Fill jobs
    for _, log := range logs {
        jobs <- log
    }
    close(jobs)

    var wg sync.WaitGroup

    for w := 0; w < workers; w++ {
        wg.Add(1)
        go func(id int) {
            defer wg.Done()
            for log := range jobs {
                results <- log.Process()
            }
        }(w)
    }

    // Wait for all workers to finish, then close results
    go func() {
        wg.Wait()
        close(results)
    }()

    // Collect results
    processed := make([]ProcessedLog, 0, len(logs))
    for r := range results {
        processed = append(processed, r)
    }
    return processed
}

func main() {
    const logCount = 10000  // 10k for demo; can scale to 100k

    levels := []LogLevel{Info, Warn, Error, Debug}
    sources := []string{"api-gateway", "user-service", "payment", "database"}

    logs := make([]LogLine, logCount)
    for i := 0; i < logCount; i++ {
        logs[i] = LogLine{
            Timestamp: time.Now(),
            Level:     levels[rand.Intn(len(levels))],
            Message:   fmt.Sprintf("log-entry-%d", i),
            Source:    sources[rand.Intn(len(sources))],
        }
    }

    fmt.Printf("Log count: %d\n", logCount)
    fmt.Printf("CPU cores: %d, GOMAXPROCS: %d\n", runtime.NumCPU(), runtime.GOMAXPROCS(0))

    // Serial processing
    start := time.Now()
    serialResults := processSerial(logs)
    serialTime := time.Since(start)
    fmt.Printf("\nSerial: %v (%d entries/sec)\n", serialTime,
        int(float64(logCount)/serialTime.Seconds()))

    // Concurrent processing (10 workers)
    runtime.GC()
    start = time.Now()
    concurrentResults := processConcurrent(logs, 10)
    concurrentTime := time.Since(start)
    fmt.Printf("Concurrent(10): %v (%d entries/sec)\n", concurrentTime,
        int(float64(logCount)/concurrentTime.Seconds()))

    // Concurrent processing (50 workers)
    runtime.GC()
    start = time.Now()
    concurrentResults = processConcurrent(logs, 50)
    concurrentTime = time.Since(start)
    fmt.Printf("Concurrent(50): %v (%d entries/sec)\n", concurrentTime,
        int(float64(logCount)/concurrentTime.Seconds()))

    fmt.Printf("\nResult validation: serial=%d, concurrent=%d\n",
        len(serialResults), len(concurrentResults))
    fmt.Printf("Current goroutine count: %d\n", runtime.NumGoroutine())

    // Count alerts
    alertCount := 0
    for _, r := range concurrentResults {
        if r.Alert {
            alertCount++
        }
    }
    fmt.Printf("Alert count: %d\n", alertCount)
}

Log count: 10000
CPU cores: 8, GOMAXPROCS: 8

Serial: 5.2s (1923 entries/sec)
Concurrent(10): 520ms (19230 entries/sec)
Concurrent(50): 110ms (90909 entries/sec)

Result validation: serial=10000, concurrent=10000
Current goroutine count: 1
Alert count: 2453