Merge pull request #933 from molecula/qcx_no_auto_reset

pilosa: no automatic reuse of Qcx

field_internal_test.go

@@ -609,12 +609,14 @@ func TestBSIGroup_importValue(t *testing.T) {
 			t.Fatalf("test %d, importing values: %s", i, err.Error())
 		}
 		panicOn(qcx.Finish())
+		qcx.Reset()
 		if row, err := f.Range(qcx, f.name, pql.EQ, tt.checkVal); err != nil {
 			t.Fatalf("test %d, getting range: %s", i, err.Error())
 		} else if !reflect.DeepEqual(row.Columns(), tt.expCols) {
 			t.Fatalf("test %d, expected columns: %v, but got: %v", i, tt.expCols, row.Columns())
 		}
 		panicOn(qcx.Finish())
+		qcx.Reset()
 	} // loop
 }
 
@@ -679,6 +681,7 @@ func TestIntField_MinMaxForShard(t *testing.T) {
 				t.Fatalf("test %d, importing values: %s", i, err.Error())
 			}
 			panicOn(qcx.Finish())
+			qcx.Reset()
 
 			shard := uint64(0)
 			tx := f.idx.holder.txf.NewTx(Txo{Write: !writable, Index: f.idx, Field: f.Field, Shard: shard})
@@ -905,6 +908,7 @@ func TestBSIGroup_TxReopenDB(t *testing.T) {
 			t.Fatalf("test %d, importing values: %s", i, err.Error())
 		}
 		panicOn(qcx.Finish())
+		qcx.Reset()
 
 		if row, err := f.Range(qcx, f.name, pql.EQ, tt.checkVal); err != nil {
 			t.Fatalf("test %d, getting range: %s", i, err.Error())
@@ -912,6 +916,7 @@ func TestBSIGroup_TxReopenDB(t *testing.T) {
 			t.Fatalf("test %d, expected columns: %v, but got: %v", i, tt.expCols, row.Columns())
 		}
 		panicOn(qcx.Finish())
+		qcx.Reset()
 	} // loop
 
 	// the test: can we re-open a BSI fragment under Tx store

txfactory.go

@@ -138,6 +138,9 @@ type Qcx struct {
 	// writable tx for all reads and writes on each given
 	// shard
 	write bool
+
+	// don't allow automatic reuse now. Must manually call Reset, or NewQcx().
+	done bool
 }
 
 // Finish commits/rollsback all stored Tx and resets the
@@ -153,7 +156,7 @@ func (q *Qcx) Finish() (err error) {
 		}
 	}
 	err2 := q.Grp.FinishGroup()
-	q.reset()
+	q.done = true
 
 	if err != nil {
 		return err
@@ -171,16 +174,25 @@ func (q *Qcx) Abort() {
 	}
 	q.Grp.AbortGroup()
 
-	q.reset()
+	q.done = true
 }
 
-// reset forgets everything are starts fresh with an empty
+// Reset forgets everything are starts fresh with an empty
 // group, ready for use again as if NewQcx() had been called.
-// q.mu must be held
-func (q *Qcx) reset() {
+func (q *Qcx) Reset() {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+	if !q.done {
+		panic("must call Qcx.Abort() or Qcx.Finish() before calling Reset().")
+	}
+	q.unprotected_reset()
+}
+
+func (q *Qcx) unprotected_reset() {
 	q.RequiredForAtomicWriteTx = nil
 	q.RequiredTxo = nil
 	q.Grp = q.Txf.NewTxGroup()
+	q.done = false
 }
 
 // NewQcxWithGroup allocates a freshly allocated and empty Grp.
@@ -236,6 +248,11 @@ func (qcx *Qcx) GetTx(o Txo) (tx Tx, finisher func(perr *error)) {
 	qcx.mu.Lock()
 	defer qcx.mu.Unlock()
 
+	if qcx.done {
+		panic(fmt.Sprintf("cannot call GetTx on a Qcx that is already done. "+
+			"Must call Reset() or txf.NewQcx(); stack='%v'", stack()))
+	}
+
 	// Use direct option if set on QCX.
 	if qcx.Direct {
 		o.Direct = true
@@ -861,7 +878,7 @@ func (f *TxFactory) NewTx(o Txo) (txn Tx) {
 
 	tx, err := dbs.NewTx(o.Write, indexName, o)
 	if err != nil {
-		panic(errors.Wrap(err, "rbfDB.NewRBFTx transaction errored"))
+		panic(errors.Wrap(err, "dbs.NewTx transaction errored"))
 	}
 	return tx
 }

txfactory_internal_test.go

@@ -74,6 +74,7 @@ func Test_TxFactory_Qcx_query_context(t *testing.T) {
 			finisher(nil) // hit the write tx.Commit path
 			// commit the change, and verify it is still there
 			panicOn(qcx.Finish())
+			qcx.Reset()
 
 			tx, finread := qcx.GetTx(Txo{Write: !writable, Index: idx, Fragment: f, Shard: f.shard})
 			if n := f.mustRow(tx, 120).Count(); n != 2 {
@@ -83,6 +84,7 @@ func Test_TxFactory_Qcx_query_context(t *testing.T) {
 			}
 			finread(nil) // no-op on reads that are in a group, so must qcx.Abort() to stop them.
 			qcx.Abort()
+			qcx.Reset()
 		}
 	}
 	N := 1000