WIP: Move Run loop body into own function to recover panic

Author: dmiller

stm/locking_transaction.go

@@ -239,95 +239,113 @@ func RunInTransaction(fn TxFn) (interface{}, error) {
 func (tx *Tx) Run(fn TxFn) (interface{}, error) {
 
 	done := false
-	var ret interface{}
 	locked := make([]*Ref, 0, 10)
 	// notify := make([]*Notify)
 
 	defer func() {
 		for k := len(locked) - 1; k >= 0; k-- {
 			locked[k].exitWriteLock()
 		}
+
+		locked = nil
+		for r, _ := range tx.ensures {
+			r.exitReadLock()
+		}
+		tx.ensures = nil
+		if done {
+			tx.Stop(txCommitted)
+		} else {
+			tx.Stop(txRetry)
+		}
+		if done {
+			// do notifies and agent actions, if we every implement
+		}
+
 	}()
 
-	locked = nil
-	for r, _ := range tx.ensures {
-		r.exitReadLock()
-	}
-	tx.ensures = nil
-	if done {
-		tx.Stop(txCommitted)
-	} else {
-		tx.Stop(txRetry)
-	}
-	if done {
-		// do notifies and agent actions, if we every implement
+	for i := 0; !done && i < retryLimit; i++ {
+		ret, err := tx.tryRun(i, fn, &locked)
+		if err == nil {
+			done = true
+			return ret, nil
+		}
 	}
 
-	for i := 0; !done && i < retryLimit; i++ {
+	return nil, errors.New("Transaction failed after reaching retry limit")
+}
+
+// One iteration of the Run loop
+// Split out so that we can catch a retry panic
+func (tx *Tx) tryRun(i int, fn TxFn, locked *[]*Ref) (ret interface{}, err error) {
 
-		tx.getReadPoint()
-		if i == 0 {
-			tx.startPoint = tx.readPoint
-			tx.startTime = time.Now()
+	ret, err = nil, nil
+
+	defer func() {
+		r := recover()
+		if r == retryError {
+			ret, err = nil, retryError
+		} else if r != nil {
+			panic(r)
 		}
+	}()
 
-		tx.info = newTxInfo(txRunning, tx.startPoint)
-		ret = fn(tx)
+	tx.getReadPoint()
+	if i == 0 {
+		tx.startPoint = tx.readPoint
+		tx.startTime = time.Now()
+	}
 
-		// make sure no one has killed us before this point, and can't from now on
-		if atomic.CompareAndSwapUint32(&tx.info.status, txRunning, txCommitting) {
-			for r, calls := range tx.commutes {
-				if _, ok := tx.sets[r]; ok {
-					continue
-				}
-				_, wasEnsured := tx.ensures[r]
-				tx.releaseIfEnsured(r)
-				tryWriteLock(r)
-				locked = append(locked, r)
-				if wasEnsured && r.currValPoint() > tx.readPoint {
-					panic(retryError)
-				}
+	tx.info = newTxInfo(txRunning, tx.startPoint)
+	ret = fn(tx)
 
-				refInfo := r.tinfo
-				if refInfo != nil && refInfo != tx.info && refInfo.isRunning() {
-					if !tx.barge(refInfo) {
-						panic(retryError)
-					}
-				}
-				val := r.tryGetVal()
-				tx.vals[r] = val
-				for _, call := range calls {
-					tx.vals[r] = call.fn(tx.vals[r], call.args...)
-				}
+	// make sure no one has killed us before this point, and can't from now on
+	if atomic.CompareAndSwapUint32(&tx.info.status, txRunning, txCommitting) {
+		for r, calls := range tx.commutes {
+			if _, ok := tx.sets[r]; ok {
+				continue
 			}
-
-			for r, _ := range tx.sets {
-				tryWriteLock(r)
-				locked = append(locked, r)
+			_, wasEnsured := tx.ensures[r]
+			tx.releaseIfEnsured(r)
+			tryWriteLock(r)
+			*locked = append(*locked, r)
+			if wasEnsured && r.currValPoint() > tx.readPoint {
+				panic(retryError)
 			}
 
-			// if we do validations for refs, it goes here
-
-			// at this point,
-			//    all values are calculated,
-			//    all refs to be written are locked
-			//    no more client code to be called
-			commitPoint := getCommitPoint()
-			for r, newV := range tx.vals {
-				//oldV := r.tryGetVal()
-				r.setValue(newV, commitPoint)
-				// todo: call notifies
+			refInfo := r.tinfo
+			if refInfo != nil && refInfo != tx.info && refInfo.isRunning() {
+				if !tx.barge(refInfo) {
+					panic(retryError)
+				}
+			}
+			val := r.tryGetVal()
+			tx.vals[r] = val
+			for _, call := range calls {
+				tx.vals[r] = call.fn(tx.vals[r], call.args...)
 			}
-			done = true
-			atomic.StoreUint32(&tx.info.status, txCommitted)
 		}
 
-		if done {
-			return ret, nil
+		for r, _ := range tx.sets {
+			tryWriteLock(r)
+			*locked = append(*locked, r)
 		}
 
+		// if we do validations for refs, it goes here
+
+		// at this point,
+		//    all values are calculated,
+		//    all refs to be written are locked
+		//    no more client code to be called
+		commitPoint := getCommitPoint()
+		for r, newV := range tx.vals {
+			//oldV := r.tryGetVal()
+			r.setValue(newV, commitPoint)
+			// todo: call notifies
+		}
+		atomic.StoreUint32(&tx.info.status, txCommitted)
 	}
-	return nil, errors.New("Transaction failed after reaching retry limit")
+
+	return
 }
 
 // Get the value of a Ref (most recently sent in this transaction or value prior to entering)

stm/tx_test.go

@@ -5,7 +5,10 @@
 package stm
 
 import (
+	"runtime"
+	"sync"
 	"testing"
+	"time"
 )
 
 func TestSimpleTxCall(t *testing.T) {
@@ -225,5 +228,78 @@ func TestSimpleCommute(t *testing.T) {
 	if save2 != init1+add1+add2 {
 		t.Errorf("Expected save to have value %v, got %v", init1+add1+add2, save2)
 	}
+}
+
+func TestSimpleInterference(t *testing.T) {
+
+	runtime.GOMAXPROCS(4)
+
+	init1 := 12
+	val1 := 100
+	val2 := 200
+	add1 := 1
+
+	r1 := NewRef(init1)
+	ret := 300
+
+	fc := func(old interface{}, args ...interface{}) interface{} {
+		return old.(int) + args[0].(int)
+	}
+
+	var done sync.WaitGroup
+	done.Add(2)
+
+	var once sync.Once
+
+	nfEnter, nfExit := 0, 0
+	ngEnter, ngExit := 0, 0
+
+	ch := make(chan bool, 1)
+
+	signalG := func() {
+		ch <- true
+	}
+
+	f := func(tx *Tx) interface{} {
+		t.Logf("F: Entering")
+		nfEnter++
+		r1.Set(tx, val1)
+		t.Logf("F: between sets")
+		once.Do(signalG)
+		time.Sleep(20 * time.Nanosecond)
+		r1.Set(tx, val2)
+		t.Logf("F: after sets")
+		nfExit++
+		return ret
+	}
+
+	go func() {
+		defer done.Done()
+		t.Logf("F: before TX")
+		RunInTransaction(f)
+		t.Logf("F: after TX")
+	}()
+
+	g := func(tx *Tx) interface{} {
+		t.Logf("G: entering")
+		ngEnter++
+		r1.Commute(tx, fc, add1)
+		t.Logf("G: after commute")
+		ngExit++
+		return ret
+	}
+
+	go func() {
+		defer done.Done()
+		<-ch
+		t.Logf("G: before TX")
+		RunInTransaction(g)
+		t.Logf("G: after TX")
+	}()
+
+	time.Sleep(time.Nanosecond)
+	done.Wait()
+
+	t.Errorf("%v %v %v %v %v", ngEnter, ngExit, r1.Deref(nil), nfEnter, nfExit)
 
 }