multi: add buffer.Write and pool.WriteBuffer, make GCQueue generic

pool/recycle.go

@@ -0,0 +1,52 @@
+package pool
+
+import (
+	"time"
+
+	"github.com/lightningnetwork/lnd/queue"
+)
+
+// Recycler is an interface that allows an object to be reclaimed without
+// needing to be returned to the runtime.
+type Recycler interface {
+	// Recycle resets the object to its default state.
+	Recycle()
+}
+
+// Recycle is a generic queue for recycling objects implementing the Recycler
+// interface. It is backed by an underlying queue.GCQueue, and invokes the
+// Recycle method on returned objects before returning them to the queue.
+type Recycle struct {
+	queue *queue.GCQueue
+}
+
+// NewRecycle initializes a fresh Recycle instance.
+func NewRecycle(newItem func() interface{}, returnQueueSize int,
+	gcInterval, expiryInterval time.Duration) *Recycle {
+
+	return &Recycle{
+		queue: queue.NewGCQueue(
+			newItem, returnQueueSize,
+			gcInterval, expiryInterval,
+		),
+	}
+}
+
+// Take returns an element from the pool.
+func (r *Recycle) Take() interface{} {
+	return r.queue.Take()
+}
+
+// Return returns an item implementing the Recycler interface to the pool. The
+// Recycle method is invoked before returning the item to improve performance
+// and utilization under load.
+func (r *Recycle) Return(item Recycler) {
+	// Recycle the item to ensure that a dirty instance is never offered
+	// from Take. The call is done here so that the CPU cycles spent
+	// clearing the buffer are owned by the caller, and not by the queue
+	// itself. This makes the queue more likely to be available to deliver
+	// items in the free list.
+	item.Recycle()
+
+	r.queue.Return(item)
+}

pool/recycle_test.go

@@ -0,0 +1,193 @@
+package pool_test
+
+import (
+	"bytes"
+	"testing"
+	"time"
+
+	"github.com/lightningnetwork/lnd/buffer"
+	"github.com/lightningnetwork/lnd/pool"
+)
+
+type mockRecycler bool
+
+func (m *mockRecycler) Recycle() {
+	*m = false
+}
+
+// TestRecyclers verifies that known recyclable types properly return to their
+// zero-value after invoking Recycle.
+func TestRecyclers(t *testing.T) {
+	tests := []struct {
+		name    string
+		newItem func() interface{}
+	}{
+		{
+			"mock recycler",
+			func() interface{} { return new(mockRecycler) },
+		},
+		{
+			"write_buffer",
+			func() interface{} { return new(buffer.Write) },
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			// Initialize the Recycler to test.
+			r := test.newItem().(pool.Recycler)
+
+			// Dirty the item.
+			dirtyGeneric(t, r)
+
+			// Invoke Recycle to clear the item.
+			r.Recycle()
+
+			// Assert the item is now clean.
+			isCleanGeneric(t, r)
+		})
+	}
+}
+
+type recyclePoolTest struct {
+	name    string
+	newPool func() interface{}
+}
+
+// TestGenericRecyclePoolTests generically tests that pools derived from the
+// base Recycle pool properly are properly configured.
+func TestConcreteRecyclePoolTests(t *testing.T) {
+	const (
+		gcInterval     = time.Second
+		expiryInterval = 250 * time.Millisecond
+	)
+
+	tests := []recyclePoolTest{
+		{
+			name: "write buffer pool",
+			newPool: func() interface{} {
+				return pool.NewWriteBuffer(
+					gcInterval, expiryInterval,
+				)
+			},
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			testRecyclePool(t, test)
+		})
+	}
+}
+
+func testRecyclePool(t *testing.T, test recyclePoolTest) {
+	p := test.newPool()
+
+	// Take an item from the pool.
+	r1 := takeGeneric(t, p)
+
+	// Dirty the item.
+	dirtyGeneric(t, r1)
+
+	// Return the item to the pool.
+	returnGeneric(t, p, r1)
+
+	// Take items from the pool until we find the original. We expect at
+	// most two, in the event that a fresh item is populated after the
+	// first is taken.
+	for i := 0; i < 2; i++ {
+		// Wait a small duration to ensure the tests are reliable, and
+		// don't to active the non-blocking case unintentionally.
+		<-time.After(time.Millisecond)
+
+		r2 := takeGeneric(t, p)
+
+		// Take an item, skipping those whose pointer does not match the
+		// one we dirtied.
+		if r1 != r2 {
+			continue
+		}
+
+		// Finally, verify that the item has been properly cleaned.
+		isCleanGeneric(t, r2)
+
+		return
+	}
+
+	t.Fatalf("original item not found")
+}
+
+func takeGeneric(t *testing.T, p interface{}) pool.Recycler {
+	t.Helper()
+
+	switch pp := p.(type) {
+	case *pool.WriteBuffer:
+		return pp.Take()
+
+	default:
+		t.Fatalf("unknown pool type: %T", p)
+	}
+
+	return nil
+}
+
+func returnGeneric(t *testing.T, p, item interface{}) {
+	t.Helper()
+
+	switch pp := p.(type) {
+	case *pool.WriteBuffer:
+		pp.Return(item.(*buffer.Write))
+
+	default:
+		t.Fatalf("unknown pool type: %T", p)
+	}
+}
+
+func dirtyGeneric(t *testing.T, i interface{}) {
+	t.Helper()
+
+	switch item := i.(type) {
+	case *mockRecycler:
+		*item = true
+
+	case *buffer.Write:
+		dirtySlice(item[:])
+
+	default:
+		t.Fatalf("unknown item type: %T", i)
+	}
+
+}
+
+func dirtySlice(slice []byte) {
+	for i := range slice {
+		slice[i] = 0xff
+	}
+}
+
+func isCleanGeneric(t *testing.T, i interface{}) {
+	t.Helper()
+
+	switch item := i.(type) {
+	case *mockRecycler:
+		if isDirty := *item; isDirty {
+			t.Fatalf("mock recycler still diry")
+		}
+
+	case *buffer.Write:
+		isCleanSlice(t, item[:])
+
+	default:
+		t.Fatalf("unknown item type: %T", i)
+	}
+}
+
+func isCleanSlice(t *testing.T, slice []byte) {
+	t.Helper()
+
+	expSlice := make([]byte, len(slice))
+	if !bytes.Equal(expSlice, slice) {
+		t.Fatalf("slice not recycled, want: %v, got: %v",
+			expSlice, slice)
+	}
+}

pool/write_buffer.go

@@ -0,0 +1,48 @@
+package pool
+
+import (
+	"time"
+
+	"github.com/lightningnetwork/lnd/buffer"
+)
+
+const (
+	// DefaultWriteBufferGCInterval is the default interval that a Write
+	// will perform a sweep to see which expired buffer.Writes can be
+	// released to the runtime.
+	DefaultWriteBufferGCInterval = 15 * time.Second
+
+	// DefaultWriteBufferExpiryInterval is the default, minimum interval
+	// that must elapse before a Write will release a buffer.Write.  The
+	// maximum time before the buffer can be released is equal to the expiry
+	// interval plus the gc interval.
+	DefaultWriteBufferExpiryInterval = 30 * time.Second
+)
+
+// WriteBuffer is a pool of recycled buffer.Write items, that dynamically
+// allocates and reclaims buffers in response to load.
+type WriteBuffer struct {
+	pool *Recycle
+}
+
+// NewWriteBuffer returns a freshly instantiated WriteBuffer, using the given
+// gcInterval and expiryIntervals.
+func NewWriteBuffer(gcInterval, expiryInterval time.Duration) *WriteBuffer {
+	return &WriteBuffer{
+		pool: NewRecycle(
+			func() interface{} { return new(buffer.Write) },
+			100, gcInterval, expiryInterval,
+		),
+	}
+}
+
+// Take returns a fresh buffer.Write to the caller.
+func (p *WriteBuffer) Take() *buffer.Write {
+	return p.pool.Take().(*buffer.Write)
+}
+
+// Return returns the buffer.Write to the pool, so that it can be recycled or
+// released.
+func (p *WriteBuffer) Return(buf *buffer.Write) {
+	p.pool.Return(buf)
+}