This commit is contained in:
parent
ce03757deb
commit
0e2d33ad86
@ -17,6 +17,7 @@ import (
|
|||||||
"github.com/github/gh-ost/go/sql"
|
"github.com/github/gh-ost/go/sql"
|
||||||
|
|
||||||
"github.com/outbrain/golib/sqlutils"
|
"github.com/outbrain/golib/sqlutils"
|
||||||
|
"sync"
|
||||||
)
|
)
|
||||||
|
|
||||||
const (
|
const (
|
||||||
@ -787,7 +788,7 @@ func (this *Applier) CreateAtomicCutOverSentryTable() error {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// AtomicCutOverMagicLock
|
// AtomicCutOverMagicLock
|
||||||
func (this *Applier) AtomicCutOverMagicLock(sessionIdChan chan int64, tableLocked chan<- error, okToUnlockTable <-chan bool, tableUnlocked chan<- error) error {
|
func (this *Applier) AtomicCutOverMagicLock(sessionIdChan chan int64, tableLocked chan<- error, okToUnlockTable <-chan bool, tableUnlocked chan<- error, dropCutOverSentryTableOnce *sync.Once) error {
|
||||||
tx, err := this.db.Begin()
|
tx, err := this.db.Begin()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
tableLocked <- err
|
tableLocked <- err
|
||||||
@ -865,10 +866,13 @@ func (this *Applier) AtomicCutOverMagicLock(sessionIdChan chan int64, tableLocke
|
|||||||
sql.EscapeName(this.migrationContext.DatabaseName),
|
sql.EscapeName(this.migrationContext.DatabaseName),
|
||||||
sql.EscapeName(this.migrationContext.GetOldTableName()),
|
sql.EscapeName(this.migrationContext.GetOldTableName()),
|
||||||
)
|
)
|
||||||
if _, err := tx.Exec(query); err != nil {
|
|
||||||
this.migrationContext.Log.Errore(err)
|
dropCutOverSentryTableOnce.Do(func() {
|
||||||
// We DO NOT return here because we must `UNLOCK TABLES`!
|
if _, err := tx.Exec(query); err != nil {
|
||||||
}
|
this.migrationContext.Log.Errore(err)
|
||||||
|
// 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",
|
||||||
|
@ -11,6 +11,7 @@ import (
|
|||||||
"math"
|
"math"
|
||||||
"os"
|
"os"
|
||||||
"strings"
|
"strings"
|
||||||
|
"sync"
|
||||||
"sync/atomic"
|
"sync/atomic"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
@ -606,9 +607,12 @@ 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
|
||||||
this.applier.DropAtomicCutOverSentryTableIfExists()
|
dropCutOverSentryTableOnce.Do(func() {
|
||||||
|
this.applier.DropAtomicCutOverSentryTableIfExists()
|
||||||
|
})
|
||||||
}()
|
}()
|
||||||
|
|
||||||
atomic.StoreInt64(&this.migrationContext.AllEventsUpToLockProcessedInjectedFlag, 0)
|
atomic.StoreInt64(&this.migrationContext.AllEventsUpToLockProcessedInjectedFlag, 0)
|
||||||
@ -617,7 +621,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); err != nil {
|
if err := this.applier.AtomicCutOverMagicLock(lockOriginalSessionIdChan, tableLocked, okToUnlockTable, tableUnlocked, &dropCutOverSentryTableOnce); err != nil {
|
||||||
this.migrationContext.Log.Errore(err)
|
this.migrationContext.Log.Errore(err)
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
|
Loading…
Reference in New Issue
Block a user