CR Revisions: Make concurrent operations safe

Author: nikhilmat

go/logic/applier.go

@@ -34,14 +34,14 @@ type Applier struct {
 	db                *gosql.DB
 	singletonDB       *gosql.DB
 	migrationContext  *base.MigrationContext
-	finishedMigrating bool
+	finishedMigrating int64
 }
 
 func NewApplier(migrationContext *base.MigrationContext) *Applier {
 	return &Applier{
 		connectionConfig:  migrationContext.ApplierConnectionConfig,
 		migrationContext:  migrationContext,
-		finishedMigrating: false,
+		finishedMigrating: 0,
 	}
 }
 
@@ -312,7 +312,7 @@ func (this *Applier) InitiateHeartbeat() {
 
 	heartbeatTick := time.Tick(time.Duration(this.migrationContext.HeartbeatIntervalMilliseconds) * time.Millisecond)
 	for range heartbeatTick {
-		if this.finishedMigrating {
+		if atomic.LoadInt64(&this.finishedMigrating) > 0 {
 			return
 		}
 		// Generally speaking, we would issue a goroutine, but I'd actually rather
@@ -1049,5 +1049,5 @@ func (this *Applier) Teardown() {
 	log.Debugf("Tearing down...")
 	this.db.Close()
 	this.singletonDB.Close()
-	this.finishedMigrating = true
+	atomic.StoreInt64(&this.finishedMigrating, 1)
 }

go/logic/migrator.go

@@ -85,7 +85,7 @@ type Migrator struct {
 
 	handledChangelogStates map[string]bool
 
-	finishedMigrating bool
+	finishedMigrating int64
 }
 
 func NewMigrator(context *base.MigrationContext) *Migrator {
@@ -100,7 +100,7 @@ func NewMigrator(context *base.MigrationContext) *Migrator {
 		copyRowsQueue:          make(chan tableWriteFunc),
 		applyEventsQueue:       make(chan *applyEventStruct, base.MaxEventsBatchSize),
 		handledChangelogStates: make(map[string]bool),
-		finishedMigrating:      false,
+		finishedMigrating:      0,
 	}
 	return migrator
 }
@@ -727,7 +727,7 @@ func (this *Migrator) initiateStatus() error {
 	this.printStatus(ForcePrintStatusAndHintRule)
 	statusTick := time.Tick(1 * time.Second)
 	for range statusTick {
-		if this.finishedMigrating {
+		if atomic.LoadInt64(&this.finishedMigrating) > 0 {
 			return nil
 		}
 		go this.printStatus(HeuristicPrintStatusRule)
@@ -954,7 +954,7 @@ func (this *Migrator) initiateStreaming() error {
 	go func() {
 		ticker := time.Tick(1 * time.Second)
 		for range ticker {
-			if this.finishedMigrating {
+			if atomic.LoadInt64(&this.finishedMigrating) > 0 {
 				return
 			}
 			this.migrationContext.SetRecentBinlogCoordinates(*this.eventsStreamer.GetCurrentBinlogCoordinates())
@@ -1147,7 +1147,7 @@ func (this *Migrator) executeWriteFuncs() error {
 		return nil
 	}
 	for {
-		if this.finishedMigrating {
+		if atomic.LoadInt64(&this.finishedMigrating) > 0 {
 			return nil
 		}
 
@@ -1232,7 +1232,7 @@ func (this *Migrator) finalCleanup() error {
 }
 
 func (this *Migrator) teardown() {
-	this.finishedMigrating = true
+	atomic.StoreInt64(&this.finishedMigrating, 1)
 
 	if this.inspector != nil {
 		log.Infof("Tearing down inspector")