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
|
// Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
#include <vespa/log/log.h>
LOG_SETUP("job_tracked_maintenance_test");
#include <vespa/searchcore/proton/server/i_blockable_maintenance_job.h>
#include <vespa/searchcore/proton/server/job_tracked_maintenance_job.h>
#include <vespa/searchcore/proton/test/simple_job_tracker.h>
#include <vespa/vespalib/testkit/testapp.h>
#include <vespa/vespalib/util/closuretask.h>
#include <vespa/vespalib/util/gate.h>
#include <vespa/vespalib/util/threadstackexecutor.h>
using namespace proton;
using namespace vespalib;
using test::SimpleJobTracker;
using GateUP = std::unique_ptr<Gate>;
using GateVector = std::vector<GateUP>;
using BlockedReason = IBlockableMaintenanceJob::BlockedReason;
GateVector
getGateVector(size_t size)
{
GateVector retval;
for (size_t i = 0; i < size; ++i) {
retval.push_back(std::move(GateUP(new Gate())));
}
return retval;
}
struct MyMaintenanceJob : public IBlockableMaintenanceJob
{
GateVector _runGates;
size_t _runIdx;
bool _blocked;
MyMaintenanceJob(size_t numRuns)
: IBlockableMaintenanceJob("myjob", 10, 20),
_runGates(getGateVector(numRuns)),
_runIdx(0),
_blocked(false)
{}
void block() { setBlocked(BlockedReason::RESOURCE_LIMITS); }
void unBlock() { unBlock(BlockedReason::RESOURCE_LIMITS); }
virtual void setBlocked(BlockedReason) override { _blocked = true; }
virtual void unBlock(BlockedReason) override { _blocked = false; }
virtual bool isBlocked() const override { return _blocked; }
virtual bool run() override {
_runGates[_runIdx++]->await(5000);
return _runIdx == _runGates.size();
}
};
struct Fixture
{
SimpleJobTracker::SP _tracker;
IMaintenanceJob::UP _job;
MyMaintenanceJob *_myJob;
IMaintenanceJob::UP _trackedJob;
bool _runRetval;
GateVector _runGates;
size_t _runIdx;
ThreadStackExecutor _exec;
Fixture(size_t numRuns = 1)
: _tracker(new SimpleJobTracker(1)),
_job(new MyMaintenanceJob(numRuns)),
_myJob(static_cast<MyMaintenanceJob *>(_job.get())),
_trackedJob(new JobTrackedMaintenanceJob(_tracker, std::move(_job))),
_runRetval(false),
_runGates(getGateVector(numRuns)),
_runIdx(0),
_exec(1, 64000)
{
}
void runJob() {
_runRetval = _trackedJob->run();
_runGates[_runIdx++]->countDown();
}
void assertTracker(size_t startedGateCount, size_t endedGateCount) {
EXPECT_EQUAL(startedGateCount, _tracker->_started.getCount());
EXPECT_EQUAL(endedGateCount, _tracker->_ended.getCount());
}
void runJobAndWait(size_t runIdx, size_t startedGateCount, size_t endedGateCount) {
_exec.execute(makeTask(makeClosure(this, &Fixture::runJob)));
_tracker->_started.await(5000);
assertTracker(startedGateCount, endedGateCount);
_myJob->_runGates[runIdx]->countDown();
_runGates[runIdx]->await(5000);
}
};
TEST_F("require that maintenance job name, delay and interval are preserved", Fixture)
{
EXPECT_EQUAL("myjob", f._trackedJob->getName());
EXPECT_EQUAL(10, f._trackedJob->getDelay());
EXPECT_EQUAL(20, f._trackedJob->getInterval());
}
TEST_F("require that maintenance job that needs 1 run is tracked", Fixture)
{
f.assertTracker(1, 1);
f.runJobAndWait(0, 0, 1);
f.assertTracker(0, 0);
EXPECT_TRUE(f._runRetval);
}
TEST_F("require that maintenance job that needs several runs is tracked", Fixture(2))
{
f.assertTracker(1, 1);
f.runJobAndWait(0, 0, 1);
f.assertTracker(0, 1);
EXPECT_FALSE(f._runRetval);
f.runJobAndWait(1, 0, 1);
f.assertTracker(0, 0);
EXPECT_TRUE(f._runRetval);
}
TEST_F("require that maintenance job that is destroyed is tracked", Fixture(2))
{
f.assertTracker(1, 1);
f.runJobAndWait(0, 0, 1);
f.assertTracker(0, 1);
EXPECT_FALSE(f._runRetval);
f._trackedJob.reset();
f.assertTracker(0, 0);
}
TEST_F("require that block calls are sent to underlying jobs", Fixture)
{
EXPECT_FALSE(f._trackedJob->isBlocked());
EXPECT_TRUE(f._trackedJob->asBlockable() != nullptr);
f._myJob->block();
EXPECT_TRUE(f._myJob->isBlocked());
EXPECT_TRUE(f._trackedJob->isBlocked());
f._myJob->unBlock();
EXPECT_FALSE(f._myJob->isBlocked());
EXPECT_FALSE(f._trackedJob->isBlocked());
}
TEST_MAIN() { TEST_RUN_ALL(); }
|
