peer

Documentation

Index

• func WithCalculateDigest(enable bool) func(*pieceManager)
• func WithLimiter(limiter *rate.Limiter) func(*pieceManager)
• func WithTransport(rt http.RoundTripper) func(*pieceDownloader) error
• func WithTransportOption(opt *config.TransportOption) func(*pieceManager)
• type Bitmap
  • func NewBitmap() *Bitmap
  • func NewBitmapWithCap(c int32) *Bitmap
  • func (b *Bitmap) Clean(i int32)
  • func (b *Bitmap) IsSet(i int32) bool
  • func (b *Bitmap) Set(i int32)
  • func (b *Bitmap) Sets(xs ...int32)
  • func (b *Bitmap) Settled() int32
• type DownloadPieceRequest
• type DownloadPieceResult
• type FileTask
• type FileTaskProgress
• type FileTaskRequest
• type IDGenerator
  • func NewPeerIDGenerator(ip string) IDGenerator
• type Logger
• type PieceDownloader
  • func NewPieceDownloader(timeout time.Duration, opts ...func(*pieceDownloader) error) (PieceDownloader, error)
• type PieceInfo
• type PieceManager
  • func NewPieceManager(pieceDownloadTimeout time.Duration, opts ...func(*pieceManager)) (PieceManager, error)
• type ProgressState
• type StreamTask
• type StreamTaskRequest
• type SubscribeResult
• type Task
• type TaskManager
  • func NewPeerTaskManager(host *scheduler.PeerHost, pieceManager PieceManager, ...) (TaskManager, error)
• type TinyData

Functions

func WithCalculateDigest

func WithCalculateDigest(enable bool) func(*pieceManager)

func WithLimiter

func WithLimiter(limiter *rate.Limiter) func(*pieceManager)

WithLimiter sets upload rate limiter, the burst size must be bigger than piece size

func WithTransport

func WithTransport(rt http.RoundTripper) func(*pieceDownloader) error

func WithTransportOption

func WithTransportOption(opt *config.TransportOption) func(*pieceManager)

Types

type Bitmap

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

func NewBitmap

func NewBitmap() *Bitmap

func NewBitmapWithCap

func NewBitmapWithCap(c int32) *Bitmap

func (*Bitmap) Clean

func (b *Bitmap) Clean(i int32)

func (*Bitmap) IsSet

func (b *Bitmap) IsSet(i int32) bool

func (*Bitmap) Set

func (b *Bitmap) Set(i int32)

func (*Bitmap) Sets

func (b *Bitmap) Sets(xs ...int32)

func (*Bitmap) Settled

func (b *Bitmap) Settled() int32

type DownloadPieceRequest

type DownloadPieceRequest struct {
	TaskID     string
	PeerID     string
	DstPid     string
	DstAddr    string
	CalcDigest bool
	// contains filtered or unexported fields
}

type DownloadPieceResult

type DownloadPieceResult struct {
	// Size of piece
	Size int64
	// BeginTime nanosecond
	BeginTime int64
	// FinishTime nanosecond
	FinishTime int64
}

type FileTask

type FileTask interface {
	Start(ctx context.Context) (chan *FileTaskProgress, error)
}

FileTask represents a peer task to download a file

type FileTaskProgress

type FileTaskProgress struct {
	State           *ProgressState
	TaskID          string
	PeerID          string
	ContentLength   int64
	CompletedLength int64
	PeerTaskDone    bool
	DoneCallback    func()
}

type FileTaskRequest

type FileTaskRequest struct {
	scheduler.PeerTaskRequest
	Output             string
	Limit              float64
	DisableBackSource  bool
	Pattern            string
	Callsystem         string
	Range              *clientutil.Range
	KeepOriginalOffset bool
}

type IDGenerator

type IDGenerator interface {
	PeerID() string
}

func NewPeerIDGenerator

func NewPeerIDGenerator(ip string) IDGenerator

type Logger

type Logger interface {
	Log() *logger.SugaredLoggerOnWith
}

type PieceDownloader

type PieceDownloader interface {
	DownloadPiece(context.Context, *DownloadPieceRequest) (io.Reader, io.Closer, error)
}

func NewPieceDownloader

func NewPieceDownloader(timeout time.Duration, opts ...func(*pieceDownloader) error) (PieceDownloader, error)

type PieceInfo

type PieceInfo struct {
	// Num is the current piece num
	Num int32
	// OrderedNum is the max pieces num with ordered, eg: 0 1 2 3 5 7 8, the OrderedNum is 3
	OrderedNum int32
	Finished   bool
}

type PieceManager

type PieceManager interface {
	DownloadSource(ctx context.Context, pt Task, request *scheduler.PeerTaskRequest) error
	DownloadPiece(ctx context.Context, request *DownloadPieceRequest) (*DownloadPieceResult, error)
}

func NewPieceManager

func NewPieceManager(pieceDownloadTimeout time.Duration, opts ...func(*pieceManager)) (PieceManager, error)

type ProgressState

type ProgressState struct {
	Success bool
	Code    base.Code
	Msg     string
}

type StreamTask

type StreamTask interface {
	// Start starts the special peer task, return an io.Reader for stream io
	// when all data transferred, reader return an io.EOF
	// attribute stands some extra data, like HTTP response Header
	Start(ctx context.Context) (rc io.ReadCloser, attribute map[string]string, err error)
}

StreamTask represents a peer task with stream io for reading directly without once more disk io

type StreamTaskRequest

type StreamTaskRequest struct {
	// universal resource locator for different kind of storage
	URL string
	// url meta info
	URLMeta *base.UrlMeta
	// http range
	Range *clientutil.Range
	// peer's id and must be global uniqueness
	PeerID string
}

type SubscribeResult

type SubscribeResult struct {
	Storage          storage.TaskStorageDriver
	PieceInfoChannel chan *PieceInfo
	Success          chan struct{}
	Fail             chan struct{}
}

type Task

type Task interface {
	Logger
	Context() context.Context
	Log() *logger.SugaredLoggerOnWith

	GetStorage() storage.TaskStorageDriver

	GetPeerID() string
	GetTaskID() string

	GetTotalPieces() int32
	SetTotalPieces(int32)

	GetContentLength() int64
	SetContentLength(int64)

	AddTraffic(uint64)
	GetTraffic() uint64

	SetPieceMd5Sign(string)
	GetPieceMd5Sign() string

	PublishPieceInfo(pieceNum int32, size uint32)
	ReportPieceResult(request *DownloadPieceRequest, result *DownloadPieceResult, err error)
}

Task represents common interface to operate a peer task

type TaskManager

type TaskManager interface {
	// StartFileTask starts a peer task to download a file
	// return a progress channel for request download progress
	// tiny stands task file is tiny and task is done
	StartFileTask(ctx context.Context, req *FileTaskRequest) (
		progress chan *FileTaskProgress, tiny *TinyData, err error)
	// StartStreamTask starts a peer task with stream io
	// tiny stands task file is tiny and task is done
	StartStreamTask(ctx context.Context, req *StreamTaskRequest) (
		readCloser io.ReadCloser, attribute map[string]string, err error)

	Subscribe(request *base.PieceTaskRequest) (*SubscribeResult, bool)

	IsPeerTaskRunning(id string) bool

	// Stop stops the PeerTaskManager
	Stop(ctx context.Context) error
}

TaskManager processes all peer tasks request

func NewPeerTaskManager

func NewPeerTaskManager(
	host *scheduler.PeerHost,
	pieceManager PieceManager,
	storageManager storage.Manager,
	schedulerClient schedulerclient.SchedulerClient,
	schedulerOption config.SchedulerOption,
	perPeerRateLimit rate.Limit,
	multiplex bool,
	prefetch bool,
	calculateDigest bool,
	getPiecesMaxRetry int) (TaskManager, error)

type TinyData

type TinyData struct {
	TaskID  string
	PeerID  string
	Content []byte
}