rapid

Documentation

Index

• func GetTokenSource(ctx context.Context, keyFile string) (tokenSrc oauth2.TokenSource, err error)
• type DownloadTask
  • func NewDownloadTask(downloadRange Range, pool *MRDPool, writer io.Writer, ...) *DownloadTask
  • func (dt *DownloadTask) Execute()
• type MRDPool
  • func NewMRDPool(config *MRDPoolConfig) (*MRDPool, error)
  • func (p *MRDPool) Add(output io.Writer, offset, length int64, callback func(int64, int64, error)) error
  • func (p *MRDPool) Close() error
  • func (p *MRDPool) Error() error
  • func (p *MRDPool) GetHandle() []byte
  • func (p *MRDPool) GetStats() PoolStats
  • func (p *MRDPool) IsClosed() bool
  • func (p *MRDPool) PoolSize() int
  • func (p *MRDPool) Wait()
• type MRDPoolConfig
• type MultiRangeDownloader
• type PoolStats
• type Range

Examples

• MRDPool

Functions

func GetTokenSource

func GetTokenSource(
	ctx context.Context,
	keyFile string,
) (tokenSrc oauth2.TokenSource, err error)

GetTokenSource returns a TokenSource for GCS API given a key file, or with the default credentials.

Types

type DownloadTask

type DownloadTask struct {
	// contains filtered or unexported fields
}

DownloadTask represents a task that downloads a range using an MRD pool and writes the result to an io.Writer. It implements the workerpool.Task interface.

func NewDownloadTask

func NewDownloadTask(downloadRange Range, pool *MRDPool, writer io.Writer, callback func(int64, int64, error)) *DownloadTask

NewDownloadTask creates a new download task that can be scheduled to a worker pool.

func (*DownloadTask) Execute

func (dt *DownloadTask) Execute()

Execute implements the workerpool.Task interface. It schedules the download of the range using the MRD pool.

type MRDPool

type MRDPool struct {
	// contains filtered or unexported fields
}

MRDPool manages a pool of MultiRangeDownloader instances and distributes requests across them using round-robin scheduling.

Example

// Create a storage client (in real usage)
// client, err := storage.NewClient(context.Background())
// if err != nil {
//     log.Fatal(err)
// }
// defer client.Close()

// Create MRD pool with 5 downloaders
config := &MRDPoolConfig{
	PoolSize: 5,
	Client:   &storage.Client{}, // Use actual client in production
}

pool, err := NewMRDPool(config)
if err != nil {
	fmt.Printf("Failed to create pool: %v\n", err)
	return
}
defer pool.Close()

// Use the pool for downloading
var buf bytes.Buffer
completed := false

err = pool.Add(&buf, 0, 1024, func(offset, length int64, err error) {
	if err != nil {
		fmt.Printf("Download failed: %v\n", err)
	} else {
		fmt.Printf("Downloaded %d bytes at offset %d\n", length, offset)
		completed = true
	}
})

if err != nil {
	fmt.Printf("Add failed: %v\n", err)
	return
}

// Wait for all downloads to complete
pool.Wait()

// Check for errors
if err := pool.Error(); err != nil {
	fmt.Printf("Pool error: %v\n", err)
	return
}

// Get pool statistics
stats := pool.GetStats()
fmt.Printf("Pool size: %d, Total requests: %d, Completed: %v\n", stats.PoolSize, stats.RequestCount, completed)

func NewMRDPool

func NewMRDPool(config *MRDPoolConfig) (*MRDPool, error)

NewMRDPool creates a new pool of MultiRangeDownloader instances.

func (*MRDPool) Add

func (p *MRDPool) Add(output io.Writer, offset, length int64, callback func(int64, int64, error)) error

Add adds a download task to one of the MRD instances using round-robin selection. The output writer receives the downloaded data, and the callback is invoked when the download completes with the offset, length, and any error. If the selected downloader is in error state, it attempts to recreate it and retry up to maxRetries times across different downloaders.

func (*MRDPool) Close

func (p *MRDPool) Close() error

Close closes all MultiRangeDownloader instances in the pool.

func (*MRDPool) Error

func (p *MRDPool) Error() error

Error returns any error from the MRD instances. Returns the first error encountered, or nil if no errors.

func (*MRDPool) GetHandle

func (p *MRDPool) GetHandle() []byte

GetHandle returns the handle from the first MRD instance in the pool. This is primarily for compatibility with the MultiRangeDownloader interface.

func (*MRDPool) GetStats

func (p *MRDPool) GetStats() PoolStats

GetStats returns current pool statistics.

func (*MRDPool) IsClosed

func (p *MRDPool) IsClosed() bool

IsClosed returns whether the pool is closed.

func (*MRDPool) PoolSize

func (p *MRDPool) PoolSize() int

PoolSize returns the size of the pool.

func (*MRDPool) Wait

func (p *MRDPool) Wait()

Wait waits for all downloads on all MRD instances to complete.

type MRDPoolConfig

type MRDPoolConfig struct {
	// PoolSize is the number of MultiRangeDownloader instances in the pool
	PoolSize int

	// Client is the GCS storage client used to create MRD instances
	Client *storage.Client

	Bucket string
	Object string
}

MRDPoolConfig contains configuration for the MRD pool.

type MultiRangeDownloader

type MultiRangeDownloader interface {
	Add(output io.Writer, offset, length int64, callback func(int64, int64, error))
	Close() error
	Wait()
	Error() error
	GetHandle() []byte
}

An interface to generalize the MultiRangeDownloader structure in go-storage module to ease our testing.

type PoolStats

type PoolStats struct {
	PoolSize     int
	RequestCount uint64
	Closed       bool
}

Stats returns statistics about the pool usage.

type Range

type Range struct {
	Offset int64 // Starting byte offset in the object
	Length int64 // Number of bytes to read
}

Range represents a contiguous range of data to be downloaded.