diff --git a/field_internal_test.go b/field_internal_test.go index deea358e4..d187d5ab9 100644 --- a/field_internal_test.go +++ b/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 diff --git a/txfactory.go b/txfactory.go index e4a868806..4fc489b53 100644 --- a/txfactory.go +++ b/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 } diff --git a/txfactory_internal_test.go b/txfactory_internal_test.go index 26416f95a..293e6fe21 100644 --- a/txfactory_internal_test.go +++ b/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