openlang

Please review my go pipeline implementation (still learning the language)

Hello Everybody,

I am learning go and its concurrency features, and “pipeline” is one of the usual patterns. Here's my version of that pattern, I'd appreciate a review and suggestions about anything I could improve

package main

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

func generate(ctx context.Context, inputs chan<- int) {
    defer close(inputs)
    item := 0
    for {
        select {
        case inputs ← item:
            item++
            time.Sleep(60 * time.Millisecond)
        case <-ctx.Done():
            fmt.Println("Ending producer goroutine and inputs channel gracefuly.")
            return
        }
    }
}

func process(inputs <-chan int, outputs chan<- int, group *sync.WaitGroup) {
    defer group.Done()
    for item := range inputs {
        outputs ← item * item
    }
}

func collect(outputs <-chan int) {
    for square := range outputs {
        fmt.Printf("square %d\n", square)
    }
}

func finish(group *sync.WaitGroup, outputs chan<- int) {
    group.Wait()
    close(outputs)
    fmt.Println("Ending coordinator/worker goroutines and outputs channel gracefuly.")
}

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

    inputs := make(chan int, 960)
    outputs := make(chan int, 960)

    go generate(ctx, inputs)

    var group sync.WaitGroup
    for range 4 {
        group.Add(1)
        go process(inputs, outputs, &group)
    }

    go finish(&group, outputs)

    collect(outputs)

    fmt.Printf("Pipeline drained completly !\n")
    fmt.Printf("Goroutine count : %d\n", runtime.NumGoroutine()) // expected to be 1
}

Also, what other important, widely used concurrency patterns should I learn, pubsub for example.

Cheers,

FV

The conversation continues 3

Follow the reasoning
  1. Permalink

    Maybe look at alternative Go pipelines and the ideas underlying the pattern. You can then get a better understanding of the problem before deciding what changes your implementation needs…

    That's my little contribution…

  2. Permalink

    If “pipeline” here means a pipeline of the Linux shell variety, this isn't one. What makes those pipelines so good is their concealment of the synchronization mechanism. That's why seeing Sync code here confuses me.

Other sides of the same question

  1. Converting functions into continuations for deep handlers

    In the documentation's deep-handler example, there's this paragraph: A computation f can itself perform effects. When f perfo…

    Replies 3Latest activity ocaml