通道和优雅关闭死锁

Question

运行以下程序并运行 CTRL + C，handle 例程在尝试发送到通道时被阻止，但 process 例程已关闭。有什么更好的并发设计来解决这个问题？

编辑了程序以描述应用此处建议的规则https://stackoverflow.com/a/66708290/4106031的问题

package main

import (
    "context"
    "fmt"
    "os"
    "os/signal"
    "sync"
    "syscall"
    "time"
)

func process(ctx context.Context, c chan string) {
    fmt.Println("process: processing (select)")
    for {
        select {
        case <-ctx.Done():
            fmt.Printf("process: ctx done bye\n")
            return
        case i := <-c:
            fmt.Printf("process: received i: %v\n", i)
        }
    }
}

func handle(ctx context.Context, readChan <-chan string) {
    c := make(chan string, 1)
    wg := &sync.WaitGroup{}
    wg.Add(1)
    go func() {
        process(ctx, c)
        wg.Done()
    }()
    defer wg.Wait()

    for i := 0; ; i++ {
        select {
        case <-ctx.Done():
            fmt.Printf("handle: ctx done bye\n")
            return
        case i := <-readChan:
            fmt.Printf("handle: received: %v\n", i)
            fmt.Printf("handle: sending for processing: %v\n", i)
            // suppose huge time passes here
            // to cause the issue we want to happen
            // we want the process() to exit due to ctx
            // cancellation before send to it happens, this creates deadlock
            time.Sleep(5 * time.Second)
            // deadlock
            c <- i
        }
    }
}

func main() {
    wg := &sync.WaitGroup{}
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    readChan := make(chan string, 10)
    wg.Add(1)
    go func() {
        defer wg.Done()
        for i := 0; ; i++ {
            select {
            case <-ctx.Done():
                fmt.Printf("read: ctx done bye\n")
                return
            case readChan <- fmt.Sprintf("%d", i):
                fmt.Printf("read: sent msg: %v\n", i)
            }
        }
    }()

    wg.Add(1)
    go func() {
        handle(ctx, readChan)
        wg.Done()
    }()

    go func() {
        sigterm := make(chan os.Signal, 1)
        signal.Notify(sigterm, syscall.SIGINT, syscall.SIGTERM)
        select {
        case <-sigterm:
            fmt.Printf("SIGTERM signal received\n")
            cancel()
        }
    }()

    wg.Wait()
}

输出

$ go run chan-shared.go
read: sent msg: 0
read: sent msg: 1
read: sent msg: 2
read: sent msg: 3
process: processing (select)
read: sent msg: 4
read: sent msg: 5
read: sent msg: 6
handle: received: 0
handle: sending for processing: 0
read: sent msg: 7
read: sent msg: 8
read: sent msg: 9
read: sent msg: 10
handle: received: 1
handle: sending for processing: 1
read: sent msg: 11
process: received i: 0
process: received i: 1
read: sent msg: 12
handle: received: 2
handle: sending for processing: 2
^CSIGTERM signal received
process: ctx done bye
read: ctx done bye
handle: received: 3
handle: sending for processing: 3


Killed: 9

Answer 1

一步一步的回顾

• 不管你怎么想，总是取消上下文。

ctx, cancel := context.WithCancel(context.Background())
defer cancel()

• 开始例行程序后不要 wd.Add

    wg.Add(1)
    go handle(ctx, wg)

• 不要稀疏地使用等待组

    wg.Add(1)
    go func() {
        handle(ctx)
        wg.Done()
    }()

• 不要在默认情况下在通道上循环。只需从中读取并让它解除阻塞

    <-sigterm
    fmt.Printf("SIGTERM signal received\n")

• main 从不阻塞信号，主阻塞处理例程。信令应该只是做信令，即取消上下文。

    go func() {
        sigterm := make(chan os.Signal, 1)
        signal.Notify(sigterm, syscall.SIGINT, syscall.SIGTERM)
        <-sigterm
        fmt.Printf("SIGTERM signal received\n")
        cancel()
    }()

• 可以在通道写入时检查上下文取消。

        select {
        case <-ctx.Done():
            fmt.Printf("process: ctx done bye\n")
            return
        case c <- fmt.Sprintf("%d", i):
            fmt.Printf("handled: sent to channel: %v\n", i)
        }

• Dont time.Sleep，你不能用它来测试上下文取消。

        select {
        case <-ctx.Done():
            fmt.Printf("process: ctx done bye\n")
            return
        case <-time.After(time.Second * 5):
        }

因此，应用了这些不同规则的代码的完整修订版本为我们提供了

package main

import (
    "context"
    "fmt"
    "os"
    "os/signal"
    "sync"
    "syscall"
    "time"
)

func process(ctx context.Context, c chan string) {
    fmt.Println("process: processing (select)")
    for {
        select {
        case <-ctx.Done():
            fmt.Printf("process: ctx done bye\n")
            return
        case msg := <-c:
            fmt.Printf("process: got msg: %v\n", msg)
        }
    }
}

func handle(ctx context.Context) {
    c := make(chan string, 3)
    wg := &sync.WaitGroup{}
    wg.Add(1)
    go func() {
        process(ctx, c)
        wg.Done()
    }()
    defer wg.Wait()

    for i := 0; ; i++ {
        select {
        case <-ctx.Done():
            fmt.Printf("process: ctx done bye\n")
            return
        case <-time.After(time.Second * 5):
        }
        select {
        case <-ctx.Done():
            fmt.Printf("process: ctx done bye\n")
            return
        case c <- fmt.Sprintf("%d", i):
            fmt.Printf("handled: sent to channel: %v\n", i)
        }
    }
}

func main() {
    wg := &sync.WaitGroup{}
    ctx, cancel := context.WithCancel(context.Background())
    defer cancel()

    wg.Add(1)
    go func() {
        handle(ctx)
        wg.Done()
    }()

    go func() {
        sigterm := make(chan os.Signal, 1)
        signal.Notify(sigterm, syscall.SIGINT, syscall.SIGTERM)
        <-sigterm
        fmt.Printf("SIGTERM signal received\n")
        cancel()
    }()
    wg.Wait()
}

还有更多关于退出条件的信息，但这取决于要求。

Answer 2

如https://stackoverflow.com/a/66708290/4106031 所述，此更改已解决了我的问题。也感谢 mh-cbon 的规则！