blob: 896443117c9eb78cb2728e49101692990fe83067 (
plain) (
blame)
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
|
// Copyright Verizon Media. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
package com.yahoo.concurrent.maintenance;
import com.yahoo.transaction.Mutex;
import java.util.Collections;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentSkipListMap;
/**
* Provides status and control over running maintenance jobs.
*
* This is multi-thread safe.
*
* @author bratseth
*/
public class JobControl {
/** This is not persisted as all nodes start all jobs */
private final Map<String, Maintainer> startedJobs = new ConcurrentSkipListMap<>();
/** Used to store deactivation in a persistent shared database to make changes take effect on all nodes */
private final Db db;
public JobControl(Db db) {
this.db = db;
}
/** Notifies this that a job was started */
public void started(String jobSimpleClassName, Maintainer maintainer) {
startedJobs.put(jobSimpleClassName, maintainer);
}
/**
* Returns a snapshot of the set of jobs started on this system (whether deactivated or not).
* Each job is represented by its simple (omitting package) class name.
*/
public Set<String> jobs() { return Collections.unmodifiableSet(startedJobs.keySet()); }
/** Returns a snapshot containing the currently inactive jobs in this */
public Set<String> inactiveJobs() { return db.readInactiveJobs(); }
/** Returns true if this job is not currently deactivated */
public boolean isActive(String jobSimpleClassName) {
return ! inactiveJobs().contains(jobSimpleClassName);
}
/** Set a job active or inactive */
public void setActive(String jobSimpleClassName, boolean active) {
try (var lock = db.lockInactiveJobs()) {
Set<String> inactiveJobs = db.readInactiveJobs();
if (active)
inactiveJobs.remove(jobSimpleClassName);
else
inactiveJobs.add(jobSimpleClassName);
db.writeInactiveJobs(inactiveJobs);
}
}
/** Run given job (inactive or not) immediately */
public void run(String jobSimpleClassName) {
var job = startedJobs.get(jobSimpleClassName);
if (job == null) throw new IllegalArgumentException("No such job '" + jobSimpleClassName + "'");
job.lockAndMaintain();
}
/** Acquire lock for running given job */
public Mutex lockJob(String jobSimpleClassName) {
return db.lockMaintenanceJob(jobSimpleClassName);
}
/** The database used for managing job state and synchronization */
public interface Db {
/** Returns the set of jobs that are temporarily inactive */
Set<String> readInactiveJobs();
/** Make given jobs as inactive */
void writeInactiveJobs(Set<String> inactiveJobs);
/** Acquire lock for changing jobs */
Mutex lockInactiveJobs();
/** Acquire lock for running given job */
Mutex lockMaintenanceJob(String job);
}
}
|
