Go: أنماط التوازي في Go

// Bad code: single-threaded, not utilizing CPU
func processLogs(logs []string) {
    for _, log := range logs {
        parsed := parse(log)       // CPU-intensive
        enriched := enrich(parsed) // CPU-intensive
        save(enriched)              // I/O-intensive
    }
    // 10 million log entries: 3 hours!
}

// pipeline.go
package main

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

type LogEntry struct {
    Raw     string
    Level   string
    Message string
    Time    time.Time
}

func main() {
    logs := make([]string, 100)
    for i := range logs {
        logs[i] = fmt.Sprintf("log-%d", i)
    }
    start := time.Now()

    // Pipeline: generator → parse → enrich → save
    rawCh := generator(logs)
    parsedCh := parseStage(rawCh, 4)       // 4 parse workers
    enrichedCh := enrichStage(parsedCh, 4)  // 4 enrich workers
    saveStage(enrichedCh, 2)                // 2 save workers

    fmt.Printf("Processing complete, elapsed: %v\n", time.Since(start))
}

// Stage 1: Generator (data source)
func generator(logs []string) <-chan string {
    out := make(chan string, 100)
    go func() {
        defer close(out)
        for _, log := range logs {
            out <- log
        }
    }()
    return out
}

// Stage 2: Parse (Fan-out)
func parseStage(in <-chan string, workers int) <-chan LogEntry {
    out := make(chan LogEntry, 100)
    for i := 0; i < workers; i++ {
        go func() {
            for raw := range in {
                time.Sleep(1 * time.Millisecond) // Simulate parsing
                out <- LogEntry{Raw: raw, Level: "INFO", Message: raw, Time: time.Now()}
            }
        }()
    }
    return out
}

// Stage 3: Enrich (Fan-out)
func enrichStage(in <-chan LogEntry, workers int) <-chan LogEntry {
    out := make(chan LogEntry, 100)
    for i := 0; i < workers; i++ {
        go func() {
            for entry := range in {
                time.Sleep(2 * time.Millisecond) // Simulate data enrichment
                out <- entry
            }
        }()
    }
    return out
}

// Stage 4: Save (Fan-in to 2 workers)
func saveStage(in <-chan LogEntry, workers int) {
    var wg sync.WaitGroup
    for i := 0; i < workers; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for entry := range in {
                _ = entry // Simulate writing to database
                time.Sleep(500 * time.Microsecond)
            }
        }()
    }
    wg.Wait()
}

package main

import (
    "fmt"
)

// Stage 1: Generate
func generate(nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for _, n := range nums {
            out <- n
        }
    }()
    return out
}

// Stage 2: Square
func square(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in {
            out <- n * n
        }
    }()
    return out
}

// Stage 3: Double
func double(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in {
            out <- n * 2
        }
    }()
    return out
}

func main() {
    // Pipeline: generate → square → double
    for result := range double(square(generate(1, 2, 3, 4, 5))) {
        fmt.Println(result)
    }
    // Output: 2, 8, 18, 32, 50  (n²×2)
}

package main

import (
    "fmt"
    "sync"
)

// Stage 1: Generate
func generate(nums ...int) <-chan int {
    out := make(chan int, 10)
    go func() {
        defer close(out)
        for _, n := range nums {
            out <- n
        }
    }()
    return out
}

// Stage 2: Worker (Fan-out target)
func worker(id int, in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for n := range in {
            result := n * n
            fmt.Printf("Worker %d: %d^2 = %d\n", id, n, result)
            out <- result
        }
    }()
    return out
}

// Fan-in: merge multiple channels
func fanIn(channels ...<-chan int) <-chan int {
    out := make(chan int)
    var wg sync.WaitGroup

    for _, ch := range channels {
        wg.Add(1)
        go func(c <-chan int) {
            defer wg.Done()
            for v := range c {
                out <- v
            }
        }(ch)
    }

    go func() {
        wg.Wait()
        close(out)
    }()

    return out
}

func main() {
    // Fan-out: one generator distributes to 3 workers
    in := generate(1, 2, 3, 4, 5, 6)
    w1 := worker(1, in)
    w2 := worker(2, in)
    w3 := worker(3, in)

    // Fan-in: merge results from 3 workers into one channel
    results := fanIn(w1, w2, w3)

    for result := range results {
        fmt.Printf("Result: %d\n", result)
    }
}

package main

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

type Job struct {
    ID      int
    Payload string
}

type Result struct {
    JobID  int
    Output string
    Err    error
}

func worker(id int, jobs <-chan Job, results chan<- Result, wg *sync.WaitGroup) {
    defer wg.Done()
    for job := range jobs {
        fmt.Printf("Worker %d processing job %d\n", id, job.ID)
        time.Sleep(100 * time.Millisecond) // Simulate work
        results <- Result{
            JobID:  job.ID,
            Output: fmt.Sprintf("processed by worker %d", id),
        }
    }
}

func main() {
    const numJobs = 20
    const numWorkers = 5

    jobs := make(chan Job, numJobs)
    results := make(chan Result, numJobs)

    // Start Worker Pool
    var wg sync.WaitGroup
    for i := 0; i < numWorkers; i++ {
        wg.Add(1)
        go worker(i, jobs, results, &wg)
    }

    // Send jobs
    for i := 0; i < numJobs; i++ {
        jobs <- Job{ID: i, Payload: fmt.Sprintf("data-%d", i)}
    }
    close(jobs)

    // Wait for all workers to complete
    wg.Wait()
    close(results)

    // Collect results
    for result := range results {
        fmt.Printf("Job %d → %s\n", result.JobID, result.Output)
    }
}

package main

import (
    "fmt"
    "time"
)

// orDone wraps a channel, making it support cancellation via a done channel
func orDone(done <-chan struct{}, c <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for {
            select {
            case <-done:
                return
            case v, ok := <-c:
                if !ok {
                    return
                }
                select {
                case out <- v:
                case <-done:
                    return
                }
            }
        }
    }()
    return out
}

func generateWithCancel(done <-chan struct{}, nums ...int) <-chan int {
    out := make(chan int)
    go func() {
        defer close(out)
        for _, n := range nums {
            select {
            case <-done:
                return
            case out <- n:
            }
        }
    }()
    return out
}

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

    // Start generator (will be canceled after reaching 5)
    nums := generateWithCancel(done, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10)

    // Consume safely via orDone
    for n := range orDone(done, nums) {
        fmt.Printf("Processing: %d\n", n)
        if n >= 5 {
            close(done) // Cancel all operations
        }
    }

    fmt.Println("Canceled")
    time.Sleep(100 * time.Millisecond) // Wait for goroutines to exit
}

package main

import (
    "fmt"
    "time"

    "golang.org/x/sync/errgroup"
)

func main() {
    // Create an errgroup
    // The first goroutine to return an error will cancel the others
    g := errgroup.Group{}

    // Start 3 goroutines
    for i := 0; i < 3; i++ {
        id := i
        g.Go(func() error {
            return doWork(id)
        })
    }

    // Wait for all goroutines to complete, get the first error
    if err := g.Wait(); err != nil {
        fmt.Printf("Task failed: %v\n", err)
    } else {
        fmt.Println("All succeeded")
    }
}

func doWork(id int) error {
    fmt.Printf("Worker %d starting\n", id)
    time.Sleep(time.Duration(id+1) * 500 * time.Millisecond)

    if id == 1 {
        return fmt.Errorf("worker %d failed", id)
    }

    fmt.Printf("Worker %d complete\n", id)
    return nil
}

package main

import (
    "context"
    "fmt"
    "time"

    "golang.org/x/sync/errgroup"
)

func main() {
    // errgroup with Context
    ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
    defer cancel()

    g, ctx := errgroup.WithContext(ctx)

    // Workers check if Context is canceled
    for i := 0; i < 5; i++ {
        id := i
        g.Go(func() error {
            select {
            case <-time.After(time.Duration(id+1) * time.Second):
                fmt.Printf("Worker %d complete\n", id)
                return nil
            case <-ctx.Done():
                fmt.Printf("Worker %d canceled: %v\n", id, ctx.Err())
                return ctx.Err()
            }
        })
    }

    if err := g.Wait(); err != nil {
        fmt.Printf("errgroup stopped: %v\n", err)
    }
}

// data_pipeline.go
package main

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

// ---------- Data types ----------

type DataPoint struct {
    ID        int
    Value     float64
    Timestamp time.Time
}

type ProcessedData struct {
    DataPoint
    Normalized bool
    Anomaly    bool
    Category   string
}

// ---------- Pipeline Stages ----------

// Stage 1: Source (data source)
func source(ctx context.Context, count int) <-chan DataPoint {
    out := make(chan DataPoint, 100)
    go func() {
        defer close(out)
        for i := 0; i < count; i++ {
            select {
            case <-ctx.Done():
                return
            case out <- DataPoint{
                ID:        i,
                Value:     rand.Float64() * 100,
                Timestamp: time.Now(),
            }:
            }
        }
    }()
    return out
}

// Stage 2: Normalize (Fan-out)
func normalize(ctx context.Context, in <-chan DataPoint, workers int) <-chan ProcessedData {
    out := make(chan ProcessedData, 100)
    var wg sync.WaitGroup

    for i := 0; i < workers; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for dp := range orDoneDataPoint(ctx.Done(), in) {
                select {
                case <-ctx.Done():
                    return
                case out <- ProcessedData{
                    DataPoint:  dp,
                    Normalized: true,
                    Category:   categorize(dp.Value),
                }:
                }
            }
        }()
    }

    go func() {
        wg.Wait()
        close(out)
    }()

    return out
}

// Stage 3: Detect Anomaly (Fan-out)
func detectAnomaly(ctx context.Context, in <-chan ProcessedData, workers int) <-chan ProcessedData {
    out := make(chan ProcessedData, 100)
    var wg sync.WaitGroup

    for i := 0; i < workers; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for pd := range orDoneProcessedData(ctx.Done(), in) {
                pd.Anomaly = pd.Value > 90 || pd.Value < 10
                select {
                case <-ctx.Done():
                    return
                case out <- pd:
                }
            }
        }()
    }

    go func() {
        wg.Wait()
        close(out)
    }()

    return out
}

// Stage 4: Sink (output, Fan-in)
func sink(ctx context.Context, in <-chan ProcessedData) {
    var normalCount, anomalyCount int
    for pd := range orDoneProcessedData(ctx.Done(), in) {
        if pd.Anomaly {
            anomalyCount++
            fmt.Printf("[ANOMALY] ID=%d, Value=%.2f, Category=%s\n",
                pd.ID, pd.Value, pd.Category)
        } else {
            normalCount++
        }
    }
    fmt.Printf("\nSummary: normal=%d, anomaly=%d\n", normalCount, anomalyCount)
}

// ---------- Helper functions ----------

func orDoneDataPoint(done <-chan struct{}, in <-chan DataPoint) <-chan DataPoint {
    out := make(chan DataPoint)
    go func() {
        defer close(out)
        for {
            select {
            case <-done:
                return
            case v, ok := <-in:
                if !ok {
                    return
                }
                select {
                case out <- v:
                case <-done:
                    return
                }
            }
        }
    }()
    return out
}

func orDoneProcessedData(done <-chan struct{}, in <-chan ProcessedData) <-chan ProcessedData {
    out := make(chan ProcessedData)
    go func() {
        defer close(out)
        for {
            select {
            case <-done:
                return
            case v, ok := <-in:
                if !ok {
                    return
                }
                select {
                case out <- v:
                case <-done:
                    return
                }
            }
        }
    }()
    return out
}

func categorize(value float64) string {
    switch {
    case value < 30:
        return "low"
    case value < 70:
        return "medium"
    default:
        return "high"
    }
}

func main() {
    ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
    defer cancel()

    fmt.Println("Starting real-time data processing pipeline...")

    // Build Pipeline
    data := source(ctx, 1000)
    normalized := normalize(ctx, data, 5)           // 5 normalize workers
    detected := detectAnomaly(ctx, normalized, 3)   // 3 anomaly detection workers
    sink(ctx, detected)

    fmt.Println("Pipeline processing complete")
}