3 * BackupDumper that postprocesses XML dumps from dumpBackup.php to add page text
5 * Copyright (C) 2005 Brion Vibber <brion@pobox.com>
6 * http://www.mediawiki.org/
8 * This program is free software; you can redistribute it and/or modify
9 * it under the terms of the GNU General Public License as published by
10 * the Free Software Foundation; either version 2 of the License, or
11 * (at your option) any later version.
13 * This program is distributed in the hope that it will be useful,
14 * but WITHOUT ANY WARRANTY; without even the implied warranty of
15 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
16 * GNU General Public License for more details.
18 * You should have received a copy of the GNU General Public License along
19 * with this program; if not, write to the Free Software Foundation, Inc.,
20 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
21 * http://www.gnu.org/copyleft/gpl.html
24 * @ingroup Maintenance
29 * @ingroup Maintenance
31 class TextPassDumper extends BackupDumper {
33 var $input = "php://stdin";
34 var $history = WikiExporter::FULL;
36 var $prefetchCount = 0;
37 var $prefetchCountLast = 0;
38 var $fetchCountLast = 0;
41 var $maxConsecutiveFailedTextRetrievals = 200;
42 var $failureTimeout = 5; // Seconds to sleep after db failure
46 var $spawnProc = false;
47 var $spawnWrite = false;
48 var $spawnRead = false;
49 var $spawnErr = false;
51 var $xmlwriterobj = false;
53 // when we spend more than maxTimeAllowed seconds on this run, we continue
54 // processing until we write out the next complete page, then save output file(s),
55 // rename it/them and open new one(s)
56 var $maxTimeAllowed = 0; // 0 = no limit
57 var $timeExceeded = false;
58 var $firstPageWritten = false;
59 var $lastPageWritten = false;
60 var $checkpointJustWritten = false;
61 var $checkpointFiles = array();
70 * Drop the database connection $this->db and try to get a new one.
72 * This function tries to get a /different/ connection if this is
73 * possible. Hence, (if this is possible) it switches to a different
74 * failover upon each call.
76 * This function resets $this->lb and closes all connections on it.
81 // Cleaning up old connections
82 if ( isset( $this->lb ) ) {
83 $this->lb->closeAll();
87 if ( $this->forcedDb !== null ) {
88 $this->db = $this->forcedDb;
92 if ( isset( $this->db ) && $this->db->isOpen() ) {
93 throw new MWException( 'DB is set and has not been closed by the Load Balancer' );
98 // Trying to set up new connection.
99 // We do /not/ retry upon failure, but delegate to encapsulating logic, to avoid
100 // individually retrying at different layers of code.
102 // 1. The LoadBalancer.
104 $this->lb = wfGetLBFactory()->newMainLB();
105 } catch ( Exception $e ) {
106 throw new MWException( __METHOD__ . " rotating DB failed to obtain new load balancer (" . $e->getMessage() . ")" );
110 // 2. The Connection, through the load balancer.
112 $this->db = $this->lb->getConnection( DB_SLAVE, 'backup' );
113 } catch ( Exception $e ) {
114 throw new MWException( __METHOD__ . " rotating DB failed to obtain new database (" . $e->getMessage() . ")" );
119 function initProgress( $history = WikiExporter::FULL ) {
120 parent::initProgress();
121 $this->timeOfCheckpoint = $this->startTime;
124 function dump( $history, $text = WikiExporter::TEXT ) {
125 // Notice messages will foul up your XML output even if they're
126 // relatively harmless.
127 if ( ini_get( 'display_errors' ) )
128 ini_set( 'display_errors', 'stderr' );
130 $this->initProgress( $this->history );
132 // We are trying to get an initial database connection to avoid that the
133 // first try of this request's first call to getText fails. However, if
134 // obtaining a good DB connection fails it's not a serious issue, as
135 // getText does retry upon failure and can start without having a working
139 } catch ( Exception $e ) {
140 // We do not even count this as failure. Just let eventual
142 $this->progress( "Getting initial DB connection failed (" .
143 $e->getMessage() . ")" );
146 $this->egress = new ExportProgressFilter( $this->sink, $this );
148 // it would be nice to do it in the constructor, oh well. need egress set
149 $this->finalOptionCheck();
151 // we only want this so we know how to close a stream :-P
152 $this->xmlwriterobj = new XmlDumpWriter();
154 $input = fopen( $this->input, "rt" );
155 $result = $this->readDump( $input );
157 if ( WikiError::isError( $result ) ) {
158 throw new MWException( $result->getMessage() );
161 if ( $this->spawnProc ) {
165 $this->report( true );
168 function processOption( $opt, $val, $param ) {
170 $url = $this->processFileOpt( $val, $param );
174 require_once "$IP/maintenance/backupPrefetch.inc";
175 $this->prefetch = new BaseDump( $url );
181 $this->maxTimeAllowed = intval( $val ) * 60;
183 case 'checkpointfile':
184 $this->checkpointFiles[] = $val;
187 $this->history = WikiExporter::CURRENT;
190 $this->history = WikiExporter::FULL;
201 function processFileOpt( $val, $param ) {
202 $fileURIs = explode( ';', $param );
203 foreach ( $fileURIs as $URI ) {
209 $newURI = "compress.zlib://$URI";
212 $newURI = "compress.bzip2://$URI";
215 $newURI = "mediawiki.compress.7z://$URI";
220 $newFileURIs[] = $newURI;
222 $val = implode( ';', $newFileURIs );
227 * Overridden to include prefetch ratio if enabled.
229 function showReport() {
230 if ( !$this->prefetch ) {
231 parent::showReport();
235 if ( $this->reporting ) {
236 $now = wfTimestamp( TS_DB );
238 $deltaAll = wfTime() - $this->startTime;
239 $deltaPart = wfTime() - $this->lastTime;
240 $this->pageCountPart = $this->pageCount - $this->pageCountLast;
241 $this->revCountPart = $this->revCount - $this->revCountLast;
244 $portion = $this->revCount / $this->maxCount;
245 $eta = $this->startTime + $deltaAll / $portion;
246 $etats = wfTimestamp( TS_DB, intval( $eta ) );
247 if ( $this->fetchCount ) {
248 $fetchRate = 100.0 * $this->prefetchCount / $this->fetchCount;
252 $pageRate = $this->pageCount / $deltaAll;
253 $revRate = $this->revCount / $deltaAll;
261 if ( $this->fetchCountLast ) {
262 $fetchRatePart = 100.0 * $this->prefetchCountLast / $this->fetchCountLast;
264 $fetchRatePart = '-';
266 $pageRatePart = $this->pageCountPart / $deltaPart;
267 $revRatePart = $this->revCountPart / $deltaPart;
270 $fetchRatePart = '-';
274 $this->progress( sprintf( "%s: %s (ID %d) %d pages (%0.1f|%0.1f/sec all|curr), %d revs (%0.1f|%0.1f/sec all|curr), %0.1f%%|%0.1f%% prefetched (all|curr), ETA %s [max %d]",
275 $now, wfWikiID(), $this->ID, $this->pageCount, $pageRate, $pageRatePart, $this->revCount, $revRate, $revRatePart, $fetchRate, $fetchRatePart, $etats, $this->maxCount ) );
276 $this->lastTime = $nowts;
277 $this->revCountLast = $this->revCount;
278 $this->prefetchCountLast = $this->prefetchCount;
279 $this->fetchCountLast = $this->fetchCount;
283 function setTimeExceeded() {
284 $this->timeExceeded = True;
287 function checkIfTimeExceeded() {
288 if ( $this->maxTimeAllowed && ( $this->lastTime - $this->timeOfCheckpoint > $this->maxTimeAllowed ) ) {
294 function finalOptionCheck() {
295 if ( ( $this->checkpointFiles && ! $this->maxTimeAllowed ) ||
296 ( $this->maxTimeAllowed && !$this->checkpointFiles ) ) {
297 throw new MWException( "Options checkpointfile and maxtime must be specified together.\n" );
299 foreach ( $this->checkpointFiles as $checkpointFile ) {
300 $count = substr_count ( $checkpointFile, "%s" );
302 throw new MWException( "Option checkpointfile must contain two '%s' for substitution of first and last pageids, count is $count instead, file is $checkpointFile.\n" );
306 if ( $this->checkpointFiles ) {
307 $filenameList = (array)$this->egress->getFilenames();
308 if ( count( $filenameList ) != count( $this->checkpointFiles ) ) {
309 throw new MWException( "One checkpointfile must be specified for each output option, if maxtime is used.\n" );
314 function readDump( $input ) {
316 $this->openElement = false;
317 $this->atStart = true;
319 $this->lastName = "";
323 $parser = xml_parser_create( "UTF-8" );
324 xml_parser_set_option( $parser, XML_OPTION_CASE_FOLDING, false );
326 xml_set_element_handler( $parser, array( &$this, 'startElement' ), array( &$this, 'endElement' ) );
327 xml_set_character_data_handler( $parser, array( &$this, 'characterData' ) );
329 $offset = 0; // for context extraction on error reporting
330 $bufferSize = 512 * 1024;
332 if ( $this->checkIfTimeExceeded() ) {
333 $this->setTimeExceeded();
335 $chunk = fread( $input, $bufferSize );
336 if ( !xml_parse( $parser, $chunk, feof( $input ) ) ) {
337 wfDebug( "TextDumpPass::readDump encountered XML parsing error\n" );
338 return new WikiXmlError( $parser, 'XML import parse failure', $chunk, $offset );
340 $offset += strlen( $chunk );
341 } while ( $chunk !== false && !feof( $input ) );
342 if ( $this->maxTimeAllowed ) {
343 $filenameList = (array)$this->egress->getFilenames();
344 // we wrote some stuff after last checkpoint that needs renamed
345 if ( file_exists( $filenameList[0] ) ) {
346 $newFilenames = array();
347 # we might have just written the header and footer and had no
348 # pages or revisions written... perhaps they were all deleted
349 # there's no pageID 0 so we use that. the caller is responsible
350 # for deciding what to do with a file containing only the
351 # siteinfo information and the mw tags.
352 if ( ! $this->firstPageWritten ) {
353 $firstPageID = str_pad( 0, 9, "0", STR_PAD_LEFT );
354 $lastPageID = str_pad( 0, 9, "0", STR_PAD_LEFT );
357 $firstPageID = str_pad( $this->firstPageWritten, 9, "0", STR_PAD_LEFT );
358 $lastPageID = str_pad( $this->lastPageWritten, 9, "0", STR_PAD_LEFT );
360 for ( $i = 0; $i < count( $filenameList ); $i++ ) {
361 $checkpointNameFilledIn = sprintf( $this->checkpointFiles[$i], $firstPageID, $lastPageID );
362 $fileinfo = pathinfo( $filenameList[$i] );
363 $newFilenames[] = $fileinfo['dirname'] . '/' . $checkpointNameFilledIn;
365 $this->egress->closeAndRename( $newFilenames );
368 xml_parser_free( $parser );
374 * Tries to get the revision text for a revision id.
376 * Upon errors, retries (Up to $this->maxFailures tries each call).
377 * If still no good revision get could be found even after this retrying, "" is returned.
378 * If no good revision text could be returned for
379 * $this->maxConsecutiveFailedTextRetrievals consecutive calls to getText, MWException
382 * @param $id string The revision id to get the text for
384 * @return string The revision text for $id, or ""
385 * @throws MWException
387 function getText( $id ) {
388 $prefetchNotTried = true; // Whether or not we already tried to get the text via prefetch.
389 $text = false; // The candidate for a good text. false if no proper value.
390 $failures = 0; // The number of times, this invocation of getText already failed.
392 static $consecutiveFailedTextRetrievals = 0; // The number of times getText failed without
393 // yielding a good text in between.
397 // To allow to simply return on success and do not have to worry about book keeping,
398 // we assume, this fetch works (possible after some retries). Nevertheless, we koop
399 // the old value, so we can restore it, if problems occur (See after the while loop).
400 $oldConsecutiveFailedTextRetrievals = $consecutiveFailedTextRetrievals;
401 $consecutiveFailedTextRetrievals = 0;
403 while ( $failures < $this->maxFailures ) {
405 // As soon as we found a good text for the $id, we will return immediately.
406 // Hence, if we make it past the try catch block, we know that we did not
410 // Step 1: Get some text (or reuse from previous iteratuon if checking
411 // for plausibility failed)
413 // Trying to get prefetch, if it has not been tried before
414 if ( $text === false && isset( $this->prefetch ) && $prefetchNotTried ) {
415 $prefetchNotTried = false;
416 $tryIsPrefetch = true;
417 $text = $this->prefetch->prefetch( $this->thisPage, $this->thisRev );
418 if ( $text === null ) {
423 if ( $text === false ) {
424 // Fallback to asking the database
425 $tryIsPrefetch = false;
426 if ( $this->spawn ) {
427 $text = $this->getTextSpawned( $id );
429 $text = $this->getTextDb( $id );
433 if ( $text === false ) {
434 throw new MWException( "Generic error while obtaining text for id " . $id );
437 // We received a good candidate for the text of $id via some method
439 // Step 2: Checking for plausibility and return the text if it is
441 $revID = intval( $this->thisRev );
442 if ( ! isset( $this->db ) ) {
443 throw new MWException( "No database available" );
445 $revLength = $this->db->selectField( 'revision', 'rev_len', array( 'rev_id' => $revID ) );
446 if ( strlen( $text ) == $revLength ) {
447 if ( $tryIsPrefetch ) {
448 $this->prefetchCount++;
454 throw new MWException( "Received text is unplausible for id " . $id );
456 } catch ( Exception $e ) {
457 $msg = "getting/checking text " . $id . " failed (" . $e->getMessage() . ")";
458 if ( $failures + 1 < $this->maxFailures ) {
459 $msg .= " (Will retry " . ( $this->maxFailures - $failures - 1 ) . " more times)";
461 $this->progress( $msg );
464 // Something went wrong; we did not a text that was plausible :(
468 // After backing off for some time, we try to reboot the whole process as
469 // much as possible to not carry over failures from one part to the other
471 sleep( $this->failureTimeout );
474 if ( $this->spawn ) {
478 } catch ( Exception $e ) {
479 $this->progress( "Rebooting getText infrastructure failed (" . $e->getMessage() . ")" .
480 " Trying to continue anyways" );
484 // Retirieving a good text for $id failed (at least) maxFailures times.
485 // We abort for this $id.
487 // Restoring the consecutive failures, and maybe aborting, if the dump
489 $consecutiveFailedTextRetrievals = $oldConsecutiveFailedTextRetrievals + 1;
490 if ( $consecutiveFailedTextRetrievals > $this->maxConsecutiveFailedTextRetrievals ) {
491 throw new MWException( "Graceful storage failure" );
499 * May throw a database error if, say, the server dies during query.
501 * @return bool|string
502 * @throws MWException
504 private function getTextDb( $id ) {
506 if ( ! isset( $this->db ) ) {
507 throw new MWException( __METHOD__ . "No database available" );
509 $row = $this->db->selectRow( 'text',
510 array( 'old_text', 'old_flags' ),
511 array( 'old_id' => $id ),
513 $text = Revision::getRevisionText( $row );
514 if ( $text === false ) {
517 $stripped = str_replace( "\r", "", $text );
518 $normalized = $wgContLang->normalize( $stripped );
522 private function getTextSpawned( $id ) {
523 wfSuppressWarnings();
524 if ( !$this->spawnProc ) {
528 $text = $this->getTextSpawnedOnce( $id );
533 function openSpawn() {
536 if ( file_exists( "$IP/../multiversion/MWScript.php" ) ) {
538 array_map( 'wfEscapeShellArg',
541 "$IP/../multiversion/MWScript.php",
543 '--wiki', wfWikiID() ) ) );
547 array_map( 'wfEscapeShellArg',
550 "$IP/maintenance/fetchText.php",
551 '--wiki', wfWikiID() ) ) );
554 0 => array( "pipe", "r" ),
555 1 => array( "pipe", "w" ),
556 2 => array( "file", "/dev/null", "a" ) );
559 $this->progress( "Spawning database subprocess: $cmd" );
560 $this->spawnProc = proc_open( $cmd, $spec, $pipes );
561 if ( !$this->spawnProc ) {
563 $this->progress( "Subprocess spawn failed." );
567 $this->spawnWrite, // -> stdin
568 $this->spawnRead, // <- stdout
574 private function closeSpawn() {
575 wfSuppressWarnings();
576 if ( $this->spawnRead )
577 fclose( $this->spawnRead );
578 $this->spawnRead = false;
579 if ( $this->spawnWrite )
580 fclose( $this->spawnWrite );
581 $this->spawnWrite = false;
582 if ( $this->spawnErr )
583 fclose( $this->spawnErr );
584 $this->spawnErr = false;
585 if ( $this->spawnProc )
586 pclose( $this->spawnProc );
587 $this->spawnProc = false;
591 private function getTextSpawnedOnce( $id ) {
594 $ok = fwrite( $this->spawnWrite, "$id\n" );
595 // $this->progress( ">> $id" );
596 if ( !$ok ) return false;
598 $ok = fflush( $this->spawnWrite );
599 // $this->progress( ">> [flush]" );
600 if ( !$ok ) return false;
602 // check that the text id they are sending is the one we asked for
603 // this avoids out of sync revision text errors we have encountered in the past
604 $newId = fgets( $this->spawnRead );
605 if ( $newId === false ) {
608 if ( $id != intval( $newId ) ) {
612 $len = fgets( $this->spawnRead );
613 // $this->progress( "<< " . trim( $len ) );
614 if ( $len === false ) return false;
616 $nbytes = intval( $len );
617 // actual error, not zero-length text
618 if ( $nbytes < 0 ) return false;
622 // Subprocess may not send everything at once, we have to loop.
623 while ( $nbytes > strlen( $text ) ) {
624 $buffer = fread( $this->spawnRead, $nbytes - strlen( $text ) );
625 if ( $buffer === false ) break;
629 $gotbytes = strlen( $text );
630 if ( $gotbytes != $nbytes ) {
631 $this->progress( "Expected $nbytes bytes from database subprocess, got $gotbytes " );
635 // Do normalization in the dump thread...
636 $stripped = str_replace( "\r", "", $text );
637 $normalized = $wgContLang->normalize( $stripped );
641 function startElement( $parser, $name, $attribs ) {
642 $this->checkpointJustWritten = false;
644 $this->clearOpenElement( null );
645 $this->lastName = $name;
647 if ( $name == 'revision' ) {
648 $this->state = $name;
649 $this->egress->writeOpenPage( null, $this->buffer );
651 } elseif ( $name == 'page' ) {
652 $this->state = $name;
653 if ( $this->atStart ) {
654 $this->egress->writeOpenStream( $this->buffer );
656 $this->atStart = false;
660 if ( $name == "text" && isset( $attribs['id'] ) ) {
661 $text = $this->getText( $attribs['id'] );
662 $this->openElement = array( $name, array( 'xml:space' => 'preserve' ) );
663 if ( strlen( $text ) > 0 ) {
664 $this->characterData( $parser, $text );
667 $this->openElement = array( $name, $attribs );
671 function endElement( $parser, $name ) {
672 $this->checkpointJustWritten = false;
674 if ( $this->openElement ) {
675 $this->clearOpenElement( "" );
677 $this->buffer .= "</$name>";
680 if ( $name == 'revision' ) {
681 $this->egress->writeRevision( null, $this->buffer );
684 } elseif ( $name == 'page' ) {
685 if ( ! $this->firstPageWritten ) {
686 $this->firstPageWritten = trim( $this->thisPage );
688 $this->lastPageWritten = trim( $this->thisPage );
689 if ( $this->timeExceeded ) {
690 $this->egress->writeClosePage( $this->buffer );
691 // nasty hack, we can't just write the chardata after the
692 // page tag, it will include leading blanks from the next line
693 $this->egress->sink->write( "\n" );
695 $this->buffer = $this->xmlwriterobj->closeStream();
696 $this->egress->writeCloseStream( $this->buffer );
699 $this->thisPage = "";
700 // this could be more than one file if we had more than one output arg
702 $filenameList = (array)$this->egress->getFilenames();
703 $newFilenames = array();
704 $firstPageID = str_pad( $this->firstPageWritten, 9, "0", STR_PAD_LEFT );
705 $lastPageID = str_pad( $this->lastPageWritten, 9, "0", STR_PAD_LEFT );
706 for ( $i = 0; $i < count( $filenameList ); $i++ ) {
707 $checkpointNameFilledIn = sprintf( $this->checkpointFiles[$i], $firstPageID, $lastPageID );
708 $fileinfo = pathinfo( $filenameList[$i] );
709 $newFilenames[] = $fileinfo['dirname'] . '/' . $checkpointNameFilledIn;
711 $this->egress->closeRenameAndReopen( $newFilenames );
712 $this->buffer = $this->xmlwriterobj->openStream();
713 $this->timeExceeded = false;
714 $this->timeOfCheckpoint = $this->lastTime;
715 $this->firstPageWritten = false;
716 $this->checkpointJustWritten = true;
719 $this->egress->writeClosePage( $this->buffer );
721 $this->thisPage = "";
724 } elseif ( $name == 'mediawiki' ) {
725 $this->egress->writeCloseStream( $this->buffer );
730 function characterData( $parser, $data ) {
731 $this->clearOpenElement( null );
732 if ( $this->lastName == "id" ) {
733 if ( $this->state == "revision" ) {
734 $this->thisRev .= $data;
735 } elseif ( $this->state == "page" ) {
736 $this->thisPage .= $data;
739 // have to skip the newline left over from closepagetag line of
740 // end of checkpoint files. nasty hack!!
741 if ( $this->checkpointJustWritten ) {
742 if ( $data[0] == "\n" ) {
743 $data = substr( $data, 1 );
745 $this->checkpointJustWritten = false;
747 $this->buffer .= htmlspecialchars( $data );
750 function clearOpenElement( $style ) {
751 if ( $this->openElement ) {
752 $this->buffer .= Xml::element( $this->openElement[0], $this->openElement[1], $style );
753 $this->openElement = false;