<commit_before>package scheduler

import (
	"context"
	"database/sql"
	"fmt"
	"time"

	"github.com/go-gorp/gorp"
	"github.com/gorhill/cronexpr"
	"github.com/lib/pq"

	"github.com/ovh/cds/sdk"
	"github.com/ovh/cds/sdk/log"
)

var schedulerStatus = "Not Running"

//Scheduler is the goroutine which compute date of next execution for pipeline scheduler
func Scheduler(c context.Context, DBFunc func() *gorp.DbMap) {
	tick := time.NewTicker(2 * time.Second).C
	for {
		select {
		case <-c.Done():
			if c.Err() != nil {
				log.Error("Exiting scheduler.Scheduler: %v", c.Err())
				return
			}
		case <-tick:
			_, status, err := Run(DBFunc())

			if err != nil {
				log.Error("%s: %s", status, err)
			}
			schedulerStatus = status
		}
	}
}

//Run is the core function of Scheduler goroutine
func Run(db *gorp.DbMap) ([]sdk.PipelineSchedulerExecution, string, error) {
	//Load unscheduled pipelines
	ps, errl := LoadUnscheduledPipelines(db)
	if errl != nil {
		return nil, "Run> Unable to load unscheduled pipelines : %s", errl
	}

	execs := []sdk.PipelineSchedulerExecution{}

	for i := range ps {
		tx, errb := db.Begin()
		if errb != nil {
			return nil, "Run> Unable to start a transaction", errb
		}

		query := `
			SELECT pipeline_scheduler.* 
			FROM pipeline_scheduler 
			JOIN (
				SELECT 	pipeline_scheduler_id 
				FROM  	pipeline_scheduler_execution 
				WHERE 	pipeline_scheduler_id = $1
				AND 	executed = 'true' 
				ORDER BY  execution_planned_date DESC 
				LIMIT 1
				) execs ON pipeline_scheduler.id = execs.pipeline_scheduler_id 
			WHERE pipeline_scheduler.id = $1
			FOR UPDATE NOWAIT`

		var gorpPS = &PipelineScheduler{}
		if err := tx.SelectOne(gorpPS, query, ps[i].ID); err != nil {
			if pqerr, ok := err.(*pq.Error); ok && pqerr.Code != "55P03" || err != sql.ErrNoRows {
				log.Error("Run> Unable to lock to pipeline_scheduler %s", err)
			}
			_ = tx.Rollback()
			continue
		}

		s := sdk.PipelineScheduler(*gorpPS)
		//Skip disabled scheduler
		if s.Disabled {
			_ = tx.Rollback()
			continue
		}

		//Compute a new execution
		e, errn := Next(tx, &s)
		if errn != nil {
			//Nothing to compute<commit_msg>fix (api): scheduler execution lock (#1226)<commit_after>package scheduler

import (
	"context"
	"database/sql"
	"fmt"
	"time"

	"github.com/go-gorp/gorp"
	"github.com/gorhill/cronexpr"
	"github.com/lib/pq"

	"github.com/ovh/cds/sdk"
	"github.com/ovh/cds/sdk/log"
)

var schedulerStatus = "Not Running"

//Scheduler is the goroutine which compute date of next execution for pipeline scheduler
func Scheduler(c context.Context, DBFunc func() *gorp.DbMap) {
	tick := time.NewTicker(2 * time.Second).C
	for {
		select {
		case <-c.Done():
			if c.Err() != nil {
				log.Error("Exiting scheduler.Scheduler: %v", c.Err())
				return
			}
		case <-tick:
			_, status, err := Run(DBFunc())

			if err != nil {
				log.Error("%s: %s", status, err)
			}
			schedulerStatus = status
		}
	}
}

//Run is the core function of Scheduler goroutine
func Run(db *gorp.DbMap) ([]sdk.PipelineSchedulerExecution, string, error) {
	//Load unscheduled pipelines
	ps, errl := LoadUnscheduledPipelines(db)
	if errl != nil {
		return nil, "Run> Unable to load unscheduled pipelines : %s", errl
	}

	execs := []sdk.PipelineSchedulerExecution{}

	for i := range ps {
		tx, errb := db.Begin()
		if errb != nil {
			return nil, "Run> Unable to start a transaction", errb
		}

		query := `
			SELECT pipeline_scheduler.* 
			FROM pipeline_scheduler ,
			(
				SELECT 	count(pipeline_scheduler_id) as total
				FROM  	pipeline_scheduler_execution 
				WHERE 	pipeline_scheduler_id = $1
				AND     executed = 'false'
			) nb_execs 
			WHERE   pipeline_scheduler.id = $1
			AND     nb_execs.total = 0
			FOR UPDATE NOWAIT`

		var gorpPS = &PipelineScheduler{}
		if err := tx.SelectOne(gorpPS, query, ps[i].ID); err != nil {
			if pqerr, ok := err.(*pq.Error); ok && pqerr.Code != "55P03" || err != sql.ErrNoRows {
				log.Error("Run> Unable to lock to pipeline_scheduler %s", err)
			}
			_ = tx.Rollback()
			continue
		}

		s := sdk.PipelineScheduler(*gorpPS)
		//Skip disabled scheduler
		if s.Disabled {
			_ = tx.Rollback()
			continue
		}

		//Compute a new execution
		e, errn := Next(tx, &s)
		if errn != nil {
			//Nothing to compute
{'fraction_non_alphanumeric': 0.13358419567262464, 'fraction_numerical': 0.004703668861712135, 'mean_word_length': 3.165523996082272, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5504663', 'n_tokens_mistral': 1779, 'n_tokens_neox': 1563, 'n_words': 403}