Merge pull request #448 from kzys/buf-reset

Reset bytes.Buffer before returning it to sync.Pool

cache/cache.go

@@ -132,6 +132,7 @@ func NewDirectoryCache(directory string, config DirectoryCacheConfig) (BlobCache
 		}
 		dataCache = lrucache.New(maxEntry)
 		dataCache.OnEvicted = func(key string, value interface{}) {
+			value.(*bytes.Buffer).Reset()
 			bufPool.Put(value)
 		}
 	}
@@ -296,7 +297,6 @@ func (dc *directoryCache) Add(key string, opts ...Option) (Writer, error) {
 	}
 
 	b := dc.bufPool.Get().(*bytes.Buffer)
-	b.Reset()
 	memW := &writer{
 		WriteCloser: nopWriteCloser(io.Writer(b)),
 		commitFunc: func() error {
@@ -306,7 +306,7 @@ func (dc *directoryCache) Add(key string, opts ...Option) (Writer, error) {
 			}
 			cached, done, added := dc.cache.Add(key, b)
 			if !added {
-				dc.bufPool.Put(b) // already exists in the cache. abort it.
+				dc.putBuffer(b) // already exists in the cache. abort it.
 			}
 			commit := func() error {
 				defer done()
@@ -331,14 +331,19 @@ func (dc *directoryCache) Add(key string, opts ...Option) (Writer, error) {
 		abortFunc: func() error {
 			defer w.Close()
 			defer w.Abort()
-			dc.bufPool.Put(b) // abort it.
+			dc.putBuffer(b) // abort it.
 			return nil
 		},
 	}
 
 	return memW, nil
 }
 
+func (dc *directoryCache) putBuffer(b *bytes.Buffer) {
+	b.Reset()
+	dc.bufPool.Put(b)
+}
+
 func (dc *directoryCache) Close() error {
 	dc.closedMu.Lock()
 	defer dc.closedMu.Unlock()

fs/layer/layer.go

@@ -192,6 +192,7 @@ func newCache(root string, cacheType string, cfg config.Config) (cache.BlobCache
 	}
 	dCache, fCache := lrucache.New(maxDataEntry), lrucache.New(maxFdEntry)
 	dCache.OnEvicted = func(key string, value interface{}) {
+		value.(*bytes.Buffer).Reset()
 		bufPool.Put(value)
 	}
 	fCache.OnEvicted = func(key string, value interface{}) {

fs/reader/reader.go

@@ -335,6 +335,11 @@ func (gr *reader) cacheWithReader(ctx context.Context, currentDepth int, eg *err
 	return
 }
 
+func (gr *reader) putBuffer(b *bytes.Buffer) {
+	b.Reset()
+	gr.bufPool.Put(b)
+}
+
 type file struct {
 	name   string
 	digest string
@@ -406,11 +411,10 @@ func (sf *file) ReadAt(p []byte, offset int64) (int, error) {
 
 		// Use temporally buffer for aligning this chunk
 		b := sf.gr.bufPool.Get().(*bytes.Buffer)
-		b.Reset()
 		b.Grow(int(ce.ChunkSize))
 		ip := b.Bytes()[:ce.ChunkSize]
 		if _, err := sf.ra.ReadAt(ip, ce.ChunkOffset); err != nil && err != io.EOF {
-			sf.gr.bufPool.Put(b)
+			sf.gr.putBuffer(b)
 			return 0, errors.Wrap(err, "failed to read data")
 		}
 
@@ -421,7 +425,7 @@ func (sf *file) ReadAt(p []byte, offset int64) (int, error) {
 
 		// Verify this chunk
 		if err := sf.verify(ip, ce); err != nil {
-			sf.gr.bufPool.Put(b)
+			sf.gr.putBuffer(b)
 			return 0, errors.Wrap(err, "invalid chunk")
 		}
 
@@ -435,7 +439,7 @@ func (sf *file) ReadAt(p []byte, offset int64) (int, error) {
 			w.Close()
 		}
 		n := copy(p[nr:], ip[lowerDiscard:ce.ChunkSize-upperDiscard])
-		sf.gr.bufPool.Put(b)
+		sf.gr.putBuffer(b)
 		if int64(n) != expectedSize {
 			return 0, fmt.Errorf("unexpected final data size %d; want %d", n, expectedSize)
 		}