Merge pull request #85 from Workiva/try_pools

Try pools

Author: dustinhiatt-wf

btree/palm/tree.go

@@ -35,13 +35,21 @@ const (
 	apply
 )
 
-const multiThreadAt = 1000 // number of keys before we multithread lookups
+const multiThreadAt = 400 // number of keys before we multithread lookups
 
 type keyBundle struct {
 	key         common.Comparator
 	left, right *node
 }
 
+func (kb *keyBundle) dispose(ptree *ptree) {
+	if ptree.kbRing.Len() == ptree.kbRing.Cap() {
+		return
+	}
+	kb.key, kb.left, kb.right = nil, nil, nil
+	ptree.kbRing.Put(kb)
+}
+
 type ptree struct {
 	root            *node
 	_padding0       [8]uint64
@@ -54,6 +62,10 @@ type ptree struct {
 	disposed        uint64
 	buffer1         [8]uint64
 	running         uint64
+	_padding2       [8]uint64
+	kbRing          *queue.RingBuffer
+	disposeChannel  chan bool
+	mpChannel       chan map[*node][]*keyBundle
 }
 
 func (ptree *ptree) checkAndRun(action action) {
@@ -110,6 +122,29 @@ func (ptree *ptree) init(bufferSize, ary uint64) {
 	ptree.cache = make([]interface{}, 0, bufferSize)
 	ptree.root = newNode(true, newKeys(ary), newNodes(ary))
 	ptree.actions = queue.NewRingBuffer(ptree.bufferSize)
+	ptree.kbRing = queue.NewRingBuffer(1024)
+	for i := uint64(0); i < ptree.kbRing.Cap(); i++ {
+		ptree.kbRing.Put(&keyBundle{})
+	}
+	ptree.disposeChannel = make(chan bool)
+	ptree.mpChannel = make(chan map[*node][]*keyBundle, 1024)
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go ptree.disposer(&wg)
+	wg.Wait()
+}
+
+func (ptree *ptree) newKeyBundle(key common.Comparator) *keyBundle {
+	if ptree.kbRing.Len() == 0 {
+		return &keyBundle{key: key}
+	}
+	ifc, err := ptree.kbRing.Get()
+	if err != nil {
+		return nil
+	}
+	kb := ifc.(*keyBundle)
+	kb.key = key
+	return kb
 }
 
 func (ptree *ptree) operationRunner(xns interfaces, threaded bool) {
@@ -153,12 +188,12 @@ func (ptree *ptree) fetchKeys(xns interfaces, inParallel bool) (map[*node][]*key
 		switch action.operation() {
 		case add:
 			for i, n := range action.nodes() {
-				writeOperations[n] = append(writeOperations[n], &keyBundle{key: action.keys()[i]})
+				writeOperations[n] = append(writeOperations[n], ptree.newKeyBundle(action.keys()[i]))
 			}
 			toComplete = append(toComplete, action)
 		case remove:
 			for i, n := range action.nodes() {
-				deleteOperations[n] = append(deleteOperations[n], &keyBundle{key: action.keys()[i]})
+				deleteOperations[n] = append(deleteOperations[n], ptree.newKeyBundle(action.keys()[i]))
 			}
 			toComplete = append(toComplete, action)
 		case get, apply:
@@ -188,6 +223,19 @@ func (ptree *ptree) apply(n *node, aa *applyAction) {
 	}
 }
 
+func (ptree *ptree) disposer(wg *sync.WaitGroup) {
+	wg.Done()
+
+	for {
+		select {
+		case mp := <-ptree.mpChannel:
+			ptree.cleanMap(mp)
+		case <-ptree.disposeChannel:
+			return
+		}
+	}
+}
+
 func (ptree *ptree) fetchKeysInSerial(xns interfaces) {
 	for _, ifc := range xns {
 		action := ifc.(action)
@@ -340,6 +388,14 @@ func (ptree *ptree) applyNode(n *node, adds, deletes []*keyBundle) {
 	}
 }
 
+func (ptree *ptree) cleanMap(op map[*node][]*keyBundle) {
+	for _, bundles := range op {
+		for _, kb := range bundles {
+			kb.dispose(ptree)
+		}
+	}
+}
+
 func (ptree *ptree) recursiveMutate(adds, deletes map[*node][]*keyBundle, setRoot, inParallel bool) {
 	if len(adds) == 0 && len(deletes) == 0 {
 		return
@@ -413,12 +469,18 @@ func (ptree *ptree) recursiveMutate(adds, deletes map[*node][]*keyBundle, setRoo
 			ptree.splitNode(n, parent, &nodes, &keys)
 			write.Lock()
 			for i, k := range keys {
-				nextLayerWrite[parent] = append(nextLayerWrite[parent], &keyBundle{key: k, left: nodes[i*2], right: nodes[i*2+1]})
+				kb := ptree.newKeyBundle(k)
+				kb.left = nodes[i*2]
+				kb.right = nodes[i*2+1]
+				nextLayerWrite[parent] = append(nextLayerWrite[parent], kb)
 			}
 			write.Unlock()
 		}
 	})
 
+	ptree.mpChannel <- adds
+	ptree.mpChannel <- deletes
+
 	ptree.recursiveMutate(nextLayerWrite, nextLayerDelete, setRoot, inParallel)
 }
 
@@ -467,8 +529,12 @@ func (ptree *ptree) Query(start, stop common.Comparator) common.Comparators {
 // Dispose will clean up any resources used by this tree.  This
 // must be called to prevent a memory leak.
 func (ptree *ptree) Dispose() {
+	if atomic.LoadUint64(&ptree.disposed) == 1 {
+		return
+	}
 	ptree.actions.Dispose()
 	atomic.StoreUint64(&ptree.disposed, 1)
+	close(ptree.disposeChannel)
 }
 
 func (ptree *ptree) print(output *log.Logger) {

btree/palm/tree_test.go

@@ -437,7 +437,7 @@ func BenchmarkReadAndWrites(b *testing.B) {
 		keys = append(keys, generateRandomKeys(numItems))
 	}
 
-	tree := newTree(16, 8)
+	tree := newTree(8, 8)
 	b.ResetTimer()
 
 	for i := 0; i < b.N; i++ {
@@ -447,25 +447,27 @@ func BenchmarkReadAndWrites(b *testing.B) {
 }
 
 func BenchmarkSimultaneousReadsAndWrites(b *testing.B) {
-	numItems := 1000
+	numItems := 10000
 	numRoutines := 8
-	keys := make([]common.Comparators, 0, numRoutines)
-	for i := 0; i < numRoutines; i++ {
-		keys = append(keys, generateRandomKeys(numItems))
+	keys := generateRandomKeys(numItems)
+	chunks := chunkKeys(keys, int64(numRoutines))
+
+	trees := make([]*ptree, 0, numItems)
+	for i := 0; i < b.N; i++ {
+		trees = append(trees, newTree(8, 8))
 	}
 
-	tree := newTree(16, 8)
 	var wg sync.WaitGroup
 	b.ResetTimer()
 
 	for i := 0; i < b.N; i++ {
 		wg.Add(numRoutines)
 		for j := 0; j < numRoutines; j++ {
-			go func(j int) {
-				tree.Insert(keys[j]...)
-				tree.Get(keys[j]...)
+			go func(i, j int) {
+				trees[i].Insert(chunks[j]...)
+				trees[i].Get(chunks[j]...)
 				wg.Done()
-			}(j)
+			}(i, j)
 		}
 
 		wg.Wait()
@@ -475,12 +477,15 @@ func BenchmarkSimultaneousReadsAndWrites(b *testing.B) {
 func BenchmarkBulkAdd(b *testing.B) {
 	numItems := 10000
 	keys := generateRandomKeys(numItems)
+	trees := make([]*ptree, 0, b.N)
+	for i := 0; i < b.N; i++ {
+		trees = append(trees, newTree(8, 8))
+	}
 
 	b.ResetTimer()
 
 	for i := 0; i < b.N; i++ {
-		tree := newTree(8, 8)
-		tree.Insert(keys...)
+		trees[i].Insert(keys...)
 	}
 }
 
@@ -515,7 +520,7 @@ func BenchmarkBulkAddToExisting(b *testing.B) {
 func BenchmarkGet(b *testing.B) {
 	numItems := 10000
 	keys := generateRandomKeys(numItems)
-	tree := newTree(32, 8)
+	tree := newTree(8, 8)
 	tree.Insert(keys...)
 
 	b.ResetTimer()

btree/plus/btree_test.go

@@ -17,6 +17,7 @@ limitations under the License.
 package plus
 
 import (
+	"sync"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
@@ -394,3 +395,34 @@ func BenchmarkReadAndWrites(b *testing.B) {
 		tree.Get(ks[i]...)
 	}
 }
+
+func BenchmarkSimultaneousReadsAndWrites(b *testing.B) {
+	numItems := 10000
+	numRoutines := 8
+	keys := constructRandomMockKeys(numItems)
+	chunks := chunkKeys(keys, int64(numRoutines))
+
+	trees := make([]*btree, 0, numItems)
+	for i := 0; i < b.N; i++ {
+		trees = append(trees, newBTree(8))
+	}
+
+	var wg sync.WaitGroup
+	var lock sync.Mutex
+	b.ResetTimer()
+
+	for i := 0; i < b.N; i++ {
+		wg.Add(numRoutines)
+		for j := 0; j < numRoutines; j++ {
+			go func(i, j int) {
+				lock.Lock()
+				trees[i].Insert(chunks[j]...)
+				trees[i].Get(chunks[j]...)
+				lock.Unlock()
+				wg.Done()
+			}(i, j)
+		}
+
+		wg.Wait()
+	}
+}

btree/plus/mock_test.go

@@ -16,6 +16,14 @@ limitations under the License.
 
 package plus
 
+func chunkKeys(ks keys, numParts int64) []keys {
+	parts := make([]keys, numParts)
+	for i := int64(0); i < numParts; i++ {
+		parts[i] = ks[i*int64(len(ks))/numParts : (i+1)*int64(len(ks))/numParts]
+	}
+	return parts
+}
+
 type mockKey struct {
 	value int
 }

queue/ring.go

@@ -49,13 +49,13 @@ type nodes []*node
 // described here: http://www.1024cores.net/home/lock-free-algorithms/queues/bounded-mpmc-queue
 // with some minor additions.
 type RingBuffer struct {
-	_buffer0       [8]uint64
+	_padding0      [8]uint64
 	queue          uint64
-	_buffer1       [8]uint64
+	_padding1      [8]uint64
 	dequeue        uint64
-	_buffer2       [8]uint64
+	_padding2      [8]uint64
 	mask, disposed uint64
-	_buffer3       [8]uint64
+	_padding3      [8]uint64
 	nodes          nodes
 }
 
@@ -74,6 +74,7 @@ func (rb *RingBuffer) init(size uint64) {
 func (rb *RingBuffer) Put(item interface{}) error {
 	var n *node
 	pos := atomic.LoadUint64(&rb.queue)
+	i := 0
 L:
 	for {
 		if atomic.LoadUint64(&rb.disposed) == 1 {
@@ -92,7 +93,13 @@ L:
 		default:
 			pos = atomic.LoadUint64(&rb.queue)
 		}
-		runtime.Gosched() // free up the cpu before the next iteration
+
+		if i == 10000 {
+			runtime.Gosched() // free up the cpu before the next iteration
+			i = 0
+		} else {
+			i++
+		}
 	}
 
 	n.data = item
@@ -107,6 +114,7 @@ L:
 func (rb *RingBuffer) Get() (interface{}, error) {
 	var n *node
 	pos := atomic.LoadUint64(&rb.dequeue)
+	i := 0
 L:
 	for {
 		if atomic.LoadUint64(&rb.disposed) == 1 {
@@ -125,7 +133,13 @@ L:
 		default:
 			pos = atomic.LoadUint64(&rb.dequeue)
 		}
-		runtime.Gosched() // free up the cpu before the next iteration
+
+		if i == 10000 {
+			runtime.Gosched() // free up the cpu before the next iteration
+			i = 0
+		} else {
+			i++
+		}
 	}
 	data := n.data
 	n.data = nil