From 7fd1f6a17b9f8eda998d2564c32e2432f1700c41 Mon Sep 17 00:00:00 2001 From: Petr Pchelko Date: Thu, 25 May 2017 15:20:16 -0700 Subject: [PATCH] JobQueue: Create a debugging queue As a part of creating the kafka-based JobQueue implementation, we need to transition to the new queue smoothly and be able to first debug the event production side, then try executing a no-op job in parallel with each of the real jobs (to be able to check deduplication and all the jpb management code) and only then try switching jobs one-by-one to the new backend queue. In order to achieve that at first we need to temporary produce both to the old job pipeline and to the new one - thus creating a queue 'splitter' - it delegates all the methods to the main queue while the produce requests also go to the EventBus. This class it temporary, so it's marked deprecated right away. It will not be needed once the queue transition is over. Bug: T163380 Change-Id: I483b8f00f1c593ff3af6ba6826868b553339525f --- autoload.php | 1 + includes/jobqueue/JobQueueFederated.php | 2 +- includes/jobqueue/JobQueueSecondTestQueue.php | 280 ++++++++++++++++++ 3 files changed, 282 insertions(+), 1 deletion(-) create mode 100644 includes/jobqueue/JobQueueSecondTestQueue.php diff --git a/autoload.php b/autoload.php index a6128a43b9..e0cab4c734 100644 --- a/autoload.php +++ b/autoload.php @@ -679,6 +679,7 @@ $wgAutoloadLocalClasses = [ 'JobQueueMemory' => __DIR__ . '/includes/jobqueue/JobQueueMemory.php', 'JobQueueReadOnlyError' => __DIR__ . '/includes/jobqueue/JobQueue.php', 'JobQueueRedis' => __DIR__ . '/includes/jobqueue/JobQueueRedis.php', + 'JobQueueSecondTestQueue' => __DIR__ . '/includes/jobqueue/JobQueueSecondTestQueue.php', 'JobRunner' => __DIR__ . '/includes/jobqueue/JobRunner.php', 'JobSpecification' => __DIR__ . '/includes/jobqueue/JobSpecification.php', 'JpegHandler' => __DIR__ . '/includes/media/Jpeg.php', diff --git a/includes/jobqueue/JobQueueFederated.php b/includes/jobqueue/JobQueueFederated.php index 7fdd617ab3..44721d9e8f 100644 --- a/includes/jobqueue/JobQueueFederated.php +++ b/includes/jobqueue/JobQueueFederated.php @@ -28,7 +28,7 @@ * For example, one can set $wgJobTypeConf['refreshLinks'] to point to a * JobQueueFederated instance, which itself would consist of three JobQueueRedis * instances, each using their own redis server. This would allow for the jobs - * to be split (evenly or based on weights) accross multiple servers if a single + * to be split (evenly or based on weights) across multiple servers if a single * server becomes impractical or expensive. Different JobQueue classes can be mixed. * * The basic queue configuration (e.g. "order", "claimTTL") of a federated queue diff --git a/includes/jobqueue/JobQueueSecondTestQueue.php b/includes/jobqueue/JobQueueSecondTestQueue.php new file mode 100644 index 0000000000..a1935dfa0d --- /dev/null +++ b/includes/jobqueue/JobQueueSecondTestQueue.php @@ -0,0 +1,280 @@ + $params['wiki'], 'type' => $params['type'] ]; + $this->mainQueue = JobQueue::factory( $params['mainqueue'] + $conf ); + $this->debugQueue = JobQueue::factory( $params['debugqueue'] + $conf ); + + // We need to construct parent after creating the main and debug queue + // because super constructor calls some methods we delegate to the main queue. + parent::__construct( $params ); + } + + /** + * Get the allowed queue orders for configuration validation + * + * @return array Subset of (random, timestamp, fifo, undefined) + */ + protected function supportedOrders() { + return $this->mainQueue->supportedOrders(); + } + + /** + * Get the default queue order to use if configuration does not specify one + * + * @return string One of (random, timestamp, fifo, undefined) + */ + protected function optimalOrder() { + return $this->mainQueue->optimalOrder(); + } + + /** + * Find out if delayed jobs are supported for configuration validation + * + * @return bool Whether delayed jobs are supported + */ + protected function supportsDelayedJobs() { + return $this->mainQueue->supportsDelayedJobs(); + } + + /** + * @see JobQueue::isEmpty() + * @return bool + */ + protected function doIsEmpty() { + return $this->mainQueue->doIsEmpty(); + } + + /** + * @see JobQueue::getSize() + * @return int + */ + protected function doGetSize() { + return $this->mainQueue->doGetSize(); + } + + /** + * @see JobQueue::getAcquiredCount() + * @return int + */ + protected function doGetAcquiredCount() { + return $this->mainQueue->doGetAcquiredCount(); + } + + /** + * @see JobQueue::getDelayedCount() + * @return int + */ + protected function doGetDelayedCount() { + return $this->mainQueue->doGetDelayedCount(); + } + + /** + * @see JobQueue::getAbandonedCount() + * @return int + */ + protected function doGetAbandonedCount() { + return $this->mainQueue->doGetAbandonedCount(); + } + + /** + * @see JobQueue::batchPush() + * @param IJobSpecification[] $jobs + * @param int $flags + */ + protected function doBatchPush( array $jobs, $flags ) { + $this->mainQueue->doBatchPush( $jobs, $flags ); + + try { + $this->debugQueue->doBatchPush( $jobs, $flags ); + } catch ( Exception $exception ) { + MWExceptionHandler::logException( $exception ); + } + } + + /** + * @see JobQueue::pop() + * @return Job|bool + */ + protected function doPop() { + return $this->mainQueue->doPop(); + } + + /** + * @see JobQueue::ack() + * @param Job $job + */ + protected function doAck( Job $job ) { + return $this->mainQueue->doAck( $job ); + } + + /** + * @see JobQueue::deduplicateRootJob() + * @param IJobSpecification $job + * @throws MWException + * @return bool + */ + protected function doDeduplicateRootJob( IJobSpecification $job ) { + return $this->mainQueue->doDeduplicateRootJob( $job ); + } + + /** + * @see JobQueue::isRootJobOldDuplicate() + * @param Job $job + * @return bool + */ + protected function doIsRootJobOldDuplicate( Job $job ) { + return $this->mainQueue->doIsRootJobOldDuplicate( $job ); + } + + /** + * @param string $signature Hash identifier of the root job + * @return string + */ + protected function getRootJobCacheKey( $signature ) { + return $this->mainQueue->getRootJobCacheKey( $signature ); + } + + /** + * @see JobQueue::delete() + * @throws MWException + */ + protected function doDelete() { + return $this->mainQueue->doDelete(); + } + + /** + * @see JobQueue::waitForBackups() + * @return void + */ + protected function doWaitForBackups() { + $this->mainQueue->doWaitForBackups(); + } + + /** + * @see JobQueue::flushCaches() + * @return void + */ + protected function doFlushCaches() { + $this->mainQueue->doFlushCaches(); + } + + /** + * Get an iterator to traverse over all available jobs in this queue. + * This does not include jobs that are currently acquired or delayed. + * Note: results may be stale if the queue is concurrently modified. + * + * @return Iterator + * @throws JobQueueError + */ + public function getAllQueuedJobs() { + return $this->mainQueue->getAllQueuedJobs(); + } + + /** + * Get an iterator to traverse over all delayed jobs in this queue. + * Note: results may be stale if the queue is concurrently modified. + * + * @return Iterator + * @throws JobQueueError + * @since 1.22 + */ + public function getAllDelayedJobs() { + return $this->mainQueue->getAllDelayedJobs(); + } + + /** + * Get an iterator to traverse over all claimed jobs in this queue + * + * Callers should be quick to iterator over it or few results + * will be returned due to jobs being acknowledged and deleted + * + * @return Iterator + * @throws JobQueueError + * @since 1.26 + */ + public function getAllAcquiredJobs() { + return $this->mainQueue->getAllAcquiredJobs(); + } + + /** + * Get an iterator to traverse over all abandoned jobs in this queue + * + * @return Iterator + * @throws JobQueueError + * @since 1.25 + */ + public function getAllAbandonedJobs() { + return $this->mainQueue->getAllAbandonedJobs(); + } + + /** + * Do not use this function outside of JobQueue/JobQueueGroup + * + * @return string + * @since 1.22 + */ + public function getCoalesceLocationInternal() { + return $this->mainQueue->getCoalesceLocationInternal(); + } + + /** + * @see JobQueue::getSiblingQueuesWithJobs() + * @param array $types List of queues types + * @return array|null (list of queue types) or null if unsupported + */ + protected function doGetSiblingQueuesWithJobs( array $types ) { + return $this->mainQueue->doGetSiblingQueuesWithJobs( $types ); + } + + /** + * @see JobQueue::getSiblingQueuesSize() + * @param array $types List of queues types + * @return array|null (list of queue types) or null if unsupported + */ + protected function doGetSiblingQueueSizes( array $types ) { + return $this->mainQueue->doGetSiblingQueueSizes( $types ); + } + + /** + * @throws JobQueueReadOnlyError + */ + protected function assertNotReadOnly() { + $this->mainQueue->assertNotReadOnly(); + } +} -- 2.20.1