Merge pull request #1810 from dnephin/add-staticcheck-linter

Fix usage of sync.Pool

.gometalinter.json

@@ -13,6 +13,7 @@
     "structcheck",
     "unused",
     "varcheck",
+    "staticcheck",
 
     "gofmt",
     "goimports",

archive/tar.go

@@ -19,13 +19,12 @@ import (
 	"github.com/pkg/errors"
 )
 
-var (
-	bufferPool = &sync.Pool{
-		New: func() interface{} {
-			return make([]byte, 32*1024)
-		},
-	}
-)
+var bufferPool = &sync.Pool{
+	New: func() interface{} {
+		buffer := make([]byte, 32*1024)
+		return &buffer
+	},
+}
 
 // Diff returns a tar stream of the computed filesystem
 // difference between the provided directories.
@@ -404,8 +403,8 @@ func (cw *changeWriter) HandleChange(k fs.ChangeKind, p string, f os.FileInfo, e
 			}
 			defer file.Close()
 
-			buf := bufferPool.Get().([]byte)
-			n, err := io.CopyBuffer(cw.tw, file, buf)
+			buf := bufferPool.Get().(*[]byte)
+			n, err := io.CopyBuffer(cw.tw, file, *buf)
 			bufferPool.Put(buf)
 			if err != nil {
 				return errors.Wrap(err, "failed to copy")
@@ -529,7 +528,7 @@ func createTarFile(ctx context.Context, path, extractDir string, hdr *tar.Header
 }
 
 func copyBuffered(ctx context.Context, dst io.Writer, src io.Reader) (written int64, err error) {
-	buf := bufferPool.Get().([]byte)
+	buf := bufferPool.Get().(*[]byte)
 	defer bufferPool.Put(buf)
 
 	for {
@@ -540,9 +539,9 @@ func copyBuffered(ctx context.Context, dst io.Writer, src io.Reader) (written in
 		default:
 		}
 
-		nr, er := src.Read(buf)
+		nr, er := src.Read(*buf)
 		if nr > 0 {
-			nw, ew := dst.Write(buf[0:nr])
+			nw, ew := dst.Write((*buf)[0:nr])
 			if nw > 0 {
 				written += int64(nw)
 			}

content/helpers.go

@@ -10,13 +10,12 @@ import (
 	"github.com/pkg/errors"
 )
 
-var (
-	bufPool = sync.Pool{
-		New: func() interface{} {
-			return make([]byte, 1<<20)
-		},
-	}
-)
+var bufPool = sync.Pool{
+	New: func() interface{} {
+		buffer := make([]byte, 1<<20)
+		return &buffer
+	},
+}
 
 // NewReader returns a io.Reader from a ReaderAt
 func NewReader(ra ReaderAt) io.Reader {
@@ -88,10 +87,10 @@ func Copy(ctx context.Context, cw Writer, r io.Reader, size int64, expected dige
 		}
 	}
 
-	buf := bufPool.Get().([]byte)
+	buf := bufPool.Get().(*[]byte)
 	defer bufPool.Put(buf)
 
-	if _, err := io.CopyBuffer(cw, r, buf); err != nil {
+	if _, err := io.CopyBuffer(cw, r, *buf); err != nil {
 		return err
 	}
 

content/helpers_test.go

@@ -0,0 +1,112 @@
+package content
+
+import (
+	"bytes"
+	"context"
+	"io"
+	"strings"
+	"testing"
+
+	"github.com/containerd/containerd/errdefs"
+	"github.com/opencontainers/go-digest"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+type copySource struct {
+	reader io.Reader
+	size   int64
+	digest digest.Digest
+}
+
+func TestCopy(t *testing.T) {
+	defaultSource := newCopySource("this is the source to copy")
+
+	var testcases = []struct {
+		name     string
+		source   copySource
+		writer   fakeWriter
+		expected string
+	}{
+		{
+			name:     "copy no offset",
+			source:   defaultSource,
+			writer:   fakeWriter{},
+			expected: "this is the source to copy",
+		},
+		{
+			name:     "copy with offset from seeker",
+			source:   defaultSource,
+			writer:   fakeWriter{status: Status{Offset: 8}},
+			expected: "the source to copy",
+		},
+		{
+			name:     "copy with offset from unseekable source",
+			source:   copySource{reader: bytes.NewBufferString("foo"), size: 3},
+			writer:   fakeWriter{status: Status{Offset: 8}},
+			expected: "foo",
+		},
+		{
+			name:   "commit already exists",
+			source: defaultSource,
+			writer: fakeWriter{commitFunc: func() error {
+				return errdefs.ErrAlreadyExists
+			}},
+		},
+	}
+
+	for _, testcase := range testcases {
+		t.Run(testcase.name, func(t *testing.T) {
+			err := Copy(context.Background(),
+				&testcase.writer,
+				testcase.source.reader,
+				testcase.source.size,
+				testcase.source.digest)
+
+			require.NoError(t, err)
+			assert.Equal(t, testcase.source.digest, testcase.writer.commitedDigest)
+			assert.Equal(t, testcase.expected, testcase.writer.String())
+		})
+	}
+}
+
+func newCopySource(raw string) copySource {
+	return copySource{
+		reader: strings.NewReader(raw),
+		size:   int64(len(raw)),
+		digest: digest.FromBytes([]byte(raw)),
+	}
+}
+
+type fakeWriter struct {
+	bytes.Buffer
+	commitedDigest digest.Digest
+	status         Status
+	commitFunc     func() error
+}
+
+func (f *fakeWriter) Close() error {
+	f.Buffer.Reset()
+	return nil
+}
+
+func (f *fakeWriter) Commit(ctx context.Context, size int64, expected digest.Digest, opts ...Opt) error {
+	f.commitedDigest = expected
+	if f.commitFunc == nil {
+		return nil
+	}
+	return f.commitFunc()
+}
+
+func (f *fakeWriter) Digest() digest.Digest {
+	return f.commitedDigest
+}
+
+func (f *fakeWriter) Status() (Status, error) {
+	return f.status, nil
+}
+
+func (f *fakeWriter) Truncate(size int64) error {
+	f.Buffer.Truncate(int(size))
+	return nil
+}

content/local/store.go

@@ -21,13 +21,12 @@ import (
 	"github.com/pkg/errors"
 )
 
-var (
-	bufPool = sync.Pool{
-		New: func() interface{} {
-			return make([]byte, 1<<20)
-		},
-	}
-)
+var bufPool = sync.Pool{
+	New: func() interface{} {
+		buffer := make([]byte, 1<<20)
+		return &buffer
+	},
+}
 
 // LabelStore is used to store mutable labels for digests
 type LabelStore interface {
@@ -463,10 +462,10 @@ func (s *store) writer(ctx context.Context, ref string, total int64, expected di
 		}
 		defer fp.Close()
 
-		p := bufPool.Get().([]byte)
+		p := bufPool.Get().(*[]byte)
 		defer bufPool.Put(p)
 
-		offset, err = io.CopyBuffer(digester.Hash(), fp, p)
+		offset, err = io.CopyBuffer(digester.Hash(), fp, *p)
 		if err != nil {
 			return nil, err
 		}

events/exchange/exchange_test.go

@@ -2,7 +2,6 @@ package exchange
 
 import (
 	"context"
-	"fmt"
 	"reflect"
 	"sync"
 	"testing"
@@ -39,13 +38,14 @@ func TestExchangeBasic(t *testing.T) {
 	t.Log("publish")
 	var wg sync.WaitGroup
 	wg.Add(1)
+	errChan := make(chan error)
 	go func() {
 		defer wg.Done()
+		defer close(errChan)
 		for _, event := range testevents {
-			fmt.Println("publish", event)
 			if err := exchange.Publish(ctx, "/test", event); err != nil {
-				fmt.Println("publish error", err)
-				t.Fatal(err)
+				errChan <- err
+				return
 			}
 		}
 
@@ -54,6 +54,9 @@ func TestExchangeBasic(t *testing.T) {
 
 	t.Log("waiting")
 	wg.Wait()
+	if err := <-errChan; err != nil {
+		t.Fatal(err)
+	}
 
 	for _, subscriber := range []struct {
 		eventq <-chan *events.Envelope

fs/copy.go

@@ -9,13 +9,12 @@ import (
 	"github.com/pkg/errors"
 )
 
-var (
-	bufferPool = &sync.Pool{
-		New: func() interface{} {
-			return make([]byte, 32*1024)
-		},
-	}
-)
+var bufferPool = &sync.Pool{
+	New: func() interface{} {
+		buffer := make([]byte, 32*1024)
+		return &buffer
+	},
+}
 
 // CopyDir copies the directory from src to dst.
 // Most efficient copy of files is attempted.

fs/copy_linux.go

@@ -43,8 +43,8 @@ func copyFileContent(dst, src *os.File) error {
 			return errors.Wrap(err, "copy file range failed")
 		}
 
-		buf := bufferPool.Get().([]byte)
-		_, err = io.CopyBuffer(dst, src, buf)
+		buf := bufferPool.Get().(*[]byte)
+		_, err = io.CopyBuffer(dst, src, *buf)
 		bufferPool.Put(buf)
 		return err
 	}

fs/copy_unix.go

@@ -34,8 +34,8 @@ func copyFileInfo(fi os.FileInfo, name string) error {
 }
 
 func copyFileContent(dst, src *os.File) error {
-	buf := bufferPool.Get().([]byte)
-	_, err := io.CopyBuffer(dst, src, buf)
+	buf := bufferPool.Get().(*[]byte)
+	_, err := io.CopyBuffer(dst, src, *buf)
 	bufferPool.Put(buf)
 
 	return err

fs/copy_windows.go

@@ -18,8 +18,8 @@ func copyFileInfo(fi os.FileInfo, name string) error {
 }
 
 func copyFileContent(dst, src *os.File) error {
-	buf := bufferPool.Get().([]byte)
-	_, err := io.CopyBuffer(dst, src, buf)
+	buf := bufferPool.Get().(*[]byte)
+	_, err := io.CopyBuffer(dst, src, *buf)
 	bufferPool.Put(buf)
 	return err
 }

gc/scheduler/scheduler.go

@@ -250,7 +250,6 @@ func (s *gcScheduler) run(ctx context.Context) {
 				schedC, nextCollection = schedule(interval)
 				continue
 			}
-			break
 		case e := <-s.eventC:
 			if lastCollection != nil && lastCollection.After(e.ts) {
 				continue

rootfs/init.go

@@ -69,12 +69,12 @@ func createInitLayer(ctx context.Context, parent, initName string, initFn func(s
 	if err != nil {
 		return "", err
 	}
+
 	defer func() {
 		if err != nil {
-			// TODO: once implemented uncomment
-			//if rerr := snapshotter.Remove(ctx, td); rerr != nil {
-			//	log.G(ctx).Errorf("Failed to remove snapshot %s: %v", td, merr)
-			//}
+			if rerr := snapshotter.Remove(ctx, td); rerr != nil {
+				log.G(ctx).Errorf("Failed to remove snapshot %s: %v", td, rerr)
+			}
 		}
 	}()
 

services/content/service.go

@@ -29,7 +29,8 @@ type service struct {
 
 var bufPool = sync.Pool{
 	New: func() interface{} {
-		return make([]byte, 1<<20)
+		buffer := make([]byte, 1<<20)
+		return &buffer
 	},
 }
 
@@ -179,7 +180,7 @@ func (s *service) Read(req *api.ReadContentRequest, session api.Content_ReadServ
 
 		// TODO(stevvooe): Using the global buffer pool. At 32KB, it is probably
 		// little inefficient for work over a fast network. We can tune this later.
-		p = bufPool.Get().([]byte)
+		p = bufPool.Get().(*[]byte)
 	)
 	defer bufPool.Put(p)
 
@@ -195,13 +196,10 @@ func (s *service) Read(req *api.ReadContentRequest, session api.Content_ReadServ
 		return status.Errorf(codes.OutOfRange, "read past object length %v bytes", oi.Size)
 	}
 
-	if _, err := io.CopyBuffer(
+	_, err = io.CopyBuffer(
 		&readResponseWriter{session: session},
-		io.NewSectionReader(ra, offset, size), p); err != nil {
-		return err
-	}
-
-	return nil
+		io.NewSectionReader(ra, offset, size), *p)
+	return err
 }
 
 // readResponseWriter is a writer that places the output into ReadContentRequest messages.