storage

Documentation

Index

• Constants
• Variables
• func WithGCInterval(gcInterval time.Duration) func(*storageManager) error
• func WithPeerSearchBroadcaster(peerSearchBroadcaster pex.PeerSearchBroadcaster) func(*storageManager) error
• func WithStorageOption(opt *config.StorageOption) func(*storageManager) error
• func WithWriteBufferSize(size int64) func(*storageManager) error
• type CommonTaskRequest
• type GCCallback
• type Manager
  • func NewStorageManager(storeStrategy config.StoreStrategy, opt *config.StorageOption, ...) (Manager, error)
• type PeerTaskMetadata
• type PieceMetadata
• type ReadAllPiecesRequest
• type ReadPieceRequest
• type Reclaimer
• type RegisterSubTaskRequest
• type RegisterTaskRequest
• type ReusePeerTask
• type StoreRequest
• type TaskStorageDriver
• type UpdateTaskRequest
• type WritePieceRequest

Constants

const (
	GCName = "StorageManager"
)

Variables

var (
	ErrTaskNotFound     = errors.New("task not found")
	ErrPieceNotFound    = errors.New("piece not found")
	ErrPieceCountNotSet = errors.New("total piece count not set")
	ErrDigestNotSet     = errors.New("digest not set")
	ErrInvalidDigest    = errors.New("invalid digest")
	ErrBadRequest       = errors.New("bad request")
)

var (
	ErrShortRead = errors.New("short read")
)

Functions

func WithGCInterval

func WithGCInterval(gcInterval time.Duration) func(*storageManager) error

func WithPeerSearchBroadcaster

func WithPeerSearchBroadcaster(peerSearchBroadcaster pex.PeerSearchBroadcaster) func(*storageManager) error

func WithStorageOption

func WithStorageOption(opt *config.StorageOption) func(*storageManager) error

func WithWriteBufferSize

func WithWriteBufferSize(size int64) func(*storageManager) error

Types

type CommonTaskRequest

type CommonTaskRequest struct {
	PeerID      string `json:"peerID,omitempty"`
	TaskID      string `json:"taskID,omitempty"`
	Destination string
}

type GCCallback

type GCCallback func(request CommonTaskRequest)

type Manager

type Manager interface {
	TaskStorageDriver
	// KeepAlive tests if storage is used in given time duration
	util.KeepAlive
	// RegisterTask registers a task in storage driver
	RegisterTask(ctx context.Context, req *RegisterTaskRequest) (TaskStorageDriver, error)
	// RegisterSubTask registers a subtask in storage driver
	RegisterSubTask(ctx context.Context, req *RegisterSubTaskRequest) (TaskStorageDriver, error)
	// UnregisterTask unregisters a task in storage driver
	UnregisterTask(ctx context.Context, req CommonTaskRequest) error
	// FindCompletedTask try to find a completed task for fast path
	FindCompletedTask(taskID string) *ReusePeerTask
	// FindCompletedSubTask try to find a completed subtask for fast path
	FindCompletedSubTask(taskID string) *ReusePeerTask
	// FindPartialCompletedTask try to find a partial completed task for fast path
	FindPartialCompletedTask(taskID string, rg *nethttp.Range) *ReusePeerTask
	// CleanUp cleans all storage data
	CleanUp()
	// ListAllPeers return all peers info
	ListAllPeers(perGroupCount int) [][]*dfdaemonv1.PeerMetadata
}

func NewStorageManager

func NewStorageManager(storeStrategy config.StoreStrategy, opt *config.StorageOption, gcCallback GCCallback, dirMode fs.FileMode, moreOpts ...func(*storageManager) error) (Manager, error)

type PeerTaskMetadata

type PeerTaskMetadata struct {
	PeerID string `json:"peerID,omitempty"`
	TaskID string `json:"taskID,omitempty"`
}

type PieceMetadata

type PieceMetadata struct {
	Num    int32               `json:"num,omitempty"`
	Md5    string              `json:"md5,omitempty"`
	Offset uint64              `json:"offset,omitempty"`
	Range  http.Range          `json:"range,omitempty"`
	Style  commonv1.PieceStyle `json:"style,omitempty"`
	// time(nanosecond) consumed
	Cost uint64 `json:"cost,omitempty"`
}

type ReadAllPiecesRequest

type ReadAllPiecesRequest struct {
	PeerTaskMetadata
	Range *http.Range
}

type ReadPieceRequest

type ReadPieceRequest struct {
	PeerTaskMetadata
	PieceMetadata
}

type Reclaimer

type Reclaimer interface {
	// CanReclaim indicates whether the storage can be reclaimed
	CanReclaim() bool

	// MarkReclaim marks the storage which will be reclaimed
	MarkReclaim()

	// Reclaim reclaims the storage
	Reclaim() error
}

Reclaimer stands storage reclaimer

type RegisterSubTaskRequest

type RegisterSubTaskRequest struct {
	Parent  PeerTaskMetadata
	SubTask PeerTaskMetadata
	Range   *http.Range
}

type RegisterTaskRequest

type RegisterTaskRequest struct {
	PeerTaskMetadata
	DesiredLocation string
	ContentLength   int64
	TotalPieces     int32
	PieceMd5Sign    string
}

type ReusePeerTask

type ReusePeerTask struct {
	PeerTaskMetadata
	ContentLength int64
	TotalPieces   int32
	PieceMd5Sign  string
	Header        *source.Header
	Storage       TaskStorageDriver
}

type StoreRequest

type StoreRequest struct {
	CommonTaskRequest
	MetadataOnly bool
	// StoreDataOnly stands save file only without save metadata, used in reuse cases
	StoreDataOnly bool
	TotalPieces   int32
	// OriginalOffset stands keep original offset in the target file, if the target file is not original file, return error
	OriginalOffset bool
}

type TaskStorageDriver

type TaskStorageDriver interface {
	// WritePiece put a piece of a task to storage
	WritePiece(ctx context.Context, req *WritePieceRequest) (int64, error)

	// ReadPiece get a piece data reader of a task from storage
	// return a Reader and a Closer from task data with sought, caller should read bytes and close it.
	// If req.Num is equal to -1, range has a fixed value.
	ReadPiece(ctx context.Context, req *ReadPieceRequest) (io.Reader, io.Closer, error)

	ReadAllPieces(ctx context.Context, req *ReadAllPiecesRequest) (io.ReadCloser, error)

	GetPieces(ctx context.Context, req *commonv1.PieceTaskRequest) (*commonv1.PiecePacket, error)

	GetTotalPieces(ctx context.Context, req *PeerTaskMetadata) (int32, error)

	GetExtendAttribute(ctx context.Context, req *PeerTaskMetadata) (*commonv1.ExtendAttribute, error)

	UpdateTask(ctx context.Context, req *UpdateTaskRequest) error

	// Store stores task data to the target path
	Store(ctx context.Context, req *StoreRequest) error

	ValidateDigest(req *PeerTaskMetadata) error

	IsInvalid(req *PeerTaskMetadata) (bool, error)
}

type UpdateTaskRequest

type UpdateTaskRequest struct {
	PeerTaskMetadata
	ContentLength int64
	TotalPieces   int32
	PieceMd5Sign  string
	Header        *source.Header
}

type WritePieceRequest

type WritePieceRequest struct {
	PeerTaskMetadata
	PieceMetadata
	UnknownLength bool
	Reader        io.Reader
	// NeedGenMetadata is used after the last piece in back source case
	NeedGenMetadata func(n int64) (total int32, contentLength int64, gen bool)
}