CODEWITHSUNDEEP
gochannel
Shane Hou
Shane Hou

Reputation: 5018

How to check a channel is closed or not without reading it?

This is a good example of workers & controller mode in Go written by @Jimt, in answer to "Is there some elegant way to pause & resume any other goroutine in golang?"

package main

import (
    "fmt"
    "runtime"
    "sync"
    "time"
)

// Possible worker states.
const (
    Stopped = 0
    Paused  = 1
    Running = 2
)

// Maximum number of workers.
const WorkerCount = 1000

func main() {
    // Launch workers.
    var wg sync.WaitGroup
    wg.Add(WorkerCount + 1)

    workers := make([]chan int, WorkerCount)
    for i := range workers {
        workers[i] = make(chan int)

        go func(i int) {
            worker(i, workers[i])
            wg.Done()
        }(i)
    }

    // Launch controller routine.
    go func() {
        controller(workers)
        wg.Done()
    }()

    // Wait for all goroutines to finish.
    wg.Wait()
}

func worker(id int, ws <-chan int) {
    state := Paused // Begin in the paused state.

    for {
        select {
        case state = <-ws:
            switch state {
            case Stopped:
                fmt.Printf("Worker %d: Stopped\n", id)
                return
            case Running:
                fmt.Printf("Worker %d: Running\n", id)
            case Paused:
                fmt.Printf("Worker %d: Paused\n", id)
            }

        default:
            // We use runtime.Gosched() to prevent a deadlock in this case.
            // It will not be needed of work is performed here which yields
            // to the scheduler.
            runtime.Gosched()

            if state == Paused {
                break
            }

            // Do actual work here.
        }
    }
}

// controller handles the current state of all workers. They can be
// instructed to be either running, paused or stopped entirely.
func controller(workers []chan int) {
    // Start workers
    for i := range workers {
        workers[i] <- Running
    }

    // Pause workers.
    <-time.After(1e9)
    for i := range workers {
        workers[i] <- Paused
    }

    // Unpause workers.
    <-time.After(1e9)
    for i := range workers {
        workers[i] <- Running
    }

    // Shutdown workers.
    <-time.After(1e9)
    for i := range workers {
        close(workers[i])
    }
}

But this code also has an issue: If you want to remove a worker channel in workers when worker() exits, dead lock happens.

If you close(workers[i]), next time controller writes into it will cause a panic since go can't write into a closed channel. If you use some mutex to protect it, then it will be stuck on workers[i] <- Running since the worker is not reading anything from the channel and write will be blocked, and mutex will cause a dead lock. You can also give a bigger buffer to channel as a work-around, but it's not good enough.

So I think the best way to solve this is worker() close channel when exits, if the controller finds a channel closed, it will jump over it and do nothing. But I can't find how to check a channel is already closed or not in this situation. If I try to read the channel in controller, the controller might be blocked. So I'm very confused for now.

PS: Recovering the raised panic is what I have tried, but it will close goroutine which raised panic. In this case it will be controller so it's no use.

Still, I think it's useful for Go team to implement this function in next version of Go.

Upvotes: 141

Views: 264382

Answers (11)

karolinski
karolinski

Reputation: 577

hmm what you can do is to use following struct:

type MyChannel struct {
    C      chan T
    closed bool
    mutex  sync.Mutex
}

func NewMyChannel() *MyChannel {
    return &MyChannel{C: make(chan T)}
}

func (mc *MyChannel) SafeClose() {
    mc.mutex.Lock()
    defer mc.mutex.Unlock()
    if !mc.closed {
        close(mc.C)
        mc.closed = true
    }
}

func (mc *MyChannel) IsClosed() bool {
    mc.mutex.Lock()
    defer mc.mutex.Unlock()
    return mc.closed
}

if all the goroutines which send or receive from the channel use SafeClose method you avoid the unwanted panic.

Source: https://go101.org/article/channel-closing.html

Upvotes: 0

zzzz
zzzz

Reputation: 91243

In a hacky way it can be done for channels which one attempts to write to by recovering the raised panic. But you cannot check if a read channel is closed without reading from it.

Either you will

  • eventually read the "true" value from it (v <- c)
  • read the "true" value and 'not closed' indicator (v, ok <- c)
  • read a zero value and the 'closed' indicator (v, ok <- c) (example)
  • will block in the channel read forever (v <- c)

Only the last one technically doesn't read from the channel, but that's of little use.

Upvotes: 105

Petrucho Peres
Petrucho Peres

Reputation: 1

ch1 := make(chan int)
ch2 := make(chan int)
go func(){
    for i:=0; i<10; i++{
        ch1 <- i
    }
    close(ch1)
}()
go func(){
    for i:=10; i<15; i++{
        ch2 <- i
    }
    close(ch2)
}()
ok1, ok2 := false, false
v := 0
for{
    ok1, ok2 = true, true
    select{
        case v,ok1 = <-ch1:
        if ok1 {fmt.Println(v)}
        default:
    }
    select{
        case v,ok2 = <-ch2:
        if ok2 {fmt.Println(v)}
        default:
    }
    if !ok1 && !ok2{return}
    
}

}

Upvotes: 0

acrazing
acrazing

Reputation: 2284

Well, you can use default branch to detect it, for a closed channel will be selected, for example: the following code will select default, channel, channel, the first select is not blocked.

func main() {
    ch := make(chan int)

    go func() {
        select {
        case <-ch:
            log.Printf("1.channel")
        default:
            log.Printf("1.default")
        }
        select {
        case <-ch:
            log.Printf("2.channel")
        }
        close(ch)
        select {
        case <-ch:
            log.Printf("3.channel")
        default:
            log.Printf("3.default")
        }
    }()
    time.Sleep(time.Second)
    ch <- 1
    time.Sleep(time.Second)
}

Prints

2018/05/24 08:00:00 1.default
2018/05/24 08:00:01 2.channel
2018/05/24 08:00:01 3.channel

Note, refer to comment by @Angad under this answer:

It doesn't work if you're using a Buffered Channel and it contains unread data

Upvotes: 5

jregovic
jregovic

Reputation: 29

I have had this problem frequently with multiple concurrent goroutines.

It may or may not be a good pattern, but I define a a struct for my workers with a quit channel and field for the worker state:

type Worker struct {
    data chan struct
    quit chan bool
    stopped bool
}

Then you can have a controller call a stop function for the worker:

func (w *Worker) Stop() {
    w.quit <- true
    w.stopped = true
}

func (w *Worker) eventloop() {
    for {
        if w.Stopped {
            return
        }
        select {
            case d := <-w.data:
                //DO something
                if w.Stopped {
                    return
                }
            case <-w.quit:
                return
        }
    }
}

This gives you a pretty good way to get a clean stop on your workers without anything hanging or generating errors, which is especially good when running in a container.

Upvotes: 2

user4790427
user4790427

Reputation:

I know this answer is so late, I have wrote this solution, Hacking Go run-time, It's not safety, It may crashes:

import (
    "unsafe"
    "reflect"
)


func isChanClosed(ch interface{}) bool {
    if reflect.TypeOf(ch).Kind() != reflect.Chan {
        panic("only channels!")
    }
    
    // get interface value pointer, from cgo_export 
    // typedef struct { void *t; void *v; } GoInterface;
    // then get channel real pointer
    cptr := *(*uintptr)(unsafe.Pointer(
        unsafe.Pointer(uintptr(unsafe.Pointer(&ch)) + unsafe.Sizeof(uint(0))),
    ))
    
    // this function will return true if chan.closed > 0
    // see hchan on https://github.com/golang/go/blob/master/src/runtime/chan.go 
    // type hchan struct {
    // qcount   uint           // total data in the queue
    // dataqsiz uint           // size of the circular queue
    // buf      unsafe.Pointer // points to an array of dataqsiz elements
    // elemsize uint16
    // closed   uint32
    // **
    
    cptr += unsafe.Sizeof(uint(0))*2
    cptr += unsafe.Sizeof(unsafe.Pointer(uintptr(0)))
    cptr += unsafe.Sizeof(uint16(0))
    return *(*uint32)(unsafe.Pointer(cptr)) > 0
}

Upvotes: 9

Kasper Gyselinck
Kasper Gyselinck

Reputation: 175

You could set your channel to nil in addition to closing it. That way you can check if it is nil.

example in the playground: https://play.golang.org/p/v0f3d4DisCz

edit: This is actually a bad solution as demonstrated in the next example, because setting the channel to nil in a function would break it: https://play.golang.org/p/YVE2-LV9TOp

Upvotes: 2

Israel Barba
Israel Barba

Reputation: 1494

From the documentation:

A channel may be closed with the built-in function close. The multi-valued assignment form of the receive operator reports whether a received value was sent before the channel was closed.

https://golang.org/ref/spec#Receive_operator

Example by Golang in Action shows this case:

// This sample program demonstrates how to use an unbuffered
// channel to simulate a game of tennis between two goroutines.
package main

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

// wg is used to wait for the program to finish.
var wg sync.WaitGroup

func init() {
    rand.Seed(time.Now().UnixNano())
}

// main is the entry point for all Go programs.
func main() {
    // Create an unbuffered channel.
    court := make(chan int)
    // Add a count of two, one for each goroutine.
    wg.Add(2)
    // Launch two players.
    go player("Nadal", court)
    go player("Djokovic", court)
    // Start the set.
    court <- 1
    // Wait for the game to finish.
    wg.Wait()
}

// player simulates a person playing the game of tennis.
func player(name string, court chan int) {
    // Schedule the call to Done to tell main we are done.
    defer wg.Done()
    for {
        // Wait for the ball to be hit back to us.
        ball, ok := <-court
        fmt.Printf("ok %t\n", ok)
        if !ok {
            // If the channel was closed we won.
            fmt.Printf("Player %s Won\n", name)
            return
        }
        // Pick a random number and see if we miss the ball.
        n := rand.Intn(100)
        if n%13 == 0 {
            fmt.Printf("Player %s Missed\n", name)
            // Close the channel to signal we lost.
            close(court)
            return
        }

        // Display and then increment the hit count by one.
        fmt.Printf("Player %s Hit %d\n", name, ball)
        ball++
        // Hit the ball back to the opposing player.
        court <- ball
    }
}

Upvotes: -3

Enric
Enric

Reputation: 3

it's easier to check first if the channel has elements, that would ensure the channel is alive.

func isChanClosed(ch chan interface{}) bool {
    if len(ch) == 0 {
        select {
        case _, ok := <-ch:
            return !ok
        }
    }
    return false 
}

Upvotes: -6

Dustin
Dustin

Reputation: 90980

There's no way to write a safe application where you need to know whether a channel is open without interacting with it.

The best way to do what you're wanting to do is with two channels -- one for the work and one to indicate a desire to change state (as well as the completion of that state change if that's important).

Channels are cheap. Complex design overloading semantics isn't.

[also]

<-time.After(1e9)

is a really confusing and non-obvious way to write

time.Sleep(time.Second)

Keep things simple and everyone (including you) can understand them.

Upvotes: 98

jurka
jurka

Reputation: 13173

If you listen this channel you always can findout that channel was closed.

case state, opened := <-ws:
    if !opened {
         // channel was closed 
         // return or made some final work
    }
    switch state {
        case Stopped:

But remember, you can not close one channel two times. This will raise panic.

Upvotes: -9

Related Questions