从接收方关闭通道：从多个 goroutine 访问 sync.Mutex 时出现死锁

Question

我正在尝试从接收端实现优雅的频道关闭。

是的，我知道这违反了频道关闭规则：

...don't close a channel from the receiver side and don't close a channel if the channel has multiple concurrent senders.

但我想实现这样的逻辑。不幸的是，我在很多情况下都没有遇到死锁问题：应用程序只是无限期挂起，试图再次锁定相同的锁定Mutex。

所以，我有 2 个 goroutine：

• 将写入通道和
• 另一个将接收数据 + 将从接收端关闭通道。

我的频道用 sync.Mutex 和 closed 布尔标志包裹在结构中：

type Chan struct {
    sync.Mutex // can be replaced with deadlock.Mutex from "github.com/sasha-s/go-deadlock"
    data           chan int
    closed         bool
}

此结构上的所有Send()、Close()、IsClosed() 操作都由Mutex 保护，并使用非线程安全方法版本（send()、@ 987654333@, isClosed())。

完整源代码：

package main

import (
    "log"
    "net/http"
    "sync"
)

func main() {
    log.Println("Start")

    ch := New(0) // unbuffered channel to expose problem faster

    wg := sync.WaitGroup{}
    wg.Add(2)

    // send data:
    go func(ch *Chan) {
        for i := 0; i < 100; i++ {
            ch.Send(i)
        }
        wg.Done()
    }(ch)

    // receive data and close from receiver side:
    go func(ch *Chan) {
        for data := range ch.data {
            log.Printf("Received %d data", data)
            // Bad practice: I want to close the channel from receiver's side:
            if data > 50 {
                ch.Close()
                break
            }
        }
        wg.Done()
    }(ch)

    wg.Wait()
    log.Println("End")
}

type Chan struct {
    deadlock.Mutex //sync.Mutex
    data           chan int
    closed         bool
}

func New(size int) *Chan {
    defer func() {
        log.Printf("Channel was created")
    }()
    return &Chan{
        data: make(chan int, size),
    }
}

func (c *Chan) Send(data int) {
    c.Lock()
    c.send(data)
    c.Unlock()
}

func (c *Chan) Close() {
    c.Lock()
    c.close()
    c.Unlock()
}

func (c *Chan) IsClosed() bool {
    c.Lock()
    defer c.Unlock()
    return c.isClosed()
}

// send is internal non-threadsafe api.
func (c *Chan) send(data int) {
    if !c.closed {
        c.data <- data
        log.Printf("Data %d was sent", data)
    }
}

// close is internal non-threadsafe api.
func (c *Chan) close() {
    if !c.closed {
        close(c.data)
        c.closed = true
        log.Println("Channel was closed")
    } else {
        log.Println("Channel was already closed")
    }
}

// isClosed is internal non-threadsafe api.
func (c *Chan) isClosed() bool {
    return c.closed
}

你可以在sandbox运行这个程序。

在本地机器上，少量运行，30 秒后输出将是（使用deadlock.Mutex 而不是sync.Mutex）：

2018/04/01 11:26:22 Data 50 was sent
2018/04/01 11:26:22 Received 50 data
2018/04/01 11:26:22 Data 51 was sent
2018/04/01 11:26:22 Received 51 data
POTENTIAL DEADLOCK:
Previous place where the lock was grabbed
goroutine 35 lock 0xc42015a040
close-from-receiver-side/closeFromReceiverSideIsBadPractice.go:71 main.(*Chan).Send { c.Lock() } <<<<<
close-from-receiver-side/closeFromReceiverSideIsBadPractice.go:30 main.main.func1 { ch.Send(i) }

Have been trying to lock it again for more than 30s
goroutine 36 lock 0xc42015a040
close-from-receiver-side/closeFromReceiverSideIsBadPractice.go:77 main.(*Chan).Close { c.Lock() } <<<<<
close-from-receiver-side/closeFromReceiverSideIsBadPractice.go:44 main.main.func2 { ch.Close() }

为什么会发生这种死锁以及如何修复此实现以避免死锁？

---

关闭发送方的通道不是答案。所以，这不是我的问题的解决方法：Example of closing channel from sender side。

Answer 1

Send 获取锁，然后尝试沿通道发送数据。这可能发生在第 50 次接收操作之后。将不再有接收，因此c.data <- data 将永远阻塞，因此 Mutex 将永远保留。

要取消，请使用另一个通道（而不是布尔值）和 Send 中的 select 语句。您还可以利用the context package。

Answer 2

你可以尽情尝试：你必须从发送方关闭频道。

你也许可以在没有完全锁定的情况下让它工作，但你会泄漏 goroutine。发件人将永远阻塞，无法关闭。如果接收方想要触发关闭，它必须告诉发送方关闭通道。如何让发件人关闭：

• 您建议的布尔值（需要另一个互斥体）
• stop-channel 在关闭时向发送方发出关闭数据通道的信号（不能多次关闭）
• ctx.Context：调用cancel() 函数将通知发送方停止。 （可以多次取消，不用担心）

（仅详细说明彼得斯的正确答案）