From c90c4c275d86ef5324d8988bbd5f6fd8f3deb65a Mon Sep 17 00:00:00 2001 From: Jason Aten Date: Thu, 13 Aug 2020 03:58:37 +0000 Subject: [PATCH] lower autocommit limits to avoid ErrTxnTooBig; improve txn size estimation --- badger.go | 15 +++++++++----- badger_test.go | 2 +- bluegreentx.go | 8 ++++---- cmd/demo-lmdb/lmdb.go | 1 + cmd/demo-lmdb/vprint.go | 1 + txfactory.go | 44 +++++++++++++++++++++++++++++++++++++---- 6 files changed, 57 insertions(+), 14 deletions(-) diff --git a/badger.go b/badger.go index 187a320a6..bc217ac89 100644 --- a/badger.go +++ b/badger.go @@ -720,7 +720,8 @@ func (tx *BadgerTx) PutContainer(index, field, view string, shard uint64, ckey u panic(fmt.Sprintf("unknown roaring.Container type: %v", ct)) } tx.writeCount++ - tx.writeByteCount += len(by) + sz := len(by) + len(bkey) + 2 + tx.writeByteCount += sz tx.mu.Lock() defer tx.mu.Unlock() @@ -728,16 +729,20 @@ func (tx *BadgerTx) PutContainer(index, field, view string, shard uint64, ckey u // The integration tests do large bit level loads that exceed 10MB. // So we autocommit and start a new Txn if we are about to // write too much into one Txn. + // // The badger defaults limits are currently: // maxBatchCount:104857, maxBatchSize:10066329 - // So we stop a little before to make sure we fit. - if tx.writeCount > 100000 || tx.writeByteCount > 10000000 { + // + // However, emprirically we still get ErrTnTooBig when + // tx.writeByteCount=5884222; or when tx.writeCount=16197. + // So duck under both those thresholds by some margin. + if tx.writeCount > 10000 || tx.writeByteCount > 4000000 { // avoid ErrTxnTooBig by commiting before going over the limits, // because then we get a error: "Transaction Conflict. Please retry." panicOn(tx.tx.Commit()) tx.tx = tx.Db.db.NewTransaction(tx.write) tx.writeCount = 1 - tx.writeByteCount = len(by) + tx.writeByteCount = sz } entry := badger.NewEntry(bkey, by).WithMeta(ct) err := tx.tx.SetEntry(entry) @@ -748,7 +753,7 @@ func (tx *BadgerTx) PutContainer(index, field, view string, shard uint64, ckey u // is best to commit the transaction and start a new transaction immediately." // if err == badger.ErrTxnTooBig { - panic(fmt.Sprintf("got error badger.ErrTxnTooBig, but we shoud never get this now; len(by) = %v; vs limit is 10MB. tx.writeCount=%v; tx.writeByteCount=%v;", len(by), tx.writeCount, tx.writeByteCount)) + panic(fmt.Sprintf("got error badger.ErrTxnTooBig, but we shoud never get this now; len(by) = %v; len(bkey)=%v; vs limit is 10MB. tx.writeCount=%v; tx.writeByteCount=%v;", len(by), len(bkey), tx.writeCount, tx.writeByteCount)) } return err } diff --git a/badger_test.go b/badger_test.go index bdeafb0b8..009f38935 100644 --- a/badger_test.go +++ b/badger_test.go @@ -1278,7 +1278,7 @@ func getTestBitmapAsRawRoaring(bitsToSet ...uint64) []byte { func TestBadger_AutoCommit(t *testing.T) { // setup - dbwrap, clean := mustOpenEmptyBadgerWrapper("TestBadger_DeleteIndex") + dbwrap, clean := mustOpenEmptyBadgerWrapper("TestBadger_AutoCommit") defer clean() defer dbwrap.Close() diff --git a/bluegreentx.go b/bluegreentx.go index 1f1d9fc4e..7c62a214a 100644 --- a/bluegreentx.go +++ b/bluegreentx.go @@ -192,16 +192,16 @@ func (c *blueGreenTx) Rollback() { panic(r) } }() - fmt.Printf("blueGreenTx.Rollback() about to call (%v) a.Rollback()\n", c.as) + //fmt.Printf("blueGreenTx.Rollback() about to call (%v) a.Rollback()\n", c.as) c.a.Rollback() - fmt.Printf("blueGreenTx.Rollback() about to call (%v) b.Rollback()\n", c.bs) + //fmt.Printf("blueGreenTx.Rollback() about to call (%v) b.Rollback()\n", c.bs) c.b.Rollback() } func (c *blueGreenTx) Commit() error { c.mu.Lock() defer c.mu.Unlock() - fmt.Printf("blueGreenTx.Commit() called.\n") + //fmt.Printf("blueGreenTx.Commit() called.\n") if c.rollbackOrCommitDone { return nil } @@ -766,7 +766,7 @@ type blueGreenChecker struct { // see would mark a thing as seen. func (b *blueGreenChecker) see(index, field, view string, shard uint64) { // keep this next Printf. Useful to see the sequence of Tx operations. - fmt.Printf("blueGreenTx.%v on index='%v'\n", Caller(1), index) + //fmt.Printf("blueGreenTx.%v on index='%v'\n", Caller(1), index) b.mu.Lock() defer b.mu.Unlock() diff --git a/cmd/demo-lmdb/lmdb.go b/cmd/demo-lmdb/lmdb.go index bf91af113..1b24e06ff 100644 --- a/cmd/demo-lmdb/lmdb.go +++ b/cmd/demo-lmdb/lmdb.go @@ -13,6 +13,7 @@ // limitations under the License. // +build !386 +// +build skip_building_lmdb_for_now package main diff --git a/cmd/demo-lmdb/vprint.go b/cmd/demo-lmdb/vprint.go index d66762ef2..1e498e47c 100644 --- a/cmd/demo-lmdb/vprint.go +++ b/cmd/demo-lmdb/vprint.go @@ -23,6 +23,7 @@ // SOFTWARE. // +build !386 +// +build skip_building_lmdb_for_now package main diff --git a/txfactory.go b/txfactory.go index eb5953d7f..446a0ab9e 100644 --- a/txfactory.go +++ b/txfactory.go @@ -250,6 +250,18 @@ func (f *TxFactory) DeleteIndex(name string) error { return f.badgerDB.DeleteIndex(name) case blueGreenRoaringBadger: return f.badgerDB.DeleteIndex(name) + case blueGreenRBFBadger: + _ = f.rbfDB.DeleteIndex(name) + return f.badgerDB.DeleteIndex(name) + case blueGreenBadgerRBF: + _ = f.badgerDB.DeleteIndex(name) + return f.rbfDB.DeleteIndex(name) + case blueGreenRBFRoaring: + // roaring already done + return f.rbfDB.DeleteIndex(name) + case blueGreenRoaringRBF: + // roaring already done + return f.rbfDB.DeleteIndex(name) } panic(fmt.Sprintf("unknown f.typeOfTx type: '%v'", f.typeOfTx)) } @@ -260,8 +272,6 @@ func (f *TxFactory) DeleteFieldFromStore(index, field, fieldPath string) error { return f.roaringDB.DeleteField(index, field, fieldPath) case badgerTxn: return f.badgerDB.DeleteField(index, field, fieldPath) - //case lmdbTxn: - //return f.lmDB.DeleteField(index, field, fieldPath) case rbfTxn: return f.rbfDB.DeleteField(index, field, fieldPath) case blueGreenBadgerRoaring: @@ -270,6 +280,20 @@ func (f *TxFactory) DeleteFieldFromStore(index, field, fieldPath string) error { case blueGreenRoaringBadger: _ = f.roaringDB.DeleteField(index, field, fieldPath) return f.badgerDB.DeleteField(index, field, fieldPath) + case blueGreenRBFBadger: + _ = f.badgerDB.DeleteField(index, field, fieldPath) + return f.rbfDB.DeleteField(index, field, fieldPath) + case blueGreenBadgerRBF: + _ = f.rbfDB.DeleteField(index, field, fieldPath) + return f.badgerDB.DeleteField(index, field, fieldPath) + case blueGreenRBFRoaring: + _ = f.rbfDB.DeleteField(index, field, fieldPath) + return f.roaringDB.DeleteField(index, field, fieldPath) + case blueGreenRoaringRBF: + _ = f.roaringDB.DeleteField(index, field, fieldPath) + return f.rbfDB.DeleteField(index, field, fieldPath) + //case lmdbTxn: + //return f.lmDB.DeleteField(index, field, fieldPath) } panic(fmt.Sprintf("unknown f.typeOfTx type: '%v'", f.typeOfTx)) } @@ -282,14 +306,26 @@ func (f *TxFactory) DeleteFragmentFromStore(index, field, view string, shard uin return f.badgerDB.DeleteFragment(index, field, view, shard, frag) case rbfTxn: return f.rbfDB.DeleteFragment(index, field, view, shard, frag) - // case lmdbTxn: - // return f.lmDB.DeleteFragment(index, field, view, shard, frag) case blueGreenBadgerRoaring: _ = f.badgerDB.DeleteFragment(index, field, view, shard, frag) return f.roaringDB.DeleteFragment(index, field, view, shard, frag) case blueGreenRoaringBadger: _ = f.roaringDB.DeleteFragment(index, field, view, shard, frag) return f.badgerDB.DeleteFragment(index, field, view, shard, frag) + case blueGreenRBFBadger: + _ = f.rbfDB.DeleteFragment(index, field, view, shard, frag) + return f.badgerDB.DeleteFragment(index, field, view, shard, frag) + case blueGreenBadgerRBF: + _ = f.badgerDB.DeleteFragment(index, field, view, shard, frag) + return f.rbfDB.DeleteFragment(index, field, view, shard, frag) + case blueGreenRBFRoaring: + _ = f.rbfDB.DeleteFragment(index, field, view, shard, frag) + return f.roaringDB.DeleteFragment(index, field, view, shard, frag) + case blueGreenRoaringRBF: + _ = f.roaringDB.DeleteFragment(index, field, view, shard, frag) + return f.rbfDB.DeleteFragment(index, field, view, shard, frag) + // case lmdbTxn: + // return f.lmDB.DeleteFragment(index, field, view, shard, frag) } panic(fmt.Sprintf("unknown f.typeOfTx type: '%v'", f.typeOfTx))