Merge pull request #2036 from stevvooe/use-buffer-pools

archive, cio, cmd, linux: use buffer pools

archive/tar.go

@@ -19,7 +19,7 @@ import (
 	"github.com/pkg/errors"
 )
 
-var bufferPool = &sync.Pool{
+var bufPool = &sync.Pool{
 	New: func() interface{} {
 		buffer := make([]byte, 32*1024)
 		return &buffer
@@ -527,9 +527,7 @@ func (cw *changeWriter) HandleChange(k fs.ChangeKind, p string, f os.FileInfo, e
 			}
 			defer file.Close()
 
-			buf := bufferPool.Get().(*[]byte)
+			n, err := copyBuffered(context.TODO(), cw.tw, file)
-			n, err := io.CopyBuffer(cw.tw, file, *buf)
-			bufferPool.Put(buf)
 			if err != nil {
 				return errors.Wrap(err, "failed to copy")
 			}
@@ -589,8 +587,8 @@ func (cw *changeWriter) includeParents(hdr *tar.Header) error {
 }
 
 func copyBuffered(ctx context.Context, dst io.Writer, src io.Reader) (written int64, err error) {
-	buf := bufferPool.Get().(*[]byte)
+	buf := bufPool.Get().(*[]byte)
-	defer bufferPool.Put(buf)
+	defer bufPool.Put(buf)
 
 	for {
 		select {

archive/tar_windows.go

@@ -386,6 +386,10 @@ func writeBackupStreamFromTarFile(w io.Writer, t *tar.Reader, hdr *tar.Header) (
 			return nil, err
 		}
 	}
+
+	buf := bufPool.Get().(*[]byte)
+	defer bufPool.Put(buf)
+
 	if hdr.Typeflag == tar.TypeReg || hdr.Typeflag == tar.TypeRegA {
 		bhdr := winio.BackupHeader{
 			Id:   winio.BackupData,
@@ -395,7 +399,7 @@ func writeBackupStreamFromTarFile(w io.Writer, t *tar.Reader, hdr *tar.Header) (
 		if err != nil {
 			return nil, err
 		}
-		_, err = io.Copy(bw, t)
+		_, err = io.CopyBuffer(bw, t, *buf)
 		if err != nil {
 			return nil, err
 		}
@@ -418,7 +422,7 @@ func writeBackupStreamFromTarFile(w io.Writer, t *tar.Reader, hdr *tar.Header) (
 		if err != nil {
 			return nil, err
 		}
-		_, err = io.Copy(bw, t)
+		_, err = io.CopyBuffer(bw, t, *buf)
 		if err != nil {
 			return nil, err
 		}

cio/io.go

@@ -10,6 +10,13 @@ import (
 	"github.com/containerd/containerd/defaults"
 )
 
+var bufPool = sync.Pool{
+	New: func() interface{} {
+		buffer := make([]byte, 32<<10)
+		return &buffer
+	},
+}
+
 // Config holds the IO configurations.
 type Config struct {
 	// Terminal is true if one has been allocated

cio/io_unix.go

@@ -47,7 +47,10 @@ func copyIO(fifos *FIFOSet, ioset *Streams) (*cio, error) {
 
 	if fifos.Stdin != "" {
 		go func() {
-			io.Copy(pipes.Stdin, ioset.Stdin)
+			p := bufPool.Get().(*[]byte)
+			defer bufPool.Put(p)
+
+			io.CopyBuffer(pipes.Stdin, ioset.Stdin, *p)
 			pipes.Stdin.Close()
 		}()
 	}
@@ -55,7 +58,10 @@ func copyIO(fifos *FIFOSet, ioset *Streams) (*cio, error) {
 	var wg = &sync.WaitGroup{}
 	wg.Add(1)
 	go func() {
-		io.Copy(ioset.Stdout, pipes.Stdout)
+		p := bufPool.Get().(*[]byte)
+		defer bufPool.Put(p)
+
+		io.CopyBuffer(ioset.Stdout, pipes.Stdout, *p)
 		pipes.Stdout.Close()
 		wg.Done()
 	}()
@@ -63,7 +69,10 @@ func copyIO(fifos *FIFOSet, ioset *Streams) (*cio, error) {
 	if !fifos.Terminal {
 		wg.Add(1)
 		go func() {
-			io.Copy(ioset.Stderr, pipes.Stderr)
+			p := bufPool.Get().(*[]byte)
+			defer bufPool.Put(p)
+
+			io.CopyBuffer(ioset.Stderr, pipes.Stderr, *p)
 			pipes.Stderr.Close()
 			wg.Done()
 		}()

cio/io_windows.go

@@ -47,7 +47,11 @@ func copyIO(fifos *FIFOSet, ioset *Streams) (*cio, error) {
 				log.L.WithError(err).Errorf("failed to accept stdin connection on %s", fifos.Stdin)
 				return
 			}
-			io.Copy(c, ioset.Stdin)
+
+			p := bufPool.Get().(*[]byte)
+			defer bufPool.Put(p)
+
+			io.CopyBuffer(c, ioset.Stdin, *p)
 			c.Close()
 			l.Close()
 		}()
@@ -73,7 +77,11 @@ func copyIO(fifos *FIFOSet, ioset *Streams) (*cio, error) {
 				log.L.WithError(err).Errorf("failed to accept stdout connection on %s", fifos.Stdout)
 				return
 			}
-			io.Copy(ioset.Stdout, c)
+
+			p := bufPool.Get().(*[]byte)
+			defer bufPool.Put(p)
+
+			io.CopyBuffer(ioset.Stdout, c, *p)
 			c.Close()
 			l.Close()
 		}()
@@ -99,7 +107,11 @@ func copyIO(fifos *FIFOSet, ioset *Streams) (*cio, error) {
 				log.L.WithError(err).Errorf("failed to accept stderr connection on %s", fifos.Stderr)
 				return
 			}
-			io.Copy(ioset.Stderr, c)
+
+			p := bufPool.Get().(*[]byte)
+			defer bufPool.Put(p)
+
+			io.CopyBuffer(ioset.Stderr, c, *p)
 			c.Close()
 			l.Close()
 		}()

cmd/containerd/main.go

@@ -10,10 +10,6 @@ import (
 	"os/signal"
 	"time"
 
-	"google.golang.org/grpc/grpclog"
-
-	gocontext "golang.org/x/net/context"
-
 	"github.com/containerd/containerd/log"
 	"github.com/containerd/containerd/server"
 	"github.com/containerd/containerd/sys"
@@ -21,6 +17,8 @@ import (
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 	"github.com/urfave/cli"
+	gocontext "golang.org/x/net/context"
+	"google.golang.org/grpc/grpclog"
 )
 
 const usage = `

cmd/containerd/main_unix.go

@@ -7,11 +7,10 @@ import (
 	"os"
 	"runtime"
 
-	"github.com/sirupsen/logrus"
-	"golang.org/x/sys/unix"
-
 	"github.com/containerd/containerd/log"
 	"github.com/containerd/containerd/server"
+	"github.com/sirupsen/logrus"
+	"golang.org/x/sys/unix"
 )
 
 const defaultConfigPath = "/etc/containerd/config.toml"

cmd/ctr/commands/shim/io_unix.go

@@ -12,6 +12,13 @@ import (
 	"golang.org/x/sys/unix"
 )
 
+var bufPool = sync.Pool{
+	New: func() interface{} {
+		buffer := make([]byte, 32<<10)
+		return &buffer
+	},
+}
+
 func prepareStdio(stdin, stdout, stderr string, console bool) (wg *sync.WaitGroup, err error) {
 	wg = &sync.WaitGroup{}
 	ctx := gocontext.Background()
@@ -26,7 +33,9 @@ func prepareStdio(stdin, stdout, stderr string, console bool) (wg *sync.WaitGrou
 		}
 	}(f)
 	go func(w io.WriteCloser) {
-		io.Copy(w, os.Stdin)
+		p := bufPool.Get().(*[]byte)
+		defer bufPool.Put(p)
+		io.CopyBuffer(w, os.Stdin, *p)
 		w.Close()
 	}(f)
 

linux/bundle.go

@@ -3,9 +3,8 @@
 package linux
 
 import (
-	"bytes"
 	"context"
-	"io"
+	"io/ioutil"
 	"os"
 	"path/filepath"
 
@@ -52,12 +51,7 @@ func newBundle(id, path, workDir string, spec []byte) (b *bundle, err error) {
 	if err := os.Mkdir(filepath.Join(path, "rootfs"), 0711); err != nil {
 		return nil, err
 	}
-	f, err := os.Create(filepath.Join(path, configFilename))
+	err = ioutil.WriteFile(filepath.Join(path, configFilename), spec, 0666)
-	if err != nil {
-		return nil, err
-	}
-	defer f.Close()
-	_, err = io.Copy(f, bytes.NewReader(spec))
 	return &bundle{
 		id:      id,
 		path:    path,

linux/proc/io.go

@@ -13,6 +13,13 @@ import (
 	runc "github.com/containerd/go-runc"
 )
 
+var bufPool = sync.Pool{
+	New: func() interface{} {
+		buffer := make([]byte, 32<<10)
+		return &buffer
+	},
+}
+
 func copyPipes(ctx context.Context, rio runc.IO, stdin, stdout, stderr string, wg, cwg *sync.WaitGroup) error {
 	for name, dest := range map[string]func(wc io.WriteCloser, rc io.Closer){
 		stdout: func(wc io.WriteCloser, rc io.Closer) {
@@ -20,7 +27,9 @@ func copyPipes(ctx context.Context, rio runc.IO, stdin, stdout, stderr string, w
 			cwg.Add(1)
 			go func() {
 				cwg.Done()
-				io.Copy(wc, rio.Stdout())
+				p := bufPool.Get().(*[]byte)
+				defer bufPool.Put(p)
+				io.CopyBuffer(wc, rio.Stdout(), *p)
 				wg.Done()
 				wc.Close()
 				rc.Close()
@@ -31,7 +40,10 @@ func copyPipes(ctx context.Context, rio runc.IO, stdin, stdout, stderr string, w
 			cwg.Add(1)
 			go func() {
 				cwg.Done()
-				io.Copy(wc, rio.Stderr())
+				p := bufPool.Get().(*[]byte)
+				defer bufPool.Put(p)
+
+				io.CopyBuffer(wc, rio.Stderr(), *p)
 				wg.Done()
 				wc.Close()
 				rc.Close()
@@ -59,7 +71,10 @@ func copyPipes(ctx context.Context, rio runc.IO, stdin, stdout, stderr string, w
 	cwg.Add(1)
 	go func() {
 		cwg.Done()
-		io.Copy(rio.Stdin(), f)
+		p := bufPool.Get().(*[]byte)
+		defer bufPool.Put(p)
+
+		io.CopyBuffer(rio.Stdin(), f, *p)
 		rio.Stdin().Close()
 		f.Close()
 	}()

linux/proc/utils.go

@@ -66,7 +66,10 @@ func copyFile(to, from string) error {
 		return err
 	}
 	defer tt.Close()
-	_, err = io.Copy(tt, ff)
+
+	p := bufPool.Get().(*[]byte)
+	defer bufPool.Put(p)
+	_, err = io.CopyBuffer(tt, ff, *p)
 	return err
 }
 

linux/shim/service.go

@@ -29,7 +29,15 @@ import (
 	"google.golang.org/grpc/status"
 )
 
-var empty = &ptypes.Empty{}
+var (
+	empty   = &ptypes.Empty{}
+	bufPool = sync.Pool{
+		New: func() interface{} {
+			buffer := make([]byte, 32<<10)
+			return &buffer
+		},
+	}
+)
 
 // Config contains shim specific configuration
 type Config struct {

linux/shim/service_linux.go

@@ -33,7 +33,9 @@ func (p *linuxPlatform) CopyConsole(ctx context.Context, console console.Console
 		cwg.Add(1)
 		go func() {
 			cwg.Done()
-			io.Copy(epollConsole, in)
+			p := bufPool.Get().(*[]byte)
+			defer bufPool.Put(p)
+			io.CopyBuffer(epollConsole, in, *p)
 		}()
 	}
 
@@ -49,7 +51,9 @@ func (p *linuxPlatform) CopyConsole(ctx context.Context, console console.Console
 	cwg.Add(1)
 	go func() {
 		cwg.Done()
-		io.Copy(outw, epollConsole)
+		p := bufPool.Get().(*[]byte)
+		defer bufPool.Put(p)
+		io.CopyBuffer(outw, epollConsole, *p)
 		epollConsole.Close()
 		outr.Close()
 		outw.Close()

linux/shim/service_unix.go

@@ -24,7 +24,10 @@ func (p *unixPlatform) CopyConsole(ctx context.Context, console console.Console,
 		cwg.Add(1)
 		go func() {
 			cwg.Done()
-			io.Copy(console, in)
+			p := bufPool.Get().(*[]byte)
+			defer bufPool.Put(p)
+
+			io.CopyBuffer(console, in, *p)
 		}()
 	}
 	outw, err := fifo.OpenFifo(ctx, stdout, syscall.O_WRONLY, 0)
@@ -39,7 +42,10 @@ func (p *unixPlatform) CopyConsole(ctx context.Context, console console.Console,
 	cwg.Add(1)
 	go func() {
 		cwg.Done()
-		io.Copy(outw, console)
+		p := bufPool.Get().(*[]byte)
+		defer bufPool.Put(p)
+
+		io.CopyBuffer(outw, console, *p)
 		console.Close()
 		outr.Close()
 		outw.Close()