diff --git a/bluegreentx.go b/bluegreentx.go index 6d02c6f3f..bfb75277f 100644 --- a/bluegreentx.go +++ b/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() { diff --git a/dbshard.go b/dbshard.go index 0c578b513..7db0c4dcd 100644 --- a/dbshard.go +++ b/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 { diff --git a/fragment_internal_test.go b/fragment_internal_test.go index 9e5a567bc..addead151 100644 --- a/fragment_internal_test.go +++ b/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") } } diff --git a/rbf.go b/rbf.go index cb764bc29..0e5b1ec2a 100644 --- a/rbf.go +++ b/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() diff --git a/txfactory.go b/txfactory.go index c04889500..36ed6194d 100644 --- a/txfactory.go +++ b/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))) } diff --git a/txfactory_internal_test.go b/txfactory_internal_test.go index 0db9a5a44..15b03b59c 100644 --- a/txfactory_internal_test.go +++ b/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) + } + } +}