api: use simple streaming upload for GetObject() reader.

This change is needed to avoid the offset based ReadAt()
calls which can occur if using io.SectionReader().

Optimizing this part re-implementing a sectionReader()
based on Seeker() wasn't worth the complexity.

This PR also fixes the bug where the streaming signature
was interpreting the Read() error return in a wrong manner.

Author: harshavardhana

api-put-object-streaming.go

@@ -49,7 +49,7 @@ func (c Client) putObjectMultipartStream(bucketName, objectName string,
 	// while it is going to be duck typed similar to io.ReaderAt.
 	// It is to indicate that *minio.Object implements io.ReaderAt.
 	// and such a functionality is used in the subsequent code path.
-	if isObject(reader) || isReadAt(reader) || isFile(reader) {
+	if isFile(reader) || !isObject(reader) && isReadAt(reader) {
 		n, err = c.putObjectMultipartStreamFromReadAt(bucketName, objectName, reader.(io.ReaderAt), size, metadata, progress)
 	} else {
 		n, err = c.putObjectMultipartStreamNoChecksum(bucketName, objectName, reader, size, metadata, progress)
@@ -270,14 +270,8 @@ func (c Client) putObjectMultipartStreamNoChecksum(bucketName, objectName string
 
 		var objPart ObjectPart
 		objPart, err = c.uploadPart(bucketName, objectName, uploadID,
-			io.LimitReader(hookReader, partSize), partNumber,
-			nil, nil, partSize, metadata)
-		// For unknown size, Read EOF we break away.
-		// We do not have to upload till totalPartsCount.
-		if err == io.EOF && size < 0 {
-			break
-		}
-
+			io.LimitReader(hookReader, partSize),
+			partNumber, nil, nil, partSize, metadata)
 		if err != nil {
 			return totalUploadedSize, err
 		}
@@ -340,9 +334,8 @@ func (c Client) putObjectNoChecksum(bucketName, objectName string, reader io.Rea
 		return 0, ErrEntityTooSmall(size, bucketName, objectName)
 	}
 	if size > 0 {
-		readerAt, ok := reader.(io.ReaderAt)
-		if ok {
-			reader = io.NewSectionReader(readerAt, 0, size)
+		if isReadAt(reader) && !isObject(reader) {
+			reader = io.NewSectionReader(reader.(io.ReaderAt), 0, size)
 		}
 	}
 

pkg/s3signer/request-signature-streaming.go

@@ -254,7 +254,18 @@ func (s *StreamingReader) Read(buf []byte) (int, error) {
 		s.chunkBufLen = 0
 		for {
 			n1, err := s.baseReadCloser.Read(s.chunkBuf[s.chunkBufLen:])
-			if err == nil || err == io.ErrUnexpectedEOF {
+			// Usually we validate `err` first, but in this case
+			// we are validating n > 0 for the following reasons.
+			//
+			// 1. n > 0, err is one of io.EOF, nil (near end of stream)
+			// A Reader returning a non-zero number of bytes at the end
+			// of the input stream may return either err == EOF or err == nil
+			//
+			// 2. n == 0, err is io.EOF (actual end of stream)
+			//
+			// Callers should always process the n > 0 bytes returned
+			// before considering the error err.
+			if n1 > 0 {
 				s.chunkBufLen += n1
 				s.bytesRead += int64(n1)
 
@@ -265,25 +276,26 @@ func (s *StreamingReader) Read(buf []byte) (int, error) {
 					s.signChunk(s.chunkBufLen)
 					break
 				}
+			}
+			if err != nil {
+				if err == io.EOF {
+					// No more data left in baseReader - last chunk.
+					// Done reading the last chunk from baseReader.
+					s.done = true
+
+					// bytes read from baseReader different than
+					// content length provided.
+					if s.bytesRead != s.contentLen {
+						return 0, fmt.Errorf("Expected %d, got %d", s.bytesRead, s.contentLen)
+					}
 
-			} else if err == io.EOF {
-				// No more data left in baseReader - last chunk.
-				// Done reading the last chunk from baseReader.
-				s.done = true
-
-				// bytes read from baseReader different than
-				// content length provided.
-				if s.bytesRead != s.contentLen {
-					return 0, io.ErrUnexpectedEOF
+					// Sign the chunk and write it to s.buf.
+					s.signChunk(0)
+					break
 				}
-
-				// Sign the chunk and write it to s.buf.
-				s.signChunk(0)
-				break
-
-			} else {
 				return 0, err
 			}
+
 		}
 	}
 	return s.buf.Read(buf)