CODEWITHSUNDEEP
gochannelgoroutine
Jacek Wysocki
Jacek Wysocki

Reputation: 1090

How to safely bypass results from other goroutines when first is completed

I want to ask several servers for data (e.g. multiple read replicas). In this task most important is speed, so first result should be served and all other can be ignored.

I have problem with idiomatic way of bypassing this data. Everything with this problem is ok when it quits (all slower goroutines are not finishing their work, because main process exists). But when we uncomment last line (with Sleep) We can see that other goroutines are doing their work too.

Now I'm pushing data through channel is there any way to not push them?

What is good and safe way of dealing with this kind of problems? 

package main

import (
    "fmt"
    "log"
    "math/rand"
    "time"
)

type Result int

type Conn struct {
    Id int
}

func (c *Conn) DoQuery(params string) Result {
    log.Println("Querying start", params, c.Id)
    time.Sleep(time.Duration(rand.Int31n(1000)) * time.Millisecond)
    log.Println("Querying end", params, c.Id)

    return Result(1000 + c.Id*c.Id)
}

func Query(conns []Conn, query string) Result {
    ch := make(chan Result)
    for _, conn := range conns {
        go func(c Conn) {
            ch <- c.DoQuery(query)
        }(conn)
    }

    return <-ch
}

func main() {
    conns := []Conn{Conn{1}, Conn{2}, Conn{3}, Conn{4}, Conn{5}}
    result := Query(conns, "query!")
    fmt.Println(result)
    // time.Sleep(time.Minute)
}

Upvotes: 2

Views: 361

Answers (1)

captncraig
captncraig

Reputation: 23078

My recommendation would be to make ch a buffered channel with one space per query: ch := make(chan Result, len(conns)). This way each query can run to completion, and will not block on the channel write.

Query can read once and return the first result. When all other goroutines complete, the channel will eventually be garbage collected and everything will go away. With your unbuffered channel, you create a lot of goroutines that can never terminate.

EDIT: If you want to cancel in-flight requests, it can become significantly harder. Some operations and apis provide cancellation, and others don't. With an http request you can use Cancel field on the request struct. Simply provide a channel that you can close to cancel:

func (c *Conn) DoQuery(params string, cancel chan struct{}) Result {
    //error handling omitted. It is important to handle errors properly. 
    req, _ := http.NewRequest(...)
    req.Cancel = cancel
    resp, _ := http.DefaultClient.Do(req)
    //On Cancellation, the request will return an error of some kind.
    return readData(resp)
}
func Query(conns []Conn, query string) Result {
    ch := make(chan Result)
    cancel := make(chan struct{})
    for _, conn := range conns {
        go func(c Conn) {
            ch <- c.DoQuery(query,cancel)
        }(conn)
    }

    first := <-ch
    close(cancel)
    return first
}

This may help if there is a large request to read that you won't care about, but it may or may not actually cancel the request on the remote server. If your query is not http, but a database call or something else, you will need to look into if there is a similar cancellation mechanism you can use.

Upvotes: 5

Related Questions