mgr

Documentation

Index

• Variables
• func Register(name config.CDNPattern, builder CDNBuilder)
• type CDNBuilder
• type CDNMgr
  • func GetCDNManager(cfg *config.Config, cacheStore *store.Store, progressManager ProgressMgr, ...) (CDNMgr, error)
• type DfgetTaskMgr
• type GCMgr
• type PeerMgr
• type PeerState
• type PieceErrorMgr
• type PieceResult
• type ProgressMgr
• type SchedulerMgr
• type TaskMgr

Variables

var PieceStatusMap = map[string]int{
	types.PieceUpdateRequestPieceStatusFAILED:  config.PieceFAILED,
	types.PieceUpdateRequestPieceStatusSEMISUC: config.PieceSEMISUC,
	types.PieceUpdateRequestPieceStatusSUCCESS: config.PieceSUCCESS,
}

PieceStatusMap maintains the mapping relationship between PieceUpdateRequestResult and PieceStatus code.

Functions

func Register

func Register(name config.CDNPattern, builder CDNBuilder)

Types

type CDNBuilder

type CDNBuilder func(cfg *config.Config, cacheStore *store.Store, progressManager ProgressMgr,
	originClient httpclient.OriginHTTPClient, register prometheus.Registerer) (CDNMgr, error)

type CDNMgr

type CDNMgr interface {
	// TriggerCDN will trigger CDN to download the file from sourceUrl.
	// It includes the following steps:
	// 1). download the source file
	// 2). write the file to disk
	//
	// In fact, it's a very time consuming operation.
	// So if not necessary, it should usually be executed concurrently.
	// In addition, it's not thread-safe.
	TriggerCDN(ctx context.Context, taskInfo *types.TaskInfo) (*types.TaskInfo, error)

	// GetHTTPPath returns the http download path of taskID.
	GetHTTPPath(ctx context.Context, taskInfo *types.TaskInfo) (path string, err error)

	// GetStatus gets the status of the file.
	GetStatus(ctx context.Context, taskID string) (cdnStatus string, err error)

	// GetGCTaskIDs returns the taskIDs that should exec GC operations as a string slice.
	//
	// It should return nil when the free disk of cdn storage is lager than config.YoungGCThreshold.
	// It should return all taskIDs that are not running when the free disk of cdn storage is less than config.FullGCThreshold.
	GetGCTaskIDs(ctx context.Context, taskMgr TaskMgr) ([]string, error)

	// GetPieceMD5 gets the piece Md5 accorrding to the specified taskID and pieceNum.
	GetPieceMD5(ctx context.Context, taskID string, pieceNum int, pieceRange, source string) (pieceMd5 string, err error)

	// CheckFile checks the file whether exists.
	CheckFile(ctx context.Context, taskID string) bool

	// Delete the cdn meta with specified taskID.
	// The file on the disk will be deleted when the force is true.
	Delete(ctx context.Context, taskID string, force bool) error
}

CDNMgr as an interface defines all operations against CDN and operates on the underlying files stored on the local disk, etc.

func GetCDNManager

func GetCDNManager(cfg *config.Config, cacheStore *store.Store, progressManager ProgressMgr,
	originClient httpclient.OriginHTTPClient, register prometheus.Registerer) (CDNMgr, error)

type DfgetTaskMgr

type DfgetTaskMgr interface {
	// Add a new dfgetTask, we use clientID and taskID to identify a dfgetTask uniquely.
	// ClientID should be generated by dfget, supernode will use it directly.
	// NOTE: We should create a new dfgetTask for each download process,
	//       even if the downloads initiated by the same machine.
	Add(ctx context.Context, dfgetTask *types.DfGetTask) error

	// Get a dfgetTask info with specified clientID and taskID.
	Get(ctx context.Context, clientID, taskID string) (dfgetTask *types.DfGetTask, err error)

	// GetCIDByPeerIDAndTaskID returns cid with specified peerID and taskID.
	GetCIDByPeerIDAndTaskID(ctx context.Context, peerID, taskID string) (string, error)

	// GetCIDsByTaskID returns cids as a string slice with specified taskID.
	GetCIDsByTaskID(ctx context.Context, taskID string) ([]string, error)

	// GetCIDAndTaskIDsByPeerID returns a cid<->taskID map by specified peerID.
	GetCIDAndTaskIDsByPeerID(ctx context.Context, peerID string) (map[string]string, error)

	// List returns the list of dfgetTask.
	List(ctx context.Context, filter map[string]string) (dfgetTaskList []*types.DfGetTask, err error)

	// Delete a dfgetTask with clientID and taskID.
	Delete(ctx context.Context, clientID, taskID string) error

	// UpdateStatus updates the status of dfgetTask with specified clientID and taskID.
	// Supernode will update the status of dfgetTask in the following situations:
	// 1. after init the dfgetTask
	// 2. when success/fail to download some pieces
	// 3. when the entire download process ends in success or failure
	UpdateStatus(ctx context.Context, clientID, taskID, status string) error
}

DfgetTaskMgr as an interface defines all operations against DfgetTask. A DfgetTask represents a download process initiated by dfget or other clients.

type GCMgr

type GCMgr interface {
	// StartGC starts to execute GC with a new goroutine.
	StartGC(ctx context.Context)

	// GCTask is used to do the gc task job with specified taskID.
	// The CDN file will be deleted when the full is true.
	GCTask(ctx context.Context, taskID string, full bool)

	// GCPeer is used to do the gc peer job when a peer offline.
	GCPeer(ctx context.Context, peerID string)
}

GCMgr as an interface defines all operations about gc operation.

type PeerMgr

type PeerMgr interface {
	// Register a peer with specified peerInfo.
	// Supernode will generate a unique peerID for every Peer with PeerInfo provided.
	Register(ctx context.Context, peerCreateRequest *types.PeerCreateRequest) (peerCreateResponse *types.PeerCreateResponse, err error)

	// DeRegister offline a peer service and
	// NOTE: update the info related for scheduler.
	DeRegister(ctx context.Context, peerID string) error

	// Get the peer Info with specified peerID.
	Get(ctx context.Context, peerID string) (*types.PeerInfo, error)

	// GetAllPeerIDs returns all peerIDs.
	GetAllPeerIDs(ctx context.Context) (peerIDs []string)

	// List returns a list of peers info with filter.
	List(ctx context.Context, filter *util.PageFilter) (peerList []*types.PeerInfo, err error)
}

PeerMgr as an interface defines all operations against Peer. A Peer represents a web server that provides file downloads for others.

type PeerState

type PeerState struct {
	// PeerID identifies a peer uniquely.
	PeerID string

	// ProducerLoad is the load of download services provided by the current node.
	ProducerLoad *atomiccount.AtomicInt

	// ClientErrorCount maintains the number of times that PeerID failed to downloaded from the other peer nodes.
	ClientErrorCount *atomiccount.AtomicInt

	// ServiceErrorCount maintains the number of times that the other peer nodes failed to downloaded from the PeerID.
	ServiceErrorCount *atomiccount.AtomicInt

	// ServiceDownTime the down time of the peer service.
	ServiceDownTime int64
}

PeerState maintains peer related information.

type PieceErrorMgr

type PieceErrorMgr interface {
	// StartHandleError starts a goroutine to handle the piece error.
	StartHandleError(ctx context.Context)

	// HandlePieceError the peer should report the error with related info when
	// it failed to download a piece from supernode.
	// And the supernode should handle the piece Error and do some repair operations.
	HandlePieceError(ctx context.Context, pieceErrorRequest *types.PieceErrorRequest) error
}

PieceErrorMgr as an interface defines all operations to handle piece errors.

type PieceResult

type PieceResult struct {
	TaskID   string
	PieceNum int
	DstPID   string
}

PieceResult contains the information about which piece to download from which node.

type ProgressMgr

type ProgressMgr interface {
	// InitProgress inits the correlation information between peers and pieces, etc.
	InitProgress(ctx context.Context, taskID, peerID, clientID string) error

	// UpdateProgress updates the correlation information between peers and pieces.
	// 1. update the info about srcCID to tell the scheduler that corresponding peer has the piece now.
	// 2. update the info about dstPID to tell the scheduler that someone has downloaded the piece form here.
	// Scheduler will calculate the load and times of error/success for every peer to make better decisions.
	UpdateProgress(ctx context.Context, taskID, srcCID, srcPID, dstPID string, pieceNum, pieceStatus int) error

	// UpdateClientProgress updates the info when success to schedule peer srcCID to download from dstPID.
	UpdateClientProgress(ctx context.Context, taskID, srcCID, dstPID string, pieceNum, pieceStatus int) error

	// GetPieceProgressByCID gets all pieces progress with specified clientID.
	// The filter parameter depends on the specific implementation.
	GetPieceProgressByCID(ctx context.Context, taskID, clientID, filter string) (pieceNums []int, err error)

	// GetPeerIDsByPieceNum gets all peerIDs with specified taskID and pieceNum.
	GetPeerIDsByPieceNum(ctx context.Context, taskID string, pieceNum int) (peerIDs []string, err error)

	// DeletePeerIDByPieceNum deletes the peerID which means that
	// the peer no longer provides the service for the pieceNum of taskID.
	DeletePeerIDByPieceNum(ctx context.Context, taskID string, pieceNum int, peerID string) error

	// GetPeerStateByPeerID gets peer state with specified peerID.
	GetPeerStateByPeerID(ctx context.Context, peerID string) (peerState *PeerState, err error)

	// UpdateSuperLoad updates the superload of taskID by adding the delta.
	// The updated will be `false` if failed to do update operation.
	//
	// It's considered as a failure when then superload is greater than limit after adding delta.
	UpdatePeerServiceDown(ctx context.Context, peerID string) (err error)

	// GetPeersByTaskID gets all peers info with specified taskID.
	GetPeersByTaskID(ctx context.Context, taskID string) (peersInfo []*types.PeerInfo, err error)

	// GetBlackInfoByPeerID gets black info with specified peerID.
	GetBlackInfoByPeerID(ctx context.Context, peerID string) (dstPIDMap *syncmap.SyncMap, err error)

	// UpdateSuperLoad updates the superLoad with delta.
	//
	// The value will be rolled back if it exceeds the limit after updated and returns false.
	UpdateSuperLoad(ctx context.Context, taskID string, delta, limit int32) (updated bool, err error)

	// DeleteTaskID deletes the super progress with specified taskID.
	DeleteTaskID(ctx context.Context, taskID string, pieceTotal int) (err error)

	// DeleteCID deletes the super progress with specified clientID.
	DeleteCID(ctx context.Context, clientID string) (err error)

	// DeletePeerID deletes the peerState by PeerID.
	DeletePeerID(ctx context.Context, peerID string) (err error)
}

ProgressMgr is responsible for maintaining the correspondence between peer and pieces.

type SchedulerMgr

type SchedulerMgr interface {
	// Schedule gets scheduler result with specified taskID, clientID and peerID through some rules.
	Schedule(ctx context.Context, taskID, clientID, peerID string) ([]*PieceResult, error)
}

SchedulerMgr is responsible for calculating scheduling results according to certain rules.

type TaskMgr

type TaskMgr interface {
	// Register a task represents that someone wants to download a file.
	// Supernode will get the task file meta and return taskID.
	// NOTE: If supernode cannot find the task file, the CDN download will be triggered.
	Register(ctx context.Context, taskCreateRequest *types.TaskCreateRequest) (taskCreateResponse *types.TaskCreateResponse, err error)

	// Get the task Info with specified taskID.
	Get(ctx context.Context, taskID string) (*types.TaskInfo, error)

	// GetAccessTime gets all task accessTime.
	GetAccessTime(ctx context.Context) (*syncmap.SyncMap, error)

	// List returns the list tasks with filter.
	List(ctx context.Context, filter map[string]string) ([]*types.TaskInfo, error)

	// CheckTaskStatus checks whether the taskID corresponding file exists.
	CheckTaskStatus(ctx context.Context, taskID string) (bool, error)

	// Delete deletes a task.
	Delete(ctx context.Context, taskID string) error

	// Update updates the task info with specified info.
	// In common, there are several situations that we will use this method:
	// 1. when finished to download, update task status.
	// 2. for operation usage.
	// TODO: define a struct of TaskUpdateRequest?
	Update(ctx context.Context, taskID string, taskInfo *types.TaskInfo) error

	// GetPieces gets the pieces to be downloaded based on the scheduling result,
	// just like this: which pieces can be downloaded from which peers.
	GetPieces(ctx context.Context, taskID, clientID string, piecePullRequest *types.PiecePullRequest) (isFinished bool, data interface{}, err error)

	// UpdatePieceStatus updates the piece status with specified parameters.
	// A task file is divided into several pieces logically.
	// We use a sting called pieceRange to identify a piece.
	// A pieceRange is separated by a dash, like this: 0-45565, etc.
	UpdatePieceStatus(ctx context.Context, taskID, pieceRange string, pieceUpdateRequest *types.PieceUpdateRequest) error
}

TaskMgr as an interface defines all operations against Task. A Task will store some meta info about the taskFile, pieces and something else. A Task has a one-to-one correspondence with a file on the disk which is identified by taskID.