Method putObjectMultipartStreamParallel

api-put-object-streaming.go:453–654

putObjectMultipartStreamParallel uploads opts.NumThreads parts in parallel. This is expected to take opts.PartSize * opts.NumThreads * (GOGC / 100) bytes of buffer.

(ctx context.Context, bucketName, objectName string,
	reader io.Reader, opts PutObjectOptions,
)

// putObjectMultipartStreamParallel uploads opts.NumThreads parts in parallel.
// This is expected to take opts.PartSize * opts.NumThreads * (GOGC / 100) bytes of buffer.
func (c *Client) putObjectMultipartStreamParallel(ctx context.Context, bucketName, objectName string,
	reader io.Reader, opts PutObjectOptions,
) (info UploadInfo, err error) {
	// Input validation.
	if err = s3utils.CheckValidBucketName(bucketName); err != nil {
		return UploadInfo{}, err
	}

	if err = s3utils.CheckValidObjectName(objectName); err != nil {
		return UploadInfo{}, err
	}

	// Cancel all when an error occurs.
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()

	// Calculate the optimal parts info for a given size.
	totalPartsCount, partSize, _, err := OptimalPartInfo(-1, opts.PartSize)
	if err != nil {
		return UploadInfo{}, err
	}

	// Initiates a new multipart request
	uploadID, err := c.newUploadID(ctx, bucketName, objectName, opts)
	if err != nil {
		return UploadInfo{}, err
	}

	// Aborts the multipart upload if the function returns
	// any error, since we do not resume we should purge
	// the parts which have been uploaded to relinquish
	// storage space.
	defer func() {
		if err != nil {
			c.abortMultipartUpload(ctx, bucketName, objectName, uploadID)
		}
	}()

	// Create checksums
	// CRC32C is ~50% faster on AMD64 @ 30GB/s
	crc := opts.AutoChecksum.Hasher()

	// Total data read and written to server. should be equal to 'size' at the end of the call.
	var totalUploadedSize int64

	// Initialize parts uploaded map.
	partsInfo := make(map[int]ObjectPart)

	// Create a buffer.
	nBuffers := int64(opts.NumThreads)
	bufs := make(chan []byte, nBuffers)
	all := make([]byte, nBuffers*partSize)
	for i := int64(0); i < nBuffers; i++ {
		bufs <- all[i*partSize : i*partSize+partSize]
	}

	var wg sync.WaitGroup
	var mu sync.Mutex