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

context dump serialized with table writes; avoiding sync problems

Browse Source
This commit is contained in:
Shlomi Noach 2016-12-20 16:38:58 +02:00
parent 5e0f38cc6f
commit 3223a9389e
2 changed files with 11 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
go/base/context.go
View File

@ -224,6 +224,9 @@ func GetMigrationContext() *MigrationContext {
// DumpJSON exports this config to JSON string and writes it to file
func (this *MigrationContext) ToJSON() (string, error) {
this.throttleMutex.Lock()
defer this.throttleMutex.Unlock()
if this.MigrationRangeMinValues != nil {
this.EncodedRangeValues["MigrationRangeMinValues"], _ = this.MigrationRangeMinValues.ToBase64()
}

24
go/logic/migrator.go
View File

@ -123,19 +123,6 @@ func (this *Migrator) initiateHooksExecutor() (err error) {
return nil
}
// initiateContextDump
func (this *Migrator) initiateContextDump() (err error) {
go func() {
contextDumpTick := time.Tick(contextDumpInterval)
for range contextDumpTick {
if jsonString, err := this.migrationContext.ToJSON(); err == nil {
this.applier.WriteChangelog("context", jsonString)
}
}
}()
return nil
}
// sleepWhileTrue sleeps indefinitely until the given function returns 'false'
// (or fails with error)
func (this *Migrator) sleepWhileTrue(operation func() (bool, error)) error {
@ -298,9 +285,6 @@ func (this *Migrator) Migrate() (err error) {
if err := this.initiateHooksExecutor(); err != nil {
return err
}
if err := this.initiateContextDump(); err != nil {
return err
}
if err := this.hooksExecutor.onStartup(); err != nil {
return err
}
@ -1049,12 +1033,20 @@ func (this *Migrator) executeWriteFuncs() error {
log.Debugf("Noop operation; not really executing write funcs")
return nil
}
contextDumpTick := time.Tick(contextDumpInterval)
for {
this.throttler.throttle(nil)
// We give higher priority to event processing, then secondary priority to
// rowcopy
select {
case <-contextDumpTick:
{
if jsonString, err := this.migrationContext.ToJSON(); err == nil {
this.applier.WriteChangelog("context", jsonString)
log.Debugf("Context dumped")
}
}
case applyEventFunc := <-this.applyEventsQueue:
{
if err := this.retryOperation(applyEventFunc); err != nil {