How to optimize MinIO for large file uploads and downloads?

package main

import (
    "context"
    "fmt"
    "io"
    "log"
    "os"
    "runtime"
    "sync"
    "time"

    "github.com/minio/minio-go/v7"
    "github.com/minio/minio-go/v7/pkg/credentials"
)

type OptimizedUploader struct {
    client       *minio.Client
    bucketName   string
    workers      int
    partSize     int64
    uploadQueue  chan UploadTask
    wg           sync.WaitGroup
}

type UploadTask struct {
    filePath   string
    objectName string
    fileSize   int64
}

func NewOptimizedUploader(endpoint, accessKey, secretKey, bucket string) *OptimizedUploader {
    client, err := minio.New(endpoint, &minio.Options{
        Creds:  credentials.NewStaticV4(accessKey, secretKey, ""),
        Secure: false,
    })
    if err != nil {
        log.Fatal(err)
    }

    return &OptimizedUploader{
        client:      client,
        bucketName:  bucket,
        workers:     runtime.NumCPU() * 4, // 4x CPU cores
        partSize:    128 * 1024 * 1024,    // 128MB parts
        uploadQueue: make(chan UploadTask, 1000),
    }
}

func (u *OptimizedUploader) UploadLargeFile(filePath, objectName string) error {
    file, err := os.Open(filePath)
    if err != nil {
        return err
    }
    defer file.Close()

    stat, err := file.Stat()
    if err != nil {
        return err
    }

    fileSize := stat.Size()

    options := minio.PutObjectOptions{
        PartSize:           uint64(u.partSize),
        ContentType:        "application/octet-stream",
        SendContentMd5:     true,
        DisableContentSha256: true, // Disable for performance
        ConcurrentStreamParts: true,
        NumThreads:         uint(u.workers),
    }

    // Custom reader with buffer optimization
    bufferedReader := &BufferedReader{
        reader:     file,
        bufferSize: int(u.partSize),
    }

    start := time.Now()
    _, err = u.client.PutObject(
        context.Background(),
        u.bucketName,
        objectName,
        bufferedReader,
        fileSize,
        options,
    )

    if err != nil {
        return err
    }

    duration := time.Since(start)
    throughput := float64(fileSize) / duration.Seconds() / (1024 * 1024) // MB/s

    fmt.Printf("Upload completed: %s (%.2f MB/s)\n", objectName, throughput)
    return nil
}

type BufferedReader struct {
    reader     io.Reader
    buffer     []byte
    bufferSize int
}

func (br *BufferedReader) Read(p []byte) (n int, err error) {
    if br.buffer == nil {
        br.buffer = make([]byte, br.bufferSize)
    }

    return br.reader.Read(p)
}

// Concurrent upload manager
func (u *OptimizedUploader) StartWorkers() {
    for i := 0; i < u.workers; i++ {
        go u.worker()
    }
}

func (u *OptimizedUploader) worker() {
    for task := range u.uploadQueue {
        err := u.UploadLargeFile(task.filePath, task.objectName)
        if err != nil {
            log.Printf("Upload failed for %s: %v", task.filePath, err)
        }
        u.wg.Done()
    }
}

func (u *OptimizedUploader) QueueUpload(filePath, objectName string) {
    stat, err := os.Stat(filePath)
    if err != nil {
        log.Printf("Failed to stat file %s: %v", filePath, err)
        return
    }

    task := UploadTask{
        filePath:   filePath,
        objectName: objectName,
        fileSize:   stat.Size(),
    }

    u.wg.Add(1)
    u.uploadQueue <- task
}

func (u *OptimizedUploader) WaitForCompletion() {
    u.wg.Wait()
    close(u.uploadQueue)
}

// Usage example
func main() {
    uploader := NewOptimizedUploader(
        "minio.example.com:9000",
        "access-key",
        "secret-key",
        "large-files",
    )

    uploader.StartWorkers()

    // Queue multiple large files
    uploader.QueueUpload("/path/to/large-file-1.bin", "file-1.bin")
    uploader.QueueUpload("/path/to/large-file-2.bin", "file-2.bin")
    uploader.QueueUpload("/path/to/large-file-3.bin", "file-3.bin")

    uploader.WaitForCompletion()
    fmt.Println("All uploads completed")
}

import os
import threading
import queue
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from minio import Minio
from minio.error import S3Error

class OptimizedMinIOUploader:
    def __init__(self, endpoint, access_key, secret_key, bucket_name):
        self.client = Minio(
            endpoint,
            access_key=access_key,
            secret_key=secret_key,
            secure=False
        )
        self.bucket_name = bucket_name
        self.part_size = 128 * 1024 * 1024  # 128MB
        self.max_workers = os.cpu_count() * 4

    def upload_large_file(self, file_path, object_name=None):
        """Upload a large file with optimized settings"""
        if object_name is None:
            object_name = os.path.basename(file_path)

        file_size = os.path.getsize(file_path)

        start_time = time.time()

        try:
            with open(file_path, 'rb') as file_data:
                self.client.put_object(
                    self.bucket_name,
                    object_name,
                    file_data,
                    file_size,
                    part_size=self.part_size,
                    num_parallel_uploads=self.max_workers // 2,
                    progress=self._progress_callback
                )

            end_time = time.time()
            duration = end_time - start_time
            throughput = (file_size / (1024 * 1024)) / duration  # MB/s

            print(f"Upload completed: {object_name} ({throughput:.2f} MB/s)")
            return True

        except S3Error as e:
            print(f"Upload failed for {object_name}: {e}")
            return False

    def _progress_callback(self, bytes_uploaded):
        """Progress callback for monitoring uploads"""
        pass  # Implement progress tracking if needed

    def upload_multiple_files(self, file_list, max_concurrent=None):
        """Upload multiple files concurrently"""
        if max_concurrent is None:
            max_concurrent = min(self.max_workers, len(file_list))

        results = []

        with ThreadPoolExecutor(max_workers=max_concurrent) as executor:
            future_to_file = {
                executor.submit(self.upload_large_file, file_path): file_path
                for file_path in file_list
            }

            for future in as_completed(future_to_file):
                file_path = future_to_file[future]
                try:
                    result = future.result()
                    results.append((file_path, result))
                except Exception as e:
                    print(f"Error uploading {file_path}: {e}")
                    results.append((file_path, False))

        return results

    def resume_upload(self, file_path, object_name, upload_id):
        """Resume an interrupted multipart upload"""
        try:
            # List existing parts
            parts = self.client.list_parts(self.bucket_name, object_name, upload_id)
            uploaded_parts = [(part.part_number, part.etag) for part in parts]

            # Continue upload from last part
            with open(file_path, 'rb') as file_data:
                result = self.client.put_object(
                    self.bucket_name,
                    object_name,
                    file_data,
                    os.path.getsize(file_path),
                    part_size=self.part_size,
                    sse=None,
                    progress=self._progress_callback,
                    metadata=None,
                    tags=None,
                    retention=None,
                    legal_hold=False,
                    part_number_start=len(uploaded_parts) + 1
                )

            return result

        except S3Error as e:
            print(f"Resume upload failed: {e}")
            return None

# Usage example
if __name__ == "__main__":
    uploader = OptimizedMinIOUploader(
        "minio.example.com:9000",
        "access-key",
        "secret-key",
        "large-files"
    )

    # Upload single large file
    uploader.upload_large_file("/path/to/large-file.bin")

    # Upload multiple files concurrently
    files = [
        "/path/to/file1.bin",
        "/path/to/file2.bin",
        "/path/to/file3.bin"
    ]
    results = uploader.upload_multiple_files(files, max_concurrent=8)