Golang async but sequential enqueue with buffered channel

Question

The objective of this program is to call enqueue sequentially with multiple URLs, and never block, fetching the URLs asynchronously, but on the order they were entered.

In other words: keep calling enqueue to add urls to a buffered channel (queue) (possibly with a specific callback), and consume one by one in the order they where entered in the queue. I used WaitGroup to prevent the go program from immediately exiting, but I feel I'm not taking advantage of the buffered channel at all.

package main
import (
 "fmt"
 "net/http"
 "sync"
)
type callback func(url string, res *http.Response, err error)
type callbackRequest struct {
 cb callback
 url string
}
var queue = make(chan callbackRequest, 1)
var wg sync.WaitGroup
func worker() {
 for {
 req := <-queue
 fmt.Println("processing", req.url)
 res, err := http.Get(req.url)
 req.cb(req.url, res, err)
 }
}
func enqueue(url string) {
 fmt.Println("enqueue", url)
 cb := func(url string, res *http.Response, err error) {
 if err != nil {
 fmt.Println("error")
 }
 fmt.Println(url, res.Status)
 wg.Done()
 }
 cbReq := callbackRequest{url: url, cb: cb}
 wg.Add(1)
 go func(cbr callbackRequest) {
 queue <- cbReq
 }(cbReq)
}
func main() {
 go worker()
 enqueue("http://google.com")
 enqueue("http://reddit.com")
 enqueue("http://yahoo.com")
 enqueue("http://bing.com")
 enqueue("http://google.com")
 enqueue("http://yahoo.com")
 enqueue("http://reddit.com")
 enqueue("http://bing.com")
 enqueue("http://google.com")
 wg.Wait()
}

Output:

$ go run enqueue/main.go 
enqueue http://google.com
enqueue http://reddit.com
enqueue http://yahoo.com
enqueue http://bing.com
enqueue http://google.com
enqueue http://yahoo.com
enqueue http://reddit.com
enqueue http://bing.com
enqueue http://google.com
processing http://google.com
http://google.com 200 OK
processing http://reddit.com
http://reddit.com 200 OK
processing http://yahoo.com
http://yahoo.com 200 OK
processing http://bing.com
http://bing.com 200 OK
processing http://google.com
http://google.com 200 OK
processing http://yahoo.com
http://yahoo.com 504 Gateway Timeout
processing http://reddit.com
http://reddit.com 200 OK
processing http://bing.com
http://bing.com 200 OK
processing http://google.com
http://google.com 200 OK

Answer 1

I believe that you are correct. The channel buffer really doesn't matter here because you are manually building an un-bounded buffer here:

go func(cbr callbackRequest) {
 queue <- cbReq
}(cbReq)

The queue <- cbReq line will block until queue has room, but since it is running in its own goroutine, it won't block progress for the entire program. Once there is room in queue, the goroutine will progress and immediately terminate, having deposited cbReq in the queue.

You can think about it as a bunch of people waiting in line to drop off packages at a post office for shipping. As the postal employee gets time, he removes the packages from the counter in the order they were brought in (maybe they are on a conveyor belt) and sends them on their way. If the counter (or conveyor belt) is large enough, several packages can be placed on it at one time. In this case, once a person has deposited her package on the counter, she may leave. If the counter was smaller, the people would have to wait in line longer, but their packages would still ship in the same order.

That is kind of a complicated analogy, but maybe it will help. The idea is that the queue either happens in the channel buffer, or in the blocked goroutines.

The one wrinkle here is that I am not entirely sure about the semantics of a blocked channel insertion operation. It appears from your output that the first goroutine to attempt to insert to a channel will be the first goroutine to be allowed to insert to that channel, but I couldn't find a reference to back that up.

• George Lesica: Another option might be to consider whether you need the requests to run in a particular order or whether it is just the output you care about. If the latter, you could use a pool of goroutines and a single channel to feed the pool. In this case, you would add an index to each callbackRequest that would be used to re-order the results after they've come back (changing the code would involve a bit more than that, but that's the general idea).