How to efficiently parallelize array list and control the parallelism?

Question

I have a resourceId array which I need loop in parallel. And generate URL for each resource and then put inside a map which is key (resourcId) and value is url.

I got below code which does the job but I am not sure if this is the right way to do it. I am using sizedwaitgroup here to parallelize the resourceId list. And also using lock on map while writing the data to it. I am sure this isn't efficient code as using lock and then using sizedwaitgroup will have some performance problem.

What is the best and efficient way to do this? Should I use channels here? I want to control the parallelism on how much I should have instead of running length of resourceId list. If any resourceId url generation fails, I want to log that as an error for that resourceId but do not disrupt other go routine running in parallel to get the url generated for other resourceId.

For example: If there are 10 resources, and 2 fails then log error for those 2 and map should have entry for remaining 8.

// running 20 threads in parallel
swg := sizedwaitgroup.New(20)
var mutex = &sync.Mutex{}
start := time.Now()
m := make(map[string]*customerPbV1.CustomerResponse)
for _, resources := range resourcesList {
  swg.Add()
  go func(resources string) {
    defer swg.Done()
    customerUrl, err := us.GenerateUrl(clientId, resources, appConfig)
    if err != nil {
      errs.NewWithCausef(err, "Could not generate the url for %s", resources)
    }
    mutex.Lock()
    m[resources] = customerUrl
    mutex.Unlock()
  }(resources)
}
swg.Wait()

elapsed := time.Since(start)
fmt.Println(elapsed)

Note: Above code will be called at high throughput from multiple reader threads so it needs to perform well.

Answer 1

I'm not sure what sizedwaitgroup is and it's not explained, but overall this approach doesn't look very typical of Go. For that matter, "best" is a matter of opinion, but the most typical approach in Go would be something along these lines:

func main() {
    wg := new(sync.WaitGroup)
    start := time.Now()
    numWorkers := 20
    m := make(map[string]*customerPbV1.CustomerResponse)
    work := make(chan string)
    results := make(chan result)
    for i := 0; i < numWorkers; i++ {
        wg.Add(1)
        go worker(work, results)
    }
    go func() {
        for _, resources := range resourcesList {
            work <- resources
        }
        close(work)
    }()

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

    for result := range results {
        m[result.resources] = result.response
    }

    elapsed := time.Since(start)
    fmt.Println(elapsed)
}

type result struct {
    resources string
    response  *customerPbV1.CustomerResponse
}

func worker(ch chan string, r chan result) {
    for w := range ch {
        customerUrl, err := us.GenerateUrl(clientId, w, appConfig)
        if err != nil {
            errs.NewWithCausef(err, "Could not generate the url for %s", resources)
            continue
        }
        r <- result{w, customerUrl}
    }
}

(Though, based on the name, I would assume errs.NewWithCause doesn't actually handle errors, but returns one, in which case the current code is dropping them on the floor, and a proper solution would have an additional chan error for handling errors:

func main() {
    wg := new(sync.WaitGroup)
    start := time.Now()
    numWorkers := 20
    m := make(map[string]*customerPbV1.CustomerResponse)
    work := make(chan string)
    results := make(chan result)
    errors := make(chan error)
    for i := 0; i < numWorkers; i++ {
        wg.Add(1)
        go worker(work, results, errors)
    }

    go func() {
        for _, resources := range resourcesList {
            work <- resources
        }
        close(work)
    }()

    go func() {
        wg.Wait()
        close(results)
        close(errors)
    }()

    go func() {
        for err := range errors {
            // Do something with err
        }
    }()

    for result := range results {
        m[result.resources] = result.response
    }

    elapsed := time.Since(start)
    fmt.Println(elapsed)
}

type result struct {
    resources string
    response  *customerPbV1.CustomerResponse
}

func worker(ch chan string, r chan result, errs chan error) {
    for w := range ch {
        customerUrl, err := us.GenerateUrl(clientId, w, appConfig)
        if err != nil {
            errs <- errs.NewWithCausef(err, "Could not generate the url for %s", resources)
            continue
        }
        r <- result{w, customerUrl}
    }
}

Answer 2

I have create example code with comment. please read the comment.

note: query function will sleep in 1 second.

package main

import (
    "errors"
    "fmt"
    "log"
    "math/rand"
    "runtime"
    "strconv"
    "sync"
    "time"
)

type Result struct {
    resource string
    val      int
    err      error
}

/*
CHANGE Result struct to this
result struct will collect all you need to create map
type Result struct {
    resources string
    customerUrl *customerPbV1.CustomerResponse
    err error
}
*/

// const numWorker = 8

func main() {
    now := time.Now()
    rand.Seed(time.Now().UnixNano())
    m := make(map[string]int)
    // m := make(map[string]*customerPbV1.CustomerResponse)                 // CHANGE TO THIS

    numWorker := runtime.NumCPU()
    fmt.Println(numWorker)
    chanResult := make(chan Result)

    go func() {
        for i := 0; i < 20; i++ {
            /*
             customerUrl, err := us.GenerateUrl(clientId, resources, appConfig)
             we asume i is resources
             chanResult <- Result {resource: strconv.Itoa(i)}
            */
            chanResult <- Result{ // this will block until chanResult is consume in line 68
                resource: strconv.Itoa(i),
            }
        }
        close(chanResult)
    }()

    var wg sync.WaitGroup
    cr := make(chan Result)
    wg.Add(numWorker)

    go func() {
        wg.Wait()
        close(cr) // NOTE: don't forget to close cr
    }()

    go func() {
        for i := 0; i < numWorker; i++ { // this for loop will run goroutine
            go func(x int) {
                for job := range chanResult { // unblock chan on line 49
                    log.Println("worker", x, "working on", job.resource)
                    x, err := query(job.resource) // TODO: customerUrl, err := us.GenerateUrl(clientId, resources, appConfig)
                    cr <- Result{                 // send to channel, will block until it consume. Consume is in MAIN goroutine "line 84"
                        resource: job.resource,
                        val:      x,
                        err:      err,
                    }
                }
                wg.Done()
            }(i)
        }
    }()

    counterTotal := 0
    counterSuccess := 0
    for res := range cr { // will unblock channel in line 71
        if res.err != nil {
            log.Printf("error found %s. stack trace: %s", res.resource, res.err)
        } else {
            m[res.resource] = res.val // NOTE: save to map
            counterSuccess++
        }
        counterTotal++
    }
    log.Printf("%d/%d of total job run", counterSuccess, counterTotal)
    fmt.Println("final :", m)
    fmt.Println("len m", len(m))

    fmt.Println(runtime.NumGoroutine())
    fmt.Println(time.Since(now))
}

func query(s string) (int, error) {
    time.Sleep(time.Second)
    i, err := strconv.Atoi(s)
    if err != nil {
        return 0, err
    }

    if i%3 == 0 {
        return 0, errors.New("i divided by 3")
    }
    ms := i + 500 + rand.Intn(500)
    return ms, nil
}

playground : https://go.dev/play/p/LeyE9n1hh81

Answer 3

Here is a pure channel solution (playground). I think the performance really depends on the GenerateUrl or in my code generateURL. Also one more thing I would like to point out is that correct term for this is concurrency not parallelism.

package main

import (
    "errors"
    "log"
    "strconv"
    "strings"
)

type result struct {
    resourceID, url string
    err             error
}

func generateURL(resourceID string) (string, error) {
    if strings.HasPrefix(resourceID, "error-") {
        return "", errors.New(resourceID)
    }
    return resourceID, nil
}

func main() {
    // This is the resource IDs
    resources := make([]string, 10000)
    for i := 0; i < 10000; i++ {
        s := strconv.Itoa(i)
        if i%10 == 0 {
            resources[i] = "error-" + s
        } else {
            resources[i] = "id-" + s
        }
    }

    numOfChannel := 20
    // We send result through this channel to the resourceMap
    ch := make(chan result, 10)
    // These are the channels that go routine receives resource ID from
    channels := make([]chan string, numOfChannel)
    // After processing all resources, this channel is used to signal the go routines to exit
    done := make(chan struct{})

    for i := range channels {
        c := make(chan string)
        channels[i] = c

        go func() {
            for {
                select {
                case rid := <-c:
                    u, err := generateURL(rid)
                    ch <- result{rid, u, err}
                case _, ok := <-done:
                    if !ok {
                        break
                    }
                }
            }
        }()
    }

    go func() {
        for i, r := range resources {
            channels[i%numOfChannel] <- r
        }
    }()

    resourceMap := make(map[string]string)
    i := 0
    for p := range ch {
        if p.err != nil {
            log.Println(p.resourceID, p.err)
        } else {
            resourceMap[p.resourceID] = p.url
        }
        i++
        if i == len(resources)-1 {
            break
        }
    }

    close(done)
}