supervisor

Dec 3, 2021

Changes in this version

+ const HostMaxLoad

+ const PeerGCID

+ const TaskGCID

+ var ErrCDNClientUninitialized = errors.New("cdn client is not initialized")

+ var ErrCDNDownloadFail = errors.New("cdn task download failed")

+ var ErrCDNInvokeFail = errors.New("invoke cdn interface failed")

+ var ErrCDNRegisterFail = errors.New("cdn task register failed")

+ var ErrCDNUnknown = errors.New("cdn obtain seed encounter unknown err")

+ var ErrChannelBusy = errors.New("channel busy")

+ var ErrInitCDNPeerFail = errors.New("init cdn peer failed")

+ type CDN interface

+ GetClient func() CDNDynmaicClient

+ StartSeedTask func(context.Context, *Task) (*Peer, error)

+ func NewCDN(client CDNDynmaicClient, peerManager PeerManager, hostManager HostManager) CDN

+ type CDNDynmaicClient interface

+ GetHost func(hostID string) (*Host, bool)

+ func NewCDNDynmaicClient(dynConfig config.DynconfigInterface, opts []grpc.DialOption) (CDNDynmaicClient, error)

type Channel

+ func (c *Channel) Error() error

+ type Host struct

+ CurrentUploadLoad atomic.Uint32

+ DownloadPort int32

+ HostName string

+ IDC string

+ IP string

+ IsCDN bool

+ Location string

+ NetTopology string

+ RPCPort int32

+ SecurityDomain string

+ TotalUploadLoad uint32

+ UUID string

+ func NewCDNHost(uuid, ip, hostname string, rpcPort, downloadPort int32, ...) *Host

+ func NewClientHost(uuid, ip, hostname string, rpcPort, downloadPort int32, ...) *Host

+ func (h *Host) AddPeer(peer *Peer)

+ func (h *Host) DeletePeer(id string)

+ func (h *Host) GetFreeUploadLoad() int32

+ func (h *Host) GetPeer(id string) (*Peer, bool)

+ func (h *Host) GetPeersLen() int

+ func (h *Host) Log() *logger.SugaredLoggerOnWith

+ type HostManager interface

+ Add func(*Host)

+ Delete func(string)

+ Get func(string) (*Host, bool)

+ func NewHostManager() HostManager

+ type HostOption func(rt *Host) *Host

+ func WithNetTopology(n string) HostOption

+ func WithTotalUploadLoad(load uint32) HostOption

type Peer

+ CreateAt *atomic.Time

+ ID string

+ TotalPieceCount atomic.Int32

+ func (peer *Peer) CloseChannelWithError(err error) error

+ func (peer *Peer) GetPieceCosts() []int

+ func (peer *Peer) GetRoot() *Peer

+ func (peer *Peer) GetTreeDepth() int

+ func (peer *Peer) GetTreeNodeCount() int

+ func (peer *Peer) IsAncestor(offspring *Peer) bool

+ func (peer *Peer) IsDescendant(ancestor *Peer) bool

+ func (peer *Peer) Leave()

+ func (peer *Peer) SetParent(parent *Peer)

+ func (peer *Peer) SetPieceCosts(costs ...int)

+ func (peer *Peer) SortedValue() int

+ type PeerManager interface

+ Add func(*Peer)

+ Delete func(string)

+ Get func(string) (*Peer, bool)

+ GetPeers func() *sync.Map

+ GetPeersByTask func(string) []*Peer

+ func NewPeerManager(cfg *config.GCConfig, gcManager gc.GC, hostManager HostManager) (PeerManager, error)

type Task

+ BackToSourceWeight atomic.Int32

+ CreateAt *atomic.Time

+ ID string

+ LastTriggerAt *atomic.Time

+ TotalPieceCount atomic.Int32

+ func (task *Task) AddBackToSourcePeer(peerID string)

+ func (task *Task) CanBackToSource() bool

+ func (task *Task) ContainsBackToSourcePeer(peerID string) bool

+ func (task *Task) GetBackToSourcePeers() []string

+ func (task *Task) GetOrAddPiece(p *base.PieceInfo) (*base.PieceInfo, bool)

+ func (task *Task) GetPeers() list.SortedUniqueList

+ func (task *Task) UpdateSuccess(pieceCount int32, contentLength int64)

+ type TaskManager interface

+ Add func(*Task)

+ Delete func(string)

+ Get func(string) (*Task, bool)

+ GetOrAdd func(*Task) (*Task, bool)

+ func NewTaskManager(cfg *config.GCConfig, gcManager gc.GC, peerManager PeerManager) (TaskManager, error)

Sep 9, 2021

Changes in this version

+ const TinyFileSize

+ func GetDiffPieceNum(dst *Peer, src *Peer) int32

+ type CDNMgr interface

+ StartSeedTask func(ctx context.Context, task *Task) (*Peer, error)

+ type Channel struct

+ func (c *Channel) Close()

+ func (c *Channel) CloseWithError(err error)

+ func (c *Channel) Done() <-chan struct{}

+ func (c *Channel) Err() error

+ func (c *Channel) IsClosed() bool

+ func (c *Channel) Receiver() <-chan *scheduler.PieceResult

+ func (c *Channel) Send(packet *scheduler.PeerPacket) error

+ type HostMgr interface

+ Add func(host *PeerHost)

+ Delete func(uuid string)

+ Get func(uuid string) (*PeerHost, bool)

+ type Peer struct

+ CreateTime time.Time

+ Host *PeerHost

+ PeerID string

+ Task *Task

+ func NewPeer(peerID string, task *Task, host *PeerHost) *Peer

+ func (peer *Peer) BindNewConn(stream scheduler.Scheduler_ReportPieceResultServer) *Channel

+ func (peer *Peer) CloseChannel(err error) error

+ func (peer *Peer) GetChildren() *sync.Map

+ func (peer *Peer) GetCost() int

+ func (peer *Peer) GetCostHistory() []int

+ func (peer *Peer) GetDepth() int

+ func (peer *Peer) GetFinishedNum() int32

+ func (peer *Peer) GetLastAccessTime() time.Time

+ func (peer *Peer) GetParent() *Peer

+ func (peer *Peer) GetSortKeys() (key1, key2 int)

+ func (peer *Peer) GetStatus() PeerStatus

+ func (peer *Peer) GetTreeRoot() *Peer

+ func (peer *Peer) GetWholeTreeNode() int

+ func (peer *Peer) IsAncestorOf(offspring *Peer) bool

+ func (peer *Peer) IsBad() bool

+ func (peer *Peer) IsConnected() bool

+ func (peer *Peer) IsDescendantOf(ancestor *Peer) bool

+ func (peer *Peer) IsDone() bool

+ func (peer *Peer) IsFail() bool

+ func (peer *Peer) IsLeave() bool

+ func (peer *Peer) IsRunning() bool

+ func (peer *Peer) IsSuccess() bool

+ func (peer *Peer) IsWaiting() bool

+ func (peer *Peer) Log() *logger.SugaredLoggerOnWith

+ func (peer *Peer) MarkLeave()

+ func (peer *Peer) ReplaceParent(parent *Peer)

+ func (peer *Peer) SendSchedulePacket(packet *scheduler.PeerPacket) error

+ func (peer *Peer) SetStatus(status PeerStatus)

+ func (peer *Peer) Touch()

+ func (peer *Peer) UpdateProgress(finishedCount int32, cost int)

+ type PeerHost struct

+ CDN bool

+ DownloadPort int32

+ HostName string

+ IDC string

+ IP string

+ Location string

+ NetTopology string

+ RPCPort int32

+ SecurityDomain string

+ TotalUploadLoad int32

+ UUID string

+ func NewCDNPeerHost(uuid, ip, hostname string, rpcPort, downloadPort int32, ...) *PeerHost

+ func NewClientPeerHost(uuid, ip, hostname string, rpcPort, downloadPort int32, ...) *PeerHost

+ func (h *PeerHost) AddPeer(peer *Peer)

+ func (h *PeerHost) DecUploadLoad() int32

+ func (h *PeerHost) DeletePeer(peerID string)

+ func (h *PeerHost) GetCurrentUpload() int32

+ func (h *PeerHost) GetFreeUploadLoad() int

+ func (h *PeerHost) GetPeer(peerID string) (*Peer, bool)

+ func (h *PeerHost) GetPeerTaskNum() int

+ func (h *PeerHost) GetUploadLoadPercent() float64

+ func (h *PeerHost) IncUploadLoad() int32

+ func (h *PeerHost) Log() *logger.SugaredLoggerOnWith

+ type PeerMgr interface

+ Add func(peer *Peer)

+ Delete func(peerID string)

+ Get func(peerID string) (*Peer, bool)

+ ListPeers func() *sync.Map

+ ListPeersByTask func(taskID string) []*Peer

+ type PeerStatus uint8

+ const PeerStatusFail

+ const PeerStatusRunning

+ const PeerStatusSuccess

+ const PeerStatusWaiting

+ const PeerStatusZombie

+ func (status PeerStatus) String() string

+ type Task struct

+ ContentLength int64

+ CreateTime time.Time

+ DirectPiece []byte

+ PieceTotal int32

+ TaskID string

+ URL string

+ URLMeta *base.UrlMeta

+ func NewTask(taskID, url string, meta *base.UrlMeta) *Task

+ func (task *Task) AddPeer(peer *Peer)

+ func (task *Task) AddPiece(p *base.PieceInfo)

+ func (task *Task) CanSchedule() bool

+ func (task *Task) DeletePeer(peer *Peer)

+ func (task *Task) GetBackSourcePeers() []string

+ func (task *Task) GetLastAccessTime() time.Time

+ func (task *Task) GetLastTriggerTime() time.Time

+ func (task *Task) GetPiece(pieceNum int32) *base.PieceInfo

+ func (task *Task) GetStatus() TaskStatus

+ func (task *Task) IncreaseBackSourcePeer(peerID string)

+ func (task *Task) IsBackSourcePeer(peerID string) bool

+ func (task *Task) IsFail() bool

+ func (task *Task) IsHealth() bool

+ func (task *Task) IsSuccess() bool

+ func (task *Task) IsWaiting() bool

+ func (task *Task) ListPeers() *sortedlist.SortedList

+ func (task *Task) Lock()

+ func (task *Task) Log() *logger.SugaredLoggerOnWith

+ func (task *Task) NeedClientBackSource() bool

+ func (task *Task) Pick(limit int, pickFn func(peer *Peer) bool) (pickedPeers []*Peer)

+ func (task *Task) PickReverse(limit int, pickFn func(peer *Peer) bool) (pickedPeers []*Peer)

+ func (task *Task) RLock()

+ func (task *Task) RUnlock()

+ func (task *Task) SetClientBackSourceStatusAndLimit(backSourceLimit int32)

+ func (task *Task) SetStatus(status TaskStatus)

+ func (task *Task) Touch()

+ func (task *Task) UnLock()

+ func (task *Task) UpdateLastTriggerTime(lastTriggerTime time.Time)

+ func (task *Task) UpdatePeer(peer *Peer)

+ func (task *Task) UpdateTaskSuccessResult(pieceTotal int32, contentLength int64)

+ type TaskMgr interface

+ Add func(task *Task)

+ Delete func(taskID string)

+ Get func(taskID string) (task *Task, ok bool)

+ GetOrAdd func(task *Task) (actual *Task, loaded bool)

+ type TaskStatus uint8

+ const TaskStatusFail

+ const TaskStatusRunning

+ const TaskStatusSeeding

+ const TaskStatusSuccess

+ const TaskStatusWaiting

+ const TaskStatusZombie

+ func (status TaskStatus) String() string