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.
CodePudding user response：
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)
}
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)
}
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}
}
}
CodePudding user response：
Don't set map in goroutine.
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