You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 

127 lines
3.2 KiB

// Copyright 2019 PingCAP, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// See the License for the specific language governing permissions and
// limitations under the License.
package statistics
import (
"sort"
"sync"
"time"
)
type analyzeJobs struct {
sync.Mutex
jobs map[*AnalyzeJob]struct{}
history []*AnalyzeJob
}
var analyzeStatus = analyzeJobs{jobs: make(map[*AnalyzeJob]struct{}), history: make([]*AnalyzeJob, 0, numMaxHistoryJobs)}
// AnalyzeJob is used to represent the status of one analyze job.
type AnalyzeJob struct {
sync.Mutex
DBName string
TableName string
PartitionName string
JobInfo string
RowCount int64
StartTime time.Time
State string
updateTime time.Time
}
const (
pending = "pending"
running = "running"
finished = "finished"
failed = "failed"
)
// AddNewAnalyzeJob adds new analyze job.
func AddNewAnalyzeJob(job *AnalyzeJob) {
analyzeStatus.Lock()
job.updateTime = time.Now()
job.State = pending
analyzeStatus.jobs[job] = struct{}{}
analyzeStatus.Unlock()
}
const numMaxHistoryJobs = 20
// MoveToHistory moves the analyze job to history.
func MoveToHistory(job *AnalyzeJob) {
analyzeStatus.Lock()
delete(analyzeStatus.jobs, job)
analyzeStatus.history = append(analyzeStatus.history, job)
numJobs := len(analyzeStatus.history)
if numJobs > numMaxHistoryJobs {
analyzeStatus.history = analyzeStatus.history[numJobs-numMaxHistoryJobs:]
}
analyzeStatus.Unlock()
}
// ClearHistoryJobs clears all history jobs.
func ClearHistoryJobs() {
analyzeStatus.Lock()
analyzeStatus.history = analyzeStatus.history[:0]
analyzeStatus.Unlock()
}
// GetAllAnalyzeJobs gets all analyze jobs.
func GetAllAnalyzeJobs() []*AnalyzeJob {
analyzeStatus.Lock()
jobs := make([]*AnalyzeJob, 0, len(analyzeStatus.jobs)+len(analyzeStatus.history))
for job := range analyzeStatus.jobs {
jobs = append(jobs, job)
}
jobs = append(jobs, analyzeStatus.history...)
analyzeStatus.Unlock()
sort.Slice(jobs, func(i int, j int) bool { return jobs[i].getUpdateTime().Before(jobs[j].getUpdateTime()) })
return jobs
}
// Start marks status of the analyze job as running and update the start time.
func (job *AnalyzeJob) Start() {
job.Mutex.Lock()
job.State = running
now := time.Now()
job.StartTime = now
job.updateTime = now
job.Mutex.Unlock()
}
// Update updates the row count of analyze job.
func (job *AnalyzeJob) Update(rowCount int64) {
job.Mutex.Lock()
job.RowCount += rowCount
job.updateTime = time.Now()
job.Mutex.Unlock()
}
// Finish update the status of analyze job to finished or failed according to `meetError`.
func (job *AnalyzeJob) Finish(meetError bool) {
job.Mutex.Lock()
if meetError {
job.State = failed
} else {
job.State = finished
}
job.updateTime = time.Now()
job.Mutex.Unlock()
}
func (job *AnalyzeJob) getUpdateTime() time.Time {
job.Mutex.Lock()
defer job.Mutex.Unlock()
return job.updateTime
}