Merge pull request #1004 from jaten-molecula/bluegreenlock

introduce a per shard blue-green RWMutex

dbshard.go

@@ -75,6 +75,7 @@ type DBShard struct {
 	mut sync.RWMutex
 
 	types      []txtype
+	stypes     []string
 	hasRoaring bool // if either of the types is roaringTxn
 
 	W             []DBWrapper
@@ -85,6 +86,8 @@ type DBShard struct {
 
 	useOpenList int
 	closed      bool
+
+	isBlueGreen bool
 }
 
 func (dbs *DBShard) DeleteFragment(index, field, view string, shard uint64, frag interface{}) (err error) {
@@ -133,24 +136,25 @@ func (dbs *DBShard) Cleanup(tx Tx) {
 	if dbs == nil {
 		return // some tests are using Tx only, no dbs available.
 	}
-	if useRWLock {
-		if !dbs.hasRoaring {
-			if tx.Readonly() {
-				dbs.mut.RUnlock()
-			} else {
-				dbs.mut.Unlock()
+	if !dbs.hasRoaring {
+		if dbs.isBlueGreen {
+			// only release on the 2nd Tx's cleanup
+			if tx.Type() == dbs.stypes[1] {
+				if tx.Readonly() {
+					dbs.mut.RUnlock()
+				} else {
+					dbs.mut.Unlock()
+				}
 			}
 		}
 	}
 }
 
-// experimental feature, off for now.
-const useRWLock = false
-
 func (dbs *DBShard) NewTx(write bool, initialIndexName string, o Txo) (tx Tx, err error) {
-	if useRWLock {
+
+	if dbs.isBlueGreen {
 		// enforce only one writer at a time. The dbs.mut is held until
-		// the Tx finishes.
+		// the Tx finishes. This makes the two Tx in the blue-green Tx atomic.
 		if !dbs.hasRoaring {
 			if write {
 				dbs.mut.Lock()
@@ -178,7 +182,8 @@ func (dbs *DBShard) NewTx(write bool, initialIndexName string, o Txo) (tx Tx, er
 		return
 	}
 	// blue green
-	return dbs.per.txf.newBlueGreenTx(txns[0], txns[1], o.Index, o), nil
+	tx, err = dbs.per.txf.newBlueGreenTx(txns[0], txns[1], o.Index, o), nil
+	return
 }
 
 func (dbs *DBShard) DeleteDBPath() (err error) {
@@ -459,7 +464,13 @@ func (per *DBPerShard) GetDBShard(index string, shard uint64, idx *Index) (dbs *
 			per:           per,
 			useOpenList:   per.useOpenList,
 			hasRoaring:    per.hasRoaring,
+			isBlueGreen:   len(per.types) > 1,
 		}
+		dbs.stypes = make([]string, len(per.types))
+		for i, ty := range per.types {
+			dbs.stypes[i] = ty.String()
+		}
+
 		dbi.Shard[shard] = dbs
 	}
 	if !dbs.Open {

fragment_internal_test.go

@@ -113,7 +113,7 @@ func TestFragment_ClearBit(t *testing.T) {
 	// In that spirit, we will check that the Tx Commit is visible afterwards.
 	panicOn(tx.Commit())
 
-	tx = idx.holder.txf.NewTx(Txo{Write: writable, Index: idx, Fragment: f, Shard: f.shard})
+	tx = idx.holder.txf.NewTx(Txo{Write: !writable, Index: idx, Fragment: f, Shard: f.shard})
 	defer tx.Rollback()
 
 	// Close and reopen the fragment & verify the data.
@@ -198,7 +198,7 @@ func TestFragment_ClearRow(t *testing.T) {
 		t.Fatalf("unexpected count: %d", n)
 	}
 	panicOn(tx.Commit())
-	tx = idx.holder.txf.NewTx(Txo{Write: writable, Index: idx, Fragment: f, Shard: f.shard})
+	tx = idx.holder.txf.NewTx(Txo{Write: !writable, Index: idx, Fragment: f, Shard: f.shard})
 	defer tx.Rollback()
 
 	// Close and reopen the fragment & verify the data.
@@ -258,8 +258,7 @@ func TestFragment_SetRow(t *testing.T) {
 	}
 
 	panicOn(tx.Commit())
-	tx = idx.holder.txf.NewTx(Txo{Write: writable, Index: idx, Fragment: f, Shard: f.shard})
-	defer tx.Rollback()
+	tx = idx.holder.txf.NewTx(Txo{Write: !writable, Index: idx, Fragment: f, Shard: f.shard})
 
 	// Close and reopen the fragment & verify the data.
 	if err := f.Reopen(); err != nil {
@@ -268,6 +267,10 @@ func TestFragment_SetRow(t *testing.T) {
 		t.Fatalf("unexpected count (reopen): %d", n)
 	}
 
+	tx.Rollback()
+	tx = idx.holder.txf.NewTx(Txo{Write: writable, Index: idx, Fragment: f, Shard: f.shard})
+	defer tx.Rollback()
+
 	// verify that setting something from a row which lacks a segment for
 	// this fragment's shard still clears this fragment correctly.
 	notOurs := NewRow(8*ShardWidth + 1024)
@@ -1515,12 +1518,17 @@ func TestFragment_Checksum(t *testing.T) {
 	f, idx, tx := mustOpenFragment(t, "i", "f", viewStandard, 0, "")
 	_ = idx
 	defer f.Clean(t)
+	tx.Rollback() // allow f.Checksum to make its read tx.
 
 	// Retrieve checksum and set bits.
 	orig, err := f.Checksum()
 	if err != nil {
 		t.Fatal(err)
 	}
+
+	tx = idx.holder.txf.NewTx(Txo{Write: writable, Index: idx, Fragment: f, Shard: f.shard})
+	defer tx.Rollback()
+
 	if _, err := f.setBit(tx, 1, 200); err != nil {
 		t.Fatal(err)
 	} else if _, err := f.setBit(tx, HashBlockSize*2, 200); err != nil {
@@ -5412,8 +5420,6 @@ func TestFragment_Bug_Q2DoubleDelete(t *testing.T) {
 func notBlueGreenTest(t *testing.T) {
 	src := os.Getenv("PILOSA_TXSRC")
 	if strings.Contains(src, "_") {
-		if strings.Contains(src, "roaring") {
-			t.Skip("skip under blue green with roaring")
-		}
+		t.Skip("skip under blue green")
 	}
 }

rbf.go

@@ -72,6 +72,7 @@ func (w *RbfDBWrapper) CleanupTx(tx Tx) {
 	w.muDb.Lock()
 
 	delete(w.openTx, r)
+	//vv("rbf CleanupTx gid %v about to call r.o.dbs.Cleanup(tx.Sn=%v)", curGID(), tx.Sn())
 	r.o.dbs.Cleanup(tx) // release the read/write lock.
 
 	w.muDb.Unlock()

txfactory.go

@@ -276,7 +276,7 @@ func (qcx *Qcx) GetTx(o Txo) (tx Tx, finisher func(perr *error)) {
 
 	if !o.Write && qcx.Grp != nil {
 		// read, with a group in place.
-		finisher = func(perr *error) {}
+		finisher = func(perr *error) {} // finisher is a returned value
 
 		already := false
 		tx, already = qcx.Grp.AlreadyHaveTx(o)
@@ -380,7 +380,7 @@ func (qcx *Qcx) ListOpenTx() string {
 type TxFactory struct {
 	typeOfTx string
 
-	mu sync.Mutex // group protection
+	mu sync.Mutex
 
 	types []txtype // blue-green split individually here
 
@@ -395,6 +395,8 @@ type TxFactory struct {
 	// allow holder to activate blue-green checking only
 	// once we have synced both sides at start up time.
 	blueGreenOff bool
+
+	isBlueGreen bool
 }
 
 func (f *TxFactory) Types() []txtype {
@@ -506,6 +508,7 @@ func NewTxFactory(txsrc string, holderDir string, holder *Holder) (f *TxFactory,
 	}
 	if len(types) == 2 {
 		f.blueGreenReg = newBlueGreenReg(types)
+		f.isBlueGreen = true
 	}
 	f.dbPerShard = f.NewDBPerShard(types, holderDir, holder)
 
@@ -789,16 +792,15 @@ func (g *TxGroup) AbortGroup() {
 }
 
 func (f *TxFactory) NewTx(o Txo) (txn Tx) {
-	f.mu.Lock()
-	defer f.mu.Unlock()
-
 	defer func() {
 		if globalUseStatTx {
 			txn = newStatTx(txn)
 		}
 	}()
 
+	f.mu.Lock()
 	o.blueGreenOff = f.blueGreenOff
+	f.mu.Unlock()
 
 	indexName := ""
 	if o.Index != nil {
@@ -834,14 +836,15 @@ func (f *TxFactory) NewTx(o Txo) (txn Tx) {
 	return tx
 }
 
+// has to match the const strings at the top of the file.
 func (ty txtype) String() string {
 	switch ty {
 	case noneTxn:
 		return "noneTxn"
 	case roaringTxn:
-		return "roaringTxn"
+		return "roaring"
 	case rbfTxn:
-		return "rbfTxn"
+		return "rbf"
 	case lmdbTxn:
 		return "lmdbTxn"
 	case boltTxn: