3 * Job queue aggregator code that uses PhpRedis.
5 * This program is free software; you can redistribute it and/or modify
6 * it under the terms of the GNU General Public License as published by
7 * the Free Software Foundation; either version 2 of the License, or
8 * (at your option) any later version.
10 * This program is distributed in the hope that it will be useful,
11 * but WITHOUT ANY WARRANTY; without even the implied warranty of
12 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13 * GNU General Public License for more details.
15 * You should have received a copy of the GNU General Public License along
16 * with this program; if not, write to the Free Software Foundation, Inc.,
17 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
18 * http://www.gnu.org/copyleft/gpl.html
22 use Psr\Log\LoggerInterface
;
25 * Class to handle tracking information about all queues using PhpRedis
27 * The mediawiki/services/jobrunner background service must be set up and running.
33 class JobQueueAggregatorRedis
extends JobQueueAggregator
{
34 /** @var RedisConnectionPool */
36 /** @var LoggerInterface */
38 /** @var array List of Redis server addresses */
42 * @param array $params Possible keys:
43 * - redisConfig : An array of parameters to RedisConnectionPool::__construct().
44 * - redisServers : Array of server entries, the first being the primary and the
45 * others being fallback servers. Each entry is either a hostname/port
46 * combination or the absolute path of a UNIX socket.
47 * If a hostname is specified but no port, the standard port number
48 * 6379 will be used. Required.
50 public function __construct( array $params ) {
51 parent
::__construct( $params );
52 $this->servers
= isset( $params['redisServers'] )
53 ?
$params['redisServers']
54 : [ $params['redisServer'] ]; // b/c
55 $params['redisConfig']['serializer'] = 'none';
56 $this->redisPool
= RedisConnectionPool
::singleton( $params['redisConfig'] );
57 $this->logger
= \MediaWiki\Logger\LoggerFactory
::getInstance( 'redis' );
60 protected function doNotifyQueueEmpty( $wiki, $type ) {
61 return true; // managed by the service
64 protected function doNotifyQueueNonEmpty( $wiki, $type ) {
65 return true; // managed by the service
68 protected function doGetAllReadyWikiQueues() {
69 $conn = $this->getConnection();
74 $map = $conn->hGetAll( $this->getReadyQueueKey() );
76 if ( is_array( $map ) && isset( $map['_epoch'] ) ) {
77 unset( $map['_epoch'] ); // ignore
78 $pendingDBs = []; // (type => list of wikis)
79 foreach ( $map as $key => $time ) {
80 list( $type, $wiki ) = $this->decodeQueueName( $key );
81 $pendingDBs[$type][] = $wiki;
84 throw new UnexpectedValueException(
85 "No queue listing found; make sure redisJobChronService is running."
90 } catch ( RedisException
$e ) {
91 $this->redisPool
->handleError( $conn, $e );
97 protected function doPurge() {
98 return true; // fully and only refreshed by the service
102 * Get a connection to the server that handles all sub-queues for this queue
104 * @return RedisConnRef|bool Returns false on failure
105 * @throws MWException
107 protected function getConnection() {
109 foreach ( $this->servers
as $server ) {
110 $conn = $this->redisPool
->getConnection( $server, $this->logger
);
122 private function getReadyQueueKey() {
123 return "jobqueue:aggregator:h-ready-queues:v2"; // global
127 * @param string $name
130 private function decodeQueueName( $name ) {
131 list( $type, $wiki ) = explode( '/', $name, 2 );
133 return [ rawurldecode( $type ), rawurldecode( $wiki ) ];