Come utilizzare correttamente sync.Cond?

Question

Non riesco a capire come utilizzare correttamente sync.Cond. Da quello che posso dire, esiste una condizione di competizione tra il blocco dell'armadietto e l'invocazione del metodo di attesa della condizione. Questo esempio aggiunge un ritardo artificiale tra le due linee nel goroutine principale per simulare la condizione di competizione:

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]}

questo provoca un panico immediato:

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

Nei sto sbagliando? Come posso evitare questa apparente condizione di gara? C'è un migliore costrutto di sincronizzazione che dovrei usare?

---

Edit: mi rendo conto che dovrei avere una migliore spiegato il problema che sto cercando di risolvere qui. Ho una goroutine di lunga data che scarica un file di grandi dimensioni e un certo numero di altre goroutine che hanno bisogno di accedere alle intestazioni HTTP quando sono disponibili. Questo problema è più difficile di quanto sembri.

Non riesco a utilizzare i canali poiché solo una goroutine riceverà il valore. E alcune delle altre goroutine potrebbero cercare di recuperare le intestazioni per molto tempo dopo che sono già disponibili.

La goroutine di download potrebbe semplicemente memorizzare le intestazioni HTTP in una variabile e utilizzare un mutex per salvaguardare l'accesso ad esse. Tuttavia, questo non fornisce un modo per le altre goroutine di "aspettare" che diventino disponibili.

Avevo pensato che sia un sync.Mutex sia uno sync.Cond insieme potessero raggiungere questo obiettivo ma sembra che questo non sia possibile.

Answer 1

Alla fine ho scoperto un modo per farlo e non comporta affatto lo sync.Cond - solo il 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 
} 

Come funziona?

Il mutex è bloccato all'inizio dell'attività, assicurando che tutto ciò che chiama WaitFor() verrà bloccato. Una volta che le intestazioni sono disponibili e il mutex sbloccato dalla goroutine, ciascuna chiamata a WaitFor() verrà eseguita una alla volta. Tutte le chiamate future (anche dopo la fine della goroutine) non avranno problemi a bloccare il mutex, poiché rimarrà sempre sbloccato.

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

Sembra che tu c.Wait per il Broadcast che non sarebbe mai accade con i tuoi intervalli di tempo. Con

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

tuo snippet sembra funzionare http://play.golang.org/p/OE8aP4i6gY .O mi sto perdendo qualcosa che si tenta di achive?

Answer 4

OP ha risposto da solo, ma non ha risposto direttamente alla domanda originale, ho intenzione di pubblicare come utilizzare correttamente sync.Cond.

Non hai davvero bisogno di sync.Cond se hai una goroutine per ogni scrittura e lettura - un singolo sync.Mutex è sufficiente per comunicare tra di loro. sync.Cond potrebbe essere utile in situazioni in cui più lettori attendono che le risorse condivise siano disponibili.

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

Detto questo, utilizzando canali è ancora il metodo consigliato per passare dati intorno se la situazione permettendo.

Nota: sync.WaitGroup qui viene utilizzato solo per attendere che le goroutine completino le esecuzioni.

Answer 5

È necessario assicurarsi che c.Broadcast si chiami dopo la chiamata a c.Wait. La versione corretta del programma sarebbe:

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