Repository: bmatsuo/lmdb-go
Pull request #113 (2017-02-15 09:10:06)

lmdbpool: Improvements to TxnPool
Checking of Txn.ID() for lmdb.Readonly transactions now occurs at the correct location (as a Txn would enter the pool as opposed to when it exits the pool).

There is also an optimization that allows Txn.ID() to cache the value it gets for future calls. This requires some bookkeeping because there are times when a transaction's ID may change. But the logic behind this in LMDB is straight forward and there are some tests to make sure that Txn.ID() does transparently represent the value returned by `mdb_txn_id`.


Patch:
diff --git a/exp/lmdbpool/txnpool.go b/exp/lmdbpool/txnpool.go
index f17ff26..dd259b9 100644
--- a/exp/lmdbpool/txnpool.go
+++ b/exp/lmdbpool/txnpool.go
@@ -13,20 +13,32 @@ import (
 // transactions when an environment update occurs.  Applications with a high
 // rate of large updates may need to choose non-default settings to reduce
 // their storage requirements at the cost of read throughput.
+//
+// The zero-value of UpdateHandling causes a TxnPool to ignore all updates and
+// defers to the application and the lmdb.Txn finalizers clear stale readers
+// (pulling an lmdb.Readonly transaction out of the pool is enough to release
+// its stale pages).
 type UpdateHandling uint
 
 const (
-	// AbortOutstanding causes a TxnPool to abort any lmdb.Readonly
+	// HandleOutstanding causes a TxnPool to abort any lmdb.Readonly
 	// transactions that are being returned to the pool after an update.
-	AbortOutstanding UpdateHandling = 1 << iota
-	RenewOutstanding
+	HandleOutstanding UpdateHandling = 1 << iota
 
-	// AbortIdle causes a TxnPool to actively attempt aborting idle
+	// HandleIdle causes a TxnPool to actively attempt aborting idle
 	// transactions in the sync.Pool after an update has been committed.  There
 	// is no guarantee when using AbortIdle that all idle readers will be
 	// aborted.
-	AbortIdle
-	RenewOutstanding
+	HandleIdle
+
+	// HandleRenew modifies how other UpdateHandling flags are interpretted and
+	// causes a TxnPool to renew transactions and put them back in the pool
+	// instead of aborting them.
+	HandleRenew
+
+	// HandleAll is a convenient alias for the combination of HandleOutstanding
+	// and HandleIdle.
+	HandleAll = HandleOutstanding | HandleIdle
 )
 
 // TxnPool is a pool for reusing transactions through their Reset and Renew
@@ -38,8 +50,13 @@ const (
 // updates and prevent long-lived updates from causing excessive disk
 // utilization.
 type TxnPool struct {
-	env    *lmdb.Env
+	// UpdateHandling determines how a TxnPool behaves after updates have been
+	// committed.  It is not safe to modify UpdateHandling if TxnPool is being
+	// used concurrently.
+	UpdateHandling UpdateHandling
+
 	lastid uintptr
+	env    *lmdb.Env
 	pool   sync.Pool
 }
 
@@ -83,29 +100,19 @@ func (p *TxnPool) beginReadonly() (*lmdb.Txn, error) {
 		return p.env.BeginTxn(nil, lmdb.Readonly)
 	}
 
-	// Abort the pooled transaction if it is causing LMDB to hold onto old
-	// pages.
-	id := txn.ID()
-	lastid := atomic.LoadUintptr(&p.lastid)
-	if id < lastid {
-		txn.Abort()
-
-		// Beginning a new transaction without continuing to read from the pool
-		// is lazy.  But it is likely that remaining Txn objects in the pool
-		// are holding stale pages and polling them out would be slow.
-		// Instead, we hope the Txn finalizer will pick them up before any
-		// other caller would.
-		return p.env.BeginTxn(nil, lmdb.Readonly)
-	}
-
+	// If txn was holding stale pages the call to txn.Renew() should release
+	// them when txn aquires a new lock (this is an implication made by the
+	// LMDB documentation as of 0.9.19).
 	err := txn.Renew()
 	if err != nil {
+		p.renewError(err)
+
+		// Nothing we can do with txn now other than destroy it.
 		txn.Abort()
-		log.Printf("lmdbpool: failed to renew transaction: %v", err)
 
-		// It's not clear for now what better handling of a renew error would
-		// be so we just try to create a new transaction.  Presumably it will
-		// fail with the same error.
+		// For now it's not clear what better handling of a renew error would
+		// entail so we just try to create a new transaction.  It is assumed
+		// that it will fail with the same error... But maybe not?
 		return p.env.BeginTxn(nil, lmdb.Readonly)
 	}
 
@@ -118,21 +125,68 @@ func (p *TxnPool) beginReadonly() (*lmdb.Txn, error) {
 	return txn, nil
 }
 
+func (p *TxnPool) renewError(err error) {
+	// TODO:
+	// When this is integrated directly in the lmdb package this can use
+	// the same logging functionality that the Txn finalizer uses.
+	log.Printf("lmdbpool: failed to renew transaction: %v", err)
+}
+
 func (p *TxnPool) abortReadonly(txn *lmdb.Txn) {
+	if !returnTxnToPool {
+		// If the pool is disabled from race detection then we just abort the
+		// Txn instead of waiting for the finalizer.  See the files put.go and
+		// putrace.go for more information.
+		txn.Abort()
+		return
+	}
+
+	// We want to make sure that we handle updates in some way before we call
+	// either txn.ID() or p.getLastID() as both (can) incurr overhead.
+	if p.handlesUpdates() && txn.ID() < p.getLastID() {
+		ok, err := p.handleReadonly(txn, HandleOutstanding)
+		if err != nil {
+			// We attempted to renew the transaction but failed and the
+			// transaction was automatically aborted.
+			p.renewError(err)
+			return
+		}
+		if !ok {
+			// The transaction was aborted instead of being renewed.
+			return
+		}
+	}
+
 	// Don't waste cycles resetting RawRead here -- the cost be paid when the
 	// Txn is reused (if at all).  All we need to do is set txn.Pooled to avoid
 	// any warning emitted from the Txn finalizer.
 	txn.Pooled = true
-
 	txn.Reset()
-	if returnTxnToPool {
-		p.pool.Put(txn)
-	} else {
-		// If the pool is disabled from race detection then we just abort the
-		// Txn instead of waiting for the finalizer.  See the files put.go and
-		// putrace.go for more information.
-		txn.Abort()
+	p.pool.Put(txn)
+}
+
+func (p *TxnPool) handleReadonly(txn *lmdb.Txn, condition UpdateHandling) (renewed bool, err error) {
+	if p.UpdateHandling&condition == 0 {
+		return
 	}
+
+	if p.UpdateHandling&HandleRenew != 0 {
+		err = txn.Renew()
+		if err != nil {
+			// There is not much to do with txn other than abort it.
+			txn.Abort()
+		}
+		return true, err
+	}
+	txn.Abort()
+	return false, nil
+}
+
+// getLastID safely retrieves the value of p.lastid so routines operating on
+// the sync.Pool know if a transaction can continue to be used without bloating
+// the database.
+func (p *TxnPool) getLastID() uintptr {
+	return atomic.LoadUintptr(&p.lastid)
 }
 
 // CommitID sets the TxnPool's last-known transaction id to invalidate
@@ -140,13 +194,39 @@ func (p *TxnPool) abortReadonly(txn *lmdb.Txn) {
 //
 // CommitID should only be called if p is not used to create/commit update
 // transactions.
+//
+// BUG:
+// HandleIdle is not checked.
 func (p *TxnPool) CommitID(id uintptr) {
+	if !p.handlesUpdates() {
+		return
+	}
+
 	// As long as we think we are holding a newer id than lastid we keep trying
 	// to CAS until we see a newer id or the CAS succeeds.
 	lastid := atomic.LoadUintptr(&p.lastid)
 	for lastid < id && !atomic.CompareAndSwapUintptr(&p.lastid, lastid, id) {
 		lastid = atomic.LoadUintptr(&p.lastid)
 	}
+
+	// TODO:
+	// Presuming the CAS succeeded, do we now try to asynchronously terminate
+	// transactions in the pool?  Alternative to terminating the transactions,
+	// we could go through the pool and renew/reset any stale txns we find?
+	//
+	// In the case where a single transaction enters and exits the pool
+	// repeatedly we are actually doing a disservice to the application because
+	// it will need to allocate more Txns than it would otherwise if we were to
+	// terminate them. Renewing them preemptively runs the risk of wasting
+	// resources.
+	//
+	// The questions surrounding this require more benchmarks and real world
+	// experimentation.
+}
+
+// handlesUpdates returns if updates are handled in any way.
+func (p *TxnPool) handlesUpdates() bool {
+	return p.UpdateHandling&HandleAll != 0
 }
 
 // Abort aborts the txn and allows it to be reused if possible.  Abort must
diff --git a/lmdb/txn.go b/lmdb/txn.go
index e4c89ca..3f19270 100644
--- a/lmdb/txn.go
+++ b/lmdb/txn.go
@@ -52,11 +52,18 @@ type Txn struct {
 
 	managed  bool
 	readonly bool
-	env      *Env
-	_txn     *C.MDB_txn
-	errLogf  func(format string, v ...interface{})
-	key      *C.MDB_val
-	val      *C.MDB_val
+
+	// The value of Txn.ID() is cached so that the cost of cgo does not have to
+	// be paid.  The id of a Txn cannot change over its life, even if it is
+	// reset/renewed
+	id uintptr
+
+	env  *Env
+	_txn *C.MDB_txn
+	key  *C.MDB_val
+	val  *C.MDB_val
+
+	errLogf func(format string, v ...interface{})
 }
 
 // beginTxn does not lock the OS thread which is a prerequisite for creating a
@@ -102,6 +109,19 @@ func beginTxn(env *Env, parent *Txn, flags uint) (*Txn, error) {
 //
 // See mdb_txn_id.
 func (txn *Txn) ID() uintptr {
+	// It is possible for a txn to legitimately have ID 0 if it a readonly txn
+	// created before any updates.  In practice this does not really happen
+	// because an application typically must do an initial update to initialize
+	// application dbis.  Even so, calling C.mdb_txn_id excessively isn't
+	// actually harmful, it is just slow.
+	if txn.id == 0 {
+		txn.id = txn.getID()
+	}
+
+	return txn.id
+}
+
+func (txn *Txn) getID() uintptr {
 	return uintptr(C.mdb_txn_id(txn._txn))
 }
 
@@ -171,7 +191,7 @@ func (txn *Txn) Commit() error {
 
 func (txn *Txn) commit() error {
 	ret := C.mdb_txn_commit(txn._txn)
-	txn._txn = nil
+	txn.clearTxn()
 	return operrno("mdb_txn_commit", ret)
 }
 
@@ -201,9 +221,27 @@ func (txn *Txn) abort() {
 	}
 	txn.env.closeLock.RUnlock()
 
+	txn.clearTxn()
+}
+
+func (txn *Txn) clearTxn() {
 	// Clear the C object to prevent any potential future use of the freed
 	// pointer.
 	txn._txn = nil
+
+	// Clear txn.id because it no longer matches the value of txn._txn (and
+	// future calls to txn.ID() should not see the stale id).  Instead of
+	// returning the old ID future calls to txn.ID() will query LMDB to make
+	// sure the value returned for an invalid Txn is more or less consistent
+	// for people familiar with the C semantics.
+	txn.resetID()
+}
+
+// resetID has to be called anytime the value of Txn.getID() may change
+// otherwise the cached value may diverge from the actual value and the
+// abstraction has failed.
+func (txn *Txn) resetID() {
+	txn.id = 0
 }
 
 // Reset aborts the transaction clears internal state so the transaction may be
@@ -238,6 +276,16 @@ func (txn *Txn) Renew() error {
 
 func (txn *Txn) renew() error {
 	ret := C.mdb_txn_renew(txn._txn)
+
+	// mdb_txn_renew causes txn._txn to pick up a new transaction ID.  It's
+	// slightly confusing in the LMDB docs.  Txn ID corresponds to database
+	// snapshot the reader is holding, which is good because renewed
+	// transactions can see updates which happened since they were created (or
+	// since they were last renewed).  It should follow that renewing a Txn
+	// results in the freeing of stale pages the Txn has been holding, though
+	// this has not been confirmed in any way by bmatsuo as of 2017-02-15.
+	txn.resetID()
+
 	return operrno("mdb_txn_renew", ret)
 }
 


Test patch:
diff --git a/lmdb/txn_test.go b/lmdb/txn_test.go
index 4f61c2a..b8316cb 100644
--- a/lmdb/txn_test.go
+++ b/lmdb/txn_test.go
@@ -15,9 +15,37 @@ func TestTxn_ID(t *testing.T) {
 	env := setup(t)
 	defer clean(env, t)
 
-	var id1, id2, id3 uintptr
+	var id0, id1, id2, id3 uintptr
 	var txnInvalid *Txn
-	err := env.Update(func(txn *Txn) (err error) {
+	err := env.View(func(txn *Txn) (err error) {
+		id0 = txn.ID()
+		return nil
+	})
+	if err != nil {
+		t.Error(err)
+	}
+	if 0 != id0 {
+		t.Errorf("unexpected readonly id (before update): %v (!= %v)", id0, 0)
+	}
+
+	txnCached, err := env.BeginTxn(nil, Readonly)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	defer txnCached.Abort()
+	if 0 != txnCached.ID() {
+		t.Errorf("unexpected readonly id (before update): %v (!= %v)", txnCached.ID(), 0)
+	}
+	if txnCached.getID() != txnCached.ID() {
+		t.Errorf("unexpected readonly id (before update): %v (!= %v)", txnCached.ID(), txnCached.getID())
+	}
+	txnCached.Reset()
+	if txnCached.getID() != txnCached.ID() {
+		t.Errorf("unexpected reset id: %v (!= %v)", txnCached.ID(), txnCached.getID())
+	}
+
+	err = env.Update(func(txn *Txn) (err error) {
 		dbi, err := txn.OpenRoot(0)
 		if err != nil {
 			return err
@@ -39,15 +67,35 @@ func TestTxn_ID(t *testing.T) {
 		return
 	}
 	id3 = txnInvalid.ID()
+
+	// The ID of txnCached will actually change during the call to
+	// txnCached.Renew().  It's imperitive that any ID cached in the Txn object
+	// does not diverge.
+	if txnCached.ID() != txnCached.getID() {
+		t.Errorf("unexpected invalid id: %v (!= %v)", txnCached.ID(), txnCached.getID())
+	}
+	err = txnCached.Renew()
+	if err != nil {
+		t.Error(err)
+		return
+	}
+	if txnCached.ID() != txnCached.getID() {
+		t.Errorf("unexpected invalid id: %v (!= %v)", txnCached.ID(), txnCached.getID())
+	}
+
+	t.Logf("ro txn id:: %v", id1)
 	t.Logf("txn id: %v", id1)
 	t.Logf("ro txn id: %v", id2)
 	t.Logf("bad txn id: %v", id3)
 	if id1 != id2 {
 		t.Errorf("unexpected readonly id: %v (!= %v)", id2, id1)
 	}
-	if id2 == id3 {
+	if 0 != id3 {
 		t.Errorf("unexpected invalid id: %v (!= %v)", id3, 0)
 	}
+	if id1 != txnCached.ID() {
+		t.Errorf("unexpected invalid id: %v (!= %v)", txnCached.ID(), 0)
+	}
 }
 
 func TestTxn_errLogf(t *testing.T) {