From 81013999e56de6456be2cbf93c9f0c91a15659f3 Mon Sep 17 00:00:00 2001 From: "Jason E. Aten" Date: Thu, 15 Oct 2020 18:33:33 -0500 Subject: [PATCH 1/2] introduce a per shard blue-green RWMutex - allows blue-green testing with concurrent readers/writers. - otherwise we don't start/end the blue and green Tx together, and they get split by a read/write concurrently. --- dbshard.go | 35 +++++++++++++++++++++++------------ fragment_internal_test.go | 20 +++++++++++++------- rbf.go | 1 + txfactory.go | 17 ++++++++++------- 4 files changed, 47 insertions(+), 26 deletions(-) diff --git a/dbshard.go b/dbshard.go index 0c578b513..151ef52a7 100644 --- a/dbshard.go +++ b/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 { 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..09c572782 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,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: From d9783406bd74a62be95bddb7637be523d1bec71f Mon Sep 17 00:00:00 2001 From: "Jason E. Aten" Date: Mon, 19 Oct 2020 18:28:45 -0500 Subject: [PATCH 2/2] Fix blue-green Tx cleanup and document single import at once - correct string constants for txtype so that blue-green cleanup correctly detects when 2nd transaction in a pair has Committed and thus the blue-green RWMutex can be relased - test that txtype.String() is consistent with the corresponding string constants. - document in bluegreentx.go the current limitations of blue-green testing: only one github archive import (a single writing client) is supported by blue-green testing. Multiple importers will deadlock eventually on the DBShard.mut RWMutex. We could fix this by ordering the write locks and obtaining them in strictly increasing order (by shard number), but that would require alot of change to the executor and that would introduce more risk for a test-only pathway. --- bluegreentx.go | 12 ++++++++++++ dbshard.go | 12 ++++++++---- txfactory.go | 4 ++-- txfactory_internal_test.go | 15 +++++++++++++++ 4 files changed, 37 insertions(+), 6 deletions(-) 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 151ef52a7..7db0c4dcd 100644 --- a/dbshard.go +++ b/dbshard.go @@ -68,10 +68,8 @@ 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 @@ -136,14 +134,17 @@ func (dbs *DBShard) Cleanup(tx Tx) { if dbs == nil { return // some tests are using Tx only, no dbs available. } + //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) } } } @@ -158,7 +159,9 @@ func (dbs *DBShard) NewTx(write bool, initialIndexName string, o Txo) (tx Tx, er 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() } } @@ -183,6 +186,7 @@ func (dbs *DBShard) NewTx(write bool, initialIndexName string, o Txo) (tx Tx, er } // blue green 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 } diff --git a/txfactory.go b/txfactory.go index 09c572782..36ed6194d 100644 --- a/txfactory.go +++ b/txfactory.go @@ -846,9 +846,9 @@ func (ty txtype) String() string { case 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) + } + } +}