From 55d4c29933adf75a56beb143d24aad9980fb06fb Mon Sep 17 00:00:00 2001
From: Jason Aten <jason.aten@molecula.com>
Date: Sat, 19 Sep 2020 20:12:03 -0500
Subject: [PATCH] turn off debug machinery on tx backends  - enable row cache
 again. Was off for tx perf measurement.  - centralize UseRowCache choice to
 just one point, in rbf.EnableRowCache

---
 badger.go      |  3 +-
 lmdb.go        | 82 +++++++++++++++++++++++++++-----------------------
 pprof.go       | 37 +++++++++++++++++++++--
 rbf.go         |  2 +-
 rbf/cursorx.go |  2 +-
 rrtx.go        |  3 +-
 6 files changed, 85 insertions(+), 44 deletions(-)

diff --git a/badger.go b/badger.go
index f4d16facc..f667da77c 100644
--- a/badger.go
+++ b/badger.go
@@ -31,6 +31,7 @@ import (
 	badger "github.com/dgraph-io/badger/v2"
 	badgeroptions "github.com/dgraph-io/badger/v2/options"
 	"github.com/pilosa/pilosa/v2/hash"
+	"github.com/pilosa/pilosa/v2/rbf"
 	"github.com/pilosa/pilosa/v2/roaring"
 	"github.com/pilosa/pilosa/v2/testhook"
 	"github.com/pilosa/pilosa/v2/txkey"
@@ -660,7 +661,7 @@ func (tx *BadgerTx) Type() string {
 
 func (tx *BadgerTx) UseRowCache() bool {
 	//the row cache speeds up queries.
-	return false
+	return rbf.EnableRowCache
 }
 
 // overWriteOurAllocs provides detection of memory
diff --git a/lmdb.go b/lmdb.go
index 4758c4b26..73167e4ac 100644
--- a/lmdb.go
+++ b/lmdb.go
@@ -33,6 +33,7 @@ import (
 
 	"github.com/glycerine/lmdb-go/lmdb"
 	"github.com/pilosa/pilosa/v2/hash"
+	"github.com/pilosa/pilosa/v2/rbf"
 	"github.com/pilosa/pilosa/v2/roaring"
 	"github.com/pilosa/pilosa/v2/txkey"
 	"github.com/pkg/errors"
@@ -274,10 +275,7 @@ func (w *LMDBWrapper) CleanupTx(tx Tx) {
 }
 
 func (tx *LMDBTx) IsDone() (done bool) {
-	tx.mu.Lock()
-	done = tx.unlocked
-	tx.mu.Unlock()
-	return
+	return atomic.LoadInt64(&tx.unlocked) == 1
 }
 
 func (w *LMDBWrapper) OpenListString() (r string) {
@@ -434,9 +432,11 @@ func (w *LMDBWrapper) NewTx(write bool, initialIndexName string, o Txo) (tx Tx,
 	}
 	tx = ltx
 
-	w.muDb.Lock()
-	w.openTx[ltx] = true
-	w.muDb.Unlock()
+	if isDebugRun {
+		w.muDb.Lock()
+		w.openTx[ltx] = true
+		w.muDb.Unlock()
+	}
 	return
 }
 
@@ -445,13 +445,14 @@ func (w *LMDBWrapper) Close() (err error) {
 	w.muDb.Lock()
 	defer w.muDb.Unlock()
 	if !w.closed {
-		// complain if there are still Tx in flight, b/c otherwise we will see
-		// the somewhat mysterious 'panic: should not be in ReadSlot.free() with slot still owned by gid=107043; refCount=1'
-		if len(w.openTx) > 0 {
-			AlwaysPrintf("error: cannot close LMDBWrapper with Tx still in flight.")
-			return
+		if isDebugRun {
+			// complain if there are still Tx in flight, b/c otherwise we will see
+			// the somewhat mysterious 'panic: should not be in ReadSlot.free() with slot still owned by gid=107043; refCount=1'
+			if len(w.openTx) > 0 {
+				AlwaysPrintf("error: cannot close LMDBWrapper with Tx still in flight.")
+				return
+			}
 		}
-
 		w.reg.unregister(w)
 		w.closed = true
 		w.env.CloseDBI(w.dbi)
@@ -494,7 +495,7 @@ type LMDBTx struct {
 
 	DeleteEmptyContainer bool
 
-	unlocked bool // runtime.UnlockOSThread has been done.
+	unlocked int64 // runtime.UnlockOSThread has been done if > 0
 
 	o Txo
 
@@ -535,7 +536,7 @@ func (tx *LMDBTx) Type() string {
 }
 
 func (tx *LMDBTx) UseRowCache() bool {
-	return false
+	return rbf.EnableRowCache
 }
 
 // Pointer gives us a memory address for the underlying transaction for debugging.
@@ -545,15 +546,21 @@ func (tx *LMDBTx) Pointer() string {
 	return fmt.Sprintf("%p", tx)
 }
 
+const isDebugRun = false
+
 // Rollback rolls back the transaction.
 func (tx *LMDBTx) Rollback() {
-	tx.sanity()
-
+	alreadyDone := atomic.CompareAndSwapInt64(&tx.unlocked, 0, 1)
+	if !alreadyDone {
+		return
+	}
 	//vv("lmdb rollback tx _sn_ %v; stack \n%v", tx.sn) // , stack())
-
-	tx.Db.muDb.Lock()
-	delete(tx.Db.openTx, tx)
-	tx.Db.muDb.Unlock()
+	if isDebugRun {
+		tx.sanity()
+		tx.Db.muDb.Lock()
+		delete(tx.Db.openTx, tx)
+		tx.Db.muDb.Unlock()
+	}
 
 	tx.mu.Lock()
 	defer tx.mu.Unlock()
@@ -561,25 +568,26 @@ func (tx *LMDBTx) Rollback() {
 	//tx.debugOnlyGidcheck()
 	tx.tx.Abort() // must hold tx.mu mutex lock
 
-	if !tx.unlocked {
-		runtime.UnlockOSThread()
-		tx.unlocked = true
-		tx.o.dbs.Cleanup(tx)
-	}
+	// use CAS above instead of testing a bool unlocked.
+	runtime.UnlockOSThread()
+	tx.o.dbs.Cleanup(tx)
 }
 
 // Commit commits the transaction to permanent storage.
 // Commits can handle up to 100k updates to fragments
 // at once, but not more. This is a LMDBDB imposed limit.
 func (tx *LMDBTx) Commit() error {
-	tx.sanity()
-
+	alreadyDone := atomic.CompareAndSwapInt64(&tx.unlocked, 0, 1)
+	if !alreadyDone {
+		return nil
+	}
 	//vv("lmdb commit tx _sn_ %v; stack \n%v", tx.sn, stack())
-
-	tx.Db.muDb.Lock()
-	delete(tx.Db.openTx, tx)
-	tx.Db.muDb.Unlock()
-
+	if isDebugRun {
+		tx.sanity()
+		tx.Db.muDb.Lock()
+		delete(tx.Db.openTx, tx)
+		tx.Db.muDb.Unlock()
+	}
 	tx.mu.Lock()
 	defer tx.mu.Unlock()
 
@@ -587,11 +595,9 @@ func (tx *LMDBTx) Commit() error {
 	err := tx.tx.Commit() // must hold tx.mu mutex lock
 	panicOn(err)
 
-	if !tx.unlocked {
-		runtime.UnlockOSThread()
-		tx.unlocked = true
-		tx.o.dbs.Cleanup(tx)
-	}
+	// replace the if !tx.unlocked with the CAS on tx.unlocked above.
+	runtime.UnlockOSThread()
+	tx.o.dbs.Cleanup(tx)
 	return err
 }
 
diff --git a/pprof.go b/pprof.go
index 7e9f721f0..5c9b0b339 100644
--- a/pprof.go
+++ b/pprof.go
@@ -29,7 +29,7 @@ func CPUProfileForDur(dur time.Duration, outpath string) {
 	// per-query pprof output:
 	txsrc := os.Getenv("PILOSA_TXSRC")
 	if txsrc == "" {
-		txsrc = "roaring"
+		txsrc = DefaultTxsrc
 	}
 	path := outpath + "." + txsrc
 	f, err := os.Create(path)
@@ -53,7 +53,7 @@ func MemProfileForDur(dur time.Duration, outpath string) {
 	// per-query pprof output:
 	txsrc := os.Getenv("PILOSA_TXSRC")
 	if txsrc == "" {
-		txsrc = "roaring"
+		txsrc = DefaultTxsrc
 	}
 	path := outpath + "." + txsrc
 	f, err := os.Create(path)
@@ -73,3 +73,36 @@ func MemProfileForDur(dur time.Duration, outpath string) {
 		AlwaysPrintf("wrote memory profile after dur '%v', output: '%v'", dur, path)
 	}()
 }
+
+type pprofProfile struct {
+	fdCpu *os.File
+}
+
+var _ = newPprof
+var _ = pprofProfile{}
+
+// for manually calling Close() to stop profiling.
+func newPprof() (pp *pprofProfile) {
+	pp = &pprofProfile{}
+	f, err := os.Create("cpu.manual.pprof")
+	panicOn(err)
+	pp.fdCpu = f
+
+	_ = pprof.StartCPUProfile(pp.fdCpu)
+	return
+}
+
+func (pp *pprofProfile) Close() {
+
+	pprof.StopCPUProfile()
+	pp.fdCpu.Close()
+
+	f, err := os.Create("mem.manual.pprof")
+	panicOn(err)
+
+	runtime.GC() // get up-to-date statistics
+	if err := pprof.WriteHeapProfile(f); err != nil {
+		panic(fmt.Sprintf("could not write memory profile: %v", err))
+	}
+	f.Close()
+}
diff --git a/rbf.go b/rbf.go
index 79284de3c..36eebf64d 100644
--- a/rbf.go
+++ b/rbf.go
@@ -68,7 +68,7 @@ func (w *RbfDBWrapper) CleanupTx(tx Tx) {
 	r.done = true
 	r.mu.Unlock()
 
-	// try not to old r.mu while locking w.muDb
+	// try not to hold r.mu while locking w.muDb
 	w.muDb.Lock()
 
 	delete(w.openTx, r)
diff --git a/rbf/cursorx.go b/rbf/cursorx.go
index 5d57bdfd9..26d0c002d 100644
--- a/rbf/cursorx.go
+++ b/rbf/cursorx.go
@@ -26,7 +26,7 @@ import (
 
 // if enableRowCache, then we must not return mmap-ed memory
 // directly, but only a copy.
-const EnableRowCache = false
+const EnableRowCache = true
 
 // DoAllocZero means we copy mmap read data and
 // wipe it afterwards to catch retention of data
diff --git a/rrtx.go b/rrtx.go
index 7ea06f659..4868e443e 100644
--- a/rrtx.go
+++ b/rrtx.go
@@ -25,6 +25,7 @@ import (
 	"sync"
 	"sync/atomic"
 
+	"github.com/pilosa/pilosa/v2/rbf"
 	"github.com/pilosa/pilosa/v2/roaring"
 	"github.com/pkg/errors"
 )
@@ -62,7 +63,7 @@ func (tx *RoaringTx) Dump(short bool, shard uint64) {
 }
 
 func (tx *RoaringTx) UseRowCache() bool {
-	return false
+	return rbf.EnableRowCache
 }
 
 func (tx *RoaringTx) SliceOfShards(index, field, view, optionalViewPath string) (sliceOfShards []uint64, err error) {