client: AppendService owns its buffer sync.Pool

broker/client/append_service.go

@@ -37,6 +37,7 @@ type AppendService struct {
 	appends map[pb.Journal]*AsyncAppend // Index of the most-recent AsyncAppend.
 	errs    map[pb.Journal]error        // Index of terminal errors.
 	mu      sync.Mutex                  // Guards |appends| and |errs|.
+	pool    *sync.Pool                  // Pool of appendBuffers.
 }
 
 // NewAppendService returns an AppendService with the provided Context and BrokerClient.
@@ -46,6 +47,7 @@ func NewAppendService(ctx context.Context, client pb.RoutedJournalClient) *Appen
 		RoutedJournalClient: client,
 		appends:             make(map[pb.Journal]*AsyncAppend),
 		errs:                make(map[pb.Journal]error),
+		pool:                newAppendBufferPool(),
 	}
 }
 
@@ -207,7 +209,7 @@ func (s *AppendService) StartAppend(req pb.AppendRequest, dependencies OpFutures
 		// StartAppend. Initialize its appendBuffer, which also signals that this
 		// |aa| has been returned by StartAppend, and that a client may be
 		// waiting on its RPC response.
-		aa.fb = appendBufferPool.Get().(*appendBuffer)
+		aa.fb = s.pool.Get().(*appendBuffer)
 	}
 	return aa
 }
@@ -402,7 +404,7 @@ var serveAppends = func(s *AppendService, aa *AsyncAppend, err error) {
 				}, aa.app.Request.Journal, "failed to append to journal")
 			}
 
-			releaseFileBuffer(aa.fb)
+			aa.fb.releaseToPool()
 		}
 
 		aa.mu.Lock()
@@ -427,6 +429,7 @@ type appendBuffer struct {
 	}
 	offset int64
 	buf    *bufio.Writer
+	pool   *sync.Pool
 }
 
 func (fb *appendBuffer) Write(p []byte) (n int, err error) {
@@ -496,23 +499,28 @@ func (r returnNReader) Read(p []byte) (int, error) {
 	return r.n, io.EOF
 }
 
-func releaseFileBuffer(fb *appendBuffer) {
+func (fb *appendBuffer) releaseToPool() {
 	fb.offset = 0
 	retryUntil(fb.seek, "", "failed to seek appendBuffer")
-	appendBufferPool.Put(fb)
+	fb.pool.Put(fb)
 }
 
-var appendBufferPool = sync.Pool{
-	New: func() interface{} {
+func newAppendBufferPool() *sync.Pool {
+	var pool = new(sync.Pool)
+
+	pool.New = func() interface{} {
 		var fb *appendBuffer
 
 		retryUntil(func() (err error) {
 			fb, err = newAppendBuffer()
 			return
 		}, "", "failed to create appendBuffer")
 
+		fb.pool = pool
 		return fb
-	},
+	}
+
+	return pool
 }
 
 func retryUntil(fn func() error, journal pb.Journal, msg string) {

broker/client/append_service_test.go

@@ -253,7 +253,7 @@ func (s *AppendServiceSuite) TestAppendContextCancellation(c *gc.C) {
 func (s *AppendServiceSuite) TestFlushErrorHandlingCases(c *gc.C) {
 	var mf = mockFile{n: 6}
 
-	var fb = &appendBuffer{file: &mf}
+	var fb = &appendBuffer{file: &mf, pool: nil}
 	fb.buf = bufio.NewWriterSize(fb, 8)
 
 	// Case 1: flush succeeds.
@@ -361,7 +361,7 @@ func (s *AppendServiceSuite) TestReleaseChecksForWriteErrorAndRecovers(c *gc.C)
 	// Begin a write with a small buffer fixture which will fail.
 	var aa = as.StartAppend(pb.AppendRequest{Journal: "a/journal"}, nil)
 	var mf = mockFile{n: 2}
-	aa.fb = &appendBuffer{file: &mf}
+	aa.fb = &appendBuffer{file: &mf, pool: as.pool}
 	aa.fb.buf = bufio.NewWriterSize(aa.fb, 7)
 
 	var _, err = aa.Writer().WriteString("write spills to |mf| and fails")
@@ -486,15 +486,16 @@ func (s *AppendServiceSuite) TestDependencyError(c *gc.C) {
 }
 
 func (s *AppendServiceSuite) TestBufferPooling(c *gc.C) {
-	var ab = appendBufferPool.Get().(*appendBuffer)
+	var pool = newAppendBufferPool()
+	var ab = pool.Get().(*appendBuffer)
 
 	// Precondition: write some content.
 	_, _ = ab.buf.WriteString("foobar")
 	c.Check(ab.buf.Flush(), gc.IsNil)
 	c.Check(ab.offset, gc.Equals, int64(6))
 
-	releaseFileBuffer(ab)
-	ab = appendBufferPool.Get().(*appendBuffer)
+	ab.releaseToPool()
+	ab = pool.Get().(*appendBuffer)
 
 	// Post-condition: a released and re-fetched buffer is zeroed.
 	var n, err = ab.file.Seek(0, io.SeekCurrent)
@@ -505,7 +506,9 @@ func (s *AppendServiceSuite) TestBufferPooling(c *gc.C) {
 
 func (s *AppendServiceSuite) TestSubsetCases(c *gc.C) {
 	var mkAA = func(name pb.Journal) *AsyncAppend {
-		return &AsyncAppend{app: *NewAppender(nil, nil, pb.AppendRequest{Journal: name})}
+		return &AsyncAppend{
+			app: *NewAppender(context.Background(), nil, pb.AppendRequest{Journal: name}),
+		}
 	}
 	var set = func(l ...OpFuture) (out OpFutures) {
 		out = make(OpFutures)
@@ -608,12 +611,4 @@ func buildAppendResponseFixture(ep interface{ Endpoint() pb.Endpoint }) pb.Appen
 	}
 }
 
-type testOpFuture struct {
-	done chan struct{}
-	err  error
-}
-
-func (t testOpFuture) Done() <-chan struct{} { return t.done }
-func (t testOpFuture) Err() error            { return t.err }
-
 var _ = gc.Suite(&AppendServiceSuite{})