69 lines
1.4 KiB
Go
69 lines
1.4 KiB
Go
|
package stm
|
||
|
|
|
||
|
|
import (
|
||
|
|
"sync"
|
||
|
|
"sync/atomic"
|
||
|
|
)
|
||
|
|
|
||
|
|
// Holds an STM variable.
|
||
|
|
type Var struct {
|
||
|
|
value atomic.Value
|
||
|
|
watchers sync.Map
|
||
|
|
mu sync.Mutex
|
||
|
|
}
|
||
|
|
|
||
|
|
func (v *Var) changeValue(new interface{}) {
|
||
|
|
old := v.value.Load().(VarValue)
|
||
|
|
newVarValue := old.Set(new)
|
||
|
|
v.value.Store(newVarValue)
|
||
|
|
if old.Changed(newVarValue) {
|
||
|
|
go v.wakeWatchers(newVarValue)
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
func (v *Var) wakeWatchers(new VarValue) {
|
||
|
|
v.watchers.Range(func(k, _ interface{}) bool {
|
||
|
|
tx := k.(*Tx)
|
||
|
|
// We have to lock here to ensure that the Tx is waiting before we signal it. Otherwise we
|
||
|
|
// could signal it before it goes to sleep and it will miss the notification.
|
||
|
|
tx.mu.Lock()
|
||
|
|
if read := tx.reads[v]; read != nil && read.Changed(new) {
|
||
|
|
tx.cond.Broadcast()
|
||
|
|
for !tx.waiting && !tx.completed {
|
||
|
|
tx.cond.Wait()
|
||
|
|
}
|
||
|
|
}
|
||
|
|
tx.mu.Unlock()
|
||
|
|
return !v.value.Load().(VarValue).Changed(new)
|
||
|
|
})
|
||
|
|
}
|
||
|
|
|
||
|
|
type varSnapshot struct {
|
||
|
|
val interface{}
|
||
|
|
version uint64
|
||
|
|
}
|
||
|
|
|
||
|
|
// Returns a new STM variable.
|
||
|
|
func NewVar(val interface{}) *Var {
|
||
|
|
v := &Var{}
|
||
|
|
v.value.Store(versionedValue{
|
||
|
|
value: val,
|
||
|
|
})
|
||
|
|
return v
|
||
|
|
}
|
||
|
|
|
||
|
|
func NewCustomVar(val interface{}, changed func(interface{}, interface{}) bool) *Var {
|
||
|
|
v := &Var{}
|
||
|
|
v.value.Store(customVarValue{
|
||
|
|
value: val,
|
||
|
|
changed: changed,
|
||
|
|
})
|
||
|
|
return v
|
||
|
|
}
|
||
|
|
|
||
|
|
func NewBuiltinEqVar(val interface{}) *Var {
|
||
|
|
return NewCustomVar(val, func(a, b interface{}) bool {
|
||
|
|
return a != b
|
||
|
|
})
|
||
|
|
}
|