Add nested buckets.

This commit adds the ability to create buckets inside of other buckets.
It also replaces the buckets page with a root bucket.

Fixes etcd-io#56.

Makefile

@@ -10,7 +10,7 @@ cloc:
 	@cloc --not-match-f='Makefile|_test.go' .
 
 cover: fmt
-	go test -coverprofile=$(COVERPROFILE) -test.run=$(TEST) .
+	go test -coverprofile=$(COVERPROFILE) -test.run=$(TEST) $(COVERFLAG) .
 	go tool cover -html=$(COVERPROFILE)
 	rm $(COVERPROFILE)
 
@@ -32,6 +32,6 @@ test: fmt errcheck
 	@echo ""
 	@echo ""
 	@echo "=== RACE DETECTOR ==="
-	@go test -v -race -test.run=Parallel
+	@go test -v -race -test.run="TestSimulate_(100op|1000op|10000op)"
 
 .PHONY: bench cloc cover cpuprofile fmt memprofile test

bolt_test.go

@@ -0,0 +1,327 @@
+package bolt
+
+import (
+	"bytes"
+	"fmt"
+	"math/rand"
+	"sync"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestSimulate_1op_1p(t *testing.T)     { testSimulate(t, 100, 1) }
+func TestSimulate_10op_1p(t *testing.T)    { testSimulate(t, 10, 1) }
+func TestSimulate_100op_1p(t *testing.T)   { testSimulate(t, 100, 1) }
+func TestSimulate_1000op_1p(t *testing.T)  { testSimulate(t, 1000, 1) }
+func TestSimulate_10000op_1p(t *testing.T) { testSimulate(t, 10000, 1) }
+
+func TestSimulate_10op_10p(t *testing.T)    { testSimulate(t, 10, 10) }
+func TestSimulate_100op_10p(t *testing.T)   { testSimulate(t, 100, 10) }
+func TestSimulate_1000op_10p(t *testing.T)  { testSimulate(t, 1000, 10) }
+func TestSimulate_10000op_10p(t *testing.T) { testSimulate(t, 10000, 10) }
+
+func TestSimulate_100op_100p(t *testing.T)   { testSimulate(t, 100, 100) }
+func TestSimulate_1000op_100p(t *testing.T)  { testSimulate(t, 1000, 100) }
+func TestSimulate_10000op_100p(t *testing.T) { testSimulate(t, 10000, 100) }
+
+func TestSimulate_10000op_1000p(t *testing.T) { testSimulate(t, 10000, 1000) }
+
+// Randomly generate operations on a given database with multiple clients to ensure consistency and thread safety.
+func testSimulate(t *testing.T, threadCount, parallelism int) {
+	if testing.Short() {
+		t.Skip("skipping test in short mode.")
+	}
+
+	rand.Seed(int64(qseed))
+
+	// A list of operations that readers and writers can perform.
+	var readerHandlers = []simulateHandler{simulateGetHandler}
+	var writerHandlers = []simulateHandler{simulateGetHandler, simulatePutHandler}
+
+	var versions = make(map[txid]*QuickDB)
+	versions[1] = NewQuickDB()
+	withOpenDB(func(db *DB, path string) {
+		var mutex sync.Mutex
+
+		// Run n threads in parallel, each with their own operation.
+		var wg sync.WaitGroup
+		var threads = make(chan bool, parallelism)
+		var i int
+		for {
+			threads <- true
+			wg.Add(1)
+			writable := ((rand.Int() % 100) < 20) // 20% writers
+
+			// Choose an operation to execute.
+			var handler simulateHandler
+			if writable {
+				handler = writerHandlers[rand.Intn(len(writerHandlers))]
+			} else {
+				handler = readerHandlers[rand.Intn(len(readerHandlers))]
+			}
+
+			// Execute a thread for the given operation.
+			go func(writable bool, handler simulateHandler) {
+				defer wg.Done()
+
+				// Start transaction.
+				tx, err := db.Begin(writable)
+				if err != nil {
+					t.Fatal("tx begin: ", err)
+				}
+
+				// Obtain current state of the dataset.
+				mutex.Lock()
+				var qdb = versions[tx.id()]
+				if writable {
+					qdb = versions[tx.id()-1].Copy()
+				}
+				mutex.Unlock()
+
+				// Make sure we commit/rollback the tx at the end and update the state.
+				if writable {
+					defer func() {
+						mutex.Lock()
+						versions[tx.id()] = qdb
+						mutex.Unlock()
+
+						assert.NoError(t, tx.Commit())
+					}()
+				} else {
+					defer tx.Rollback()
+				}
+
+				// Ignore operation if we don't have data yet.
+				if qdb == nil {
+					return
+				}
+
+				// Execute handler.
+				handler(tx, qdb)
+
+				// Release a thread back to the scheduling loop.
+				<-threads
+			}(writable, handler)
+
+			i++
+			if i > threadCount {
+				break
+			}
+		}
+
+		// Wait until all threads are done.
+		wg.Wait()
+	})
+}
+
+type simulateHandler func(tx *Tx, qdb *QuickDB)
+
+// Retrieves a key from the database and verifies that it is what is expected.
+func simulateGetHandler(tx *Tx, qdb *QuickDB) {
+	// Randomly retrieve an existing exist.
+	keys := qdb.Rand()
+	if len(keys) == 0 {
+		return
+	}
+
+	// Retrieve root bucket.
+	b := tx.Bucket(keys[0])
+	if b == nil {
+		panic(fmt.Sprintf("bucket[0] expected: %v\n", keys[0]))
+	}
+
+	// Drill into nested buckets.
+	for _, key := range keys[1 : len(keys)-1] {
+		b = b.Bucket(key)
+		if b == nil {
+			panic(fmt.Sprintf("bucket[n] expected: %v -> %v\n", keys, key))
+		}
+	}
+
+	// Verify key/value on the final bucket.
+	expected := qdb.Get(keys)
+	actual := b.Get(keys[len(keys)-1])
+	if !bytes.Equal(actual, expected) {
+		fmt.Println("=== EXPECTED ===")
+		fmt.Println(expected)
+		fmt.Println("=== ACTUAL ===")
+		fmt.Println(actual)
+		fmt.Println("=== END ===")
+		panic("value mismatch")
+	}
+}
+
+// Inserts a key into the database.
+func simulatePutHandler(tx *Tx, qdb *QuickDB) {
+	keys, value := randKeys(), randValue()
+
+	// Retrieve root bucket.
+	b := tx.Bucket(keys[0])
+	if b == nil {
+		if err := tx.CreateBucket(keys[0]); err != nil {
+			panic("create bucket: " + err.Error())
+		}
+		b = tx.Bucket(keys[0])
+		if b == nil {
+			panic(fmt.Sprintf("bucket[0] nil: %v", keys[0]))
+		}
+	}
+
+	// Create nested buckets, if necessary.
+	for _, key := range keys[1 : len(keys)-1] {
+		child := b.Bucket(key)
+		if child != nil {
+			b = child
+		} else {
+			if err := b.CreateBucket(key); err != nil {
+				panic("create bucket: " + err.Error())
+			}
+			b = b.Bucket(key)
+		}
+	}
+
+	// Insert into database.
+	if err := b.Put(keys[len(keys)-1], value); err != nil {
+		panic("put: " + err.Error())
+	}
+
+	// Insert into in-memory database.
+	qdb.Put(keys, value)
+}
+
+// QuickDB is an in-memory database that replicates the functionality of the
+// Bolt DB type except that it is entirely in-memory. It is meant for testing
+// that the Bolt database is consistent.
+type QuickDB struct {
+	sync.RWMutex
+	m map[string]interface{}
+}
+
+// NewQuickDB returns an instance of QuickDB.
+func NewQuickDB() *QuickDB {
+	return &QuickDB{m: make(map[string]interface{})}
+}
+
+// Get retrieves the value at a key path.
+func (db *QuickDB) Get(keys [][]byte) []byte {
+	db.RLock()
+	defer db.RUnlock()
+
+	m := db.m
+	for _, key := range keys[:len(keys)-1] {
+		value := m[string(key)]
+		if value == nil {
+			return nil
+		}
+		switch value := value.(type) {
+		case map[string]interface{}:
+			m = value
+		case []byte:
+			return nil
+		}
+	}
+
+	// Only return if it's a simple value.
+	if value, ok := m[string(keys[len(keys)-1])].([]byte); ok {
+		return value
+	}
+	return nil
+}
+
+// Put inserts a value into a key path.
+func (db *QuickDB) Put(keys [][]byte, value []byte) {
+	db.Lock()
+	defer db.Unlock()
+
+	// Build buckets all the way down the key path.
+	m := db.m
+	for _, key := range keys[:len(keys)-1] {
+		if _, ok := m[string(key)].([]byte); ok {
+			return // Keypath intersects with a simple value. Do nothing.
+		}
+
+		if m[string(key)] == nil {
+			m[string(key)] = make(map[string]interface{})
+		}
+		m = m[string(key)].(map[string]interface{})
+	}
+
+	// Insert value into the last key.
+	m[string(keys[len(keys)-1])] = value
+}
+
+// Rand returns a random key path that points to a simple value.
+func (db *QuickDB) Rand() [][]byte {
+	db.RLock()
+	defer db.RUnlock()
+	if len(db.m) == 0 {
+		return nil
+	}
+	var keys [][]byte
+	db.rand(db.m, &keys)
+	return keys
+}
+
+func (db *QuickDB) rand(m map[string]interface{}, keys *[][]byte) {
+	i, index := 0, rand.Intn(len(m))
+	for k, v := range m {
+		if i == index {
+			*keys = append(*keys, []byte(k))
+			if v, ok := v.(map[string]interface{}); ok {
+				db.rand(v, keys)
+			}
+			return
+		}
+		i++
+	}
+	panic("quickdb rand: out-of-range")
+}
+
+// Copy copies the entire database.
+func (db *QuickDB) Copy() *QuickDB {
+	db.RLock()
+	defer db.RUnlock()
+	return &QuickDB{m: db.copy(db.m)}
+}
+
+func (db *QuickDB) copy(m map[string]interface{}) map[string]interface{} {
+	clone := make(map[string]interface{}, len(m))
+	for k, v := range m {
+		switch v := v.(type) {
+		case map[string]interface{}:
+			clone[k] = db.copy(v)
+		default:
+			clone[k] = v
+		}
+	}
+	return clone
+}
+
+func randKey() []byte {
+	var min, max = 1, 1024
+	n := rand.Intn(max-min) + min
+	b := make([]byte, n)
+	for i := 0; i < n; i++ {
+		b[i] = byte(rand.Intn(255))
+	}
+	return b
+}
+
+func randKeys() [][]byte {
+	var keys [][]byte
+	var count = rand.Intn(2) + 2
+	for i := 0; i < count; i++ {
+		keys = append(keys, randKey())
+	}
+	return keys
+}
+
+func randValue() []byte {
+	n := rand.Intn(8192)
+	b := make([]byte, n)
+	for i := 0; i < n; i++ {
+		b[i] = byte(rand.Intn(255))
+	}
+	return b
+}