Priority in Go select statement workaround

Question

I wish to have a go routine listening on two channels, blocked when both channels are drained. However, if both channels contains data, I want one to be drained before the other is handled.

In the working example below I wish all out to be drained before exit is handled. I use a select-statement which doesn't have any priority order. How might I get around the problem, making all 10 out-values be handled before the exit?

package main

import "fmt"

func sender(out chan int, exit chan bool){
    for i := 1; i <= 10; i++ {
        out <- i
    } 
    exit <- true
}

func main(){
    out := make(chan int, 10)
    exit := make(chan bool)

    go sender(out, exit)

    L:
    for {
        select {
            case i := <-out:
                fmt.Printf("Value: %d\n", i)
            case <-exit:
                fmt.Println("Exiting")
                break L
        }
    }
    fmt.Println("Did we get all 10? Most likely not")
}

Answer 1

The language supports this natively and no workaround is required. It's very simple: the quit channel should only be visible to the producer. On quit, the producer closes the channel. Only when the channel is empty and closed does the consumer quit. This is made possible by ranging over the channel.

Here is an example to illustrate:

package main

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

var (
    produced  = 0
    processed = 0
)

func produceEndlessly(out chan int, quit chan bool) {
    defer close(out)
    for {
        select {
        case <-quit:
            fmt.Println("RECV QUIT")
            return
        default:
            out <- rand.Int()
            time.Sleep(time.Duration(rand.Int63n(5e6)))
            produced++
        }
    }
}

func quitRandomly(quit chan bool) {
    d := time.Duration(rand.Int63n(5e9))
    fmt.Println("SLEEP", d)
    time.Sleep(d)
    fmt.Println("SEND QUIT")
    quit <- true
}

func main() {
    vals, quit := make(chan int, 10), make(chan bool)
    go produceEndlessly(vals, quit)
    go quitRandomly(quit)
    for x := range vals {
        fmt.Println(x)
        processed++
        time.Sleep(time.Duration(rand.Int63n(5e8)))
    }
    fmt.Println("Produced:", produced)
    fmt.Println("Processed:", processed)
}

Answer 2

package main

import "fmt"

func sender(out chan int, exit chan bool) {
    for i := 1; i <= 10; i++ {
        out <- i
    }
    exit <- true
}

func main() {
    out := make(chan int, 10)
    exit := make(chan bool)

    go sender(out, exit)

    for {
        select {
        case i := <-out:
            fmt.Printf("Value: %d\n", i)
            continue
        default:
        }
        select {
        case i := <-out:
            fmt.Printf("Value: %d\n", i)
            continue
        case <-exit:
            fmt.Println("Exiting")
        }
        break
    }
    fmt.Println("Did we get all 10? I think so!")
}

The default case of the first select makes it non-blocking. The select will drain the out channel without looking at the exit channel, but otherwise will not wait. If the out channel is empty, it immediately drops to the second select. The second select is blocking. It will wait for data on either channel. If an exit comes, it handles it and allows the loop to exit. If data comes, it goes back up the top of the loop and back into drain mode.

Answer 3

Another approach:

package main

import "fmt"

func sender(c chan int) chan int {
        go func() {
                for i := 1; i <= 15; i++ {
                        c <- i
                }
                close(c)
        }()
        return c
}

func main() {
        for i := range sender(make(chan int, 10)) {
                fmt.Printf("Value: %d\n", i)
        }
        fmt.Println("Did we get all 15? Surely yes")
}

---

$ go run main.go
Value: 1
Value: 2
Value: 3
Value: 4
Value: 5
Value: 6
Value: 7
Value: 8
Value: 9
Value: 10
Value: 11
Value: 12
Value: 13
Value: 14
Value: 15
Did we get all 15? Surely yes
$ 

Answer 4

Here's a general idiom that solves the select's priority problem.

Yes, it's not nice to say a least, but does what is needed for 100%, no pitfalls and no hidden limitations.

Here's a short code example, and explanation follows.

package main

import(
    "fmt"
    "time"
)

func sender(out chan int, exit chan bool) {
    for i := 1; i <= 10; i++ {
        out <- i
    }

    time.Sleep(2000 * time.Millisecond)
    out <- 11
    exit <- true
}

func main(){
    out := make(chan int, 20)
    exit := make(chan bool)

    go sender(out, exit)

    time.Sleep(500 * time.Millisecond)

    L:
    for {
        select {
        case i := <-out:
            fmt.Printf("Value: %d\n", i)
        default:
            select {
            case i := <-out:
                fmt.Printf("Value: %d\n", i)
            case <-exit:
                select {
                case i := <-out:
                    fmt.Printf("Value: %d\n", i)
                default:
                    fmt.Println("Exiting")
                    break L
                }
            }
        }
    }
    fmt.Println("Did we get all 10? Yes.")
    fmt.Println("Did we get 11? DEFINITELY YES")
}

And, here's how it works, the main() from above, annotated:

func main(){
    out := make(chan int, 20)
    exit := make(chan bool)
    go sender(out, exit)
    time.Sleep(500 * time.Millisecond)
    L:
    for {
        select {

            // here we go when entering next loop iteration
            // and check if the out has something to be read from

            // this select is used to handle buffered data in a loop

        case i := <-out:
            fmt.Printf("Value: %d\n", i)
        default:
            // else we fallback in here

            select {

                // this select is used to block when there's no data in either chan

            case i := <-out:
            // if out has something to read, we unblock, and then go the loop round again

                fmt.Printf("Value: %d\n", i)
            case <-exit:
                select {

                    // this select is used to explicitly propritize one chan over the another,
                    // in case we woke up (unblocked up) on the low-priority case

                    // NOTE:
                    // this will prioritize high-pri one even if it came _second_, in quick
                    // succession to the first one

                case i := <-out:
                    fmt.Printf("Value: %d\n", i)
                default:
                    fmt.Println("Exiting")
                    break L
                }
            }
        }
    }

    fmt.Println("Did we get all 10? Yes.")
    fmt.Println("Did we get 11? DEFINITELY YES")
}

NOTE: Before playing tricks with prioritizations, MAKE SURE YOU ARE SOLVING THE RIGHT PROBLEM.

Chances are, it can be solved differently.

Still, to have prioritized select in Go would have been great thing. Just a dream..

NOTE: This is quite a similar answer https://stackoverflow.com/a/45854345/11729048 on this thread, but there is only two select-s are nested, not three ones as I did. What's the difference? My approach is more efficient, and there we explicitly expect to handle random choices on each loop iteration.

However, if the high-priority channel isn't buffered, and/or you don't expect bulk data on it, only the sporadic single events, then the simpler two-stage idiom (as in that answer) will suffice:

L:
for {
    select {
    case i := <-out:
        fmt.Printf("Value: %d\n", i)
    case <-exit:
        select {
        case i := <-out:
            fmt.Printf("Value: %d\n", i)
        default:
            fmt.Println("Exiting")
            break L
        }
    }
}

It is basically 2 and 3 stages, the 1 being removed.

And once again: in like 90% cases you think you do need to prioritize chan switch cases, you really don't.

And here's a one-liner, that can be wrapped in a macro:

for {
    select { case a1 := <-ch_p1: p1_action(a1); default: select { case a1 := <-ch_p1: p1_action(a1); case a2 := <-ch_p2: select { case a1 := <-ch_p1: p1_action(a1); default: p2_action(a2); }}}
}

And what if you want to prioritize more than two cases?

Then you have two options. First one - build a tree, using intermediate goroutines, so that each fork is exactly binary (the above idiom).

The second option is to make the priority-fork more then double.

Here's an example of three priorities:

for {
    select {
    case a1 := <-ch_p1:
        p1_action(a1)
    default:
        select {
        case a2 := <-ch_p2:
            p2_action(a2)
        default:
            select {    // block here, on this select
            case a1 := <-ch_p1:
                p1_action(a1)
            case a2 := <-ch_p2:
                select {
                case a1 := <-ch_p1:
                    p1_action(a1)
                default:
                    p2_action(a2)
                }
            case a3 := <-ch_p3:
                select {
                case a1 := <-ch_p1:
                    p1_action(a1)
                case a2 := <-ch_p2:
                    p1_action(a2)
                default:
                    p2_action(a3)
                }
            }
        }
    }
}

That is, the whole structure is conceptually split into three parts, as the original (binary) one.

One again: chances are, you can design your system so that you can avoid this mess.

P.S., the rhetoric question: why Golang doesn't have it built in into the language??? The question is rhetoric one.

Answer 5

Here's another option.

Consumer Code:

  go func() {
    stop := false
    for {
      select {
      case item, _ := <-r.queue:
        doWork(item)
      case <-r.stopping:
        stop = true
      }
      if stop && len(r.queue) == 0 {
        break
      }
    }
  }()

Answer 6

I have created one rather simple workaround. It does what I want, but if anyone else has a better solution, please let me know:

exiting := false
for !exiting || len(out)>0 {
    select {
        case i := <-out:
            fmt.Printf("Value: %d\n", i)
        case <-exit:
            exiting = true
            fmt.Println("Exiting")
    }
}

Instead of exiting on receiving, I flag an exit, exiting once I've made sure nothing is left in chan out.

Answer 7

I think Sonia's answer is incorrect.This is my solution,a little bit complicate.

package main

import "fmt"

func sender(out chan int, exit chan bool){
    for i := 1; i <= 10; i++ {
        out <- i
    } 
    exit <- true
}

func main(){
    out := make(chan int, 10)
    exit := make(chan bool)

    go sender(out, exit)

    L:
    for {
        select {
            case i := <-out:
                fmt.Printf("Value: %d\n", i)
            case <-exit:
                for{
                    select{
                    case i:=<-out:
                        fmt.Printf("Value: %d\n", i)
                    default:
                        fmt.Println("Exiting")
                        break L
                    }
                }
                fmt.Println("Exiting")
                break L
        }
    }
    fmt.Println("Did we get all 10? Yes!")
}

Answer 8

In my case, I really wanted to prioritise data from one channel over another, and not just have an out-of-band exit signal. For the benefit of anyone else with the same issue I think this approach works without the potential race condition:

OUTER:
for channelA != nil || channelB != nil {

    select {

    case typeA, ok := <-channelA:
        if !ok {
            channelA = nil
            continue OUTER
        }
        doSomething(typeA)

    case nodeIn, ok := <-channelB:
        if !ok {
            channelB = nil
            continue OUTER
        }

        // Looped non-blocking nested select here checks that channelA
        // really is drained before we deal with the data from channelB
        NESTED:
        for {
            select {
            case typeA, ok := <-channelA:
                if !ok {
                    channelA = nil
                    continue NESTED
                }
                doSomething(typeA)

            default:
                // We are free to process the typeB data now
                doSomethingElse(typeB)
                break NESTED
            }
        }
    }

}

Answer 9

Is there any specific reason for using a buffered channel make(chan int, 10)?

You need to use an unbuffered channel vs buffered, which you are using.

Just remove 10, it should be just make(chan int).

This way execution in the sender function can only proceed to the exit <- true statement after the last message from the out channel is dequeued by the i := <-out statement. If that statement has not been executed, there is no way the exit <- true could be reached in the goroutine.