Method executeWriteFuncs

go/logic/migrator.go:1824–1876

executeWriteFuncs writes data via applier: both the rowcopy and the events backlog. This is where the ghost table gets the data. The function fills the data single-threaded. Both event backlog and rowcopy events are polled; the backlog events have precedence.

()

// This is where the ghost table gets the data. The function fills the data single-threaded.
// Both event backlog and rowcopy events are polled; the backlog events have precedence.
func (mgtr *Migrator) executeWriteFuncs() error {
	if mgtr.migrationContext.Noop {
		mgtr.migrationContext.Log.Debugf("Noop operation; not really executing write funcs")
		return nil
	}
	for {
		if err := mgtr.checkAbort(); err != nil {
			return err
		}
		if atomic.LoadInt64(&mgtr.finishedMigrating) > 0 {
			return nil
		}

		mgtr.throttler.throttle(nil)

		// We give higher priority to event processing, then secondary priority to
		// rowcopy
		select {
		case eventStruct := <-mgtr.applyEventsQueue:
			{
				if err := mgtr.onApplyEventStruct(eventStruct); err != nil {
					return err
				}
			}
		default:
			{
				select {
				case copyRowsFunc := <-mgtr.copyRowsQueue:
					{
						copyRowsStartTime := time.Now()
						// Retries are handled within the copyRowsFunc
						if err := copyRowsFunc(); err != nil {
							return mgtr.migrationContext.Log.Errore(err)
						}
						if niceRatio := mgtr.migrationContext.GetNiceRatio(); niceRatio > 0 {
							copyRowsDuration := time.Since(copyRowsStartTime)
							sleepTimeNanosecondFloat64 := niceRatio * float64(copyRowsDuration.Nanoseconds())
							sleepTime := time.Duration(int64(sleepTimeNanosecondFloat64)) * time.Nanosecond
							time.Sleep(sleepTime)
						}
					}
				default:
					{
						// Hmmmmm... nothing in the queue; no events, but also no row copy.
						// This is possible upon load. Let's just sleep it over.
						mgtr.migrationContext.Log.Debugf("Getting nothing in the write queue. Sleeping...")
						time.Sleep(time.Second)
					}
				}
			}
		}
	}
}

func (mgtr *Migrator) executeDMLWriteFuncs() error {
	if mgtr.migrationContext.Noop {