Pattern to avoid nested selects with repeated cases

user_2545722 0
Viewed 104

I'm reading the Concurrency in Go book from O'Reilly and found this code example:

doWork := func(
    done <-chan interface{},
    pulseInterval time.Duration,
) (<-chan interface{}, <-chan time.Time) {
    heartbeat := make(chan interface{})
    results := make(chan time.Time)
    go func() {
        defer close(heartbeat)
        defer close(results)
        pulse := time.Tick(pulseInterval)
        workGen := time.Tick(2 * pulseInterval) // this just creates workload
        sendPulse := func() {
            select {
            case heartbeat <- struct{}{}:
            default:
            }
        }
        sendResult := func(r time.Time) {
            for {
                select {
                case <-done:
                    return
                case <-pulse:
                    sendPulse()
                case results <- r:
                    return
                }
            }
        }
        for {
            select {
            case <-done:
                return
            case <-pulse:
                sendPulse()
            case r := <-workGen:
                sendResult(r)
            }
        }
    }()
    return heartbeat, results
}

It seemed very strange that two select statements were being used for a simple heartbeat. Then I understood that the reason behind this is that sendResults SHOULDN'T block trying to send a result to the results channel. If nobody is receiving from that channel, it will effectively block the goroutine and stop sending the heartbeat.

Then I thought... Why didn't they code it like this then?

doWork := func(
    done <-chan interface{},
    pulseInterval time.Duration,
) (<-chan interface{}, <-chan time.Time) {
    heartbeat := make(chan interface{})
    results := make(chan time.Time)
    go func() {
        defer close(heartbeat)
        defer close(results)
        pulse := time.Tick(pulseInterval)
        workGen := time.Tick(2 * pulseInterval) // this just creates workload
        sendPulse := func() {
            select {
            case heartbeat <- struct{}{}:
            default:
            }
        }
        for {
            select {
            case <-done:
                return
            case <-pulse:
                sendPulse()
            case results <- <- workgen:
                fmt.Println("Send result")
            }
        }
    }()
    return heartbeat, results
}

And realized that it is because if workgen has an available item and I read it but then nobody is reading from results, the workgen item will be lost.

Question

Is there any other pattern to avoid these nested selects that have to repeat all the cases not to miss a value?

Thoughts

Wouldn't all these problems be solved IF a select statement first validated that the results channel is waiting to receive a value instead of executing the right hand expression first?

In other words, if I have this results <- <- workgen, it would be better if the select said "hey, nobody is reading from results, let's not try to evaluate <- workgen". This seems like a very particular design choice for me and I think it is what makes a lot of nested selects necessary to begin with.

I think a case should first try to evaluate if the channels that are in the expression are ready to receive or send data, and then execute the expressions.

go concurrency
2 Answers

Use temporary variables for workgen and results in the select. Set these variables to nil or the original values depending on the current state.

Setting the variables to nil disables the branch of the select because channel operations on nil channels block forever.

    workGenT := workGen
    var resultsT chan time.Time
    var result time.Time
    for {
        select {
        case <-done:
            return
        case <-pulse:
            sendPulse()
        case result = <-workgenT:
            workgenT = nil
            resultsT = results
        case resultsT <- result:
            resultsT = nil
            workgenT = workgen
        }
    }

Because only one of workgenT and resultsT are non-nil, the code above has two states: The program is either waiting to receive a result or waiting to send a result.

Admin1

Another simple and similar pattern.

func hBeatTicker(ctx context.Context, ch chan<- struct{}, t time.Duration) {
    tick := time.NewTicker(t)
    for {
        select {
        case <-ctx.Done():
            close(ch)
            tick.Stop()
            return
        case <-tick.C:
            ch <- struct{}{}
        }
    }
}

func workGen(ctx context.Context, ch chan<- time.Time, t time.Duration) {
    tick := time.NewTicker(2 * t)
    for {
        select {
        case <-ctx.Done():
            close(ch)
            tick.Stop()
            return
        case ch <- <-tick.C:
        }
    }
}

func do(ctx context.Context, d time.Duration) (<-chan struct{}, <-chan time.Time) {
    heartbeat, results := make(chan struct{}), make(chan time.Time)

    go workGen(ctx, results, d)
    go hBeatTicker(ctx, heartbeat, d)

    return heartbeat, results
}
user_58214080
Related