Merge branch 'master' into pqlCleanup

bluegreentx.go

@@ -36,6 +36,15 @@ import (
 // Do not run with go test -race and expect it to be race free with RoaringTx
 // on one arm.
 //
+// Note: using the dbshard.go DBShard.mut RWMutex to begin and end
+// both the A and B transactions atomically, we support a single importer and
+// lots of readers running under blue-green transactions. Two writers a.k.a. two
+// github ingests at once will deadlock eventually, but I think that may be asking
+// for more than we want to test under blue-green, as it would require a bunch of
+// test-only internal executor logic that could mess with the production path.
+// So, for now, a limitation on blue green tests is that they be single
+// writer/single importer going at once.
+//
 type blueGreenTx struct {
 	a Tx
 	b Tx // b's output is returned
@@ -111,6 +120,9 @@ func (b *blueGreenRegistry) finishedTx(tx *blueGreenTx) {
 	sn := tx.Sn()
 	delete(b.m, sn)
 	//vv("blueGreenRegistry deleted _sn_ %v", sn)
+
+	// Note that a tx.o.dbs.Cleanup(tx) call should not be needed,
+	// because the individual tx will call cleanup themselves.
 }
 
 func (b *blueGreenRegistry) Close() {

dbshard.go

@@ -68,13 +68,12 @@ type DBShard struct {
 	Shard uint64
 	Open  bool
 
-	// With RWMutex, the
-	// writer who calls Lock() automatically gets priority over
-	// any reader who arrives later, even if the lock is held
-	// by a reader to start with.
+	// With RWMutex, the blue-green Tx can start and commit
+	// atomically.
 	mut sync.RWMutex
 
 	types      []txtype
+	stypes     []string
 	hasRoaring bool // if either of the types is roaringTxn
 
 	W             []DBWrapper
@@ -85,6 +84,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,28 +134,34 @@ 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()
+	//vv("top of DBShard %v Cleanup for tx.Sn = %v; dbs=%p; is 2nd: %v; type='%v'; dbs.stypes='%#v'", dbs.Shard, tx.Sn(), dbs, tx.Type() == dbs.stypes[1], tx.Type(), dbs.stypes)
+	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()
+					//vv("gid %v released read-lock on shard %v", curGID(), dbs.Shard)
+				} else {
+					dbs.mut.Unlock()
+					//vv("gid %v released write-lock on shard %v", curGID(), dbs.Shard)
+				}
 			}
 		}
 	}
 }
 
-// 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()
+				//vv("shard %v was write locked by gid %v; stack =\n%v", dbs.Shard, curGID(), stack())
 			} else {
+				//vv("shard %v about to be read locked by gid %v; stack=\n%v", dbs.Shard, curGID(), stack())
 				dbs.mut.RLock()
 			}
 		}
@@ -178,7 +185,9 @@ 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
+	//vv("dbshard returning blue-green tx sn %v", tx.Sn())
+	return
 }
 
 func (dbs *DBShard) DeleteDBPath() (err error) {
@@ -459,7 +468,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,18 +836,19 @@ 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"
+		return "lmdb"
 	case boltTxn:
-		return "boltTxn"
+		return "bolt"
 	}
 	panic(fmt.Sprintf("unhandled ty '%v' in txtype.String()", int(ty)))
 }

txfactory_internal_test.go

@@ -380,3 +380,18 @@ func Test_TxFactory_verifyBlueEqualsGreen(t *testing.T) {
 		}
 	}
 }
+
+func Test_TxFactory_verifyStringConstantsMatch(t *testing.T) {
+	// txtype.String() method MUST return strings that match
+	// our const definitions at the top of txfactory.go, or
+	// else blue-green transactions cannot determine when
+	// the second transaction is being released in dbshard.go.
+	check := []txtype{roaringTxn, rbfTxn, lmdbTxn, boltTxn}
+	expect := []string{RoaringTxn, RBFTxn, LmdbTxn, BoltTxn}
+	for i, chk := range check {
+		obs := chk.String()
+		if obs != expect[i] {
+			t.Fatalf("expected '%v' but got '%v'", expect[i], obs)
+		}
+	}
+}