// PutObject puts a blob to the specified container on Azure
550587
func (c *Client) PutObject(ctx context.Context, key, container string, md5 hash.Hash, size int64, body io.Reader) error {
	req := newReq(ctx, c.keyURL(container, key))
	req.Method = "PUT"
	req.ContentLength = size
	if md5 != nil {
		b64 := new(bytes.Buffer)
		encoder := base64.NewEncoder(base64.StdEncoding, b64)
		encoder.Write(md5.Sum(nil))
		encoder.Close()
		req.Header.Set("Content-MD5", b64.String())
	}
	req.Header.Set("Content-Length", strconv.Itoa(int(size)))
	req.Header.Set("x-ms-blob-type", "BlockBlob")
	c.Auth.SignRequest(req)
	req.Body = ioutil.NopCloser(body)

	res, err := c.transport().RoundTrip(req)
	if res != nil && res.Body != nil {
		defer res.Body.Close()
	}
	if err != nil {
		return err
	}
	if res.StatusCode != http.StatusCreated {
		if res.StatusCode < 500 {
			aerr := getAzureError("PutObject", res)
			return aerr
		}
		return fmt.Errorf("got response code %d from Azure", res.StatusCode)
	}
	return nil
}
305796
func (batcher *UploadObjectsIterator) UploadObject() BatchUploadObject {
	object := batcher.Objects[batcher.index]
	return object
}
1172811
func (s *Snapshotter) SetBlob(ctx context.Context, key string, diffID, blobsum digest.Digest) error {
	info, err := s.opt.Content.Info(ctx, blobsum)
	if err != nil {
		return err
	}
	if _, ok := info.Labels["containerd.io/uncompressed"]; !ok {
		labels := map[string]string{
			"containerd.io/uncompressed": diffID.String(),
		}
		if _, err := s.opt.Content.Update(ctx, content.Info{
			Digest: blobsum,
			Labels: labels,
		}, "labels.containerd.io/uncompressed"); err != nil {
			return err
		}
	}
	md, _ := s.opt.MetadataStore.Get(key)

	v, err := metadata.NewValue(DiffPair{DiffID: diffID, Blobsum: blobsum})
	if err != nil {
		return err
	}
	v.Index = index(blobsum)

	return md.Update(func(b *bolt.Bucket) error {
		return md.SetValue(b, blobKey, v)
	})
}
902161
func (self *qbsWrapped) Put(id int64, value interface{}, pb PBundle) (interface{}, error) {
	return self.applyPolicy(pb, func(tx *qbs.Qbs) (interface{}, error) {
		return self.put.PutQbs(id, value, pb, tx)
	})
}
786206
func (api objectAPIHandlers) PutObjectPartHandler(w http.ResponseWriter, r *http.Request) {
	ctx := newContext(r, w, "PutObjectPart")

	defer logger.AuditLog(w, r, "PutObjectPart", mustGetClaimsFromToken(r))

	objectAPI := api.ObjectAPI()
	if objectAPI == nil {
		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrServerNotInitialized), r.URL, guessIsBrowserReq(r))
		return
	}
	if crypto.S3KMS.IsRequested(r.Header) {
		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrNotImplemented), r.URL, guessIsBrowserReq(r)) // SSE-KMS is not supported
		return
	}
	if !api.EncryptionEnabled() && hasServerSideEncryptionHeader(r.Header) {
		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrNotImplemented), r.URL, guessIsBrowserReq(r))
		return
	}
	vars := mux.Vars(r)
	bucket := vars["bucket"]
	object := vars["object"]

	// X-Amz-Copy-Source shouldn't be set for this call.
	if _, ok := r.Header["X-Amz-Copy-Source"]; ok {
		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrInvalidCopySource), r.URL, guessIsBrowserReq(r))
		return
	}

	// get Content-Md5 sent by client and verify if valid
	md5Bytes, err := checkValidMD5(r.Header)
	if err != nil {
		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrInvalidDigest), r.URL, guessIsBrowserReq(r))
		return
	}

	/// if Content-Length is unknown/missing, throw away
	size := r.ContentLength

	rAuthType := getRequestAuthType(r)
	// For auth type streaming signature, we need to gather a different content length.
	if rAuthType == authTypeStreamingSigned {
		if sizeStr, ok := r.Header["X-Amz-Decoded-Content-Length"]; ok {
			if sizeStr[0] == "" {
				writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrMissingContentLength), r.URL, guessIsBrowserReq(r))
				return
			}
			size, err = strconv.ParseInt(sizeStr[0], 10, 64)
			if err != nil {
				writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
				return
			}
		}
	}
	if size == -1 {
		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrMissingContentLength), r.URL, guessIsBrowserReq(r))
		return
	}

	/// maximum Upload size for multipart objects in a single operation
	if isMaxAllowedPartSize(size) {
		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrEntityTooLarge), r.URL, guessIsBrowserReq(r))
		return
	}

	uploadID := r.URL.Query().Get("uploadId")
	partIDString := r.URL.Query().Get("partNumber")

	partID, err := strconv.Atoi(partIDString)
	if err != nil {
		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrInvalidPart), r.URL, guessIsBrowserReq(r))
		return
	}

	// check partID with maximum part ID for multipart objects
	if isMaxPartID(partID) {
		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrInvalidMaxParts), r.URL, guessIsBrowserReq(r))
		return
	}

	var (
		md5hex    = hex.EncodeToString(md5Bytes)
		sha256hex = ""
		reader    io.Reader
		s3Error   APIErrorCode
	)
	reader = r.Body
	if s3Error = isPutAllowed(rAuthType, bucket, object, r); s3Error != ErrNone {
		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL, guessIsBrowserReq(r))
		return
	}

	switch rAuthType {
	case authTypeStreamingSigned:
		// Initialize stream signature verifier.
		reader, s3Error = newSignV4ChunkedReader(r)
		if s3Error != ErrNone {
			writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL, guessIsBrowserReq(r))
			return
		}
	case authTypeSignedV2, authTypePresignedV2:
		if s3Error = isReqAuthenticatedV2(r); s3Error != ErrNone {
			writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL, guessIsBrowserReq(r))
			return
		}
	case authTypePresigned, authTypeSigned:
		if s3Error = reqSignatureV4Verify(r, globalServerConfig.GetRegion(), serviceS3); s3Error != ErrNone {
			writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Error), r.URL, guessIsBrowserReq(r))
			return
		}

		if !skipContentSha256Cksum(r) {
			sha256hex = getContentSha256Cksum(r, serviceS3)
		}
	}

	actualSize := size

	// get encryption options
	var opts ObjectOptions
	if crypto.SSEC.IsRequested(r.Header) {
		opts, err = putOpts(ctx, r, bucket, object, nil)
		if err != nil {
			writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
			return
		}
	}
	var li ListPartsInfo
	li, err = objectAPI.ListObjectParts(ctx, bucket, object, uploadID, 0, 1, opts)
	if err != nil {
		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
		return
	}
	// Read compression metadata preserved in the init multipart for the decision.
	_, compressPart := li.UserDefined[ReservedMetadataPrefix+"compression"]

	isCompressed := false
	if objectAPI.IsCompressionSupported() && compressPart {
		actualReader, err := hash.NewReader(reader, size, md5hex, sha256hex, actualSize)
		if err != nil {
			writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
			return
		}

		// Set compression metrics.
		reader = newSnappyCompressReader(actualReader)
		size = -1   // Since compressed size is un-predictable.
		md5hex = "" // Do not try to verify the content.
		sha256hex = ""
		isCompressed = true
	}

	hashReader, err := hash.NewReader(reader, size, md5hex, sha256hex, actualSize)
	if err != nil {
		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
		return
	}
	rawReader := hashReader
	pReader := NewPutObjReader(rawReader, nil, nil)

	// Deny if WORM is enabled
	if globalWORMEnabled {
		if _, err = objectAPI.GetObjectInfo(ctx, bucket, object, opts); err == nil {
			writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrMethodNotAllowed), r.URL, guessIsBrowserReq(r))
			return
		}
	}

	isEncrypted := false
	var objectEncryptionKey []byte
	if objectAPI.IsEncryptionSupported() && !isCompressed {
		var li ListPartsInfo
		li, err = objectAPI.ListObjectParts(ctx, bucket, object, uploadID, 0, 1, ObjectOptions{})
		if err != nil {
			writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
			return
		}
		li.UserDefined = CleanMinioInternalMetadataKeys(li.UserDefined)
		if crypto.IsEncrypted(li.UserDefined) {
			if !crypto.SSEC.IsRequested(r.Header) && crypto.SSEC.IsEncrypted(li.UserDefined) {
				writeErrorResponse(ctx, w, errorCodes.ToAPIErr(ErrSSEMultipartEncrypted), r.URL, guessIsBrowserReq(r))
				return
			}

			isEncrypted = true // to detect SSE-S3 encryption
			opts, err = putOpts(ctx, r, bucket, object, li.UserDefined)
			if err != nil {
				writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
				return
			}

			var key []byte
			if crypto.SSEC.IsRequested(r.Header) {
				key, err = ParseSSECustomerRequest(r)
				if err != nil {
					writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
					return
				}
			}

			// Calculating object encryption key
			objectEncryptionKey, err = decryptObjectInfo(key, bucket, object, li.UserDefined)
			if err != nil {
				writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
				return
			}
			var partIDbin [4]byte
			binary.LittleEndian.PutUint32(partIDbin[:], uint32(partID)) // marshal part ID

			mac := hmac.New(sha256.New, objectEncryptionKey) // derive part encryption key from part ID and object key
			mac.Write(partIDbin[:])
			partEncryptionKey := mac.Sum(nil)

			reader, err = sio.EncryptReader(hashReader, sio.Config{Key: partEncryptionKey})
			if err != nil {
				writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
				return
			}
			info := ObjectInfo{Size: size}
			hashReader, err = hash.NewReader(reader, info.EncryptedSize(), "", "", size) // do not try to verify encrypted content
			if err != nil {
				writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
				return
			}
			pReader = NewPutObjReader(rawReader, hashReader, objectEncryptionKey)
		}
	}

	putObjectPart := objectAPI.PutObjectPart
	if api.CacheAPI() != nil && !isEncrypted {
		putObjectPart = api.CacheAPI().PutObjectPart
	}
	partInfo, err := putObjectPart(ctx, bucket, object, uploadID, partID, pReader, opts)
	if err != nil {
		// Verify if the underlying error is signature mismatch.
		writeErrorResponse(ctx, w, toAPIError(ctx, err), r.URL, guessIsBrowserReq(r))
		return
	}

	etag := partInfo.ETag
	if isCompressed {
		// Suppress compressed ETag.
		etag = partInfo.ETag + "-1"
	} else if isEncrypted {
		etag = tryDecryptETag(objectEncryptionKey, partInfo.ETag, crypto.SSEC.IsRequested(r.Header))
	}
	w.Header()["ETag"] = []string{"\"" + etag + "\""}

	writeSuccessResponseHeadersOnly(w)
}
1035989
func (s *xlSets) PutObjectPart(ctx context.Context, bucket, object, uploadID string, partID int, data *PutObjReader, opts ObjectOptions) (info PartInfo, err error) {
	return s.getHashedSet(object).PutObjectPart(ctx, bucket, object, uploadID, partID, data, opts)
}
935927
func (b *bucket) StoreObject(key string, data []byte) error {
	// Validate the key.
	if err := validateKey(key); err != nil {
		return err
	}

	// Build an appropriate HTTP request.
	//
	// Reference:
	//     http://docs.amazonwebservices.com/AmazonS3/latest/API/RESTObjectPUT.html
	httpReq := &http.Request{
		Verb: "PUT",
		Path: fmt.Sprintf("/%s/%s", b.name, key),
		Body: data,
		Headers: map[string]string{
			"Date": b.clock.Now().UTC().Format(sys_time.RFC1123),
		},
	}

	// Add a Content-MD5 header, as advised in the Amazon docs.
	if err := addMd5Header(httpReq, httpReq.Body); err != nil {
		return err
	}

	// Sign the request.
	if err := b.signer.Sign(httpReq); err != nil {
		return fmt.Errorf("Sign: %v", err)
	}

	// Send the request.
	httpResp, err := b.httpConn.SendRequest(httpReq)
	if err != nil {
		return fmt.Errorf("SendRequest: %v", err)
	}

	// Check the response.
	if httpResp.StatusCode != 200 {
		return fmt.Errorf("Error from server: %d %s", httpResp.StatusCode, httpResp.Body)
	}

	return nil
}
756272
func (s *StreamEncoder) AddObject(v MarshalerJSONObject) {
	if v.IsNil() {
		return
	}
	s.Encoder.writeByte('{')
	v.MarshalJSONObject(s.Encoder)
	s.Encoder.writeByte('}')
	s.Encoder.writeByte(s.delimiter)
}
375372
func (ds *datastore) PutObjectAtomic(kvObject KVObject) error {
	var (
		previous *store.KVPair
		pair     *store.KVPair
		err      error
	)
	if ds.sequential {
		ds.Lock()
		defer ds.Unlock()
	}

	if kvObject == nil {
		return types.BadRequestErrorf("invalid KV Object : nil")
	}

	kvObjValue := kvObject.Value()

	if kvObjValue == nil {
		return types.BadRequestErrorf("invalid KV Object with a nil Value for key %s", Key(kvObject.Key()...))
	}

	if kvObject.Skip() {
		goto add_cache
	}

	if kvObject.Exists() {
		previous = &store.KVPair{Key: Key(kvObject.Key()...), LastIndex: kvObject.Index()}
	} else {
		previous = nil
	}

	_, pair, err = ds.store.AtomicPut(Key(kvObject.Key()...), kvObjValue, previous, nil)
	if err != nil {
		if err == store.ErrKeyExists {
			return ErrKeyModified
		}
		return err
	}

	kvObject.SetIndex(pair.LastIndex)

add_cache:
	if ds.cache != nil {
		// If persistent store is skipped, sequencing needs to
		// happen in cache.
		return ds.cache.add(kvObject, kvObject.Skip())
	}

	return nil
}
273229
func (c *Client) PutObject(path, objectName string, object []byte) error {
	r := bytes.NewReader(object)
	req := request{
		method:         client.PUT,
		url:            makeURL(apiStorage, path, objectName),
		reqReader:      r,
		reqLength:      len(object),
		expectedStatus: http.StatusNoContent,
	}
	if _, err := c.sendRequest(req); err != nil {
		return errors.Newf(err, "failed to create object: %s/%s", path, objectName)
	}
	return nil
}
802557
func PutOp(bin *Bin) *Operation {
	return &Operation{opType: _WRITE, binName: bin.Name, binValue: bin.Value}
}
366891
func (s *Document) SetS3Object(v *S3Object) *Document {
	s.S3Object = v
	return s
}
996028
func (c Client) putObjectNoChecksum(ctx context.Context, bucketName, objectName string, reader io.Reader, size int64, opts PutObjectOptions) (n int64, err error) {
	// Input validation.
	if err := s3utils.CheckValidBucketName(bucketName); err != nil {
		return 0, err
	}
	if err := s3utils.CheckValidObjectName(objectName); err != nil {
		return 0, err
	}

	// Size -1 is only supported on Google Cloud Storage, we error
	// out in all other situations.
	if size < 0 && !s3utils.IsGoogleEndpoint(*c.endpointURL) {
		return 0, ErrEntityTooSmall(size, bucketName, objectName)
	}
	if size > 0 {
		if isReadAt(reader) && !isObject(reader) {
			seeker, _ := reader.(io.Seeker)
			offset, err := seeker.Seek(0, io.SeekCurrent)
			if err != nil {
				return 0, ErrInvalidArgument(err.Error())
			}
			reader = io.NewSectionReader(reader.(io.ReaderAt), offset, size)
		}
	}

	// Update progress reader appropriately to the latest offset as we
	// read from the source.
	readSeeker := newHook(reader, opts.Progress)

	// This function does not calculate sha256 and md5sum for payload.
	// Execute put object.
	st, err := c.putObjectDo(ctx, bucketName, objectName, readSeeker, "", "", size, opts)
	if err != nil {
		return 0, err
	}
	if st.Size != size {
		return 0, ErrUnexpectedEOF(st.Size, size, bucketName, objectName)
	}
	return size, nil
}
451531
func (s *BatchWriteOperationResponse) SetCreateObject(v *BatchCreateObjectResponse) *BatchWriteOperationResponse {
	s.CreateObject = v
	return s
}
61472
func (c cacheObjects) PutObject(ctx context.Context, bucket, object string, r *PutObjReader, opts ObjectOptions) (objInfo ObjectInfo, err error) {
	putObjectFn := c.PutObjectFn
	data := r.Reader
	dcache, err := c.cache.getCacheFS(ctx, bucket, object)
	if err != nil {
		// disk cache could not be located,execute backend call.
		return putObjectFn(ctx, bucket, object, r, opts)
	}
	size := r.Size()

	// fetch from backend if there is no space on cache drive
	if !dcache.diskAvailable(size) {
		return putObjectFn(ctx, bucket, object, r, opts)
	}
	// fetch from backend if cache exclude pattern or cache-control
	// directive set to exclude
	if c.isCacheExclude(bucket, object) || filterFromCache(opts.UserDefined) {
		dcache.Delete(ctx, bucket, object)
		return putObjectFn(ctx, bucket, object, r, opts)
	}
	objInfo = ObjectInfo{}
	// Initialize pipe to stream data to backend
	pipeReader, pipeWriter := io.Pipe()
	hashReader, err := hash.NewReader(pipeReader, size, data.MD5HexString(), data.SHA256HexString(), data.ActualSize())
	if err != nil {
		return ObjectInfo{}, err
	}
	// Initialize pipe to stream data to cache
	rPipe, wPipe := io.Pipe()
	cHashReader, err := hash.NewReader(rPipe, size, data.MD5HexString(), data.SHA256HexString(), data.ActualSize())
	if err != nil {
		return ObjectInfo{}, err
	}
	oinfoCh := make(chan ObjectInfo)
	errCh := make(chan error)
	go func() {
		oinfo, perr := putObjectFn(ctx, bucket, object, NewPutObjReader(hashReader, nil, nil), opts)
		if perr != nil {
			pipeWriter.CloseWithError(perr)
			wPipe.CloseWithError(perr)
			close(oinfoCh)
			errCh <- perr
			return
		}
		close(errCh)
		oinfoCh <- oinfo
	}()

	go func() {
		if err = dcache.Put(ctx, bucket, object, NewPutObjReader(cHashReader, nil, nil), opts); err != nil {
			wPipe.CloseWithError(err)
			return
		}
	}()

	mwriter := io.MultiWriter(pipeWriter, wPipe)
	_, err = io.Copy(mwriter, data)
	if err != nil {
		err = <-errCh
		return objInfo, err
	}
	pipeWriter.Close()
	wPipe.Close()
	objInfo = <-oinfoCh
	return objInfo, err
}
1087205
func (b *Bucket) PutCopy(path string, perm ACL, options CopyOptions, source string) (*CopyObjectResult, error) {
	headers := make(http.Header)

	headers.Set("x-oss-object-acl", string(perm))
	headers.Set("x-oss-copy-source", source)

	options.addHeaders(headers)
	req := &request{
		method:  "PUT",
		bucket:  b.Name,
		path:    path,
		headers: headers,
		timeout: 5 * time.Minute,
	}
	resp := &CopyObjectResult{}
	err := b.Client.query(req, resp)
	if err != nil {
		return resp, err
	}
	return resp, nil
}
885449
func (xl xlObjects) PutObjectPart(ctx context.Context, bucket, object, uploadID string, partID int, r *PutObjReader, opts ObjectOptions) (pi PartInfo, e error) {
	data := r.Reader
	if err := checkPutObjectPartArgs(ctx, bucket, object, xl); err != nil {
		return pi, err
	}

	// Validate input data size and it can never be less than zero.
	if data.Size() < -1 {
		logger.LogIf(ctx, errInvalidArgument)
		return pi, toObjectErr(errInvalidArgument)
	}

	var partsMetadata []xlMetaV1
	var errs []error
	uploadIDPath := xl.getUploadIDDir(bucket, object, uploadID)
	uploadIDLockPath := xl.getUploadIDLockPath(bucket, object, uploadID)

	// pre-check upload id lock.
	preUploadIDLock := xl.nsMutex.NewNSLock(minioMetaMultipartBucket, uploadIDLockPath)
	if err := preUploadIDLock.GetRLock(globalOperationTimeout); err != nil {
		return pi, err
	}

	// Validates if upload ID exists.
	if err := xl.checkUploadIDExists(ctx, bucket, object, uploadID); err != nil {
		preUploadIDLock.RUnlock()
		return pi, toObjectErr(err, bucket, object, uploadID)
	}

	// Read metadata associated with the object from all disks.
	partsMetadata, errs = readAllXLMetadata(ctx, xl.getDisks(), minioMetaMultipartBucket,
		uploadIDPath)

	// get Quorum for this object
	_, writeQuorum, err := objectQuorumFromMeta(ctx, xl, partsMetadata, errs)
	if err != nil {
		preUploadIDLock.RUnlock()
		return pi, toObjectErr(err, bucket, object)
	}

	reducedErr := reduceWriteQuorumErrs(ctx, errs, objectOpIgnoredErrs, writeQuorum)
	if reducedErr == errXLWriteQuorum {
		preUploadIDLock.RUnlock()
		return pi, toObjectErr(reducedErr, bucket, object)
	}
	preUploadIDLock.RUnlock()

	// List all online disks.
	onlineDisks, modTime := listOnlineDisks(xl.getDisks(), partsMetadata, errs)

	// Pick one from the first valid metadata.
	xlMeta, err := pickValidXLMeta(ctx, partsMetadata, modTime, writeQuorum)
	if err != nil {
		return pi, err
	}

	onlineDisks = shuffleDisks(onlineDisks, xlMeta.Erasure.Distribution)

	// Need a unique name for the part being written in minioMetaBucket to
	// accommodate concurrent PutObjectPart requests

	partSuffix := fmt.Sprintf("part.%d", partID)
	tmpPart := mustGetUUID()
	tmpPartPath := path.Join(tmpPart, partSuffix)

	// Delete the temporary object part. If PutObjectPart succeeds there would be nothing to delete.
	defer xl.deleteObject(ctx, minioMetaTmpBucket, tmpPart, writeQuorum, false)

	erasure, err := NewErasure(ctx, xlMeta.Erasure.DataBlocks, xlMeta.Erasure.ParityBlocks, xlMeta.Erasure.BlockSize)
	if err != nil {
		return pi, toObjectErr(err, bucket, object)
	}

	// Fetch buffer for I/O, returns from the pool if not allocates a new one and returns.
	var buffer []byte
	switch size := data.Size(); {
	case size == 0:
		buffer = make([]byte, 1) // Allocate atleast a byte to reach EOF
	case size == -1 || size >= blockSizeV1:
		buffer = xl.bp.Get()
		defer xl.bp.Put(buffer)
	case size < blockSizeV1:
		// No need to allocate fully blockSizeV1 buffer if the incoming data is smaller.
		buffer = make([]byte, size, 2*size)
	}

	if len(buffer) > int(xlMeta.Erasure.BlockSize) {
		buffer = buffer[:xlMeta.Erasure.BlockSize]
	}
	writers := make([]io.Writer, len(onlineDisks))
	for i, disk := range onlineDisks {
		if disk == nil {
			continue
		}
		writers[i] = newBitrotWriter(disk, minioMetaTmpBucket, tmpPartPath, erasure.ShardFileSize(data.Size()), DefaultBitrotAlgorithm, erasure.ShardSize())
	}

	n, err := erasure.Encode(ctx, data, writers, buffer, erasure.dataBlocks+1)
	closeBitrotWriters(writers)
	if err != nil {
		return pi, toObjectErr(err, bucket, object)
	}

	// Should return IncompleteBody{} error when reader has fewer bytes
	// than specified in request header.
	if n < data.Size() {
		return pi, IncompleteBody{}
	}

	for i := range writers {
		if writers[i] == nil {
			onlineDisks[i] = nil
		}
	}

	// post-upload check (write) lock
	postUploadIDLock := xl.nsMutex.NewNSLock(minioMetaMultipartBucket, uploadIDLockPath)
	if err = postUploadIDLock.GetLock(globalOperationTimeout); err != nil {
		return pi, err
	}
	defer postUploadIDLock.Unlock()

	// Validates if upload ID exists.
	if err := xl.checkUploadIDExists(ctx, bucket, object, uploadID); err != nil {
		return pi, toObjectErr(err, bucket, object, uploadID)
	}

	// Rename temporary part file to its final location.
	partPath := path.Join(uploadIDPath, partSuffix)
	onlineDisks, err = rename(ctx, onlineDisks, minioMetaTmpBucket, tmpPartPath, minioMetaMultipartBucket, partPath, false, writeQuorum, nil)
	if err != nil {
		return pi, toObjectErr(err, minioMetaMultipartBucket, partPath)
	}

	// Read metadata again because it might be updated with parallel upload of another part.
	partsMetadata, errs = readAllXLMetadata(ctx, onlineDisks, minioMetaMultipartBucket, uploadIDPath)
	reducedErr = reduceWriteQuorumErrs(ctx, errs, objectOpIgnoredErrs, writeQuorum)
	if reducedErr == errXLWriteQuorum {
		return pi, toObjectErr(reducedErr, bucket, object)
	}

	// Get current highest version based on re-read partsMetadata.
	onlineDisks, modTime = listOnlineDisks(onlineDisks, partsMetadata, errs)

	// Pick one from the first valid metadata.
	xlMeta, err = pickValidXLMeta(ctx, partsMetadata, modTime, writeQuorum)
	if err != nil {
		return pi, err
	}

	// Once part is successfully committed, proceed with updating XL metadata.
	xlMeta.Stat.ModTime = UTCNow()

	md5hex := r.MD5CurrentHexString()

	// Add the current part.
	xlMeta.AddObjectPart(partID, partSuffix, md5hex, n, data.ActualSize())

	for i, disk := range onlineDisks {
		if disk == OfflineDisk {
			continue
		}
		partsMetadata[i].Stat = xlMeta.Stat
		partsMetadata[i].Parts = xlMeta.Parts
		partsMetadata[i].Erasure.AddChecksumInfo(ChecksumInfo{partSuffix, DefaultBitrotAlgorithm, bitrotWriterSum(writers[i])})
	}

	// Write all the checksum metadata.
	tempXLMetaPath := mustGetUUID()

	// Cleanup in case of xl.json writing failure
	defer xl.deleteObject(ctx, minioMetaTmpBucket, tempXLMetaPath, writeQuorum, false)

	// Writes a unique `xl.json` each disk carrying new checksum related information.
	if onlineDisks, err = writeUniqueXLMetadata(ctx, onlineDisks, minioMetaTmpBucket, tempXLMetaPath, partsMetadata, writeQuorum); err != nil {
		return pi, toObjectErr(err, minioMetaTmpBucket, tempXLMetaPath)
	}

	if _, err = commitXLMetadata(ctx, onlineDisks, minioMetaTmpBucket, tempXLMetaPath, minioMetaMultipartBucket, uploadIDPath, writeQuorum); err != nil {
		return pi, toObjectErr(err, minioMetaMultipartBucket, uploadIDPath)
	}

	fi, err := xl.statPart(ctx, bucket, object, uploadID, partSuffix)
	if err != nil {
		return pi, toObjectErr(err, minioMetaMultipartBucket, partSuffix)
	}

	// Return success.
	return PartInfo{
		PartNumber:   partID,
		LastModified: fi.ModTime,
		ETag:         md5hex,
		Size:         fi.Size,
		ActualSize:   data.ActualSize(),
	}, nil
}