Method executeInParallel

executor/executor.go:131–167

(logs []*OplogRecord)

}

func (batchExecutor *BatchGroupExecutor) executeInParallel(logs []*OplogRecord) {
	// prepare execution monitor
	latch := new(sync.WaitGroup)
	latch.Add(len(logs))
	// shard oplogRecords by _id primary key and make up callback chain
	var buffer = make([][]*OplogRecord, len(batchExecutor.executors))
	shardKey := oplog.PrimaryKeyHasher{}
	var completionList []func()
	for _, log := range logs {
		selected := shardKey.DistributeOplogByMod(log.original.partialLog, len(batchExecutor.executors))
		buffer[selected] = append(buffer[selected], log)
		if log.original.callback != nil {
			// should be ordered by the incoming sequence
			completionList = append(completionList, log.original.callback)
		}
	}
	for index, buf := range buffer {
		if len(buf) != 0 {
			nimo.AssertTrue(len(batchExecutor.executors[index].batchBlock) == 0, "executors buffer is not empty!")
			nimo.AssertTrue(batchExecutor.executors[index].finisher == nil, "executors await status is wrong!")
			batchExecutor.executors[index].finisher = latch
			// follow the MEMORY MODEL : finisher should be assigned
			// before batchBlock channel. it read after <- batchBlock
			batchExecutor.executors[index].batchBlock <- buf
		}
	}
	// wait for execute completely
	latch.Wait()
	// invoke all callbacks
	for _, callback := range completionList {
		callback()
	}
	// sweep executors' block buffer and await
	for _, exec := range batchExecutor.executors {
		exec.finisher = nil
	}
}

type Executor struct {
	// sequence index id in each replayer