Cách sử dụng đúng cách sync.Cond?

Question

Tôi đang gặp sự cố khi tìm hiểu cách sử dụng chính xác sync.Cond. Từ những gì tôi có thể nói, một điều kiện chủng tộc tồn tại giữa khóa Locker và gọi phương thức Wait của điều kiện. Ví dụ này cho biết thêm một sự chậm trễ nhân tạo giữa hai dòng trong goroutine chính để mô phỏng các điều kiện chủng tộc:

package main 

import (
    "sync" 
    "time" 
) 

func main() { 
    m := sync.Mutex{} 
    c := sync.NewCond(&m) 
    go func() { 
     time.Sleep(1 * time.Second) 
     c.Broadcast() 
    }() 
    m.Lock() 
    time.Sleep(2 * time.Second) 
    c.Wait() 
} 

^{[Run on the Go Playground]}

này gây ra hoảng loạn ngay lập tức:

fatal error: all goroutines are asleep - deadlock! 

goroutine 1 [semacquire]: 
sync.runtime_Syncsemacquire(0x10330208, 0x1) 
    /usr/local/go/src/runtime/sema.go:241 +0x2e0 
sync.(*Cond).Wait(0x10330200, 0x0) 
    /usr/local/go/src/sync/cond.go:63 +0xe0 
main.main() 
    /tmp/sandbox301865429/main.go:17 +0x1a0

gì tôi đang làm sai à? Làm cách nào để tránh tình trạng cuộc đua rõ ràng này? Có một cấu trúc đồng bộ hóa tốt hơn tôi nên sử dụng?

---

Edit: Tôi nhận ra tôi nên có tốt hơn giải thích vấn đề tôi đang cố gắng để giải quyết đây. Tôi có một goroutine chạy dài tải xuống một tệp lớn và một số goroutine khác cần quyền truy cập vào các tiêu đề HTTP khi chúng có sẵn. Vấn đề này khó hơn âm thanh.

Tôi không thể sử dụng kênh vì chỉ một goroutine sau đó sẽ nhận được giá trị. Và một số goroutines khác sẽ cố gắng lấy lại các tiêu đề lâu sau khi chúng đã có sẵn.

Trình tải xuống goroutine có thể chỉ lưu trữ các tiêu đề HTTP trong một biến và sử dụng một mutex để bảo vệ quyền truy cập vào chúng. Tuy nhiên, điều này không cung cấp cách để các goroutines khác "chờ" để chúng trở nên khả dụng.

Tôi đã nghĩ rằng cả hai sync.Mutex và sync.Cond cùng nhau có thể đạt được mục tiêu này nhưng dường như điều này là không thể.

Answer 1

Cuối cùng tôi đã phát hiện ra một cách để làm điều này và nó không liên quan đến tất cả - chỉ là mutex.

type Task struct { 
    m  sync.Mutex 
    headers http.Header 
} 

func NewTask() *Task { 
    t := &Task{} 
    t.m.Lock() 
    go func() { 
     defer t.m.Unlock() 
     // ...do stuff... 
    }() 
    return t 
} 

func (t *Task) WaitFor() http.Header { 
    t.m.Lock() 
    defer t.m.Unlock() 
    return t.headers 
} 

Tính năng này hoạt động như thế nào?

Mutex bị khóa ở đầu nhiệm vụ, đảm bảo rằng mọi thứ gọi WaitFor() sẽ bị chặn. Khi các tiêu đề có sẵn và mutex được mở khóa bởi goroutine, mỗi cuộc gọi đến WaitFor() sẽ thực thi một lần tại một thời điểm. Tất cả các cuộc gọi trong tương lai (ngay cả sau khi goroutine kết thúc) sẽ không có vấn đề gì với việc khóa mutex, vì nó sẽ luôn luôn được mở khóa bên trái.

Answer 2

package main 

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

func main() { 
    m := sync.Mutex{} 
    m.Lock() // main gouroutine is owner of lock 
    c := sync.NewCond(&m) 
    go func() { 
     m.Lock() // obtain a lock 
     defer m.Unlock() 
     fmt.Println("3. goroutine is owner of lock") 
     time.Sleep(2 * time.Second) // long computing - because you are the owner, you can change state variable(s) 
     c.Broadcast()    // State has been changed, publish it to waiting goroutines 
     fmt.Println("4. goroutine will release lock soon (deffered Unlock") 
    }() 
    fmt.Println("1. main goroutine is owner of lock") 
    time.Sleep(1 * time.Second) // initialization 
    fmt.Println("2. main goroutine is still lockek") 
    c.Wait() // Wait temporarily release a mutex during wating and give opportunity to other goroutines to change the state. 
    // Because you don't know, whether this is state, that you are waiting for, is usually called in loop. 
    m.Unlock() 
    fmt.Println("Done") 
} 

http://play.golang.org/p/fBBwoL7_pm

Answer 3

Hình như bạn c.Wait cho Broadcast mà sẽ không bao giờ xảy ra với khoảng thời gian của bạn. Với

time.Sleep(3 * time.Second) //Broadcast after any Wait for it 
c.Broadcast() 

đoạn mã của bạn dường như làm việc http://play.golang.org/p/OE8aP4i6gY .Or tôi thiếu một cái gì đó mà bạn cố gắng để đạt được?

Answer 4

OP tự trả lời, nhưng không trực tiếp trả lời câu hỏi gốc, tôi sẽ đăng cách sử dụng chính xác sync.Cond.

Bạn không thực sự cần sync.Cond nếu bạn có một goroutine cho mỗi lần viết và đọc - một sync.Mutex sẽ đủ để giao tiếp giữa chúng. sync.Cond có thể hữu ích trong trường hợp nhiều người đọc đợi tài nguyên được chia sẻ khả dụng.

var sharedRsc = make(map[string]interface{}) 
func main() { 
    var wg sync.WaitGroup 
    wg.Add(2) 
    m := sync.Mutex{} 
    c := sync.NewCond(&m) 
    go func() { 
     // this go routine wait for changes to the sharedRsc 
     c.L.Lock() 
     for len(sharedRsc) == 0 { 
      c.Wait() 
     } 
     fmt.Println(sharedRsc["rsc1"]) 
     c.L.Unlock() 
     wg.Done() 
    }() 

    go func() { 
     // this go routine wait for changes to the sharedRsc 
     c.L.Lock() 
     for len(sharedRsc) == 0 { 
      c.Wait() 
     } 
     fmt.Println(sharedRsc["rsc2"]) 
     c.L.Unlock() 
     wg.Done() 
    }() 

    // this one writes changes to sharedRsc 
    c.L.Lock() 
    sharedRsc["rsc1"] = "foo" 
    sharedRsc["rsc2"] = "bar" 
    c.Broadcast() 
    c.L.Unlock() 
    wg.Wait() 
} 

Playground

Có nói rằng, sử dụng các kênh truyền hình vẫn là cách khuyến khích để truyền dữ liệu xung quanh nếu tình hình cho phép.

Lưu ý: sync.WaitGroup ở đây chỉ được sử dụng để chờ các goroutines hoàn tất việc thực thi của chúng.

Answer 5

Bạn cần đảm bảo rằng c.Broadcast được gọi là sau cuộc gọi của bạn tới c.Wait. Đúng phiên bản của chương trình của bạn sẽ là:

package main 

import (
    "fmt" 
    "sync" 
) 

func main() { 
    m := &sync.Mutex{} 
    c := sync.NewCond(m) 
    m.Lock() 
    go func() { 
     m.Lock() // Wait for c.Wait() 
     c.Broadcast() 
     m.Unlock() 
    }() 
    c.Wait() // Unlocks m 
} 

https://play.golang.org/p/O1r8v8yW6h