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 ( isset( $this->db ) && $this->db->isOpen() ) {
88 throw new MWException( 'DB is set and has not been closed by the Load Balancer' );
93 // Trying to set up new connection.
94 // We do /not/ retry upon failure, but delegate to encapsulating logic, to avoid
95 // individually retrying at different layers of code.
97 // 1. The LoadBalancer.
99 $this->lb = wfGetLBFactory()->newMainLB();
100 } catch ( Exception $e ) {
101 throw new MWException( __METHOD__ . " rotating DB failed to obtain new load balancer (" . $e->getMessage() . ")" );
105 // 2. The Connection, through the load balancer.
107 $this->db = $this->lb->getConnection( DB_SLAVE, 'backup' );
108 } catch ( Exception $e ) {
109 throw new MWException( __METHOD__ . " rotating DB failed to obtain new database (" . $e->getMessage() . ")" );
114 function initProgress( $history = WikiExporter::FULL ) {
115 parent::initProgress();
116 $this->timeOfCheckpoint = $this->startTime;
119 function dump( $history, $text = WikiExporter::TEXT ) {
120 // Notice messages will foul up your XML output even if they're
121 // relatively harmless.
122 if ( ini_get( 'display_errors' ) )
123 ini_set( 'display_errors', 'stderr' );
125 $this->initProgress( $this->history );
127 // We are trying to get an initial database connection to avoid that the
128 // first try of this request's first call to getText fails. However, if
129 // obtaining a good DB connection fails it's not a serious issue, as
130 // getText does retry upon failure and can start without having a working
134 } catch ( Exception $e ) {
135 // We do not even count this as failure. Just let eventual
137 $this->progress( "Getting initial DB connection failed (" .
138 $e->getMessage() . ")" );
141 $this->egress = new ExportProgressFilter( $this->sink, $this );
143 // it would be nice to do it in the constructor, oh well. need egress set
144 $this->finalOptionCheck();
146 // we only want this so we know how to close a stream :-P
147 $this->xmlwriterobj = new XmlDumpWriter();
149 $input = fopen( $this->input, "rt" );
150 $result = $this->readDump( $input );
152 if ( WikiError::isError( $result ) ) {
153 throw new MWException( $result->getMessage() );
156 if ( $this->spawnProc ) {
160 $this->report( true );
163 function processOption( $opt, $val, $param ) {
165 $url = $this->processFileOpt( $val, $param );
169 require_once "$IP/maintenance/backupPrefetch.inc";
170 $this->prefetch = new BaseDump( $url );
176 $this->maxTimeAllowed = intval( $val ) * 60;
178 case 'checkpointfile':
179 $this->checkpointFiles[] = $val;
182 $this->history = WikiExporter::CURRENT;
185 $this->history = WikiExporter::FULL;
196 function processFileOpt( $val, $param ) {
197 $fileURIs = explode( ';', $param );
198 foreach ( $fileURIs as $URI ) {
204 $newURI = "compress.zlib://$URI";
207 $newURI = "compress.bzip2://$URI";
210 $newURI = "mediawiki.compress.7z://$URI";
215 $newFileURIs[] = $newURI;
217 $val = implode( ';', $newFileURIs );
222 * Overridden to include prefetch ratio if enabled.
224 function showReport() {
225 if ( !$this->prefetch ) {
226 parent::showReport();
230 if ( $this->reporting ) {
231 $now = wfTimestamp( TS_DB );
233 $deltaAll = wfTime() - $this->startTime;
234 $deltaPart = wfTime() - $this->lastTime;
235 $this->pageCountPart = $this->pageCount - $this->pageCountLast;
236 $this->revCountPart = $this->revCount - $this->revCountLast;
239 $portion = $this->revCount / $this->maxCount;
240 $eta = $this->startTime + $deltaAll / $portion;
241 $etats = wfTimestamp( TS_DB, intval( $eta ) );
242 if ( $this->fetchCount ) {
243 $fetchRate = 100.0 * $this->prefetchCount / $this->fetchCount;
247 $pageRate = $this->pageCount / $deltaAll;
248 $revRate = $this->revCount / $deltaAll;
256 if ( $this->fetchCountLast ) {
257 $fetchRatePart = 100.0 * $this->prefetchCountLast / $this->fetchCountLast;
259 $fetchRatePart = '-';
261 $pageRatePart = $this->pageCountPart / $deltaPart;
262 $revRatePart = $this->revCountPart / $deltaPart;
265 $fetchRatePart = '-';
269 $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]",
270 $now, wfWikiID(), $this->ID, $this->pageCount, $pageRate, $pageRatePart, $this->revCount, $revRate, $revRatePart, $fetchRate, $fetchRatePart, $etats, $this->maxCount ) );
271 $this->lastTime = $nowts;
272 $this->revCountLast = $this->revCount;
273 $this->prefetchCountLast = $this->prefetchCount;
274 $this->fetchCountLast = $this->fetchCount;
278 function setTimeExceeded() {
279 $this->timeExceeded = True;
282 function checkIfTimeExceeded() {
283 if ( $this->maxTimeAllowed && ( $this->lastTime - $this->timeOfCheckpoint > $this->maxTimeAllowed ) ) {
289 function finalOptionCheck() {
290 if ( ( $this->checkpointFiles && ! $this->maxTimeAllowed ) ||
291 ( $this->maxTimeAllowed && !$this->checkpointFiles ) ) {
292 throw new MWException( "Options checkpointfile and maxtime must be specified together.\n" );
294 foreach ( $this->checkpointFiles as $checkpointFile ) {
295 $count = substr_count ( $checkpointFile, "%s" );
297 throw new MWException( "Option checkpointfile must contain two '%s' for substitution of first and last pageids, count is $count instead, file is $checkpointFile.\n" );
301 if ( $this->checkpointFiles ) {
302 $filenameList = (array)$this->egress->getFilenames();
303 if ( count( $filenameList ) != count( $this->checkpointFiles ) ) {
304 throw new MWException( "One checkpointfile must be specified for each output option, if maxtime is used.\n" );
309 function readDump( $input ) {
311 $this->openElement = false;
312 $this->atStart = true;
314 $this->lastName = "";
318 $parser = xml_parser_create( "UTF-8" );
319 xml_parser_set_option( $parser, XML_OPTION_CASE_FOLDING, false );
321 xml_set_element_handler( $parser, array( &$this, 'startElement' ), array( &$this, 'endElement' ) );
322 xml_set_character_data_handler( $parser, array( &$this, 'characterData' ) );
324 $offset = 0; // for context extraction on error reporting
325 $bufferSize = 512 * 1024;
327 if ( $this->checkIfTimeExceeded() ) {
328 $this->setTimeExceeded();
330 $chunk = fread( $input, $bufferSize );
331 if ( !xml_parse( $parser, $chunk, feof( $input ) ) ) {
332 wfDebug( "TextDumpPass::readDump encountered XML parsing error\n" );
333 return new WikiXmlError( $parser, 'XML import parse failure', $chunk, $offset );
335 $offset += strlen( $chunk );
336 } while ( $chunk !== false && !feof( $input ) );
337 if ( $this->maxTimeAllowed ) {
338 $filenameList = (array)$this->egress->getFilenames();
339 // we wrote some stuff after last checkpoint that needs renamed
340 if ( file_exists( $filenameList[0] ) ) {
341 $newFilenames = array();
342 # we might have just written the header and footer and had no
343 # pages or revisions written... perhaps they were all deleted
344 # there's no pageID 0 so we use that. the caller is responsible
345 # for deciding what to do with a file containing only the
346 # siteinfo information and the mw tags.
347 if ( ! $this->firstPageWritten ) {
348 $firstPageID = str_pad( 0, 9, "0", STR_PAD_LEFT );
349 $lastPageID = str_pad( 0, 9, "0", STR_PAD_LEFT );
352 $firstPageID = str_pad( $this->firstPageWritten, 9, "0", STR_PAD_LEFT );
353 $lastPageID = str_pad( $this->lastPageWritten, 9, "0", STR_PAD_LEFT );
355 for ( $i = 0; $i < count( $filenameList ); $i++ ) {
356 $checkpointNameFilledIn = sprintf( $this->checkpointFiles[$i], $firstPageID, $lastPageID );
357 $fileinfo = pathinfo( $filenameList[$i] );
358 $newFilenames[] = $fileinfo['dirname'] . '/' . $checkpointNameFilledIn;
360 $this->egress->closeAndRename( $newFilenames );
363 xml_parser_free( $parser );
369 * Tries to get the revision text for a revision id.
371 * Upon errors, retries (Up to $this->maxFailures tries each call).
372 * If still no good revision get could be found even after this retrying, "" is returned.
373 * If no good revision text could be returned for
374 * $this->maxConsecutiveFailedTextRetrievals consecutive calls to getText, MWException
377 * @param $id string The revision id to get the text for
379 * @return string The revision text for $id, or ""
380 * @throws MWException
382 function getText( $id ) {
383 $prefetchNotTried = true; // Whether or not we already tried to get the text via prefetch.
384 $text = false; // The candidate for a good text. false if no proper value.
385 $failures = 0; // The number of times, this invocation of getText already failed.
387 static $consecutiveFailedTextRetrievals = 0; // The number of times getText failed without
388 // yielding a good text in between.
392 // To allow to simply return on success and do not have to worry about book keeping,
393 // we assume, this fetch works (possible after some retries). Nevertheless, we koop
394 // the old value, so we can restore it, if problems occur (See after the while loop).
395 $oldConsecutiveFailedTextRetrievals = $consecutiveFailedTextRetrievals;
396 $consecutiveFailedTextRetrievals = 0;
398 while ( $failures < $this->maxFailures ) {
400 // As soon as we found a good text for the $id, we will return immediately.
401 // Hence, if we make it past the try catch block, we know that we did not
405 // Step 1: Get some text (or reuse from previous iteratuon if checking
406 // for plausibility failed)
408 // Trying to get prefetch, if it has not been tried before
409 if ( $text === false && isset( $this->prefetch ) && $prefetchNotTried ) {
410 $prefetchNotTried = false;
411 $tryIsPrefetch = true;
412 $text = $this->prefetch->prefetch( $this->thisPage, $this->thisRev );
413 if ( $text === null ) {
418 if ( $text === false ) {
419 // Fallback to asking the database
420 $tryIsPrefetch = false;
421 if ( $this->spawn ) {
422 $text = $this->getTextSpawned( $id );
424 $text = $this->getTextDb( $id );
428 if ( $text === false ) {
429 throw new MWException( "Generic error while obtaining text for id " . $id );
432 // We received a good candidate for the text of $id via some method
434 // Step 2: Checking for plausibility and return the text if it is
436 $revID = intval( $this->thisRev );
437 if ( ! isset( $this->db ) ) {
438 throw new MWException( "No database available" );
440 $revLength = $this->db->selectField( 'revision', 'rev_len', array( 'rev_id' => $revID ) );
441 if ( strlen( $text ) == $revLength ) {
442 if ( $tryIsPrefetch ) {
443 $this->prefetchCount++;
449 throw new MWException( "Received text is unplausible for id " . $id );
451 } catch ( Exception $e ) {
452 $msg = "getting/checking text " . $id . " failed (" . $e->getMessage() . ")";
453 if ( $failures + 1 < $this->maxFailures ) {
454 $msg .= " (Will retry " . ( $this->maxFailures - $failures - 1 ) . " more times)";
456 $this->progress( $msg );
459 // Something went wrong; we did not a text that was plausible :(
463 // After backing off for some time, we try to reboot the whole process as
464 // much as possible to not carry over failures from one part to the other
466 sleep( $this->failureTimeout );
469 if ( $this->spawn ) {
473 } catch ( Exception $e ) {
474 $this->progress( "Rebooting getText infrastructure failed (" . $e->getMessage() . ")" .
475 " Trying to continue anyways" );
479 // Retirieving a good text for $id failed (at least) maxFailures times.
480 // We abort for this $id.
482 // Restoring the consecutive failures, and maybe aborting, if the dump
484 $consecutiveFailedTextRetrievals = $oldConsecutiveFailedTextRetrievals + 1;
485 if ( $consecutiveFailedTextRetrievals > $this->maxConsecutiveFailedTextRetrievals ) {
486 throw new MWException( "Graceful storage failure" );
494 * May throw a database error if, say, the server dies during query.
496 * @return bool|string
497 * @throws MWException
499 private function getTextDb( $id ) {
501 if ( ! isset( $this->db ) ) {
502 throw new MWException( __METHOD__ . "No database available" );
504 $row = $this->db->selectRow( 'text',
505 array( 'old_text', 'old_flags' ),
506 array( 'old_id' => $id ),
508 $text = Revision::getRevisionText( $row );
509 if ( $text === false ) {
512 $stripped = str_replace( "\r", "", $text );
513 $normalized = $wgContLang->normalize( $stripped );
517 private function getTextSpawned( $id ) {
518 wfSuppressWarnings();
519 if ( !$this->spawnProc ) {
523 $text = $this->getTextSpawnedOnce( $id );
528 function openSpawn() {
531 if ( file_exists( "$IP/../multiversion/MWScript.php" ) ) {
533 array_map( 'wfEscapeShellArg',
536 "$IP/../multiversion/MWScript.php",
538 '--wiki', wfWikiID() ) ) );
542 array_map( 'wfEscapeShellArg',
545 "$IP/maintenance/fetchText.php",
546 '--wiki', wfWikiID() ) ) );
549 0 => array( "pipe", "r" ),
550 1 => array( "pipe", "w" ),
551 2 => array( "file", "/dev/null", "a" ) );
554 $this->progress( "Spawning database subprocess: $cmd" );
555 $this->spawnProc = proc_open( $cmd, $spec, $pipes );
556 if ( !$this->spawnProc ) {
558 $this->progress( "Subprocess spawn failed." );
562 $this->spawnWrite, // -> stdin
563 $this->spawnRead, // <- stdout
569 private function closeSpawn() {
570 wfSuppressWarnings();
571 if ( $this->spawnRead )
572 fclose( $this->spawnRead );
573 $this->spawnRead = false;
574 if ( $this->spawnWrite )
575 fclose( $this->spawnWrite );
576 $this->spawnWrite = false;
577 if ( $this->spawnErr )
578 fclose( $this->spawnErr );
579 $this->spawnErr = false;
580 if ( $this->spawnProc )
581 pclose( $this->spawnProc );
582 $this->spawnProc = false;
586 private function getTextSpawnedOnce( $id ) {
589 $ok = fwrite( $this->spawnWrite, "$id\n" );
590 // $this->progress( ">> $id" );
591 if ( !$ok ) return false;
593 $ok = fflush( $this->spawnWrite );
594 // $this->progress( ">> [flush]" );
595 if ( !$ok ) return false;
597 // check that the text id they are sending is the one we asked for
598 // this avoids out of sync revision text errors we have encountered in the past
599 $newId = fgets( $this->spawnRead );
600 if ( $newId === false ) {
603 if ( $id != intval( $newId ) ) {
607 $len = fgets( $this->spawnRead );
608 // $this->progress( "<< " . trim( $len ) );
609 if ( $len === false ) return false;
611 $nbytes = intval( $len );
612 // actual error, not zero-length text
613 if ( $nbytes < 0 ) return false;
617 // Subprocess may not send everything at once, we have to loop.
618 while ( $nbytes > strlen( $text ) ) {
619 $buffer = fread( $this->spawnRead, $nbytes - strlen( $text ) );
620 if ( $buffer === false ) break;
624 $gotbytes = strlen( $text );
625 if ( $gotbytes != $nbytes ) {
626 $this->progress( "Expected $nbytes bytes from database subprocess, got $gotbytes " );
630 // Do normalization in the dump thread...
631 $stripped = str_replace( "\r", "", $text );
632 $normalized = $wgContLang->normalize( $stripped );
636 function startElement( $parser, $name, $attribs ) {
637 $this->checkpointJustWritten = false;
639 $this->clearOpenElement( null );
640 $this->lastName = $name;
642 if ( $name == 'revision' ) {
643 $this->state = $name;
644 $this->egress->writeOpenPage( null, $this->buffer );
646 } elseif ( $name == 'page' ) {
647 $this->state = $name;
648 if ( $this->atStart ) {
649 $this->egress->writeOpenStream( $this->buffer );
651 $this->atStart = false;
655 if ( $name == "text" && isset( $attribs['id'] ) ) {
656 $text = $this->getText( $attribs['id'] );
657 $this->openElement = array( $name, array( 'xml:space' => 'preserve' ) );
658 if ( strlen( $text ) > 0 ) {
659 $this->characterData( $parser, $text );
662 $this->openElement = array( $name, $attribs );
666 function endElement( $parser, $name ) {
667 $this->checkpointJustWritten = false;
669 if ( $this->openElement ) {
670 $this->clearOpenElement( "" );
672 $this->buffer .= "</$name>";
675 if ( $name == 'revision' ) {
676 $this->egress->writeRevision( null, $this->buffer );
679 } elseif ( $name == 'page' ) {
680 if ( ! $this->firstPageWritten ) {
681 $this->firstPageWritten = trim( $this->thisPage );
683 $this->lastPageWritten = trim( $this->thisPage );
684 if ( $this->timeExceeded ) {
685 $this->egress->writeClosePage( $this->buffer );
686 // nasty hack, we can't just write the chardata after the
687 // page tag, it will include leading blanks from the next line
688 $this->egress->sink->write( "\n" );
690 $this->buffer = $this->xmlwriterobj->closeStream();
691 $this->egress->writeCloseStream( $this->buffer );
694 $this->thisPage = "";
695 // this could be more than one file if we had more than one output arg
697 $filenameList = (array)$this->egress->getFilenames();
698 $newFilenames = array();
699 $firstPageID = str_pad( $this->firstPageWritten, 9, "0", STR_PAD_LEFT );
700 $lastPageID = str_pad( $this->lastPageWritten, 9, "0", STR_PAD_LEFT );
701 for ( $i = 0; $i < count( $filenameList ); $i++ ) {
702 $checkpointNameFilledIn = sprintf( $this->checkpointFiles[$i], $firstPageID, $lastPageID );
703 $fileinfo = pathinfo( $filenameList[$i] );
704 $newFilenames[] = $fileinfo['dirname'] . '/' . $checkpointNameFilledIn;
706 $this->egress->closeRenameAndReopen( $newFilenames );
707 $this->buffer = $this->xmlwriterobj->openStream();
708 $this->timeExceeded = false;
709 $this->timeOfCheckpoint = $this->lastTime;
710 $this->firstPageWritten = false;
711 $this->checkpointJustWritten = true;
714 $this->egress->writeClosePage( $this->buffer );
716 $this->thisPage = "";
719 } elseif ( $name == 'mediawiki' ) {
720 $this->egress->writeCloseStream( $this->buffer );
725 function characterData( $parser, $data ) {
726 $this->clearOpenElement( null );
727 if ( $this->lastName == "id" ) {
728 if ( $this->state == "revision" ) {
729 $this->thisRev .= $data;
730 } elseif ( $this->state == "page" ) {
731 $this->thisPage .= $data;
734 // have to skip the newline left over from closepagetag line of
735 // end of checkpoint files. nasty hack!!
736 if ( $this->checkpointJustWritten ) {
737 if ( $data[0] == "\n" ) {
738 $data = substr( $data, 1 );
740 $this->checkpointJustWritten = false;
742 $this->buffer .= htmlspecialchars( $data );
745 function clearOpenElement( $style ) {
746 if ( $this->openElement ) {
747 $this->buffer .= Xml::element( $this->openElement[0], $this->openElement[1], $style );
748 $this->openElement = false;