Merge pull request #620 from lowzj/increase-load

bugfix: increase load of the peer serving for others

Author: starnop

supernode/daemon/mgr/progress/progress_manager.go

@@ -254,10 +254,10 @@ func (pm *Manager) GetPeerStateByPeerID(ctx context.Context, peerID string) (*mg
 
 	return &mgr.PeerState{
 		PeerID:            peerID,
-		ServiceDownTime:   peerState.serviceDownTime,
-		ClientErrorCount:  peerState.clientErrorCount.Get(),
-		ServiceErrorCount: peerState.serviceErrorCount.Get(),
-		ProducerLoad:      peerState.producerLoad.Get(),
+		ServiceDownTime:   &peerState.serviceDownTime,
+		ClientErrorCount:  peerState.clientErrorCount,
+		ServiceErrorCount: peerState.serviceErrorCount,
+		ProducerLoad:      peerState.producerLoad,
 	}, nil
 }
 

supernode/daemon/mgr/progress/progress_util.go

@@ -123,6 +123,9 @@ func (pm *Manager) updatePeerProgress(taskID, srcPID, dstPID string, pieceNum, p
 			return err
 		}
 		if err == nil {
+			if dstPeerState.producerLoad == nil {
+				dstPeerState.producerLoad = cutil.NewAtomicInt(0)
+			}
 			updateProducerLoad(dstPeerState.producerLoad, taskID, dstPID, pieceNum, pieceStatus)
 		}
 	}
@@ -214,12 +217,6 @@ func processPeerFailInfo(srcPeerState, dstPeerState *peerState) {
 // updateProducerLoad update the load of the clientID.
 // TODO: avoid multiple calls
 func updateProducerLoad(load *cutil.AtomicInt, taskID, peerID string, pieceNum, pieceStatus int) {
-	if load == nil {
-		logrus.Warnf("client load maybe not initialized,taskID: %s,peerID: %s,pieceNum: %d",
-			taskID, peerID, pieceNum)
-		load = cutil.NewAtomicInt(0)
-	}
-
 	// increase the load of peerID when pieceStatus equals PieceRUNNING
 	if pieceStatus == config.PieceRUNNING {
 		load.Add(1)

supernode/daemon/mgr/progress_mgr.go

@@ -13,16 +13,16 @@ type PeerState struct {
 	PeerID string
 
 	// ProducerLoad is the load of download services provided by the current node.
-	ProducerLoad int32
+	ProducerLoad *cutil.AtomicInt
 
 	// ClientErrorCount maintains the number of times that PeerID failed to downloaded from the other peer nodes.
-	ClientErrorCount int32
+	ClientErrorCount *cutil.AtomicInt
 
 	// ServiceErrorCount maintains the number of times that the other peer nodes failed to downloaded from the PeerID.
-	ServiceErrorCount int32
+	ServiceErrorCount *cutil.AtomicInt
 
 	// ServiceDownTime the down time of the peer service.
-	ServiceDownTime int64
+	ServiceDownTime *int64
 }
 
 // ProgressMgr is responsible for maintaining the correspondence between peer and pieces.

supernode/daemon/mgr/scheduler/manager.go

@@ -131,8 +131,9 @@ func (sm *Manager) getPieceResults(ctx context.Context, taskID, clientID, peerID
 	if err != nil {
 		return nil, err
 	}
-	if srcPeerState.ClientErrorCount > config.FailCountLimit {
-		logrus.Warnf("peerID: %s got errors for %d times which reaches error limit: %d for taskID(%s)", peerID, srcPeerState.ClientErrorCount, config.FailCountLimit, taskID)
+	if srcPeerState.ClientErrorCount.Get() > config.FailCountLimit {
+		logrus.Warnf("peerID: %s got errors for %d times which reaches error limit: %d for taskID(%s)",
+			peerID, srcPeerState.ClientErrorCount.Get(), config.FailCountLimit, taskID)
 		useSupernode = true
 	}
 
@@ -191,13 +192,13 @@ func (sm *Manager) tryGetPID(ctx context.Context, taskID string, pieceNum int, p
 		}
 
 		// if the service has been down, and then it should not be needed.
-		if peerState.ServiceDownTime > 0 {
+		if peerState.ServiceDownTime != nil && *(peerState.ServiceDownTime) > 0 {
 			sm.deletePeerIDByPieceNum(ctx, taskID, pieceNum, peerIDs[i])
 			continue
 		}
 
 		// if service has failed for EliminationLimit times, and then it should not be needed.
-		if peerState.ServiceErrorCount >= config.EliminationLimit {
+		if peerState.ServiceErrorCount != nil && peerState.ServiceErrorCount.Get() >= config.EliminationLimit {
 			sm.deletePeerIDByPieceNum(ctx, taskID, pieceNum, peerIDs[i])
 			continue
 		}
@@ -212,8 +213,11 @@ func (sm *Manager) tryGetPID(ctx context.Context, taskID string, pieceNum int, p
 			continue
 		}
 
-		if peerState.ProducerLoad < config.PeerUpLimit {
-			return peerIDs[i]
+		if peerState.ProducerLoad != nil {
+			if peerState.ProducerLoad.Add(1) <= config.PeerUpLimit {
+				return peerIDs[i]
+			}
+			peerState.ProducerLoad.Add(-1)
 		}
 	}
 	return