octoleo/gh-ost
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge pull request #1180 from lmtwga/master

Browse Source
This commit is contained in:
Rashiq 2023-02-06 14:42:04 +01:00 committed by GitHub
commit b7db8c6ca7
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 7 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
go/logic/applier.go
View File

@ -9,7 +9,6 @@ import (
gosql "database/sql" gosql "database/sql"
"fmt" "fmt"
"strings" "strings"
"sync"
"sync/atomic" "sync/atomic"
"time" "time"
@ -935,7 +934,7 @@ func (this *Applier) CreateAtomicCutOverSentryTable() error {
} }
// AtomicCutOverMagicLock // AtomicCutOverMagicLock
func (this *Applier) AtomicCutOverMagicLock(sessionIdChan chan int64, tableLocked chan<- error, okToUnlockTable <-chan bool, tableUnlocked chan<- error, dropCutOverSentryTableOnce *sync.Once) error { func (this *Applier) AtomicCutOverMagicLock(sessionIdChan chan int64, tableLocked chan<- error, okToUnlockTable <-chan bool, tableUnlocked chan<- error) error {
tx, err := this.db.Begin() tx, err := this.db.Begin()
if err != nil { if err != nil {
tableLocked <- err tableLocked <- err
@ -946,6 +945,7 @@ func (this *Applier) AtomicCutOverMagicLock(sessionIdChan chan int64, tableLocke
tableLocked <- fmt.Errorf("Unexpected error in AtomicCutOverMagicLock(), injected to release blocking channel reads") tableLocked <- fmt.Errorf("Unexpected error in AtomicCutOverMagicLock(), injected to release blocking channel reads")
tableUnlocked <- fmt.Errorf("Unexpected error in AtomicCutOverMagicLock(), injected to release blocking channel reads") tableUnlocked <- fmt.Errorf("Unexpected error in AtomicCutOverMagicLock(), injected to release blocking channel reads")
tx.Rollback() tx.Rollback()
this.DropAtomicCutOverSentryTableIfExists()
}() }()
var sessionId int64 var sessionId int64
@ -1014,12 +1014,10 @@ func (this *Applier) AtomicCutOverMagicLock(sessionIdChan chan int64, tableLocke
sql.EscapeName(this.migrationContext.GetOldTableName()), sql.EscapeName(this.migrationContext.GetOldTableName()),
) )
dropCutOverSentryTableOnce.Do(func() {
if _, err := tx.Exec(query); err != nil { if _, err := tx.Exec(query); err != nil {
this.migrationContext.Log.Errore(err) this.migrationContext.Log.Errore(err)
// We DO NOT return here because we must `UNLOCK TABLES`! // We DO NOT return here because we must `UNLOCK TABLES`!
} }
})
// Tables still locked // Tables still locked
this.migrationContext.Log.Infof("Releasing lock from %s.%s, %s.%s", this.migrationContext.Log.Infof("Releasing lock from %s.%s, %s.%s",

7
go/logic/migrator.go
View File

@ -13,7 +13,6 @@ import (
"math" "math"
"os" "os"
"strings" "strings"
"sync"
"sync/atomic" "sync/atomic"
"time" "time"
@ -639,12 +638,8 @@ func (this *Migrator) atomicCutOver() (err error) {
defer atomic.StoreInt64(&this.migrationContext.InCutOverCriticalSectionFlag, 0) defer atomic.StoreInt64(&this.migrationContext.InCutOverCriticalSectionFlag, 0)
okToUnlockTable := make(chan bool, 4) okToUnlockTable := make(chan bool, 4)
var dropCutOverSentryTableOnce sync.Once
defer func() { defer func() {
okToUnlockTable <- true okToUnlockTable <- true
dropCutOverSentryTableOnce.Do(func() {
this.applier.DropAtomicCutOverSentryTableIfExists()
})
}() }()
atomic.StoreInt64(&this.migrationContext.AllEventsUpToLockProcessedInjectedFlag, 0) atomic.StoreInt64(&this.migrationContext.AllEventsUpToLockProcessedInjectedFlag, 0)
@ -653,7 +648,7 @@ func (this *Migrator) atomicCutOver() (err error) {
tableLocked := make(chan error, 2) tableLocked := make(chan error, 2)
tableUnlocked := make(chan error, 2) tableUnlocked := make(chan error, 2)
go func() { go func() {
if err := this.applier.AtomicCutOverMagicLock(lockOriginalSessionIdChan, tableLocked, okToUnlockTable, tableUnlocked, &dropCutOverSentryTableOnce); err != nil { if err := this.applier.AtomicCutOverMagicLock(lockOriginalSessionIdChan, tableLocked, okToUnlockTable, tableUnlocked); err != nil {
this.migrationContext.Log.Errore(err) this.migrationContext.Log.Errore(err)
} }
}() }()