go - 尝试 Recv 返回 channel 关闭，尽管它是打开的

Question

我正在尝试在 Go 中编写一个函数来监视 channel 并记录通过它发送的内容。

func monitorChannel(inChannel, outChannel reflect.Value, fid int64, cond *sync.Cond) {
    for {       
    cond.L.Lock()
    var toLog reflect.Value
    var ok bool
    for toLog, ok = inChannel.TryRecv() ; !toLog.IsValid(); { // while no value received
        if !ok {
            cond.L.Unlock()
            return
        }
        cond.Wait()
    }
    outChannel.Send(toLog)
    logMessage("a", "b", inChannel.Interface(), toLog.Interface(), fid)
    cond.L.Unlock()
}

此函数应该从 inChannel 接收，记录发送的消息并通过 outChannel 发送。由于我希望能够记录双向 channel ，因此我为每个要记录的 channel 调用此函数两次，交换 inChannel 和 outChannel。锁是为了防止两个 goroutine 相互传递消息。 “fid”只是日志文件的 ID。

但是当我运行下面的测试代码时，我遇到了死锁:

errsIn := make(chan int64)
errsOut := make(chan int64)
cond := sync.NewCond(&sync.Mutex{})
go monitorChannel(reflect.ValueOf(errsIn), reflect.ValueOf(errsOut), fid, cond)
go monitorChannel(reflect.ValueOf(errsOut), reflect.ValueOf(errsIn), fid,  cond)
errsIn <- 1
if <-errsOut != 1 {
    t.Fatal("lost value through channel send")
}
errsOut <- 1
if <-errsIn != 1 {
    t.Fatal("lost value through channel send")
}

似乎 TryRecv 在其第二个返回值上返回 false，即使我没有关闭 channel 。为什么是这样？我该怎么办？

我在 Windows 8 64 位上运行 go 1.0.3。

编辑

我后来发现 TryRecv 有一些令人困惑的行为，并设法使用反射包和两个 sync.Locker 制作了该函数的通用版本。我仍然认为 jnml 的解决方案更优雅，但如果有人遇到类似 TryRecv 的问题，请查看函数中间的注释。

func passOnAndLog(in, out reflect.Value, l1, l2 sync.Locker) {
    for {
        l1.Lock()
        val, ok := in.TryRecv()
        for !val.IsValid() { // while nothing received
            l1.Unlock()
            time.Sleep(time.Nanosecond) // pausing current thread
            l1.Lock()
            val, ok = in.TryRecv()
        }
        // if val.IsValid() == true  and ok == false ,the channel is closed
        // if val.IsValid() == false and ok == false ,the channel is open but we received nothing
        // if val.IsValid() == true  and ok == true  ,we received an actual value from the open channel
        // if val.IsValid() == false and ok == true  ,we have no idea what happened  
        if !ok {
            return
        }
        l1.Unlock()
        l2.Lock() // don't want the other thread to receive while I am sending
        out.Send(val)
        LogValue(val) // logging

        l2.Unlock()
    }
}

Answer 1

基于反射的解决方案太复杂了，我懒得弄清楚它是否正确或者是否可行。 (我怀疑不是，但只是凭直觉。)

我会以更简单但非通用的方式处理任务。让我们有一个 channel ，一些生产者将使用它来写入它，一些消费者将使用它来读取它。

c := make(chan T, N)

可以使用一个小的辅助函数来监控这个 channel ，例如:

func monitored(c chan T) chan T {
        m := make(chan T, M)
        go func() {
                for v := range c {
                        m <- v
                        logMessage(v)
                }
                close(m)
        }()
        return m
}

现在足以:

mc := monitored(c)

和

• 将 c 传递给生产者，但将 mc 传递给消费者。
• 完成后关闭 c 以防止 goroutine 泄漏。

警告:以上代码根本没有经过测试。