Function dequeuePipeline

backend/server/services/pipeline.go:253–320

(runningParallelLabels []string)

}

func dequeuePipeline(runningParallelLabels []string) (pipeline *models.Pipeline, err errors.Error) {
	txHelper := dbhelper.NewTxHelper(basicRes, &err)
	defer txHelper.End()
	tx := txHelper.Begin()
	// mysql read lock, not sure if it works for postgresql
	errors.Must(tx.LockTables(dal.LockTables{
		{Table: "_devlake_pipelines", Exclusive: false},
		{Table: "_devlake_pipeline_labels", Exclusive: false},
	}))
	// prepare query to find an appropriate pipeline to execute
	pipeline = &models.Pipeline{}
	// 1. find out the current highest priority in the queue
	top_priority := 0
	var top_priorities []int
	where_status := dal.Where("status IN ?", []string{models.TASK_CREATED, models.TASK_RERUN, models.TASK_RESUME})
	err = tx.Pluck("priority", &top_priorities, dal.From(pipeline), where_status, dal.Orderby("priority DESC"), dal.Limit(1))
	if err != nil {
		panic(err)
	}
	if len(top_priorities) > 0 {
		top_priority = top_priorities[0]
	}
	// 2. pick the earlier runnable pipeline with the highest priority
	err = tx.First(pipeline,
		where_status,
		dal.Where("priority = ?", top_priority),
		dal.Join(
			`left join _devlake_pipeline_labels ON
				_devlake_pipeline_labels.pipeline_id = _devlake_pipelines.id AND
				_devlake_pipeline_labels.name LIKE 'parallel/%' AND
				_devlake_pipeline_labels.name in ?`,
			runningParallelLabels,
		),
		dal.Groupby("id"),
		dal.Having("count(_devlake_pipeline_labels.name)=0"),
		dal.Select("id"),
		dal.Orderby("id ASC"),
		dal.Limit(1),
	)
	if err == nil {
		// mark the pipeline running, now we want a write lock
		if pipeline.BeganAt == nil {
			now := time.Now()
			pipeline.BeganAt = &now
			globalPipelineLog.Info("resumed pipeline #%d", pipeline.ID)
		}
		errors.Must(tx.LockTables(dal.LockTables{{Table: "_devlake_pipelines", Exclusive: true}}))
		err = tx.UpdateColumns(&models.Pipeline{}, []dal.DalSet{
			{ColumnName: "status", Value: models.TASK_RUNNING},
			{ColumnName: "message", Value: ""},
			{ColumnName: "began_at", Value: pipeline.BeganAt},
		}, dal.Where("id = ?", pipeline.ID))
		if err != nil {
			panic(err)
		}

		return
	}