Method processBatch

query/aql_processor.go:813–907

processBatch allocates device memory and starts async input data transferring to device memory. It then invokes previousBatchExecutor asynchronously to process the previous batch. When both async operations finish, it prepares for the current batch execution and returns it as a function closure to b

(
	batch *memCom.Batch, batchID int32, batchSize int, transferFunc batchTransferExecutor,
	customFilterFunc customFilterExecutor, previousBatchExecutor BatchExecutor, needToUnlockBatch bool)

// a function closure to be invoked later. customFilterExecutor is the executor
// to apply custom filters for live batch and archive batch.
func (qc *AQLQueryContext) processBatch(
	batch *memCom.Batch, batchID int32, batchSize int, transferFunc batchTransferExecutor,
	customFilterFunc customFilterExecutor, previousBatchExecutor BatchExecutor, needToUnlockBatch bool) BatchExecutor {
	defer func() {
		if needToUnlockBatch {
			batch.RUnlock()
		}
	}()

	if qc.Debug {
		// Finish executing previous batch first to avoid timeline overlapping
		qc.runBatchExecutor(previousBatchExecutor, false)
		previousBatchExecutor = NewDummyBatchExecutor()
	}

	// reset stats.
	qc.OOPK.currentBatch.stats = oopkBatchStats{
		batchID: batchID,
		timings: make(map[stageName]float64),
	}
	start := utils.Now()

	// Async transfer.
	stream := qc.cudaStreams[0]
	deviceSlices, hostVPs, firstColumn, startRow, totalBytes, numTransfers, sizeAfterPreFilter := transferFunc(stream)
	qc.OOPK.currentBatch.stats.bytesTransferred += totalBytes
	qc.OOPK.currentBatch.stats.numTransferCalls += numTransfers

	qc.reportTimingForCurrentBatch(stream, &start, transferTiming)

	// Async execute the previous batch.
	executionDone := make(chan struct{ error }, 1)
	go func() {
		defer func() {
			if r := recover(); r != nil {
				var err error
				// find out exactly what the error was and set err
				switch x := r.(type) {
				case string:
					err = utils.StackError(nil, x)
				case error:
					err = utils.StackError(x, "Panic happens when executing query")
				default:
					err = utils.StackError(nil, "Panic happens when executing query %v", x)
				}
				executionDone <- struct{ error }{err}
			}
		}()
		qc.runBatchExecutor(previousBatchExecutor, false)
		executionDone <- struct{ error }{}
	}()

	// Wait for data transfer of the current batch.
	cgoutils.WaitForCudaStream(stream, qc.Device)

	for _, vp := range hostVPs {
		if vp != nil {
			// only archive vector party will be returned after transfer function