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: