major refactor of kvstore/index internals, see below

In the index/store package introduce KVReader creates snapshot all read operations consistent from this snapshot must close to release introduce KVWriter only one writer active access to all operations allows for consisten read-modify-write must close to release introduce AssociativeMerge operation on batch allows efficient read-modify-write for associative operations used to consolidate updates to the term summary rows saves 1 set and 1 get op per shared instance of term in field In the index package introduced an IndexReader exposes a consisten snapshot of the index for searching At top level All searches now operate on a consisten snapshot of the index
abhishekdas29 · Sep 12, 2014 · 198ca1a · 198ca1a
1 parent 7819deb
commit 198ca1a
Show file tree

Hide file tree

Showing 62 changed files with 1,533 additions and 575 deletions.
diff --git a/index/index.go b/index/index.go
@@ -17,29 +17,38 @@ type Index interface {
 	Open() error
 	Close()
 
+	DocCount() uint64
+
 	Update(doc *document.Document) error
 	Delete(id string) error
 	Batch(batch Batch) error
 
+	SetInternal(key, val []byte) error
+	DeleteInternal(key []byte) error
+
+	DumpAll() chan interface{}
+	DumpDoc(id string) chan interface{}
+	DumpFields() chan interface{}
+
+	Reader() IndexReader
+}
+
+type IndexReader interface {
 	TermFieldReader(term []byte, field string) (TermFieldReader, error)
 	DocIDReader(start, end string) (DocIDReader, error)
 
 	FieldReader(field string, startTerm []byte, endTerm []byte) (FieldReader, error)
 
-	DocCount() uint64
-
 	Document(id string) (*document.Document, error)
 	DocumentFieldTerms(id string) (FieldTerms, error)
 
 	Fields() ([]string, error)
 
-	SetInternal(key, val []byte) error
 	GetInternal(key []byte) ([]byte, error)
-	DeleteInternal(key []byte) error
 
-	DumpAll() chan interface{}
-	DumpDoc(id string) chan interface{}
-	DumpFields() chan interface{}
+	DocCount() uint64
+
+	Close()
 }
 
 type FieldTerms map[string][]string

diff --git a/index/store/boltdb/batch.go b/index/store/boltdb/batch.go
@@ -10,6 +10,7 @@
 package boltdb
 
 import (
+	indexStore "github.com/blevesearch/bleve/index/store"
 	"github.com/boltdb/bolt"
 )
 
@@ -19,14 +20,27 @@ type op struct {
 }
 
 type Batch struct {
-	store *Store
-	ops   []op
+	store         *Store
+	alreadyLocked bool
+	ops           []op
+	merges        map[string]indexStore.AssociativeMergeChain
 }
 
 func newBatch(store *Store) *Batch {
 	rv := Batch{
-		store: store,
-		ops:   make([]op, 0),
+		store:  store,
+		ops:    make([]op, 0),
+		merges: make(map[string]indexStore.AssociativeMergeChain),
+	}
+	return &rv
+}
+
+func newBatchAlreadyLocked(store *Store) *Batch {
+	rv := Batch{
+		store:         store,
+		alreadyLocked: true,
+		ops:           make([]op, 0),
+		merges:        make(map[string]indexStore.AssociativeMergeChain),
 	}
 	return &rv
 }
@@ -39,10 +53,45 @@ func (i *Batch) Delete(key []byte) {
 	i.ops = append(i.ops, op{key, nil})
 }
 
+func (i *Batch) Merge(key []byte, oper indexStore.AssociativeMerge) {
+	opers, ok := i.merges[string(key)]
+	if !ok {
+		opers = make(indexStore.AssociativeMergeChain, 0, 1)
+	}
+	opers = append(opers, oper)
+	i.merges[string(key)] = opers
+}
+
 func (i *Batch) Execute() error {
+	if !i.alreadyLocked {
+		i.store.writer.Lock()
+		defer i.store.writer.Unlock()
+	}
 	return i.store.db.Update(func(tx *bolt.Tx) error {
 		b := tx.Bucket([]byte(i.store.bucket))
 
+		// first processed the merges
+		for k, mc := range i.merges {
+			val := b.Get([]byte(k))
+			var err error
+			val, err = mc.Merge([]byte(k), val)
+			if err != nil {
+				return err
+			}
+			if val == nil {
+				err := b.Delete([]byte(k))
+				if err != nil {
+					return err
+				}
+			} else {
+				err := b.Put([]byte(k), val)
+				if err != nil {
+					return err
+				}
+			}
+		}
+
+		// now process the regular get/set ops
 		for _, o := range i.ops {
 			if o.v == nil {
 				if err := b.Delete(o.k); err != nil {

diff --git a/index/store/boltdb/iterator.go b/index/store/boltdb/iterator.go
@@ -15,6 +15,7 @@ import (
 
 type Iterator struct {
 	store  *Store
+	ownTx  bool
 	tx     *bolt.Tx
 	cursor *bolt.Cursor
 	valid  bool
@@ -27,6 +28,18 @@ func newIterator(store *Store) *Iterator {
 	b := tx.Bucket([]byte(store.bucket))
 	cursor := b.Cursor()
 
+	return &Iterator{
+		store:  store,
+		tx:     tx,
+		ownTx:  true,
+		cursor: cursor,
+	}
+}
+
+func newIteratorExistingTx(store *Store, tx *bolt.Tx) *Iterator {
+	b := tx.Bucket([]byte(store.bucket))
+	cursor := b.Cursor()
+
 	return &Iterator{
 		store:  store,
 		tx:     tx,
@@ -69,5 +82,8 @@ func (i *Iterator) Valid() bool {
 }
 
 func (i *Iterator) Close() {
-	i.tx.Rollback()
+	// only close the transaction if we opened it
+	if i.ownTx {
+		i.tx.Rollback()
+	}
 }
diff --git a/index/store/boltdb/reader.go b/index/store/boltdb/reader.go
@@ -0,0 +1,43 @@
+//  Copyright (c) 2014 Couchbase, Inc.
+//  Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file
+//  except in compliance with the License. You may obtain a copy of the License at
+//    http://www.apache.org/licenses/LICENSE-2.0
+//  Unless required by applicable law or agreed to in writing, software distributed under the
+//  License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
+//  either express or implied. See the License for the specific language governing permissions
+//  and limitations under the License.
+
+package boltdb
+
+import (
+	"github.com/blevesearch/bleve/index/store"
+	"github.com/boltdb/bolt"
+)
+
+type Reader struct {
+	store *Store
+	tx    *bolt.Tx
+}
+
+func newReader(store *Store) *Reader {
+	tx, _ := store.db.Begin(false)
+	return &Reader{
+		store: store,
+		tx:    tx,
+	}
+}
+
+func (r *Reader) Get(key []byte) ([]byte, error) {
+	rv := r.tx.Bucket([]byte(r.store.bucket)).Get(key)
+	return rv, nil
+}
+
+func (r *Reader) Iterator(key []byte) store.KVIterator {
+	rv := newIteratorExistingTx(r.store, r.tx)
+	rv.Seek(key)
+	return rv
+}
+
+func (r *Reader) Close() error {
+	return r.tx.Rollback()
+}
diff --git a/index/store/boltdb/store.go b/index/store/boltdb/store.go
@@ -11,6 +11,7 @@ package boltdb
 
 import (
 	"fmt"
+	"sync"
 
 	"github.com/blevesearch/bleve/index/store"
 	"github.com/blevesearch/bleve/registry"
@@ -23,6 +24,7 @@ type Store struct {
 	path   string
 	bucket string
 	db     *bolt.DB
+	writer sync.Mutex
 }
 
 func Open(path string, bucket string) (*Store, error) {
@@ -49,7 +51,7 @@ func Open(path string, bucket string) (*Store, error) {
 	return &rv, nil
 }
 
-func (bs *Store) Get(key []byte) ([]byte, error) {
+func (bs *Store) get(key []byte) ([]byte, error) {
 	var rv []byte
 
 	err := bs.db.View(func(tx *bolt.Tx) error {
@@ -61,33 +63,49 @@ func (bs *Store) Get(key []byte) ([]byte, error) {
 	return rv, err
 }
 
-func (bs *Store) Set(key, val []byte) error {
+func (bs *Store) set(key, val []byte) error {
+	bs.writer.Lock()
+	defer bs.writer.Unlock()
+	return bs.setlocked(key, val)
+}
+
+func (bs *Store) setlocked(key, val []byte) error {
 	return bs.db.Update(func(tx *bolt.Tx) error {
 		return tx.Bucket([]byte(bs.bucket)).Put(key, val)
 	})
 }
 
-func (bs *Store) Delete(key []byte) error {
+func (bs *Store) delete(key []byte) error {
+	bs.writer.Lock()
+	defer bs.writer.Unlock()
+	return bs.deletelocked(key)
+}
+
+func (bs *Store) deletelocked(key []byte) error {
 	return bs.db.Update(func(tx *bolt.Tx) error {
 		return tx.Bucket([]byte(bs.bucket)).Delete(key)
 	})
 }
 
-func (bs *Store) Commit() error {
-	return nil
-}
-
 func (bs *Store) Close() error {
 	return bs.db.Close()
 }
 
-func (bs *Store) Iterator(key []byte) store.KVIterator {
+func (bs *Store) iterator(key []byte) store.KVIterator {
 	rv := newIterator(bs)
 	rv.Seek(key)
 	return rv
 }
 
-func (bs *Store) NewBatch() store.KVBatch {
+func (bs *Store) Reader() store.KVReader {
+	return newReader(bs)
+}
+
+func (bs *Store) Writer() store.KVWriter {
+	return newWriter(bs)
+}
+
+func (bs *Store) newBatch() store.KVBatch {
 	return newBatch(bs)
 }
 

diff --git a/index/store/boltdb/store_test.go b/index/store/boltdb/store_test.go
@@ -25,3 +25,13 @@ func TestStore(t *testing.T) {
 
 	store_test.CommonTestKVStore(t, s)
 }
+
+func TestReaderIsolation(t *testing.T) {
+	s, err := Open("test", "bleve")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer os.RemoveAll("test")
+
+	store_test.CommonTestReaderIsolation(t, s)
+}
diff --git a/index/store/boltdb/writer.go b/index/store/boltdb/writer.go
@@ -0,0 +1,53 @@
+//  Copyright (c) 2014 Couchbase, Inc.
+//  Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file
+//  except in compliance with the License. You may obtain a copy of the License at
+//    http://www.apache.org/licenses/LICENSE-2.0
+//  Unless required by applicable law or agreed to in writing, software distributed under the
+//  License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
+//  either express or implied. See the License for the specific language governing permissions
+//  and limitations under the License.
+
+package boltdb
+
+import (
+	"github.com/blevesearch/bleve/index/store"
+)
+
+type Writer struct {
+	store *Store
+}
+
+func newWriter(store *Store) *Writer {
+	store.writer.Lock()
+	return &Writer{
+		store: store,
+	}
+}
+
+func (w *Writer) Set(key, val []byte) error {
+	return w.store.setlocked(key, val)
+}
+
+func (w *Writer) Delete(key []byte) error {
+	return w.store.deletelocked(key)
+}
+
+func (w *Writer) NewBatch() store.KVBatch {
+	return newBatchAlreadyLocked(w.store)
+}
+
+func (w *Writer) Close() error {
+	w.store.writer.Unlock()
+	return nil
+}
+
+// these two methods can safely read using the regular
+// methods without a read transaction, because we know
+// that no one else is writing but us
+func (w *Writer) Get(key []byte) ([]byte, error) {
+	return w.store.get(key)
+}
+
+func (w *Writer) Iterator(key []byte) store.KVIterator {
+	return w.store.iterator(key)
+}