Initial work on ring buffer.

Author: dustinhiatt-wf

btree/palm/tree.go

@@ -151,8 +151,8 @@ func (ptree *ptree) runOperations() {
 		writeOperations[n] = append(writeOperations[n], toPerform.writes[i])
 	}
 
-	toPerform.signaler <- true
 	ptree.runAdds(writeOperations)
+	toPerform.signaler <- true
 }
 
 func (ptree *ptree) recursiveSplit(n, parent, left *node, nodes *[]*node, keys *Keys) {

btree/palm/tree_test.go

@@ -331,7 +331,7 @@ func BenchmarkBulkAddToExisting(b *testing.B) {
 		keySet = append(keySet, generateRandomKeys(numItems))
 	}
 
-	tree := newTree(1024)
+	tree := newTree(32768)
 
 	b.ResetTimer()
 

queue/error.go

@@ -16,8 +16,6 @@ limitations under the License.
 
 package queue
 
-type DisposedError struct{}
+import "errors"
 
-func (de DisposedError) Error() string {
-	return `Queue has been disposed.`
-}
+var disposedError = errors.New(`Queue has been disposed.`)

queue/priority_queue.go

@@ -107,7 +107,7 @@ func (pq *PriorityQueue) Put(items ...Item) error {
 	pq.lock.Lock()
 	if pq.disposed {
 		pq.lock.Unlock()
-		return DisposedError{}
+		return disposedError
 	}
 
 	for _, item := range items {
@@ -144,7 +144,7 @@ func (pq *PriorityQueue) Get(number int) ([]Item, error) {
 
 	if pq.disposed {
 		pq.lock.Unlock()
-		return nil, DisposedError{}
+		return nil, disposedError
 	}
 
 	var items []Item
@@ -159,7 +159,7 @@ func (pq *PriorityQueue) Get(number int) ([]Item, error) {
 		pq.disposeLock.Lock()
 		if pq.disposed {
 			pq.disposeLock.Unlock()
-			return nil, DisposedError{}
+			return nil, disposedError
 		}
 		pq.disposeLock.Unlock()
 

queue/queue.go

@@ -128,7 +128,7 @@ func (q *Queue) Put(items ...interface{}) error {
 
 	if q.disposed {
 		q.lock.Unlock()
-		return DisposedError{}
+		return disposedError
 	}
 
 	q.items = append(q.items, items...)
@@ -163,7 +163,7 @@ func (q *Queue) Get(number int64) ([]interface{}, error) {
 
 	if q.disposed {
 		q.lock.Unlock()
-		return nil, DisposedError{}
+		return nil, disposedError
 	}
 
 	var items []interface{}
@@ -177,7 +177,7 @@ func (q *Queue) Get(number int64) ([]interface{}, error) {
 		sema.wg.Wait()
 		// we are now inside the put's lock
 		if q.disposed {
-			return nil, DisposedError{}
+			return nil, disposedError
 		}
 		items = q.items.get(number)
 		sema.response.Done()
@@ -201,7 +201,7 @@ func (q *Queue) TakeUntil(checker func(item interface{}) bool) ([]interface{}, e
 
 	if q.disposed {
 		q.lock.Unlock()
-		return nil, DisposedError{}
+		return nil, disposedError
 	}
 
 	result := q.items.getUntil(checker)

queue/queue.prof

@@ -0,0 +1,30 @@
+/*
+Copyright 2014 Workiva, LLC
+
+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 queue
+
+type RingBuffer struct {
+	ifs               []interface{}
+	placed, retrieved uint64
+	size              int64
+}
+
+func (rb *RingBuffer) Put(items ...interface{}) error {
+	if len(items) == 0 {
+		return nil
+	}
+
+	return nil
+}