func (s *scheduler) ScheduleParent(ctx context.Context, peer *resource.Peer, blocklist set.SafeSet) {
	var n int
	for {
		select {
		case <-ctx.Done():
			peer.Log.Infof("context was done")
			return
		default:
		}

		// If the scheduling exceeds the RetryBackSourceLimit or the latest cdn peer state is PeerStateFailed,
		// peer will download the task back-to-source
		cdnPeer, ok := peer.Task.LoadCDNPeer()
		if (n >= s.config.RetryBackSourceLimit ||
			ok && cdnPeer.FSM.Is(resource.PeerStateFailed)) &&
			peer.Task.CanBackToSource() {
			stream, ok := peer.LoadStream()
			if !ok {
				peer.Log.Error("load stream failed")
				return
			}

			// Notify peer back-to-source
			if err := stream.Send(&rpcscheduler.PeerPacket{Code: base.Code_SchedNeedBackSource}); err != nil {
				peer.Log.Errorf("send packet failed: %v", err)
				return
			}
			peer.Log.Infof("peer scheduling %d times and back-to-source limit %d times, cdn peer is %#v, return code %d",
				n, s.config.RetryBackSourceLimit, cdnPeer, base.Code_SchedNeedBackSource)

			if err := peer.FSM.Event(resource.PeerEventDownloadFromBackToSource); err != nil {
				peer.Log.Errorf("peer fsm event failed: %v", err)
				return
			}

			// If the task state is TaskStateFailed,
			// peer back-to-source and reset task state to TaskStateRunning
			if peer.Task.FSM.Is(resource.TaskStateFailed) {
				if err := peer.Task.FSM.Event(resource.TaskEventDownload); err != nil {
					peer.Task.Log.Errorf("task fsm event failed: %v", err)
					return
				}
			}

			// If the peer downloads back-to-source, its parent needs to be deleted
			peer.DeleteParent()
			peer.Task.Log.Info("peer back to source successfully")
			return
		}

		// Handle peer schedule failed
		if n >= s.config.RetryLimit {
			stream, ok := peer.LoadStream()
			if !ok {
				peer.Log.Error("load stream failed")
				return
			}

			// Notify peer schedule failed
			if err := stream.Send(&rpcscheduler.PeerPacket{Code: base.Code_SchedTaskStatusError}); err != nil {
				peer.Log.Errorf("send packet failed: %v", err)
				return
			}
			peer.Log.Infof("peer scheduling exceeds the limit %d times and return code %d", s.config.RetryLimit, base.Code_SchedTaskStatusError)
			return
		}

		if _, ok := s.NotifyAndFindParent(ctx, peer, blocklist); !ok {
			n++
			peer.Log.Infof("schedule parent %d times failed", n)

			// Sleep to avoid hot looping
			time.Sleep(s.config.RetryInterval)
			continue
		}

		peer.Log.Infof("schedule parent %d times successfully", n+1)
		return
	}
}
// ScheduleParent schedule a parent and candidates to a peer
ScheduleParent schedule a parent and candidates to a peer
// ScheduleParent schedule a parent and candidates to a peer
func (s *scheduler) ScheduleParent(ctx context.Context, peer *resource.Peer, blocklist set.SafeSet) {
	var n int
	for {
		select {
		case <-ctx.Done():
			peer.Log.Infof("context was done")
			return
		default:
		}
		cdnPeer, ok := peer.Task.LoadCDNPeer()
		if (n >= s.config.RetryBackSourceLimit ||
			ok && cdnPeer.FSM.Is(resource.PeerStateFailed)) &&
			peer.Task.CanBackToSource() {
			stream, ok := peer.LoadStream()
			if !ok {
				peer.Log.Error("load stream failed")
				return
			}
			if err := stream.Send(&rpcscheduler.PeerPacket{Code: base.Code_SchedNeedBackSource}); err != nil {
				peer.Log.Errorf("send packet failed: %v", err)
				return
			}
			peer.Log.Infof("peer scheduling %d times and back-to-source limit %d times, cdn peer is %#v, return code %d",
				n, s.config.RetryBackSourceLimit, cdnPeer, base.Code_SchedNeedBackSource)
			if err := peer.FSM.Event(resource.PeerEventDownloadFromBackToSource); err != nil {
				peer.Log.Errorf("peer fsm event failed: %v", err)
				return
			}
			if peer.Task.FSM.Is(resource.TaskStateFailed) {
				if err := peer.Task.FSM.Event(resource.TaskEventDownload); err != nil {
					peer.Task.Log.Errorf("task fsm event failed: %v", err)
					return
				}
			}
			peer.DeleteParent()
			peer.Task.Log.Info("peer back to source successfully")
			return
		}
		if n >= s.config.RetryLimit {
			stream, ok := peer.LoadStream()
			if !ok {
				peer.Log.Error("load stream failed")
				return
			}
			if err := stream.Send(&rpcscheduler.PeerPacket{Code: base.Code_SchedTaskStatusError}); err != nil {
				peer.Log.Errorf("send packet failed: %v", err)
				return
			}
			peer.Log.Infof("peer scheduling exceeds the limit %d times and return code %d", s.config.RetryLimit, base.Code_SchedTaskStatusError)
			return
		}
		if _, ok := s.NotifyAndFindParent(ctx, peer, blocklist); !ok {
			n++
			peer.Log.Infof("schedule parent %d times failed", n)
			time.Sleep(s.config.RetryInterval)
			continue
		}
		peer.Log.Infof("schedule parent %d times successfully", n+1)
		return
	}
}
func ( s * scheduler ) ScheduleParent ( ctx context . Context , peer * resource . Peer , blocklist set . SafeSet ) { var n int for { select { case <- ctx . Done ( ) : peer . Log . Infof ( " " ) return default : } cdnPeer , ok := peer . Task . LoadCDNPeer ( ) if ( n >= s . config . RetryBackSourceLimit || ok && cdnPeer . FSM . Is ( resource . PeerStateFailed ) ) && peer . Task . CanBackToSource ( ) { stream , ok := peer . LoadStream ( ) if ! ok { peer . Log . Error ( " " ) return } if err := stream . Send ( & rpcscheduler . PeerPacket { Code : base . Code_SchedNeedBackSource } ) ; err != nil { peer . Log . Errorf ( " " , err ) return } peer . Log . Infof ( " " , n , s . config . RetryBackSourceLimit , cdnPeer , base . Code_SchedNeedBackSource ) if err := peer . FSM . Event ( resource . PeerEventDownloadFromBackToSource ) ; err != nil { peer . Log . Errorf ( " " , err ) return } if peer . Task . FSM . Is ( resource . TaskStateFailed ) { if err := peer . Task . FSM . Event ( resource . TaskEventDownload ) ; err != nil { peer . Task . Log . Errorf ( " " , err ) return } } peer . DeleteParent ( ) peer . Task . Log . Info ( " " ) return } if n >= s . config . RetryLimit { stream , ok := peer . LoadStream ( ) if ! ok { peer . Log . Error ( " " ) return } if err := stream . Send ( & rpcscheduler . PeerPacket { Code : base . Code_SchedTaskStatusError } ) ; err != nil { peer . Log . Errorf ( " " , err ) return } peer . Log . Infof ( " " , s . config . RetryLimit , base . Code_SchedTaskStatusError ) return } if _ , ok := s . NotifyAndFindParent ( ctx , peer , blocklist ) ; ! ok { n ++ peer . Log . Infof ( " " , n ) time . Sleep ( s . config . RetryInterval ) continue } peer . Log . Infof ( " " , n + 1 ) return } }
ScheduleParent schedule a parent and candidates to a peer
ScheduleParent schedule a parent and candidates to a peer
// If the scheduling exceeds the RetryBackSourceLimit or the latest cdn peer state is PeerStateFailed, // peer will download the task back-to-source // Notify peer back-to-source // If the task state is TaskStateFailed, // peer back-to-source and reset task state to TaskStateRunning // If the peer downloads back-to-source, its parent needs to be deleted // Handle peer schedule failed // Notify peer schedule failed // Sleep to avoid hot looping