Vendor in latest containers/storage

Fixes store.lock issue comming up when using store.Names

Signed-off-by: umohnani8 <umohnani@redhat.com>

vendor/github.com/containers/storage/pkg/ioutils/bytespipe.go

@@ -0,0 +1,186 @@
+package ioutils
+
+import (
+	"errors"
+	"io"
+	"sync"
+)
+
+// maxCap is the highest capacity to use in byte slices that buffer data.
+const maxCap = 1e6
+
+// minCap is the lowest capacity to use in byte slices that buffer data
+const minCap = 64
+
+// blockThreshold is the minimum number of bytes in the buffer which will cause
+// a write to BytesPipe to block when allocating a new slice.
+const blockThreshold = 1e6
+
+var (
+	// ErrClosed is returned when Write is called on a closed BytesPipe.
+	ErrClosed = errors.New("write to closed BytesPipe")
+
+	bufPools     = make(map[int]*sync.Pool)
+	bufPoolsLock sync.Mutex
+)
+
+// BytesPipe is io.ReadWriteCloser which works similarly to pipe(queue).
+// All written data may be read at most once. Also, BytesPipe allocates
+// and releases new byte slices to adjust to current needs, so the buffer
+// won't be overgrown after peak loads.
+type BytesPipe struct {
+	mu       sync.Mutex
+	wait     *sync.Cond
+	buf      []*fixedBuffer
+	bufLen   int
+	closeErr error // error to return from next Read. set to nil if not closed.
+}
+
+// NewBytesPipe creates new BytesPipe, initialized by specified slice.
+// If buf is nil, then it will be initialized with slice which cap is 64.
+// buf will be adjusted in a way that len(buf) == 0, cap(buf) == cap(buf).
+func NewBytesPipe() *BytesPipe {
+	bp := &BytesPipe{}
+	bp.buf = append(bp.buf, getBuffer(minCap))
+	bp.wait = sync.NewCond(&bp.mu)
+	return bp
+}
+
+// Write writes p to BytesPipe.
+// It can allocate new []byte slices in a process of writing.
+func (bp *BytesPipe) Write(p []byte) (int, error) {
+	bp.mu.Lock()
+
+	written := 0
+loop0:
+	for {
+		if bp.closeErr != nil {
+			bp.mu.Unlock()
+			return written, ErrClosed
+		}
+
+		if len(bp.buf) == 0 {
+			bp.buf = append(bp.buf, getBuffer(64))
+		}
+		// get the last buffer
+		b := bp.buf[len(bp.buf)-1]
+
+		n, err := b.Write(p)
+		written += n
+		bp.bufLen += n
+
+		// errBufferFull is an error we expect to get if the buffer is full
+		if err != nil && err != errBufferFull {
+			bp.wait.Broadcast()
+			bp.mu.Unlock()
+			return written, err
+		}
+
+		// if there was enough room to write all then break
+		if len(p) == n {
+			break
+		}
+
+		// more data: write to the next slice
+		p = p[n:]
+
+		// make sure the buffer doesn't grow too big from this write
+		for bp.bufLen >= blockThreshold {
+			bp.wait.Wait()
+			if bp.closeErr != nil {
+				continue loop0
+			}
+		}
+
+		// add new byte slice to the buffers slice and continue writing
+		nextCap := b.Cap() * 2
+		if nextCap > maxCap {
+			nextCap = maxCap
+		}
+		bp.buf = append(bp.buf, getBuffer(nextCap))
+	}
+	bp.wait.Broadcast()
+	bp.mu.Unlock()
+	return written, nil
+}
+
+// CloseWithError causes further reads from a BytesPipe to return immediately.
+func (bp *BytesPipe) CloseWithError(err error) error {
+	bp.mu.Lock()
+	if err != nil {
+		bp.closeErr = err
+	} else {
+		bp.closeErr = io.EOF
+	}
+	bp.wait.Broadcast()
+	bp.mu.Unlock()
+	return nil
+}
+
+// Close causes further reads from a BytesPipe to return immediately.
+func (bp *BytesPipe) Close() error {
+	return bp.CloseWithError(nil)
+}
+
+// Read reads bytes from BytesPipe.
+// Data could be read only once.
+func (bp *BytesPipe) Read(p []byte) (n int, err error) {
+	bp.mu.Lock()
+	if bp.bufLen == 0 {
+		if bp.closeErr != nil {
+			bp.mu.Unlock()
+			return 0, bp.closeErr
+		}
+		bp.wait.Wait()
+		if bp.bufLen == 0 && bp.closeErr != nil {
+			err := bp.closeErr
+			bp.mu.Unlock()
+			return 0, err
+		}
+	}
+
+	for bp.bufLen > 0 {
+		b := bp.buf[0]
+		read, _ := b.Read(p) // ignore error since fixedBuffer doesn't really return an error
+		n += read
+		bp.bufLen -= read
+
+		if b.Len() == 0 {
+			// it's empty so return it to the pool and move to the next one
+			returnBuffer(b)
+			bp.buf[0] = nil
+			bp.buf = bp.buf[1:]
+		}
+
+		if len(p) == read {
+			break
+		}
+
+		p = p[read:]
+	}
+
+	bp.wait.Broadcast()
+	bp.mu.Unlock()
+	return
+}
+
+func returnBuffer(b *fixedBuffer) {
+	b.Reset()
+	bufPoolsLock.Lock()
+	pool := bufPools[b.Cap()]
+	bufPoolsLock.Unlock()
+	if pool != nil {
+		pool.Put(b)
+	}
+}
+
+func getBuffer(size int) *fixedBuffer {
+	bufPoolsLock.Lock()
+	pool, ok := bufPools[size]
+	if !ok {
+		pool = &sync.Pool{New: func() interface{} { return &fixedBuffer{buf: make([]byte, 0, size)} }}
+		bufPools[size] = pool
+	}
+	bufPoolsLock.Unlock()
+	return pool.Get().(*fixedBuffer)
+}

vendor/github.com/containers/storage/pkg/ioutils/fmt.go

@@ -1,22 +0,0 @@
-package ioutils
-
-import (
-	"fmt"
-	"io"
-)
-
-// FprintfIfNotEmpty prints the string value if it's not empty
-func FprintfIfNotEmpty(w io.Writer, format, value string) (int, error) {
-	if value != "" {
-		return fmt.Fprintf(w, format, value)
-	}
-	return 0, nil
-}
-
-// FprintfIfTrue prints the boolean value if it's true
-func FprintfIfTrue(w io.Writer, format string, ok bool) (int, error) {
-	if ok {
-		return fmt.Fprintf(w, format, ok)
-	}
-	return 0, nil
-}

vendor/github.com/containers/storage/pkg/ioutils/fswriters.go

@@ -80,3 +80,83 @@ func (w *atomicFileWriter) Close() (retErr error) {
 	}
 	return nil
 }
+
+// AtomicWriteSet is used to atomically write a set
+// of files and ensure they are visible at the same time.
+// Must be committed to a new directory.
+type AtomicWriteSet struct {
+	root string
+}
+
+// NewAtomicWriteSet creates a new atomic write set to
+// atomically create a set of files. The given directory
+// is used as the base directory for storing files before
+// commit. If no temporary directory is given the system
+// default is used.
+func NewAtomicWriteSet(tmpDir string) (*AtomicWriteSet, error) {
+	td, err := ioutil.TempDir(tmpDir, "write-set-")
+	if err != nil {
+		return nil, err
+	}
+
+	return &AtomicWriteSet{
+		root: td,
+	}, nil
+}
+
+// WriteFile writes a file to the set, guaranteeing the file
+// has been synced.
+func (ws *AtomicWriteSet) WriteFile(filename string, data []byte, perm os.FileMode) error {
+	f, err := ws.FileWriter(filename, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, perm)
+	if err != nil {
+		return err
+	}
+	n, err := f.Write(data)
+	if err == nil && n < len(data) {
+		err = io.ErrShortWrite
+	}
+	if err1 := f.Close(); err == nil {
+		err = err1
+	}
+	return err
+}
+
+type syncFileCloser struct {
+	*os.File
+}
+
+func (w syncFileCloser) Close() error {
+	err := w.File.Sync()
+	if err1 := w.File.Close(); err == nil {
+		err = err1
+	}
+	return err
+}
+
+// FileWriter opens a file writer inside the set. The file
+// should be synced and closed before calling commit.
+func (ws *AtomicWriteSet) FileWriter(name string, flag int, perm os.FileMode) (io.WriteCloser, error) {
+	f, err := os.OpenFile(filepath.Join(ws.root, name), flag, perm)
+	if err != nil {
+		return nil, err
+	}
+	return syncFileCloser{f}, nil
+}
+
+// Cancel cancels the set and removes all temporary data
+// created in the set.
+func (ws *AtomicWriteSet) Cancel() error {
+	return os.RemoveAll(ws.root)
+}
+
+// Commit moves all created files to the target directory. The
+// target directory must not exist and the parent of the target
+// directory must exist.
+func (ws *AtomicWriteSet) Commit(target string) error {
+	return os.Rename(ws.root, target)
+}
+
+// String returns the location the set is writing to.
+func (ws *AtomicWriteSet) String() string {
+	return ws.root
+}

vendor/github.com/containers/storage/pkg/ioutils/multireader.go

@@ -1,226 +0,0 @@
-package ioutils
-
-import (
-	"bytes"
-	"fmt"
-	"io"
-	"os"
-)
-
-type pos struct {
-	idx    int
-	offset int64
-}
-
-type multiReadSeeker struct {
-	readers []io.ReadSeeker
-	pos     *pos
-	posIdx  map[io.ReadSeeker]int
-}
-
-func (r *multiReadSeeker) Seek(offset int64, whence int) (int64, error) {
-	var tmpOffset int64
-	switch whence {
-	case os.SEEK_SET:
-		for i, rdr := range r.readers {
-			// get size of the current reader
-			s, err := rdr.Seek(0, os.SEEK_END)
-			if err != nil {
-				return -1, err
-			}
-
-			if offset > tmpOffset+s {
-				if i == len(r.readers)-1 {
-					rdrOffset := s + (offset - tmpOffset)
-					if _, err := rdr.Seek(rdrOffset, os.SEEK_SET); err != nil {
-						return -1, err
-					}
-					r.pos = &pos{i, rdrOffset}
-					return offset, nil
-				}
-
-				tmpOffset += s
-				continue
-			}
-
-			rdrOffset := offset - tmpOffset
-			idx := i
-
-			rdr.Seek(rdrOffset, os.SEEK_SET)
-			// make sure all following readers are at 0
-			for _, rdr := range r.readers[i+1:] {
-				rdr.Seek(0, os.SEEK_SET)
-			}
-
-			if rdrOffset == s && i != len(r.readers)-1 {
-				idx++
-				rdrOffset = 0
-			}
-			r.pos = &pos{idx, rdrOffset}
-			return offset, nil
-		}
-	case os.SEEK_END:
-		for _, rdr := range r.readers {
-			s, err := rdr.Seek(0, os.SEEK_END)
-			if err != nil {
-				return -1, err
-			}
-			tmpOffset += s
-		}
-		r.Seek(tmpOffset+offset, os.SEEK_SET)
-		return tmpOffset + offset, nil
-	case os.SEEK_CUR:
-		if r.pos == nil {
-			return r.Seek(offset, os.SEEK_SET)
-		}
-		// Just return the current offset
-		if offset == 0 {
-			return r.getCurOffset()
-		}
-
-		curOffset, err := r.getCurOffset()
-		if err != nil {
-			return -1, err
-		}
-		rdr, rdrOffset, err := r.getReaderForOffset(curOffset + offset)
-		if err != nil {
-			return -1, err
-		}
-
-		r.pos = &pos{r.posIdx[rdr], rdrOffset}
-		return curOffset + offset, nil
-	default:
-		return -1, fmt.Errorf("Invalid whence: %d", whence)
-	}
-
-	return -1, fmt.Errorf("Error seeking for whence: %d, offset: %d", whence, offset)
-}
-
-func (r *multiReadSeeker) getReaderForOffset(offset int64) (io.ReadSeeker, int64, error) {
-	var rdr io.ReadSeeker
-	var rdrOffset int64
-
-	for i, rdr := range r.readers {
-		offsetTo, err := r.getOffsetToReader(rdr)
-		if err != nil {
-			return nil, -1, err
-		}
-		if offsetTo > offset {
-			rdr = r.readers[i-1]
-			rdrOffset = offsetTo - offset
-			break
-		}
-
-		if rdr == r.readers[len(r.readers)-1] {
-			rdrOffset = offsetTo + offset
-			break
-		}
-	}
-
-	return rdr, rdrOffset, nil
-}
-
-func (r *multiReadSeeker) getCurOffset() (int64, error) {
-	var totalSize int64
-	for _, rdr := range r.readers[:r.pos.idx+1] {
-		if r.posIdx[rdr] == r.pos.idx {
-			totalSize += r.pos.offset
-			break
-		}
-
-		size, err := getReadSeekerSize(rdr)
-		if err != nil {
-			return -1, fmt.Errorf("error getting seeker size: %v", err)
-		}
-		totalSize += size
-	}
-	return totalSize, nil
-}
-
-func (r *multiReadSeeker) getOffsetToReader(rdr io.ReadSeeker) (int64, error) {
-	var offset int64
-	for _, r := range r.readers {
-		if r == rdr {
-			break
-		}
-
-		size, err := getReadSeekerSize(rdr)
-		if err != nil {
-			return -1, err
-		}
-		offset += size
-	}
-	return offset, nil
-}
-
-func (r *multiReadSeeker) Read(b []byte) (int, error) {
-	if r.pos == nil {
-		r.pos = &pos{0, 0}
-	}
-
-	bCap := int64(cap(b))
-	buf := bytes.NewBuffer(nil)
-	var rdr io.ReadSeeker
-
-	for _, rdr = range r.readers[r.pos.idx:] {
-		readBytes, err := io.CopyN(buf, rdr, bCap)
-		if err != nil && err != io.EOF {
-			return -1, err
-		}
-		bCap -= readBytes
-
-		if bCap == 0 {
-			break
-		}
-	}
-
-	rdrPos, err := rdr.Seek(0, os.SEEK_CUR)
-	if err != nil {
-		return -1, err
-	}
-	r.pos = &pos{r.posIdx[rdr], rdrPos}
-	return buf.Read(b)
-}
-
-func getReadSeekerSize(rdr io.ReadSeeker) (int64, error) {
-	// save the current position
-	pos, err := rdr.Seek(0, os.SEEK_CUR)
-	if err != nil {
-		return -1, err
-	}
-
-	// get the size
-	size, err := rdr.Seek(0, os.SEEK_END)
-	if err != nil {
-		return -1, err
-	}
-
-	// reset the position
-	if _, err := rdr.Seek(pos, os.SEEK_SET); err != nil {
-		return -1, err
-	}
-	return size, nil
-}
-
-// MultiReadSeeker returns a ReadSeeker that's the logical concatenation of the provided
-// input readseekers. After calling this method the initial position is set to the
-// beginning of the first ReadSeeker. At the end of a ReadSeeker, Read always advances
-// to the beginning of the next ReadSeeker and returns EOF at the end of the last ReadSeeker.
-// Seek can be used over the sum of lengths of all readseekers.
-//
-// When a MultiReadSeeker is used, no Read and Seek operations should be made on
-// its ReadSeeker components. Also, users should make no assumption on the state
-// of individual readseekers while the MultiReadSeeker is used.
-func MultiReadSeeker(readers ...io.ReadSeeker) io.ReadSeeker {
-	if len(readers) == 1 {
-		return readers[0]
-	}
-	idx := make(map[io.ReadSeeker]int)
-	for i, rdr := range readers {
-		idx[rdr] = i
-	}
-	return &multiReadSeeker{
-		readers: readers,
-		posIdx:  idx,
-	}
-}

vendor/github.com/containers/storage/pkg/ioutils/readers.go

@@ -4,6 +4,8 @@ import (
 	"crypto/sha256"
 	"encoding/hex"
 	"io"
+
+	"golang.org/x/net/context"
 )
 
 type readCloserWrapper struct {
@@ -81,3 +83,72 @@ func (r *OnEOFReader) runFunc() {
 		r.Fn = nil
 	}
 }
+
+// cancelReadCloser wraps an io.ReadCloser with a context for cancelling read
+// operations.
+type cancelReadCloser struct {
+	cancel func()
+	pR     *io.PipeReader // Stream to read from
+	pW     *io.PipeWriter
+}
+
+// NewCancelReadCloser creates a wrapper that closes the ReadCloser when the
+// context is cancelled. The returned io.ReadCloser must be closed when it is
+// no longer needed.
+func NewCancelReadCloser(ctx context.Context, in io.ReadCloser) io.ReadCloser {
+	pR, pW := io.Pipe()
+
+	// Create a context used to signal when the pipe is closed
+	doneCtx, cancel := context.WithCancel(context.Background())
+
+	p := &cancelReadCloser{
+		cancel: cancel,
+		pR:     pR,
+		pW:     pW,
+	}
+
+	go func() {
+		_, err := io.Copy(pW, in)
+		select {
+		case <-ctx.Done():
+			// If the context was closed, p.closeWithError
+			// was already called. Calling it again would
+			// change the error that Read returns.
+		default:
+			p.closeWithError(err)
+		}
+		in.Close()
+	}()
+	go func() {
+		for {
+			select {
+			case <-ctx.Done():
+				p.closeWithError(ctx.Err())
+			case <-doneCtx.Done():
+				return
+			}
+		}
+	}()
+
+	return p
+}
+
+// Read wraps the Read method of the pipe that provides data from the wrapped
+// ReadCloser.
+func (p *cancelReadCloser) Read(buf []byte) (n int, err error) {
+	return p.pR.Read(buf)
+}
+
+// closeWithError closes the wrapper and its underlying reader. It will
+// cause future calls to Read to return err.
+func (p *cancelReadCloser) closeWithError(err error) {
+	p.pW.CloseWithError(err)
+	p.cancel()
+}
+
+// Close closes the wrapper its underlying reader. It will cause
+// future calls to Read to return io.EOF.
+func (p *cancelReadCloser) Close() error {
+	p.closeWithError(io.EOF)
+	return nil
+}