Function runPipelineTasks

backend/core/runner/run_pipeline.go:57–105

(
	basicRes context.BasicRes,
	pipelineId uint64,
	taskIds [][]uint64,
	runTasks func([]uint64) errors.Error,
)

}

func runPipelineTasks(
	basicRes context.BasicRes,
	pipelineId uint64,
	taskIds [][]uint64,
	runTasks func([]uint64) errors.Error,
) errors.Error {
	db := basicRes.GetDal()
	log := basicRes.GetLogger()
	// load pipeline from db
	dbPipeline := &models.Pipeline{}
	err := db.First(dbPipeline, dal.Where("id = ?", pipelineId))
	if err != nil {
		return err
	}

	// if pipeline has been cancelled, just return.
	if dbPipeline.Status == models.TASK_CANCELLED {
		return nil
	}

	// This double for loop executes each set of tasks sequentially while
	// executing the set of tasks concurrently.
	for i, row := range taskIds {
		// update stage
		err = db.UpdateColumns(dbPipeline, []dal.DalSet{
			{ColumnName: "status", Value: models.TASK_RUNNING},
			{ColumnName: "stage", Value: i + 1},
		})
		if err != nil {
			log.Error(err, "update pipeline state failed")
			break
		}
		// run tasks in parallel
		err = runTasks(row)
		if err != nil {
			log.Error(err, "run tasks failed")
			if errors.Is(err, gocontext.Canceled) || !dbPipeline.SkipOnFail {
				log.Info("return error")
				return err
			}
		}
	}
	if dbPipeline.BeganAt != nil {
		log.Info("pipeline finished in %d ms: %v", time.Now().UnixMilli()-dbPipeline.BeganAt.UnixMilli(), err)
	} else {
		log.Info("pipeline finished at %d ms: %v", time.Now().UnixMilli(), err)
	}
	return err
}