3 * Script 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
27 $originalDir = getcwd();
29 require_once( dirname( __FILE__
) . '/commandLine.inc' );
30 require_once( 'backup.inc' );
33 * @ingroup Maintenance
35 class TextPassDumper
extends BackupDumper
{
37 var $input = "php://stdin";
38 var $history = WikiExporter
::FULL
;
40 var $prefetchCount = 0;
41 var $prefetchCountLast = 0;
42 var $fetchCountLast = 0;
46 var $failedTextRetrievals = 0;
47 var $maxConsecutiveFailedTextRetrievals = 200;
48 var $failureTimeout = 5; // Seconds to sleep after db failure
52 var $spawnProc = false;
53 var $spawnWrite = false;
54 var $spawnRead = false;
55 var $spawnErr = false;
57 var $xmlwriterobj = false;
59 # when we spend more than maxTimeAllowed seconds on this run, we continue
60 # processing until we write out the next complete page, then save output file(s),
61 # rename it/them and open new one(s)
62 var $maxTimeAllowed = 0; // 0 = no limit
63 var $timeExceeded = false;
64 var $firstPageWritten = false;
65 var $lastPageWritten = false;
66 var $checkpointJustWritten = false;
67 var $checkpointFiles = array();
69 function initProgress( $history ) {
70 parent
::initProgress();
71 $this->timeOfCheckpoint
= $this->startTime
;
74 function dump( $history, $text = WikiExporter
::TEXT
) {
75 # This shouldn't happen if on console... ;)
76 header( 'Content-type: text/html; charset=UTF-8' );
78 # Notice messages will foul up your XML output even if they're
79 # relatively harmless.
80 if ( ini_get( 'display_errors' ) )
81 ini_set( 'display_errors', 'stderr' );
83 $this->initProgress( $this->history
);
85 $this->db
= $this->backupDb();
87 $this->egress
= new ExportProgressFilter( $this->sink
, $this );
89 # it would be nice to do it in the constructor, oh well. need egress set
90 $this->finalOptionCheck();
92 # we only want this so we know how to close a stream :-P
93 $this->xmlwriterobj
= new XmlDumpWriter();
95 $input = fopen( $this->input
, "rt" );
96 $result = $this->readDump( $input );
98 if ( WikiError
::isError( $result ) ) {
99 wfDie( $result->getMessage() );
102 if ( $this->spawnProc
) {
106 $this->report( true );
109 function processOption( $opt, $val, $param ) {
111 $url = $this->processFileOpt( $val, $param );
115 require_once "$IP/maintenance/backupPrefetch.inc";
116 $this->prefetch
= new BaseDump( $url );
122 $this->maxTimeAllowed
= intval($val)*60;
124 case 'checkpointfile':
125 $this->checkpointFiles
[] = $val;
128 $this->history
= WikiExporter
::CURRENT
;
131 $this->history
= WikiExporter
::FULL
;
142 function processFileOpt( $val, $param ) {
143 $fileURIs = explode(';',$param);
144 foreach ( $fileURIs as $URI ) {
150 $newURI = "compress.zlib://$URI";
153 $newURI = "compress.bzip2://$URI";
156 $newURI = "mediawiki.compress.7z://$URI";
161 $newFileURIs[] = $newURI;
163 $val = implode( ';', $newFileURIs );
168 * Overridden to include prefetch ratio if enabled.
170 function showReport() {
171 if ( !$this->prefetch
) {
172 return parent
::showReport();
175 if ( $this->reporting
) {
176 $now = wfTimestamp( TS_DB
);
178 $deltaAll = wfTime() - $this->startTime
;
179 $deltaPart = wfTime() - $this->lastTime
;
180 $this->pageCountPart
= $this->pageCount
- $this->pageCountLast
;
181 $this->revCountPart
= $this->revCount
- $this->revCountLast
;
184 $portion = $this->revCount
/ $this->maxCount
;
185 $eta = $this->startTime +
$deltaAll / $portion;
186 $etats = wfTimestamp( TS_DB
, intval( $eta ) );
187 if ( $this->fetchCount
) {
188 $fetchRate = 100.0 * $this->prefetchCount
/ $this->fetchCount
;
193 $pageRate = $this->pageCount
/ $deltaAll;
194 $revRate = $this->revCount
/ $deltaAll;
202 if ( $this->fetchCountLast
) {
203 $fetchRatePart = 100.0 * $this->prefetchCountLast
/ $this->fetchCountLast
;
206 $fetchRatePart = '-';
208 $pageRatePart = $this->pageCountPart
/ $deltaPart;
209 $revRatePart = $this->revCountPart
/ $deltaPart;
212 $fetchRatePart = '-';
216 $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]",
217 $now, wfWikiID(), $this->ID
, $this->pageCount
, $pageRate, $pageRatePart, $this->revCount
, $revRate, $revRatePart, $fetchRate, $fetchRatePart, $etats, $this->maxCount
) );
218 $this->lastTime
= $nowts;
219 $this->revCountLast
= $this->revCount
;
220 $this->prefetchCountLast
= $this->prefetchCount
;
221 $this->fetchCountLast
= $this->fetchCount
;
225 function setTimeExceeded() {
226 $this->timeExceeded
= True;
229 function checkIfTimeExceeded() {
230 if ( $this->maxTimeAllowed
&& ( $this->lastTime
- $this->timeOfCheckpoint
> $this->maxTimeAllowed
) ) {
236 function finalOptionCheck() {
237 if (($this->checkpointFiles
&& ! $this->maxTimeAllowed
) ||
238 ($this->maxTimeAllowed
&& !$this->checkpointFiles
)) {
239 wfDie("Options checkpointfile and maxtime must be specified together.\n");
241 foreach ($this->checkpointFiles
as $checkpointFile) {
242 $count = substr_count ($checkpointFile,"%s");
243 if (substr_count ($checkpointFile,"%s") != 2) {
244 wfDie("Option checkpointfile must contain two '%s' for substitution of first and last pageids, count is $count instead, file is $checkpointFile.\n");
248 $filenameList = $this->egress
->getFilename();
249 if (! is_array($filenameList)) {
250 $filenameList = array( $filenameList );
252 if (count($filenameList) != count($this->checkpointFiles
)) {
253 wfDie("One checkpointfile must be specified for each output option, if maxtime is used.\n");
257 function readDump( $input ) {
259 $this->openElement
= false;
260 $this->atStart
= true;
262 $this->lastName
= "";
266 $parser = xml_parser_create( "UTF-8" );
267 xml_parser_set_option( $parser, XML_OPTION_CASE_FOLDING
, false );
269 xml_set_element_handler( $parser, array( &$this, 'startElement' ), array( &$this, 'endElement' ) );
270 xml_set_character_data_handler( $parser, array( &$this, 'characterData' ) );
272 $offset = 0; // for context extraction on error reporting
273 $bufferSize = 512 * 1024;
275 if ($this->checkIfTimeExceeded()) {
276 $this->setTimeExceeded();
278 $chunk = fread( $input, $bufferSize );
279 if ( !xml_parse( $parser, $chunk, feof( $input ) ) ) {
280 wfDebug( "TextDumpPass::readDump encountered XML parsing error\n" );
281 return new WikiXmlError( $parser, 'XML import parse failure', $chunk, $offset );
283 $offset +
= strlen( $chunk );
284 } while ( $chunk !== false && !feof( $input ) );
285 if ($this->maxTimeAllowed
) {
286 $filenameList = $this->egress
->getFilename();
287 # we wrote some stuff after last checkpoint that needs renamed */
288 if (! is_array($filenameList)) {
289 $filenameList = array( $filenameList );
291 if (file_exists($filenameList[0])) {
292 $newFilenames = array();
293 $firstPageID = str_pad($this->firstPageWritten
,9,"0",STR_PAD_LEFT
);
294 $lastPageID = str_pad($this->lastPageWritten
,9,"0",STR_PAD_LEFT
);
295 for ($i =0; $i < count($filenameList); $i++
) {
296 $checkpointNameFilledIn = sprintf($this->checkpointFiles
[$i], $firstPageID, $lastPageID);
297 $fileinfo = pathinfo($filenameList[$i]);
298 $newFilenames[] = $fileinfo{'dirname'} . '/' . $checkpointNameFilledIn;
300 $this->egress
->closeAndRename( $newFilenames );
303 xml_parser_free( $parser );
308 function getText( $id ) {
310 if ( isset( $this->prefetch
) ) {
311 $text = $this->prefetch
->prefetch( $this->thisPage
, $this->thisRev
);
312 if ( $text !== null ) { // Entry missing from prefetch dump
313 $dbr = wfGetDB( DB_SLAVE
);
314 $revID = intval( $this->thisRev
);
315 $revLength = $dbr->selectField( 'revision', 'rev_len', array( 'rev_id' => $revID ) );
316 // if length of rev text in file doesn't match length in db, we reload
317 // this avoids carrying forward broken data from previous xml dumps
318 if( strlen( $text ) == $revLength ) {
319 $this->prefetchCount++
;
324 return $this->doGetText( $id );
327 private function doGetText( $id ) {
331 $ex = new MWException( "Graceful storage failure" );
333 if ( $this->spawn
) {
334 if ($this->failures
) {
335 // we don't know why it failed, could be the child process
336 // borked, could be db entry busted, could be db server out to lunch,
337 // so cover all bases
341 $text = $this->getTextSpawned( $id );
343 $text = $this->getTextDbSafe( $id );
345 if ( $text === false ) {
347 if ( $this->failures
> $this->maxFailures
) {
348 $this->progress( "Failed to retrieve revision text for text id ".
349 "$id after $this->maxFailures tries, giving up" );
350 // were there so many bad retrievals in a row we want to bail?
351 // at some point we have to declare the dump irretrievably broken
352 $this->failedTextRetrievals++
;
353 if ($this->failedTextRetrievals
> $this->maxConsecutiveFailedTextRetrievals
) {
357 // would be nice to return something better to the caller someday,
358 // log what we know about the failure and about the revision
362 $this->progress( "Error $this->failures " .
363 "of allowed $this->maxFailures retrieving revision text for text id $id! " .
364 "Pausing $this->failureTimeout seconds before retry..." );
365 sleep( $this->failureTimeout
);
368 $this->failedTextRetrievals
= 0;
376 * Fetch a text revision from the database, retrying in case of failure.
377 * This may survive some transitory errors by reconnecting, but
378 * may not survive a long-term server outage.
380 private function getTextDbSafe( $id ) {
383 $text = $this->getTextDb( $id );
384 } catch ( DBQueryError
$ex ) {
392 * May throw a database error if, say, the server dies during query.
394 private function getTextDb( $id ) {
396 $row = $this->db
->selectRow( 'text',
397 array( 'old_text', 'old_flags' ),
398 array( 'old_id' => $id ),
400 $text = Revision
::getRevisionText( $row );
401 if ( $text === false ) {
404 $stripped = str_replace( "\r", "", $text );
405 $normalized = $wgContLang->normalize( $stripped );
409 private function getTextSpawned( $id ) {
410 wfSuppressWarnings();
411 if ( !$this->spawnProc
) {
415 $text = $this->getTextSpawnedOnce( $id );
420 function openSpawn() {
424 array_map( 'wfEscapeShellArg',
427 "$IP/maintenance/fetchText.php",
428 '--wiki', wfWikiID() ) ) );
430 0 => array( "pipe", "r" ),
431 1 => array( "pipe", "w" ),
432 2 => array( "file", "/dev/null", "a" ) );
435 $this->progress( "Spawning database subprocess: $cmd" );
436 $this->spawnProc
= proc_open( $cmd, $spec, $pipes );
437 if ( !$this->spawnProc
) {
439 $this->progress( "Subprocess spawn failed." );
443 $this->spawnWrite
, // -> stdin
444 $this->spawnRead
, // <- stdout
450 private function closeSpawn() {
451 wfSuppressWarnings();
452 if ( $this->spawnRead
)
453 fclose( $this->spawnRead
);
454 $this->spawnRead
= false;
455 if ( $this->spawnWrite
)
456 fclose( $this->spawnWrite
);
457 $this->spawnWrite
= false;
458 if ( $this->spawnErr
)
459 fclose( $this->spawnErr
);
460 $this->spawnErr
= false;
461 if ( $this->spawnProc
)
462 pclose( $this->spawnProc
);
463 $this->spawnProc
= false;
467 private function getTextSpawnedOnce( $id ) {
470 $ok = fwrite( $this->spawnWrite
, "$id\n" );
471 // $this->progress( ">> $id" );
472 if ( !$ok ) return false;
474 $ok = fflush( $this->spawnWrite
);
475 // $this->progress( ">> [flush]" );
476 if ( !$ok ) return false;
478 // check that the text id they are sending is the one we asked for
479 // this avoids out of sync revision text errors we have encountered in the past
480 $newId = fgets( $this->spawnRead
);
481 if ( $newId === false ) {
484 if ( $id != intval( $newId ) ) {
488 $len = fgets( $this->spawnRead
);
489 // $this->progress( "<< " . trim( $len ) );
490 if ( $len === false ) return false;
492 $nbytes = intval( $len );
493 // actual error, not zero-length text
494 if ($nbytes < 0 ) return false;
498 // Subprocess may not send everything at once, we have to loop.
499 while ( $nbytes > strlen( $text ) ) {
500 $buffer = fread( $this->spawnRead
, $nbytes - strlen( $text ) );
501 if ( $buffer === false ) break;
505 $gotbytes = strlen( $text );
506 if ( $gotbytes != $nbytes ) {
507 $this->progress( "Expected $nbytes bytes from database subprocess, got $gotbytes " );
511 // Do normalization in the dump thread...
512 $stripped = str_replace( "\r", "", $text );
513 $normalized = $wgContLang->normalize( $stripped );
517 function startElement( $parser, $name, $attribs ) {
518 $this->checkpointJustWritten
= false;
520 $this->clearOpenElement( null );
521 $this->lastName
= $name;
523 if ( $name == 'revision' ) {
524 $this->state
= $name;
525 $this->egress
->writeOpenPage( null, $this->buffer
);
527 } elseif ( $name == 'page' ) {
528 $this->state
= $name;
529 if ( $this->atStart
) {
530 $this->egress
->writeOpenStream( $this->buffer
);
532 $this->atStart
= false;
536 if ( $name == "text" && isset( $attribs['id'] ) ) {
537 $text = $this->getText( $attribs['id'] );
538 $this->openElement
= array( $name, array( 'xml:space' => 'preserve' ) );
539 if ( strlen( $text ) > 0 ) {
540 $this->characterData( $parser, $text );
543 $this->openElement
= array( $name, $attribs );
547 function endElement( $parser, $name ) {
548 $this->checkpointJustWritten
= false;
550 if ( $this->openElement
) {
551 $this->clearOpenElement( "" );
553 $this->buffer
.= "</$name>";
556 if ( $name == 'revision' ) {
557 $this->egress
->writeRevision( null, $this->buffer
);
560 } elseif ( $name == 'page' ) {
561 if (! $this->firstPageWritten
) {
562 $this->firstPageWritten
= trim($this->thisPage
);
564 $this->lastPageWritten
= trim($this->thisPage
);
565 if ($this->timeExceeded
) {
566 $this->egress
->writeClosePage( $this->buffer
);
567 # nasty hack, we can't just write the chardata after the
568 # page tag, it will include leading blanks from the next line
569 $this->egress
->sink
->write("\n");
571 $this->buffer
= $this->xmlwriterobj
->closeStream();
572 $this->egress
->writeCloseStream( $this->buffer
);
575 $this->thisPage
= "";
576 /* this could be more than one file if we had more than one output arg */
577 $checkpointFilenames = array();
578 $filenameList = $this->egress
->getFilename();
580 if (! is_array($filenameList)) {
581 $filenameList = array( $filenameList );
583 $newFilenames = array();
584 $firstPageID = str_pad($this->firstPageWritten
,9,"0",STR_PAD_LEFT
);
585 $lastPageID = str_pad($this->lastPageWritten
,9,"0",STR_PAD_LEFT
);
586 for ($i =0; $i < count($filenameList); $i++
) {
587 $checkpointNameFilledIn = sprintf($this->checkpointFiles
[$i], $firstPageID, $lastPageID);
588 $fileinfo = pathinfo($filenameList[$i]);
589 $newFilenames[] = $fileinfo{'dirname'} . '/' . $checkpointNameFilledIn;
591 $this->egress
->closeRenameAndReopen( $newFilenames );
592 $this->buffer
= $this->xmlwriterobj
->openStream();
593 $this->timeExceeded
= false;
594 $this->timeOfCheckpoint
= $this->lastTime
;
595 $this->firstPageWritten
= false;
596 $this->checkpointJustWritten
= true;
599 $this->egress
->writeClosePage( $this->buffer
);
601 $this->thisPage
= "";
604 } elseif ( $name == 'mediawiki' ) {
605 $this->egress
->writeCloseStream( $this->buffer
);
610 function characterData( $parser, $data ) {
611 $this->clearOpenElement( null );
612 if ( $this->lastName
== "id" ) {
613 if ( $this->state
== "revision" ) {
614 $this->thisRev
.= $data;
615 } elseif ( $this->state
== "page" ) {
616 $this->thisPage
.= $data;
619 # have to skip the newline left over from closepagetag line of
620 # end of checkpoint files. nasty hack!!
621 if ($this->checkpointJustWritten
) {
622 if ($data[0] == "\n") {
623 $data = substr($data,1);
625 $this->checkpointJustWritten
= false;
627 $this->buffer
.= htmlspecialchars( $data );
630 function clearOpenElement( $style ) {
631 if ( $this->openElement
) {
632 $this->buffer
.= Xml
::element( $this->openElement
[0], $this->openElement
[1], $style );
633 $this->openElement
= false;
639 $dumper = new TextPassDumper( $argv );
641 if ( !isset( $options['help'] ) ) {
642 $dumper->dump( true );
644 $dumper->progress( <<<ENDS
645 This script postprocesses XML dumps from dumpBackup.php to add
646 page text which was stubbed out (using --stub).
648 XML input is accepted on stdin.
649 XML output is sent to stdout; progress reports are sent to stderr.
651 Usage: php dumpTextPass.php [<options>]
653 --stub=<type>:<file> To load a compressed stub dump instead of stdin
654 --prefetch=<type>:<file> Use a prior dump file as a text source, to save
655 pressure on the database.
656 (Requires the XMLReader extension)
657 --maxtime=<minutes> Write out checkpoint file after this many minutes (writing
658 out complete page, closing xml file properly, and opening new one
659 with header). This option requires the checkpointfile option.
660 --checkpointfile=<filenamepattern> Use this string for checkpoint filenames,
661 substituting first pageid written for the first %s (required) and the
662 last pageid written for the second %s if it exists.
663 --quiet Don't dump status reports to stderr.
664 --report=n Report position and speed after every n pages processed.
666 --server=h Force reading from MySQL server h
667 --current Base ETA on number of pages in database instead of all revisions
668 --spawn Spawn a subprocess for loading text records
669 --help Display this help message