add job Coordinator

go/binlog/gomysql_reader.go

@@ -106,6 +106,7 @@ func (this *GoMySQLReader) handleRowsEvent(ev *replication.BinlogEvent, rowsEven
 		)
 		switch dml {
 		case InsertDML:
+
 			{
 				binlogEntry.DmlEvent.NewColumnValues = sql.ToColumnValues(row)
 			}
@@ -286,6 +287,8 @@ groups:
 		case *replication.TableMapEvent:
 			// TODO: Can we be smart here and short circuit processing groups for tables that don't match the table in the migration context?
 
+			this.migrationContext.Log.Infof("sending transaction: %d %d", group.SequenceNumber, group.LastCommitted)
+
 			group.TableName = string(binlogEvent.Table)
 			group.DatabaseName = string(binlogEvent.Schema)
 			// we are good to send the transaction, the transaction events arrive async

go/logic/migrator.go

@@ -11,11 +11,12 @@ import (
 	"fmt"
 	"io"
 	"math"
-	"os"
 	"strings"
 	"sync/atomic"
 	"time"
 
+	"os"
+
 	"github.com/github/gh-ost/go/base"
 	"github.com/github/gh-ost/go/binlog"
 	"github.com/github/gh-ost/go/mysql"
@@ -424,8 +425,9 @@ func (this *Migrator) Migrate() (err error) {
 		return err
 	}
 
-	// TODO: configure workers
-	numApplierWorkers := 4
+	// TODO(meiji163): configure workers
+	numApplierWorkers := 5
+	this.migrationContext.Log.Info("starting applier workers")
 	go this.trxCoordinator.StartWorkers(numApplierWorkers)
 	go this.executeWriteFuncs()
 	go this.iterateChunks()
@@ -1083,14 +1085,30 @@ func (this *Migrator) initiateStreaming() error {
 	if err := this.eventsStreamer.InitDBConnections(); err != nil {
 		return err
 	}
-	this.eventsStreamer.AddListener(
+	// this.eventsStreamer.AddListener(
+	// 	false,
+	// 	this.migrationContext.DatabaseName,
+	// 	this.migrationContext.GetChangelogTableName(),
+	// 	func(dmlEvent *binlog.BinlogDMLEvent) error {
+	// 		return this.onChangelogEvent(dmlEvent)
+	// 	},
+	// )
+
+	handleChangeLogTrx := func(trx *binlog.Transaction) error {
+		for rowEvent := range trx.Changes {
+			if err := this.onChangelogEvent(rowEvent.DmlEvent); err != nil {
+				return err
+			}
+		}
+		return nil
+	}
+	this.eventsStreamer.AddTransactionListener(
 		false,
 		this.migrationContext.DatabaseName,
 		this.migrationContext.GetChangelogTableName(),
-		func(dmlEvent *binlog.BinlogDMLEvent) error {
-			return this.onChangelogEvent(dmlEvent)
-		},
+		handleChangeLogTrx,
 	)
+
 	ctx := context.Background()
 	go func() {
 		this.migrationContext.Log.Debugf("Beginning streaming")
@@ -1266,6 +1284,7 @@ func (this *Migrator) iterateChunks() error {
 }
 
 func (this *Migrator) onApplyEventStruct(eventStruct *applyEventStruct) error {
+	this.migrationContext.Log.Infof("onApplyEventStruct: %+v", eventStruct)
 	handleNonDMLEventStruct := func(eventStruct *applyEventStruct) error {
 		if eventStruct.writeFunc != nil {
 			if err := this.retryOperation(*eventStruct.writeFunc); err != nil {
@@ -1274,14 +1293,16 @@ func (this *Migrator) onApplyEventStruct(eventStruct *applyEventStruct) error {
 		}
 		return nil
 	}
-	if eventStruct.dmlEvent == nil {
-		return handleNonDMLEventStruct(eventStruct)
-	}
+	// if eventStruct.dmlEvent == nil {
+	// 	return handleNonDMLEventStruct(eventStruct)
+	// }
 	if eventStruct.trxEvent != nil {
+		this.migrationContext.Log.Infof("got transaction: %+v", eventStruct.trxEvent)
 		batchSize := int(atomic.LoadInt64(&this.migrationContext.DMLBatchSize))
 
 		processTrxFunc := func() error {
 			// process rows events from the transaction in batches
+			this.migrationContext.Log.Info("starting process transaction job")
 			for {
 				dmlEvents := make([]*binlog.BinlogDMLEvent, 0, batchSize)
 				for i := 0; i < batchSize; i++ {
@@ -1296,9 +1317,10 @@ func (this *Migrator) onApplyEventStruct(eventStruct *applyEventStruct) error {
 						}
 					}
 					dmlEvents = append(dmlEvents, binlogEntry.DmlEvent)
-					if err := this.applier.ApplyDMLEventQueries(dmlEvents); err != nil {
-						return err
-					}
+				}
+				//this.migrationContext.Log.Infof("received dmlEvents: %+v", dmlEvents)
+				if err := this.applier.ApplyDMLEventQueries(dmlEvents); err != nil {
+					return err
 				}
 			}
 		}
@@ -1308,6 +1330,7 @@ func (this *Migrator) onApplyEventStruct(eventStruct *applyEventStruct) error {
 			SequenceNumber: eventStruct.trxEvent.SequenceNumber,
 			Do:             processTrxFunc,
 		}
+		this.migrationContext.Log.Infof("submitting job: %+v", job)
 		this.trxCoordinator.SubmitJob(job)
 
 	} else if eventStruct.dmlEvent != nil {

go/logic/multi_threaded_applier.go

@@ -0,0 +1,172 @@
+package logic
+
+import (
+	"sync"
+	"fmt"
+	"time"
+)
+
+type Job struct {
+	// The sequence number of the job
+	SequenceNumber int64
+
+	// The sequence number of the job this job depends on
+	LastCommitted int64
+
+	// channel that's closed once the job's dependencies are met
+	// `nil` if the job's dependencies are already met when the job is submitted
+	waitChannel chan struct{}
+
+	// Do the job
+	Do func() error
+
+	// Data for the job
+	//Changes chan *replication.RowsEvent
+}
+
+type Coordinator struct {
+	// The queue of jobs to be executed
+	queue chan *Job
+
+	wg sync.WaitGroup
+
+	// List of workers
+	workers []*Worker
+
+	// Mutex to protect the fields below
+	mu sync.Mutex
+
+	// The low water mark. This is the sequence number of the last job that has been committed.
+	lowWaterMark int64
+
+	// This is a map of completed jobs by their sequence numbers.
+	// This is used when updating the low water mark.
+	completedJobs map[int64]bool
+
+	// These are the jobs that are waiting for a previous job to complete.
+	// They are indexed by the sequence number of the job they are waiting for.
+	waitingJobs map[int64][]*Job
+
+	// Is this the first job we're processing? If yes, we can schedule it even if the last committed
+	// sequence number is greater than the low water mark.
+	firstJob bool
+}
+
+type Worker struct {
+	executedJobs int
+}
+
+func NewCoordinator() *Coordinator {
+	return &Coordinator{
+		lowWaterMark:  0,
+		firstJob:      true,
+		completedJobs: make(map[int64]bool),
+		waitingJobs:   make(map[int64][]*Job),
+		queue:         make(chan *Job),
+	}
+}
+
+func (c *Coordinator) StartWorkers(count int) {
+	jobTimeout := 10 * time.Second
+	for i := 0; i< count; i++ {
+		go func(){
+			for job := range c.queue {
+				if job.waitChannel != nil {
+					fmt.Printf("Coordinator: Job %d is waiting for job %d to complete\n", job.SequenceNumber, job.LastCommitted)
+					select {
+					case <-job.waitChannel:
+						break
+					case <- time.After(jobTimeout):
+						// TODO: something problably went wrong here
+						fmt.Printf("Coordinator: Job %d timed out waiting for job %d\n", job.SequenceNumber, job.LastCommitted)
+						panic("worker timeout")
+					}
+					// fmt.Printf("Worker received signal for job: %d\n", job.sequenceNumber)
+ 				}
+
+				if err := job.Do(); err != nil {
+					// TODO(meiji163) handle error
+					panic(err)
+				}
+				c.markJobCompleted(job)
+			}
+		}()
+	}
+	c.wg.Wait()
+}
+
+func (c *Coordinator) SubmitJob(job *Job) {
+	c.mu.Lock()
+
+	c.wg.Add(1)
+
+	// Jobs might need for their dependencies to be met before they can be executed. We use the
+	// `waitChannel` to signal that the job's dependencies are met.
+	//
+	// We can short-circuit this if:
+	// * this is the first job we're processing
+	// * the job's last committed sequence number is less than or equal to the low water mark,
+	//   thus we know that the job's dependencies are already met.
+	// * the job's dependency has completed, but the lowWaterMark has not been updated yet because
+	//   a job with a lower sequence number is still being processed.
+	//
+	// When short-circuiting, we don't assign a `waitChannel` to the job, thus signaling that the
+	// job's dependencies are already met.
+	if c.firstJob {
+		fmt.Printf("Coordinator: Scheduling first job: %d\n", job.SequenceNumber)
+		// assume everything before the first job has been processed
+		c.lowWaterMark = job.SequenceNumber
+		c.firstJob = false
+	} else if job.LastCommitted <= c.lowWaterMark || c.completedJobs[job.LastCommitted] {
+		fmt.Printf("Coordinator: Scheduling job: %d\n", job.SequenceNumber)
+	} else {
+		job.waitChannel = make(chan struct{})
+		c.waitingJobs[job.LastCommitted] = append(c.waitingJobs[job.LastCommitted], job)
+	}
+
+	c.mu.Unlock()
+	// Add the job to the queue. This will block until a worker picks up this job.
+	c.queue <- job
+}
+
+func (c *Coordinator) markJobCompleted(job *Job) {
+	c.mu.Lock()
+	defer c.wg.Done()
+
+	fmt.Printf("Coordinator: Marking job as completed: %d\n", job.SequenceNumber)
+
+	// Mark the job as completed
+	c.completedJobs[job.SequenceNumber] = true
+
+
+	// Then, update the low water mark if possible
+
+	// TODO: this won't work because the intermediate sequence numbers
+	// can be for trxs on tables other than the one we're migrating
+	for {
+		if c.completedJobs[c.lowWaterMark+1] {
+			c.lowWaterMark++
+			delete(c.completedJobs, c.lowWaterMark)
+		} else {
+			break
+		}
+	}
+
+	jobsToNotify := make([]*Job, 0)
+
+	// TODO: fix this
+	// Schedule any jobs that were waiting for this job to complete
+	for lastCommitted, jobs := range c.waitingJobs {
+		if lastCommitted <= c.lowWaterMark {
+			jobsToNotify = append(jobsToNotify, jobs...)
+			delete(c.waitingJobs, lastCommitted)
+		}
+	}
+
+	c.mu.Unlock()
+
+	for _, waitingJob := range jobsToNotify {
+		fmt.Printf("Scheduling previously waiting job: %d - %d\n", waitingJob.SequenceNumber, waitingJob.LastCommitted)
+		waitingJob.waitChannel <- struct{}{}
+	}
+}

go/logic/streamer.go

@@ -27,11 +27,11 @@ var (
 )
 
 type BinlogEventListener struct {
-	async         bool
-	databaseName  string
-	tableName     string
-	onDmlEvent    func(event *binlog.BinlogDMLEvent) error
-	onTrxEvent    func(trx *binlog.Transaction) error
+	async        bool
+	databaseName string
+	tableName    string
+	onDmlEvent   func(event *binlog.BinlogDMLEvent) error
+	onTrxEvent   func(trx *binlog.Transaction) error
 }
 
 const (
@@ -208,11 +208,12 @@ func (evs *EventsStreamer) StreamTransactions(ctx context.Context, canStopStream
 		}
 	}()
 	for {
-		if canStopStreaming() {
-			return nil
-		}
-		// TODO: handle retry/reconnect
+		// TODO(meiji163): handle retry/reconnect
 		if err := evs.binlogReader.StreamTransactions(ctx, trxChan); err != nil {
+			if canStopStreaming() {
+				return nil
+			}
+			close(trxChan)
 			return err
 		}
 	}