1717732430

Build Type

github

Committed by GitHub

Commit Message

<a href="https://github.com/cybertec-postgresql/pg_timetable/commit/<a class=hub.com/cybertec-postgresql/pg_timetable/commit/bd27765aa129bb4f4b8eebd060d7c913aa58e4a5">bd27765aa<a href="https://github.com/cybertec-postgresql/pg_timetable/commit/bd27765aa129bb4f4b8eebd060d7c913aa58e4a5">">Merge </a><a class="double-link" href="https://github.com/cybertec-postgresql/pg_timetable/commit/<a class="double-link" href="https://github.com/cybertec-postgresql/pg_timetable/commit/c299d87abba48839dd7257c81e9999621ed85c20">c299d87ab</a>">c299d87ab</a><a href="https://github.com/cybertec-postgresql/pg_timetable/commit/bd27765aa129bb4f4b8eebd060d7c913aa58e4a5"> into a6eb7a2d3">a6eb7a2d3</a>

Pull Request Pull Request #379: [!] add REST API, resolves #378

Run Details

37 of 37 new or added lines in 2 files covered. (100.0%)

1480 of 1614 relevant lines covered (91.7%)

1.05 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

91.78

/internal/scheduler/scheduler.go

package scheduler

import (
        "context"
        "sync"
        "time"

        "github.com/cybertec-postgresql/pg_timetable/internal/config"
        "github.com/cybertec-postgresql/pg_timetable/internal/log"
        "github.com/cybertec-postgresql/pg_timetable/internal/pgengine"
)

// the main loop period. Should be 60 (sec) for release configuration. Set to 10 (sec) for debug purposes
const refetchTimeout = 60

// the min capacity of chains channels
const minChannelCapacity = 1024

// RunStatus specifies the current status of execution
type RunStatus int

const (
        // RunningStatus specifies the scheduler is in the main loop processing chains
        RunningStatus RunStatus = iota
        // ContextCancelledStatus specifies the context has been cancelled probably due to timeout
        ContextCancelledStatus
        // Shutdown specifies proper termination of the session
        ShutdownStatus
)

// Scheduler is the main class for running the tasks
type Scheduler struct {
        l          log.LoggerIface
        chainsChan chan Chain // channel for passing chains to workers
        pgengine   *pgengine.PgEngine

        exclusiveMutex sync.RWMutex //read-write mutex for running regular and exclusive chains

        // activeChains holds the map of chain ID with context cancel() function, so we can abort chain by request
        activeChains     map[int]func()
        activeChainMutex sync.Mutex

        // map of active chains, updated every minute
        intervalChains map[int]IntervalChain
        // create channel for passing interval chains to workers
        intervalChainsChan chan IntervalChain
        intervalChainMutex sync.Mutex
        shutdown           chan struct{} // closed when shutdown is called
        status             RunStatus
}

// Max returns the maximum number of two arguments
func Max(x, y int) int {
        if x < y {
                return y
        }
        return x
}

// New returns a new instance of Scheduler
func New(pge *pgengine.PgEngine, logger log.LoggerIface) *Scheduler {
        return &Scheduler{
                l:                  logger,
                pgengine:           pge,
                chainsChan:         make(chan Chain, Max(minChannelCapacity, pge.Resource.CronWorkers*2)),
                intervalChainsChan: make(chan IntervalChain, Max(minChannelCapacity, pge.Resource.IntervalWorkers*2)),
                activeChains:       make(map[int]func()), //holds cancel() functions to stop chains
                intervalChains:     make(map[int]IntervalChain),
                shutdown:           make(chan struct{}),
                status:             RunningStatus,
        }
}

// Shutdown terminates the current session
func (sch *Scheduler) Shutdown() {
        close(sch.shutdown)
}

// Config returns the current configuration for application
func (sch *Scheduler) Config() config.CmdOptions {
        return sch.pgengine.CmdOptions
}

// IsReady returns True if the scheduler is in the main loop processing chains
func (sch *Scheduler) IsReady() bool {
        return sch.status == RunningStatus
}

// Run executes jobs. Returns RunStatus why it terminated.
// There are only two possibilities: dropped connection and cancelled context.
func (sch *Scheduler) Run(ctx context.Context) RunStatus {
        // create sleeping workers waiting data on channel
        for w := 1; w <= sch.Config().Resource.CronWorkers; w++ {
                workerCtx, cancel := context.WithCancel(ctx)
                defer cancel()
                go sch.chainWorker(workerCtx, sch.chainsChan)
        }
        for w := 1; w <= sch.Config().Resource.IntervalWorkers; w++ {
                workerCtx, cancel := context.WithCancel(ctx)
                defer cancel()
                go sch.intervalChainWorker(workerCtx, sch.intervalChainsChan)
        }
        ctx = log.WithLogger(ctx, sch.l)
        /* cleanup potential database leftovers */
        sch.pgengine.FixSchedulerCrash(ctx)

        /*
                Loop forever or until we ask it to stop.
                First loop fetches notifications.
                Main loop works every refetchTimeout seconds and runs chains.
        */
        sch.l.Info("Accepting asynchronous chains execution requests...")
        go sch.retrieveAsyncChainsAndRun(ctx)

        if sch.Config().Start.Debug { //run blocking notifications receiving
                sch.pgengine.HandleNotifications(ctx)
                return ContextCancelledStatus
        }

        sch.l.Debug("Checking for @reboot task chains...")
        sch.retrieveChainsAndRun(ctx, true)

        for {
                sch.l.Debug("Checking for task chains...")
                go sch.retrieveChainsAndRun(ctx, false)
                sch.l.Debug("Checking for interval task chains...")
                go sch.retrieveIntervalChainsAndRun(ctx)

                select {
                case <-time.After(refetchTimeout * time.Second):
                        // pass
                case <-ctx.Done():
                        sch.status = ContextCancelledStatus
                case <-sch.shutdown:
                        sch.status = ShutdownStatus
                        sch.terminateChains()
                }

                if sch.status != RunningStatus {
                        return sch.status
                }
        }
}

1	package scheduler
2
3	import (
4	"context"
5	"sync"
6	"time"
7
8	"github.com/cybertec-postgresql/pg_timetable/internal/config"
9	"github.com/cybertec-postgresql/pg_timetable/internal/log"
10	"github.com/cybertec-postgresql/pg_timetable/internal/pgengine"
11	)
12
13	// the main loop period. Should be 60 (sec) for release configuration. Set to 10 (sec) for debug purposes
14	const refetchTimeout = 60
15
16	// the min capacity of chains channels
17	const minChannelCapacity = 1024
18
19	// RunStatus specifies the current status of execution
20	type RunStatus int
21
22	const (
23	// RunningStatus specifies the scheduler is in the main loop processing chains
24	RunningStatus RunStatus = iota
25	// ContextCancelledStatus specifies the context has been cancelled probably due to timeout
26	ContextCancelledStatus
27	// Shutdown specifies proper termination of the session
28	ShutdownStatus
29	)
30
31	// Scheduler is the main class for running the tasks
32	type Scheduler struct {
33	l log.LoggerIface
34	chainsChan chan Chain // channel for passing chains to workers
35	pgengine *pgengine.PgEngine
36
37	exclusiveMutex sync.RWMutex //read-write mutex for running regular and exclusive chains
38
39	// activeChains holds the map of chain ID with context cancel() function, so we can abort chain by request
40	activeChains map[int]func()
41	activeChainMutex sync.Mutex
42
43	// map of active chains, updated every minute
44	intervalChains map[int]IntervalChain
45	// create channel for passing interval chains to workers
46	intervalChainsChan chan IntervalChain
47	intervalChainMutex sync.Mutex
48	shutdown chan struct{} // closed when shutdown is called
49	status RunStatus
50	}
51
52	// Max returns the maximum number of two arguments
53	func Max(x, y int) int {	1✔
54	if x < y {	2✔
55	return y	1✔
56	}	1✔
57	return x	1✔
58	}
59
60	// New returns a new instance of Scheduler
61	func New(pge pgengine.PgEngine, logger log.LoggerIface) Scheduler {	1✔
62	return &Scheduler{	1✔
63	l: logger,	1✔
64	pgengine: pge,	1✔
65	chainsChan: make(chan Chain, Max(minChannelCapacity, pge.Resource.CronWorkers*2)),	1✔
66	intervalChainsChan: make(chan IntervalChain, Max(minChannelCapacity, pge.Resource.IntervalWorkers*2)),	1✔
67	activeChains: make(map[int]func()), //holds cancel() functions to stop chains	1✔
68	intervalChains: make(map[int]IntervalChain),	1✔
69	shutdown: make(chan struct{}),	1✔
70	status: RunningStatus,	1✔
71	}	1✔
72	}	1✔
73
74	// Shutdown terminates the current session
75	func (sch *Scheduler) Shutdown() {	1✔
76	close(sch.shutdown)	1✔
77	}	1✔
78
79	// Config returns the current configuration for application
80	func (sch *Scheduler) Config() config.CmdOptions {	1✔
81	return sch.pgengine.CmdOptions	1✔
82	}	1✔
83
84	// IsReady returns True if the scheduler is in the main loop processing chains
85	func (sch *Scheduler) IsReady() bool {	1✔
86	return sch.status == RunningStatus	1✔
87	}	1✔
88
89	// Run executes jobs. Returns RunStatus why it terminated.
90	// There are only two possibilities: dropped connection and cancelled context.
91	func (sch *Scheduler) Run(ctx context.Context) RunStatus {	1✔
92	// create sleeping workers waiting data on channel	1✔
93	for w := 1; w <= sch.Config().Resource.CronWorkers; w++ {	2✔
94	workerCtx, cancel := context.WithCancel(ctx)	1✔
95	defer cancel()	1✔
96	go sch.chainWorker(workerCtx, sch.chainsChan)	1✔
97	}	1✔
98	for w := 1; w <= sch.Config().Resource.IntervalWorkers; w++ {	2✔
99	workerCtx, cancel := context.WithCancel(ctx)	1✔
100	defer cancel()	1✔
101	go sch.intervalChainWorker(workerCtx, sch.intervalChainsChan)	1✔
102	}	1✔
103	ctx = log.WithLogger(ctx, sch.l)	1✔
104	/* cleanup potential database leftovers */	1✔
105	sch.pgengine.FixSchedulerCrash(ctx)	1✔
106		1✔
107	/*	1✔
108	Loop forever or until we ask it to stop.	1✔
109	First loop fetches notifications.	1✔
110	Main loop works every refetchTimeout seconds and runs chains.	1✔
111	*/	1✔
112	sch.l.Info("Accepting asynchronous chains execution requests...")	1✔
113	go sch.retrieveAsyncChainsAndRun(ctx)	1✔
114		1✔
115	if sch.Config().Start.Debug { //run blocking notifications receiving	1✔
116	sch.pgengine.HandleNotifications(ctx)	×
117	return ContextCancelledStatus	×
118	}	×
119
120	sch.l.Debug("Checking for @reboot task chains...")	1✔
121	sch.retrieveChainsAndRun(ctx, true)	1✔
122		1✔
123	for {	2✔
124	sch.l.Debug("Checking for task chains...")	1✔
125	go sch.retrieveChainsAndRun(ctx, false)	1✔
126	sch.l.Debug("Checking for interval task chains...")	1✔
127	go sch.retrieveIntervalChainsAndRun(ctx)	1✔
128		1✔
129	select {	1✔
130	case <-time.After(refetchTimeout * time.Second):	×
131	// pass
132	case <-ctx.Done():	×
133	sch.status = ContextCancelledStatus	×
134	case <-sch.shutdown:	1✔
135	sch.status = ShutdownStatus	1✔
136	sch.terminateChains()	1✔
137	}
138
139	if sch.status != RunningStatus {	2✔
140	return sch.status	1✔
141	}	1✔
142	}
143	}

cybertec-postgresql / pg_timetable / 1717732430

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous