func (s *Service) ReportPieceResult(stream rpcscheduler.Scheduler_ReportPieceResultServer) error {
	ctx := stream.Context()
	var (
		peer        *resource.Peer
		initialized bool
		ok          bool
	)

	for {
		select {
		case <-ctx.Done():
			logger.Infof("context was done")
			return ctx.Err()
		default:
		}

		piece, err := stream.Recv()
		if err != nil {
			if err == io.EOF {
				return nil
			}
			logger.Errorf("receive piece %#v error: %v", piece, err)
			return err
		}

		if !initialized {
			initialized = true

			// Get peer from peer manager
			peer, ok = s.resource.PeerManager().Load(piece.SrcPid)
			if !ok {
				dferr := dferrors.Newf(base.Code_SchedPeerNotFound, "peer %s not found", piece.SrcPid)
				logger.Errorf("peer %s not found", piece.SrcPid)
				return dferr
			}

			// Peer setting stream
			peer.StoreStream(stream)
			defer peer.DeleteStream()
		}

		peer.Log.Infof("receive piece: %#v %#v", piece, piece.PieceInfo)

		if piece.PieceInfo != nil {
			// Handle begin of piece
			if piece.PieceInfo.PieceNum == common.BeginOfPiece {
				s.handleBeginOfPiece(ctx, peer)
				continue
			}

			// Handle end of piece
			if piece.PieceInfo.PieceNum == common.EndOfPiece {
				s.handleEndOfPiece(ctx, peer)
				continue
			}
		}

		// Handle piece download successfully
		if piece.Success {
			s.handlePieceSuccess(ctx, peer, piece)

			// Collect peer host traffic metrics
			if s.config.Metrics != nil && s.config.Metrics.EnablePeerHost {
				metrics.PeerHostTraffic.WithLabelValues("download", peer.Host.ID, peer.Host.IP).Add(float64(piece.PieceInfo.RangeSize))
				if parent, ok := s.resource.PeerManager().Load(piece.DstPid); ok {
					metrics.PeerHostTraffic.WithLabelValues("upload", parent.Host.ID, parent.Host.IP).Add(float64(piece.PieceInfo.RangeSize))
				} else {
					peer.Log.Warnf("dst peer %s not found for piece %#v %#v", piece.DstPid, piece, piece.PieceInfo)
				}
			}
			continue
		}

		// Handle piece download code
		if piece.Code != base.Code_Success {
			// FIXME(244372610) When dfdaemon download peer return empty, retry later.
			if piece.Code == base.Code_ClientWaitPieceReady {
				peer.Log.Infof("receive piece code %d and wait for dfdaemon piece ready", piece.Code)
				continue
			}

			// Handle piece download failed
			peer.Log.Errorf("receive failed piece: %#v %#v", piece, piece.PieceInfo)
			s.handlePieceFail(ctx, peer, piece)
		}
	}
}
// ReportPieceResult handles the piece information reported by dfdaemon
ReportPieceResult handles the piece information reported by dfdaemon
// ReportPieceResult handles the piece information reported by dfdaemon
func (s *Service) ReportPieceResult(stream rpcscheduler.Scheduler_ReportPieceResultServer) error {
	ctx := stream.Context()
	var (
		peer        *resource.Peer
		initialized bool
		ok          bool
	)
	for {
		select {
		case <-ctx.Done():
			logger.Infof("context was done")
			return ctx.Err()
		default:
		}
		piece, err := stream.Recv()
		if err != nil {
			if err == io.EOF {
				return nil
			}
			logger.Errorf("receive piece %#v error: %v", piece, err)
			return err
		}
		if !initialized {
			initialized = true
			peer, ok = s.resource.PeerManager().Load(piece.SrcPid)
			if !ok {
				dferr := dferrors.Newf(base.Code_SchedPeerNotFound, "peer %s not found", piece.SrcPid)
				logger.Errorf("peer %s not found", piece.SrcPid)
				return dferr
			}
			peer.StoreStream(stream)
			defer peer.DeleteStream()
		}
		peer.Log.Infof("receive piece: %#v %#v", piece, piece.PieceInfo)
		if piece.PieceInfo != nil {
			if piece.PieceInfo.PieceNum == common.BeginOfPiece {
				s.handleBeginOfPiece(ctx, peer)
				continue
			}
			if piece.PieceInfo.PieceNum == common.EndOfPiece {
				s.handleEndOfPiece(ctx, peer)
				continue
			}
		}
		if piece.Success {
			s.handlePieceSuccess(ctx, peer, piece)
			if s.config.Metrics != nil && s.config.Metrics.EnablePeerHost {
				metrics.PeerHostTraffic.WithLabelValues("download", peer.Host.ID, peer.Host.IP).Add(float64(piece.PieceInfo.RangeSize))
				if parent, ok := s.resource.PeerManager().Load(piece.DstPid); ok {
					metrics.PeerHostTraffic.WithLabelValues("upload", parent.Host.ID, parent.Host.IP).Add(float64(piece.PieceInfo.RangeSize))
				} else {
					peer.Log.Warnf("dst peer %s not found for piece %#v %#v", piece.DstPid, piece, piece.PieceInfo)
				}
			}
			continue
		}
		if piece.Code != base.Code_Success {
			if piece.Code == base.Code_ClientWaitPieceReady {
				peer.Log.Infof("receive piece code %d and wait for dfdaemon piece ready", piece.Code)
				continue
			}
			peer.Log.Errorf("receive failed piece: %#v %#v", piece, piece.PieceInfo)
			s.handlePieceFail(ctx, peer, piece)
		}
	}
}
func ( s * Service ) ReportPieceResult ( stream rpcscheduler . Scheduler_ReportPieceResultServer ) error { ctx := stream . Context ( ) var ( peer * resource . Peer initialized bool ok bool ) for { select { case <- ctx . Done ( ) : logger . Infof ( " " ) return ctx . Err ( ) default : } piece , err := stream . Recv ( ) if err != nil { if err == io . EOF { return nil } logger . Errorf ( " " , piece , err ) return err } if ! initialized { initialized = true peer , ok = s . resource . PeerManager ( ) . Load ( piece . SrcPid ) if ! ok { dferr := dferrors . Newf ( base . Code_SchedPeerNotFound , " " , piece . SrcPid ) logger . Errorf ( " " , piece . SrcPid ) return dferr } peer . StoreStream ( stream ) defer peer . DeleteStream ( ) } peer . Log . Infof ( " " , piece , piece . PieceInfo ) if piece . PieceInfo != nil { if piece . PieceInfo . PieceNum == common . BeginOfPiece { s . handleBeginOfPiece ( ctx , peer ) continue } if piece . PieceInfo . PieceNum == common . EndOfPiece { s . handleEndOfPiece ( ctx , peer ) continue } } if piece . Success { s . handlePieceSuccess ( ctx , peer , piece ) if s . config . Metrics != nil && s . config . Metrics . EnablePeerHost { metrics . PeerHostTraffic . WithLabelValues ( " " , peer . Host . ID , peer . Host . IP ) . Add ( float64 ( piece . PieceInfo . RangeSize ) ) if parent , ok := s . resource . PeerManager ( ) . Load ( piece . DstPid ) ; ok { metrics . PeerHostTraffic . WithLabelValues ( " " , parent . Host . ID , parent . Host . IP ) . Add ( float64 ( piece . PieceInfo . RangeSize ) ) } else { peer . Log . Warnf ( " " , piece . DstPid , piece , piece . PieceInfo ) } } continue } if piece . Code != base . Code_Success { if piece . Code == base . Code_ClientWaitPieceReady { peer . Log . Infof ( " " , piece . Code ) continue } peer . Log . Errorf ( " " , piece , piece . PieceInfo ) s . handlePieceFail ( ctx , peer , piece ) } } }
ReportPieceResult handles the piece information reported by dfdaemon
ReportPieceResult handles the piece information reported by dfdaemon
// Get peer from peer manager // Peer setting stream // Handle begin of piece // Handle end of piece // Handle piece download successfully // Collect peer host traffic metrics // Handle piece download code // FIXME(244372610) When dfdaemon download peer return empty, retry later. // Handle piece download failed