-
Notifications
You must be signed in to change notification settings - Fork 133
/
comparison.go
144 lines (125 loc) · 4.19 KB
/
comparison.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
package jobdb
import (
"time"
)
type (
JobPriorityComparer struct{}
JobQueueTtlComparer struct{}
JobIdHasher struct{}
)
func (JobIdHasher) Hash(j *Job) uint32 {
var hash uint32
for _, c := range j.id {
hash = 31*hash + uint32(c)
}
return hash
}
func (JobIdHasher) Equal(a, b *Job) bool {
return a == b
}
// Compare jobs by their remaining queue time before expiry
// Invariants:
// - Job.queueTtl must be > 0
// - Job.created must be < `t`
func (j JobQueueTtlComparer) Compare(a, b *Job) int {
// Jobs with equal id are always considered equal.
// This ensures at most one job with a particular id can exist in the jobDb.
if a.id == b.id {
return 0
}
// TODO: Calling time.Now() here doesn't sound right. We should probably sort by earliest expiry time.
timeSeconds := time.Now().UTC().Unix()
aDuration := timeSeconds - (a.submittedTime / 1_000_000_000)
bDuration := timeSeconds - (b.submittedTime / 1_000_000_000)
aRemaining := max(0, a.QueueTtlSeconds()-aDuration)
bRemaining := max(0, b.QueueTtlSeconds()-bDuration)
// If jobs have different ttl remaining, they are ordered by remaining queue ttl - the smallest ttl first.
if aRemaining != bRemaining {
if aRemaining < bRemaining {
return -1
} else {
return 1
}
}
// Tie-break by logical creation timestamp.
if a.id < b.id {
return -1
} else if a.id > b.id {
return 1
}
panic("We should never get here. Since we check for job id equality at the top of this function.")
}
func max(x, y int64) int64 {
if x < y {
return y
}
return x
}
func (JobPriorityComparer) Compare(job, other *Job) int {
return SchedulingOrderCompare(job, other)
}
// SchedulingOrderCompare defines the order in which jobs in a particular queue should be scheduled,
func (job *Job) SchedulingOrderCompare(other *Job) int {
// We need this cast for now to expose this method via an interface.
// This is safe since we only ever compare jobs of the same type.
return SchedulingOrderCompare(job, other)
}
// SchedulingOrderCompare defines the order in which jobs in a queue should be scheduled
// (both when scheduling new jobs and when re-scheduling evicted jobs).
// Specifically, compare returns
// - 0 if the jobs have equal job id,
// - -1 if job should be scheduled before other,
// - +1 if other should be scheduled before other.
func SchedulingOrderCompare(job, other *Job) int {
// Jobs with equal id are always considered equal.
// This ensures at most one job with a particular id can exist in the jobDb.
if job.id == other.id {
return 0
}
// Running jobs always come before queued jobs.
// This to ensure evicted jobs are rescheduled before scheduling new jobs.
jobIsActive := job.activeRun != nil && !job.activeRun.InTerminalState()
otherIsActive := other.activeRun != nil && !other.activeRun.InTerminalState()
if jobIsActive && !otherIsActive {
return -1
} else if !jobIsActive && otherIsActive {
return 1
}
// PriorityClassPriority indicates urgency.
// Hence, jobs of higher priorityClassPriority come first.
if job.priorityClass.Priority > other.priorityClass.Priority {
return -1
} else if job.priorityClass.Priority < other.priorityClass.Priority {
return 1
}
// Jobs higher in queue-priority come first.
if job.priority < other.priority {
return -1
} else if job.priority > other.priority {
return 1
}
// If both jobs are active, order by time since the job was scheduled.
// This ensures jobs that have been running for longer are rescheduled first,
// which reduces wasted compute time when preempting.
if jobIsActive && otherIsActive {
if job.activeRunTimestamp < other.activeRunTimestamp {
return -1
} else if job.activeRunTimestamp > other.activeRunTimestamp {
return 1
}
}
// Jobs that have been queuing for longer are scheduled first.
if job.submittedTime < other.submittedTime {
return -1
} else if job.submittedTime > other.submittedTime {
return 1
}
// Tie-break by jobId, which must be unique.
// This ensures there is a total order between jobs, i.e., no jobs are equal from an ordering point of view.
if job.id < other.id {
return -1
} else if job.id > other.id {
return 1
}
panic("We should never get here. Since we check for job id equality at the top of this function.")
}