Merge branch 'master' into split_reminder_db

dapr-bot · web-flow · commit 0ad750543ab7 · 2021-07-12T11:15:36.000-07:00
diff --git a/pkg/placement/membership.go b/pkg/placement/membership.go
@@ -224,6 +224,10 @@ func (p *Service) processRaftStateCommand(stopCh chan struct{}) {
 				// until the state is consistent.
 				logApplyConcurrency <- struct{}{}
 				go func() {
+					// We lock dissemination to ensure the updates can complete before the table is disseminated.
+					p.disseminateLock.Lock()
+					defer p.disseminateLock.Unlock()
+
 					updated, raftErr := p.raftNode.ApplyCommand(op.cmdType, op.host)
 					if raftErr != nil {
 						log.Errorf("fail to apply command: %v", raftErr)
@@ -247,40 +251,44 @@ func (p *Service) processRaftStateCommand(stopCh chan struct{}) {
 			case raft.TableDisseminate:
 				// TableDisseminate will be triggered by disseminateTimer.
 				// This disseminates the latest consistent hashing tables to Dapr runtime.
-				nStreamConnPool := len(p.streamConnPool)
-				nTargetConns := len(p.raftNode.FSM().State().Members)
-
-				monitoring.RecordRuntimesCount(nStreamConnPool)
-				monitoring.RecordActorRuntimesCount(nTargetConns)
-
-				// ignore dissemination if there is no member update.
-				if cnt := p.memberUpdateCount.Load(); cnt > 0 {
-					state := p.raftNode.FSM().PlacementState()
-					log.Infof(
-						"Start disseminating tables. memberUpdateCount: %d, streams: %d, targets: %d, table generation: %s",
-						cnt, nStreamConnPool, nTargetConns, state.Version)
-					p.performTablesUpdate(p.streamConnPool, state)
-					log.Infof(
-						"Completed dissemination. memberUpdateCount: %d, streams: %d, targets: %d, table generation: %s",
-						cnt, nStreamConnPool, nTargetConns, state.Version)
-					p.memberUpdateCount.Store(0)
-
-					// set faultyHostDetectDuration to the default duration.
-					p.faultyHostDetectDuration = faultyHostDetectDefaultDuration
-				}
+				p.performTableDissemination()
 			}
 		}
 	}
 }
 
+func (p *Service) performTableDissemination() {
+	nStreamConnPool := len(p.streamConnPool)
+	nTargetConns := len(p.raftNode.FSM().State().Members)
+
+	monitoring.RecordRuntimesCount(nStreamConnPool)
+	monitoring.RecordActorRuntimesCount(nTargetConns)
+
+	// ignore dissemination if there is no member update.
+	if cnt := p.memberUpdateCount.Load(); cnt > 0 {
+		p.disseminateLock.Lock()
+		defer p.disseminateLock.Unlock()
+
+		state := p.raftNode.FSM().PlacementState()
+		log.Infof(
+			"Start disseminating tables. memberUpdateCount: %d, streams: %d, targets: %d, table generation: %s",
+			cnt, nStreamConnPool, nTargetConns, state.Version)
+		p.performTablesUpdate(p.streamConnPool, state)
+		log.Infof(
+			"Completed dissemination. memberUpdateCount: %d, streams: %d, targets: %d, table generation: %s",
+			cnt, nStreamConnPool, nTargetConns, state.Version)
+		p.memberUpdateCount.Store(0)
+
+		// set faultyHostDetectDuration to the default duration.
+		p.faultyHostDetectDuration = faultyHostDetectDefaultDuration
+	}
+}
+
 // performTablesUpdate updates the connected dapr runtimes using a 3 stage commit.
 // It first locks so no further dapr can be taken it. Once placement table is locked
 // in runtime, it proceeds to update new table to Dapr runtimes and then unlock
 // once all runtimes have been updated.
 func (p *Service) performTablesUpdate(hosts []placementGRPCStream, newTable *v1pb.PlacementTables) {
-	p.disseminateLock.Lock()
-	defer p.disseminateLock.Unlock()
-
 	// TODO: error from disseminationOperation needs to be handle properly.
 	// Otherwise, each Dapr runtime will have inconsistent hashing table.
 	p.disseminateOperation(hosts, "lock", nil)
