Lock jobs while executing them, to allow multiple executors to run in parallel
This commit is contained in:
parent
7e3ce83526
commit
d0a2fa0506
1
cron.php
1
cron.php
|
@ -138,6 +138,7 @@ try {
|
||||||
$executedJobs = [];
|
$executedJobs = [];
|
||||||
while ($job = $jobList->getNext()) {
|
while ($job = $jobList->getNext()) {
|
||||||
if (isset($executedJobs[$job->getId()])) {
|
if (isset($executedJobs[$job->getId()])) {
|
||||||
|
$jobList->unlockJob($job);
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
@ -968,12 +968,29 @@
|
||||||
</field>
|
</field>
|
||||||
|
|
||||||
<field>
|
<field>
|
||||||
|
<!-- timestamp when the job was executed the last time -->
|
||||||
<name>last_run</name>
|
<name>last_run</name>
|
||||||
<type>integer</type>
|
<type>integer</type>
|
||||||
<default></default>
|
<default></default>
|
||||||
<notnull>false</notnull>
|
<notnull>false</notnull>
|
||||||
</field>
|
</field>
|
||||||
|
|
||||||
|
<field>
|
||||||
|
<!-- timestamp when the job was checked if it needs execution the last time -->
|
||||||
|
<name>last_checked</name>
|
||||||
|
<type>integer</type>
|
||||||
|
<default></default>
|
||||||
|
<notnull>false</notnull>
|
||||||
|
</field>
|
||||||
|
|
||||||
|
<field>
|
||||||
|
<!-- timestamp when the job was reserved the last time, 1 day timeout -->
|
||||||
|
<name>reserved_at</name>
|
||||||
|
<type>integer</type>
|
||||||
|
<default></default>
|
||||||
|
<notnull>false</notnull>
|
||||||
|
</field>
|
||||||
|
|
||||||
<index>
|
<index>
|
||||||
<name>job_class_index</name>
|
<name>job_class_index</name>
|
||||||
<field>
|
<field>
|
||||||
|
|
|
@ -25,27 +25,34 @@
|
||||||
namespace OC\BackgroundJob;
|
namespace OC\BackgroundJob;
|
||||||
|
|
||||||
use OCP\AppFramework\QueryException;
|
use OCP\AppFramework\QueryException;
|
||||||
|
use OCP\AppFramework\Utility\ITimeFactory;
|
||||||
use OCP\BackgroundJob\IJob;
|
use OCP\BackgroundJob\IJob;
|
||||||
use OCP\BackgroundJob\IJobList;
|
use OCP\BackgroundJob\IJobList;
|
||||||
use OCP\AutoloadNotAllowedException;
|
use OCP\AutoloadNotAllowedException;
|
||||||
use OCP\DB\QueryBuilder\IQueryBuilder;
|
use OCP\DB\QueryBuilder\IQueryBuilder;
|
||||||
|
use OCP\IConfig;
|
||||||
|
use OCP\IDBConnection;
|
||||||
|
|
||||||
class JobList implements IJobList {
|
class JobList implements IJobList {
|
||||||
/** @var \OCP\IDBConnection */
|
|
||||||
|
/** @var IDBConnection */
|
||||||
protected $connection;
|
protected $connection;
|
||||||
|
|
||||||
/**
|
/**@var IConfig */
|
||||||
* @var \OCP\IConfig $config
|
|
||||||
*/
|
|
||||||
protected $config;
|
protected $config;
|
||||||
|
|
||||||
|
/**@var ITimeFactory */
|
||||||
|
protected $timeFactory;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* @param \OCP\IDBConnection $connection
|
* @param IDBConnection $connection
|
||||||
* @param \OCP\IConfig $config
|
* @param IConfig $config
|
||||||
|
* @param ITimeFactory $timeFactory
|
||||||
*/
|
*/
|
||||||
public function __construct($connection, $config) {
|
public function __construct(IDBConnection $connection, IConfig $config, ITimeFactory $timeFactory) {
|
||||||
$this->connection = $connection;
|
$this->connection = $connection;
|
||||||
$this->config = $config;
|
$this->config = $config;
|
||||||
|
$this->timeFactory = $timeFactory;
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
@ -71,6 +78,7 @@ class JobList implements IJobList {
|
||||||
'class' => $query->createNamedParameter($class),
|
'class' => $query->createNamedParameter($class),
|
||||||
'argument' => $query->createNamedParameter($argument),
|
'argument' => $query->createNamedParameter($argument),
|
||||||
'last_run' => $query->createNamedParameter(0, IQueryBuilder::PARAM_INT),
|
'last_run' => $query->createNamedParameter(0, IQueryBuilder::PARAM_INT),
|
||||||
|
'last_checked' => $query->createNamedParameter($this->timeFactory->getTime(), IQueryBuilder::PARAM_INT),
|
||||||
]);
|
]);
|
||||||
$query->execute();
|
$query->execute();
|
||||||
}
|
}
|
||||||
|
@ -167,45 +175,40 @@ class JobList implements IJobList {
|
||||||
* @return IJob|null
|
* @return IJob|null
|
||||||
*/
|
*/
|
||||||
public function getNext() {
|
public function getNext() {
|
||||||
$lastId = $this->getLastJob();
|
|
||||||
|
|
||||||
$query = $this->connection->getQueryBuilder();
|
$query = $this->connection->getQueryBuilder();
|
||||||
$query->select('*')
|
$query->select('*')
|
||||||
->from('jobs')
|
->from('jobs')
|
||||||
->where($query->expr()->lt('id', $query->createNamedParameter($lastId, IQueryBuilder::PARAM_INT)))
|
->where($query->expr()->lte('reserved_at', $query->createNamedParameter($this->timeFactory->getTime() - 12 * 3600, IQueryBuilder::PARAM_INT)))
|
||||||
->orderBy('id', 'DESC')
|
->orderBy('last_checked', 'ASC')
|
||||||
->setMaxResults(1);
|
->setMaxResults(1);
|
||||||
|
|
||||||
|
$update = $this->connection->getQueryBuilder();
|
||||||
|
$update->update('jobs')
|
||||||
|
->set('reserved_at', $update->createNamedParameter($this->timeFactory->getTime()))
|
||||||
|
->set('last_checked', $update->createNamedParameter($this->timeFactory->getTime()))
|
||||||
|
->where($update->expr()->eq('id', $update->createParameter('jobid')));
|
||||||
|
|
||||||
|
$this->connection->lockTable('jobs');
|
||||||
$result = $query->execute();
|
$result = $query->execute();
|
||||||
$row = $result->fetch();
|
$row = $result->fetch();
|
||||||
$result->closeCursor();
|
$result->closeCursor();
|
||||||
|
|
||||||
if ($row) {
|
if ($row) {
|
||||||
$jobId = $row['id'];
|
$update->setParameter('jobid', $row['id']);
|
||||||
|
$update->execute();
|
||||||
|
$this->connection->unlockTable();
|
||||||
|
|
||||||
$job = $this->buildJob($row);
|
$job = $this->buildJob($row);
|
||||||
} else {
|
|
||||||
//begin at the start of the queue
|
|
||||||
$query = $this->connection->getQueryBuilder();
|
|
||||||
$query->select('*')
|
|
||||||
->from('jobs')
|
|
||||||
->orderBy('id', 'DESC')
|
|
||||||
->setMaxResults(1);
|
|
||||||
$result = $query->execute();
|
|
||||||
$row = $result->fetch();
|
|
||||||
$result->closeCursor();
|
|
||||||
|
|
||||||
if ($row) {
|
if ($job === null) {
|
||||||
$jobId = $row['id'];
|
// Background job from disabled app, try again.
|
||||||
$job = $this->buildJob($row);
|
return $this->getNext();
|
||||||
} else {
|
|
||||||
return null; //empty job list
|
|
||||||
}
|
}
|
||||||
}
|
|
||||||
|
|
||||||
if (is_null($job)) {
|
|
||||||
$this->removeById($jobId);
|
|
||||||
return $this->getNext();
|
|
||||||
} else {
|
|
||||||
return $job;
|
return $job;
|
||||||
|
} else {
|
||||||
|
$this->connection->unlockTable();
|
||||||
|
return null;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -267,13 +270,30 @@ class JobList implements IJobList {
|
||||||
* @param IJob $job
|
* @param IJob $job
|
||||||
*/
|
*/
|
||||||
public function setLastJob($job) {
|
public function setLastJob($job) {
|
||||||
|
$this->unlockJob($job);
|
||||||
$this->config->setAppValue('backgroundjob', 'lastjob', $job->getId());
|
$this->config->setAppValue('backgroundjob', 'lastjob', $job->getId());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Remove the reservation for a job
|
||||||
|
*
|
||||||
|
* @param IJob $job
|
||||||
|
*/
|
||||||
|
public function unlockJob($job) {
|
||||||
|
$query = $this->connection->getQueryBuilder();
|
||||||
|
$query->update('jobs')
|
||||||
|
->set('reserved_at', $query->expr()->literal(0, IQueryBuilder::PARAM_INT))
|
||||||
|
->where($query->expr()->eq('id', $query->createNamedParameter($job->getId(), IQueryBuilder::PARAM_INT)));
|
||||||
|
$query->execute();
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* get the id of the last ran job
|
* get the id of the last ran job
|
||||||
*
|
*
|
||||||
* @return int
|
* @return int
|
||||||
|
* @deprecated 9.1.0 - The functionality behind the value is deprecated, it
|
||||||
|
* only tells you which job finished last, but since we now allow multiple
|
||||||
|
* executors to run in parallel, it's not used to calculate the next job.
|
||||||
*/
|
*/
|
||||||
public function getLastJob() {
|
public function getLastJob() {
|
||||||
return (int) $this->config->getAppValue('backgroundjob', 'lastjob', 0);
|
return (int) $this->config->getAppValue('backgroundjob', 'lastjob', 0);
|
||||||
|
|
|
@ -362,7 +362,11 @@ class Server extends ServerContainer implements IServerContainer {
|
||||||
});
|
});
|
||||||
$this->registerService('JobList', function (Server $c) {
|
$this->registerService('JobList', function (Server $c) {
|
||||||
$config = $c->getConfig();
|
$config = $c->getConfig();
|
||||||
return new \OC\BackgroundJob\JobList($c->getDatabaseConnection(), $config);
|
return new \OC\BackgroundJob\JobList(
|
||||||
|
$c->getDatabaseConnection(),
|
||||||
|
$config,
|
||||||
|
new TimeFactory()
|
||||||
|
);
|
||||||
});
|
});
|
||||||
$this->registerService('Router', function (Server $c) {
|
$this->registerService('Router', function (Server $c) {
|
||||||
$cacheFactory = $c->getMemCacheFactory();
|
$cacheFactory = $c->getMemCacheFactory();
|
||||||
|
|
|
@ -92,11 +92,22 @@ interface IJobList {
|
||||||
*/
|
*/
|
||||||
public function setLastJob($job);
|
public function setLastJob($job);
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Remove the reservation for a job
|
||||||
|
*
|
||||||
|
* @param IJob $job
|
||||||
|
* @since 9.1.0
|
||||||
|
*/
|
||||||
|
public function unlockJob($job);
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* get the id of the last ran job
|
* get the id of the last ran job
|
||||||
*
|
*
|
||||||
* @return int
|
* @return int
|
||||||
* @since 7.0.0
|
* @since 7.0.0
|
||||||
|
* @deprecated 9.1.0 - The functionality behind the value is deprecated, it
|
||||||
|
* only tells you which job finished last, but since we now allow multiple
|
||||||
|
* executors to run in parallel, it's not used to calculate the next job.
|
||||||
*/
|
*/
|
||||||
public function getLastJob();
|
public function getLastJob();
|
||||||
|
|
||||||
|
|
|
@ -9,7 +9,7 @@
|
||||||
namespace Test\BackgroundJob;
|
namespace Test\BackgroundJob;
|
||||||
|
|
||||||
use OCP\BackgroundJob\IJob;
|
use OCP\BackgroundJob\IJob;
|
||||||
use OCP\IDBConnection;
|
use OCP\DB\QueryBuilder\IQueryBuilder;
|
||||||
use Test\TestCase;
|
use Test\TestCase;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
@ -22,20 +22,31 @@ class JobListTest extends TestCase {
|
||||||
/** @var \OC\BackgroundJob\JobList */
|
/** @var \OC\BackgroundJob\JobList */
|
||||||
protected $instance;
|
protected $instance;
|
||||||
|
|
||||||
|
/** @var \OCP\IDBConnection */
|
||||||
|
protected $connection;
|
||||||
|
|
||||||
/** @var \OCP\IConfig|\PHPUnit_Framework_MockObject_MockObject */
|
/** @var \OCP\IConfig|\PHPUnit_Framework_MockObject_MockObject */
|
||||||
protected $config;
|
protected $config;
|
||||||
|
|
||||||
|
/** @var \OCP\AppFramework\Utility\ITimeFactory|\PHPUnit_Framework_MockObject_MockObject */
|
||||||
|
protected $timeFactory;
|
||||||
|
|
||||||
protected function setUp() {
|
protected function setUp() {
|
||||||
parent::setUp();
|
parent::setUp();
|
||||||
|
|
||||||
$connection = \OC::$server->getDatabaseConnection();
|
$this->connection = \OC::$server->getDatabaseConnection();
|
||||||
$this->clearJobsList($connection);
|
$this->clearJobsList();
|
||||||
$this->config = $this->getMock('\OCP\IConfig');
|
$this->config = $this->getMock('OCP\IConfig');
|
||||||
$this->instance = new \OC\BackgroundJob\JobList($connection, $this->config);
|
$this->timeFactory = $this->getMock('OCP\AppFramework\Utility\ITimeFactory');
|
||||||
|
$this->instance = new \OC\BackgroundJob\JobList(
|
||||||
|
$this->connection,
|
||||||
|
$this->config,
|
||||||
|
$this->timeFactory
|
||||||
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
protected function clearJobsList(IDBConnection $connection) {
|
protected function clearJobsList() {
|
||||||
$query = $connection->getQueryBuilder();
|
$query = $this->connection->getQueryBuilder();
|
||||||
$query->delete('jobs');
|
$query->delete('jobs');
|
||||||
$query->execute();
|
$query->execute();
|
||||||
}
|
}
|
||||||
|
@ -131,8 +142,6 @@ class JobListTest extends TestCase {
|
||||||
$this->instance->add($job, $argument);
|
$this->instance->add($job, $argument);
|
||||||
|
|
||||||
$this->assertFalse($this->instance->has($job, 10));
|
$this->assertFalse($this->instance->has($job, 10));
|
||||||
|
|
||||||
$this->instance->remove($job, $argument);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
public function testGetLastJob() {
|
public function testGetLastJob() {
|
||||||
|
@ -144,50 +153,65 @@ class JobListTest extends TestCase {
|
||||||
$this->assertEquals(15, $this->instance->getLastJob());
|
$this->assertEquals(15, $this->instance->getLastJob());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
protected function createTempJob($class, $argument, $reservedTime = 0, $lastChecked = 0) {
|
||||||
|
if ($lastChecked === 0) {
|
||||||
|
$lastChecked = time();
|
||||||
|
}
|
||||||
|
|
||||||
|
$query = $this->connection->getQueryBuilder();
|
||||||
|
$query->insert('jobs')
|
||||||
|
->values([
|
||||||
|
'class' => $query->createNamedParameter($class),
|
||||||
|
'argument' => $query->createNamedParameter($argument),
|
||||||
|
'last_run' => $query->createNamedParameter(0, IQueryBuilder::PARAM_INT),
|
||||||
|
'last_checked' => $query->createNamedParameter($lastChecked, IQueryBuilder::PARAM_INT),
|
||||||
|
'reserved_at' => $query->createNamedParameter($reservedTime, IQueryBuilder::PARAM_INT),
|
||||||
|
]);
|
||||||
|
$query->execute();
|
||||||
|
}
|
||||||
|
|
||||||
public function testGetNext() {
|
public function testGetNext() {
|
||||||
$job = new TestJob();
|
$job = new TestJob();
|
||||||
$this->instance->add($job, 1);
|
$this->createTempJob(get_class($job), 1, 0, 12345);
|
||||||
$this->instance->add($job, 2);
|
$this->createTempJob(get_class($job), 2, 0, 12346);
|
||||||
|
|
||||||
$jobs = $this->getAllSorted();
|
$jobs = $this->getAllSorted();
|
||||||
|
$savedJob1 = $jobs[0];
|
||||||
|
|
||||||
$savedJob1 = $jobs[count($jobs) - 2];
|
$this->timeFactory->expects($this->atLeastOnce())
|
||||||
$savedJob2 = $jobs[count($jobs) - 1];
|
->method('getTime')
|
||||||
|
->willReturn(123456789);
|
||||||
$this->config->expects($this->once())
|
|
||||||
->method('getAppValue')
|
|
||||||
->with('backgroundjob', 'lastjob', 0)
|
|
||||||
->will($this->returnValue($savedJob2->getId()));
|
|
||||||
|
|
||||||
$nextJob = $this->instance->getNext();
|
$nextJob = $this->instance->getNext();
|
||||||
|
|
||||||
$this->assertEquals($savedJob1, $nextJob);
|
$this->assertEquals($savedJob1, $nextJob);
|
||||||
|
|
||||||
$this->instance->remove($job, 1);
|
|
||||||
$this->instance->remove($job, 2);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
public function testGetNextWrapAround() {
|
public function testGetNextSkipReserved() {
|
||||||
$job = new TestJob();
|
$job = new TestJob();
|
||||||
$this->instance->add($job, 1);
|
$this->createTempJob(get_class($job), 1, 123456789, 12345);
|
||||||
$this->instance->add($job, 2);
|
$this->createTempJob(get_class($job), 2, 0, 12346);
|
||||||
|
|
||||||
$jobs = $this->getAllSorted();
|
|
||||||
|
|
||||||
$savedJob1 = $jobs[count($jobs) - 2];
|
|
||||||
$savedJob2 = $jobs[count($jobs) - 1];
|
|
||||||
|
|
||||||
$this->config->expects($this->once())
|
|
||||||
->method('getAppValue')
|
|
||||||
->with('backgroundjob', 'lastjob', 0)
|
|
||||||
->will($this->returnValue($savedJob1->getId()));
|
|
||||||
|
|
||||||
|
$this->timeFactory->expects($this->atLeastOnce())
|
||||||
|
->method('getTime')
|
||||||
|
->willReturn(123456789);
|
||||||
$nextJob = $this->instance->getNext();
|
$nextJob = $this->instance->getNext();
|
||||||
|
|
||||||
$this->assertEquals($savedJob2, $nextJob);
|
$this->assertEquals(get_class($job), get_class($nextJob));
|
||||||
|
$this->assertEquals(2, $nextJob->getArgument());
|
||||||
|
}
|
||||||
|
|
||||||
$this->instance->remove($job, 1);
|
public function testGetNextSkipNonExisting() {
|
||||||
$this->instance->remove($job, 2);
|
$job = new TestJob();
|
||||||
|
$this->createTempJob('\OC\Non\Existing\Class', 1, 0, 12345);
|
||||||
|
$this->createTempJob(get_class($job), 2, 0, 12346);
|
||||||
|
|
||||||
|
$this->timeFactory->expects($this->atLeastOnce())
|
||||||
|
->method('getTime')
|
||||||
|
->willReturn(123456789);
|
||||||
|
$nextJob = $this->instance->getNext();
|
||||||
|
|
||||||
|
$this->assertEquals(get_class($job), get_class($nextJob));
|
||||||
|
$this->assertEquals(2, $nextJob->getArgument());
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
@ -203,8 +227,6 @@ class JobListTest extends TestCase {
|
||||||
$addedJob = $jobs[count($jobs) - 1];
|
$addedJob = $jobs[count($jobs) - 1];
|
||||||
|
|
||||||
$this->assertEquals($addedJob, $this->instance->getById($addedJob->getId()));
|
$this->assertEquals($addedJob, $this->instance->getById($addedJob->getId()));
|
||||||
|
|
||||||
$this->instance->remove($job, $argument);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
public function testSetLastRun() {
|
public function testSetLastRun() {
|
||||||
|
@ -223,33 +245,5 @@ class JobListTest extends TestCase {
|
||||||
|
|
||||||
$this->assertGreaterThanOrEqual($timeStart, $addedJob->getLastRun());
|
$this->assertGreaterThanOrEqual($timeStart, $addedJob->getLastRun());
|
||||||
$this->assertLessThanOrEqual($timeEnd, $addedJob->getLastRun());
|
$this->assertLessThanOrEqual($timeEnd, $addedJob->getLastRun());
|
||||||
|
|
||||||
$this->instance->remove($job);
|
|
||||||
}
|
|
||||||
|
|
||||||
public function testGetNextNonExisting() {
|
|
||||||
$job = new TestJob();
|
|
||||||
$this->instance->add($job, 1);
|
|
||||||
$this->instance->add('\OC\Non\Existing\Class');
|
|
||||||
$this->instance->add($job, 2);
|
|
||||||
|
|
||||||
$jobs = $this->getAllSorted();
|
|
||||||
|
|
||||||
$savedJob1 = $jobs[count($jobs) - 2];
|
|
||||||
$savedJob2 = $jobs[count($jobs) - 1];
|
|
||||||
|
|
||||||
$this->config->expects($this->any())
|
|
||||||
->method('getAppValue')
|
|
||||||
->with('backgroundjob', 'lastjob', 0)
|
|
||||||
->will($this->returnValue($savedJob1->getId()));
|
|
||||||
|
|
||||||
$this->instance->getNext();
|
|
||||||
|
|
||||||
$nextJob = $this->instance->getNext();
|
|
||||||
|
|
||||||
$this->assertEquals($savedJob2, $nextJob);
|
|
||||||
|
|
||||||
$this->instance->remove($job, 1);
|
|
||||||
$this->instance->remove($job, 2);
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -26,7 +26,7 @@
|
||||||
// We only can count up. The 4. digit is only for the internal patchlevel to trigger DB upgrades
|
// We only can count up. The 4. digit is only for the internal patchlevel to trigger DB upgrades
|
||||||
// between betas, final and RCs. This is _not_ the public version number. Reset minor/patchlevel
|
// between betas, final and RCs. This is _not_ the public version number. Reset minor/patchlevel
|
||||||
// when updating major/minor version number.
|
// when updating major/minor version number.
|
||||||
$OC_Version = array(9, 1, 0, 2);
|
$OC_Version = array(9, 1, 0, 3);
|
||||||
|
|
||||||
// The human readable string
|
// The human readable string
|
||||||
$OC_VersionString = '9.1.0 pre alpha';
|
$OC_VersionString = '9.1.0 pre alpha';
|
||||||
|
|
Loading…
Reference in New Issue