1. First meet errgroup
WaitGroup is mainly used to control concurrent subtasks under task groups. Its specific method is to add the number of tasks through the Add method before the subtask goroutine is executed. Done is called at the end of the subtask goroutine to mark the number of completed tasks. The main task goroutine waits for all tasks to complete through the Wait method before the subsequent logic can be executed.
package main
import (
"net/http"
"sync"
)
func main() {
var wg
var urls = []string{
"/",
"/",
"http:///",
}
for _, url := range urls {
(1)
go func(url string) {
defer ()
resp, err := (url)
if err != nil {
return
}
()
}(url)
}
()
}
In the above example code, we use three goroutines to request concurrently, and the main task goroutine will stop blocking until all subtask goroutines have completed access.
But in actual project code, the execution of subtask goroutines is not always smooth, they may produce errors. WaitGroup does not tell us how to throw the child goroutine to the main task groutine when an error occurs.
You can consider using errgroup at this time
package main
import (
"fmt"
"net/http"
"/x/sync/errgroup"
)
func main() {
var urls = []string{
"/",
"/",
"http:///",
}
g := new()
for _, url := range urls {
url := url
(func() error {
resp, err := (url)
if err != nil {
(err)
return err
}
("get [%s] success: [%d] \n", url, )
return ()
})
}
if err := (); err != nil {
(err)
} else {
("All success!")
}
}
The results are as follows:
get [/] success: [200]
Get "http:///": dial tcp: lookup : no such host
Get "/": dial tcp 142.251.42.241:80: i/o timeout
Get "http:///": dial tcp: lookup : no such host
As you can see, an error occurred in the execution of the fetch and the child groutines of both urls, and the first error message was successfully captured in the main task goroutine.
In addition to having WaitGroup's control capabilities and error propagation functions, errgroup also has the most important context backpropagation mechanism. Let's take a look at its design.
2. Errgroup source code analysis
The design of errgroup is very concise, all the codes are as follows
type Group struct {
cancel func()
wg
errOnce
err error
}
func WithContext(ctx ) (*Group, ) {
ctx, cancel := (ctx)
return &Group{cancel: cancel}, ctx
}
func (g *Group) Wait() error {
()
if != nil {
()
}
return
}
func (g *Group) Go(f func() error) {
(1)
go func() {
defer ()
if err := f(); err != nil {
(func() {
= err
if != nil {
()
}
})
}
}()
}
As you can see, the implementation of errgroup relies on the structure group. It inherits the characteristics of WaitGroup through encapsulation, starts a new subtask goroutine in the Go() method, and blocks and waits through Wait in the Wait() method.
Meanwhile, Group utilizes guarantee that it has and will only retain the first child goroutine error.
Finally, the cancel function generated by the Group through the embed method can promptly propagate the cancel signal of the Context in a timely manner by calling the cancel function when an error occurs in the sub goroutine. Of course, this feature requires the cooperation of user code.
3. Errgroup context cancellation
Documentation in errgroup (//x/[email protected]/errgroup#example-Group-Pipeline), it is based on the pipeline of Go official documentation (/pipelines) implements an example of a context cancellation demonstration in a task group goroutine. However, the premise knowledge of this demo is slightly more, and this article is based on its ideas here to provide an easy-to-understand example of use.
package main
import (
"context"
"fmt"
"/x/sync/errgroup"
)
func main() {
g, ctx := (())
dataChan := make(chan int, 20)
// Data production end task sub goroutine (func() error {
defer close(dataChan)
for i := 1; ; i++ {
if i == 10 {
return ("data 10 is wrong")
}
dataChan <- i
(("sending %d", i))
}
})
// Data consumer task sub goroutine for i := 0; i < 3; i++ {
(func() error {
for j := 1; ; j++ {
select {
case <-():
return ()
case number := <-dataChan:
(("receiving %d", number))
}
}
})
}
// The main task goroutine waits for the pipeline to end the data flow err := ()
if err != nil {
(err)
}
("main goroutine done!")
}
In the above example, we simulate a data transfer pipeline. In the data production and consumption tasks concentration, there are four sub-tasks goroutines: one goroutine for production data and three goroutines for consumption data. When there is an error data on the data producer (data equals 10), we stop the production and consumption of the data and throw the error back to the execution logic of main goroutine.
As you can see, because of the embedding of the Context Cancle function in the errgroup, we can also reversely control the task context in the subtask goroutine.
When a certain operation of the program is performed, the output result is as follows:
sending 1
sending 2
sending 3
sending 4
sending 5
sending 6
sending 7
sending 8
sending 9
receiving 1
receiving 3
receiving 2
receiving 4
data 10 is wrong
main goroutine done!
4. Summary
errgroup is Go's official concurrent primitive supplement library, which is less core than the primitives provided in the standard library. Here we summarize the characteristics of errgroup.
- Inherited the functions of WaitGroup
- Error propagation: Can return the first error that occurred in the task group, but only the error can be returned.
- context signal propagation: If there is loop logic in the subtask goroutine, you can add logic. At this time, the subtask execution will be ended in advance through the cancel signal of context.
This is the end of this article about the detailed explanation of the use of errgroup in the Golang sync package. For more related Golang errgroup content, please search for my previous articles or continue browsing the related articles below. I hope everyone will support me in the future!