diff options
| -rw-r--r-- | funcs.go | 15 | ||||
| -rw-r--r-- | tx.go | 18 |
2 files changed, 25 insertions, 8 deletions
@@ -1,8 +1,10 @@ package stm import ( + "math/rand" "runtime/pprof" "sync" + "time" ) var ( @@ -43,8 +45,21 @@ func Atomically(op Operation) interface{} { expvars.Add("atomically", 1) // run the transaction tx := newTx() + tx.tries = 0 retry: + tx.tries++ tx.reset() + shift := int64(tx.tries - 1) + const maxShift = 30 + if shift > maxShift { + shift = maxShift + } + ns := int64(1) << shift + ns = rand.Int63n(ns) + if ns > 0 { + tx.updateWatchers() + time.Sleep(time.Duration(ns)) + } ret, retry := catchRetry(op, tx) if retry { expvars.Add("retries", 1) @@ -15,6 +15,7 @@ type Tx struct { locks txLocks mu sync.Mutex cond sync.Cond + tries int } // Check that none of the logged values have changed since the transaction began. @@ -35,11 +36,7 @@ func (tx *Tx) commit() { } } -// wait blocks until another transaction modifies any of the Vars read by tx. -func (tx *Tx) wait() { - if len(tx.reads) == 0 { - panic("not waiting on anything") - } +func (tx *Tx) updateWatchers() { for v := range tx.watching { if _, ok := tx.reads[v]; !ok { delete(tx.watching, v) @@ -52,6 +49,14 @@ func (tx *Tx) wait() { tx.watching[v] = struct{}{} } } +} + +// wait blocks until another transaction modifies any of the Vars read by tx. +func (tx *Tx) wait() { + if len(tx.reads) == 0 { + panic("not waiting on anything") + } + tx.updateWatchers() tx.mu.Lock() firstWait := true for tx.verify() { @@ -63,9 +68,6 @@ func (tx *Tx) wait() { firstWait = false } tx.mu.Unlock() - //for v := range tx.reads { - // v.watchers.Delete(tx) - //} } // Get returns the value of v as of the start of the transaction. |