MediaWiki  master
TextPassDumper.php
Go to the documentation of this file.
1 <?php
28 require_once __DIR__ . '/BackupDumper.php';
29 require_once __DIR__ . '/SevenZipStream.php';
30 require_once __DIR__ . '/../../includes/export/WikiExporter.php';
31 
37 
43  public $prefetch = null;
45  private $thisPage;
47  private $thisRev;
48 
49  // when we spend more than maxTimeAllowed seconds on this run, we continue
50  // processing until we write out the next complete page, then save output file(s),
51  // rename it/them and open new one(s)
52  public $maxTimeAllowed = 0; // 0 = no limit
53 
54  protected $input = "php://stdin";
56  protected $fetchCount = 0;
57  protected $prefetchCount = 0;
58  protected $prefetchCountLast = 0;
59  protected $fetchCountLast = 0;
60 
61  protected $maxFailures = 5;
63  protected $failureTimeout = 5; // Seconds to sleep after db failure
64 
65  protected $bufferSize = 524288; // In bytes. Maximum size to read from the stub in on go.
66 
68  protected $php = [];
69  protected $spawn = false;
70 
74  protected $spawnProc = false;
75 
79  protected $spawnWrite;
80 
84  protected $spawnRead;
85 
89  protected $spawnErr = false;
90 
94  protected $xmlwriterobj = false;
95 
96  protected $timeExceeded = false;
97  protected $firstPageWritten = false;
98  protected $lastPageWritten = false;
99  protected $checkpointJustWritten = false;
101  protected $checkpointFiles = [];
102 
106  protected $db;
107 
111  public function __construct( $args = null ) {
112  parent::__construct();
113 
114  $this->addDescription( <<<TEXT
115 This script postprocesses XML dumps from dumpBackup.php to add
116 page text which was stubbed out (using --stub).
117 
118 XML input is accepted on stdin.
119 XML output is sent to stdout; progress reports are sent to stderr.
120 TEXT
121  );
122  $this->stderr = fopen( "php://stderr", "wt" );
123 
124  $this->addOption( 'stub', 'To load a compressed stub dump instead of stdin. ' .
125  'Specify as --stub=<type>:<file>.', false, true );
126  $this->addOption( 'prefetch', 'Use a prior dump file as a text source, to savepressure on the ' .
127  'database. (Requires the XMLReader extension). Specify as --prefetch=<type>:<file>',
128  false, true );
129  $this->addOption( 'maxtime', 'Write out checkpoint file after this many minutes (writing' .
130  'out complete page, closing xml file properly, and opening new one' .
131  'with header). This option requires the checkpointfile option.', false, true );
132  $this->addOption( 'checkpointfile', 'Use this string for checkpoint filenames,substituting ' .
133  'first pageid written for the first %s (required) and the last pageid written for the ' .
134  'second %s if it exists.', false, true, false, true ); // This can be specified multiple times
135  $this->addOption( 'quiet', 'Don\'t dump status reports to stderr.' );
136  $this->addOption( 'full', 'Dump all revisions of every page' );
137  $this->addOption( 'current', 'Base ETA on number of pages in database instead of all revisions' );
138  $this->addOption( 'spawn', 'Spawn a subprocess for loading text records, optionally specify ' .
139  'php[,mwscript] paths' );
140  $this->addOption( 'buffersize', 'Buffer size in bytes to use for reading the stub. ' .
141  '(Default: 512KB, Minimum: 4KB)', false, true );
142 
143  if ( $args ) {
144  $this->loadWithArgv( $args );
145  $this->processOptions();
146  }
147  }
148 
152  private function getBlobStore() {
153  return MediaWikiServices::getInstance()->getBlobStore();
154  }
155 
156  public function execute() {
157  $this->processOptions();
158  $this->dump( true );
159  }
160 
161  protected function processOptions() {
162  parent::processOptions();
163 
164  if ( $this->hasOption( 'buffersize' ) ) {
165  $this->bufferSize = max( intval( $this->getOption( 'buffersize' ) ), 4 * 1024 );
166  }
167 
168  if ( $this->hasOption( 'prefetch' ) ) {
169  $url = $this->processFileOpt( $this->getOption( 'prefetch' ) );
170  $this->prefetch = new BaseDump( $url );
171  }
172 
173  if ( $this->hasOption( 'stub' ) ) {
174  $this->input = $this->processFileOpt( $this->getOption( 'stub' ) );
175  }
176 
177  if ( $this->hasOption( 'maxtime' ) ) {
178  $this->maxTimeAllowed = intval( $this->getOption( 'maxtime' ) ) * 60;
179  }
180 
181  if ( $this->hasOption( 'checkpointfile' ) ) {
182  $this->checkpointFiles = $this->getOption( 'checkpointfile' );
183  }
184 
185  if ( $this->hasOption( 'current' ) ) {
186  $this->history = WikiExporter::CURRENT;
187  }
188 
189  if ( $this->hasOption( 'full' ) ) {
190  $this->history = WikiExporter::FULL;
191  }
192 
193  if ( $this->hasOption( 'spawn' ) ) {
194  $this->spawn = true;
195  $val = $this->getOption( 'spawn' );
196  if ( $val !== 1 ) {
197  $this->php = explode( ',', $val, 2 );
198  }
199  }
200  }
201 
214  protected function rotateDb() {
215  // Cleaning up old connections
216  if ( isset( $this->lb ) ) {
217  $this->lb->closeAll();
218  unset( $this->lb );
219  }
220 
221  if ( $this->forcedDb !== null ) {
222  $this->db = $this->forcedDb;
223 
224  return;
225  }
226 
227  if ( isset( $this->db ) && $this->db->isOpen() ) {
228  throw new MWException( 'DB is set and has not been closed by the Load Balancer' );
229  }
230 
231  unset( $this->db );
232 
233  // Trying to set up new connection.
234  // We do /not/ retry upon failure, but delegate to encapsulating logic, to avoid
235  // individually retrying at different layers of code.
236 
237  try {
238  $lbFactory = MediaWikiServices::getInstance()->getDBLoadBalancerFactory();
239  $this->lb = $lbFactory->newMainLB();
240  } catch ( Exception $e ) {
241  throw new MWException( __METHOD__
242  . " rotating DB failed to obtain new load balancer (" . $e->getMessage() . ")" );
243  }
244 
245  try {
246  $this->db = $this->lb->getMaintenanceConnectionRef( DB_REPLICA, 'dump' );
247  } catch ( Exception $e ) {
248  throw new MWException( __METHOD__
249  . " rotating DB failed to obtain new database (" . $e->getMessage() . ")" );
250  }
251  }
252 
253  public function initProgress( $history = WikiExporter::FULL ) {
254  parent::initProgress();
255  $this->timeOfCheckpoint = $this->startTime;
256  }
257 
258  public function dump( $history, $text = WikiExporter::TEXT ) {
259  // Notice messages will foul up your XML output even if they're
260  // relatively harmless.
261  if ( ini_get( 'display_errors' ) ) {
262  ini_set( 'display_errors', 'stderr' );
263  }
264 
265  $this->initProgress( $this->history );
266 
267  // We are trying to get an initial database connection to avoid that the
268  // first try of this request's first call to getText fails. However, if
269  // obtaining a good DB connection fails it's not a serious issue, as
270  // getText does retry upon failure and can start without having a working
271  // DB connection.
272  try {
273  $this->rotateDb();
274  } catch ( Exception $e ) {
275  // We do not even count this as failure. Just let eventual
276  // watchdogs know.
277  $this->progress( "Getting initial DB connection failed (" .
278  $e->getMessage() . ")" );
279  }
280 
281  $this->egress = new ExportProgressFilter( $this->sink, $this );
282 
283  // it would be nice to do it in the constructor, oh well. need egress set
284  $this->finalOptionCheck();
285 
286  // we only want this so we know how to close a stream :-P
287  $this->xmlwriterobj = new XmlDumpWriter( XmlDumpWriter::WRITE_CONTENT, $this->schemaVersion );
288 
289  $input = fopen( $this->input, "rt" );
290  $this->readDump( $input );
291 
292  if ( $this->spawnProc ) {
293  $this->closeSpawn();
294  }
295 
296  $this->report( true );
297  }
298 
299  protected function processFileOpt( $opt ) {
300  $split = explode( ':', $opt, 2 );
301  $val = $split[0];
302  $param = '';
303  if ( count( $split ) === 2 ) {
304  $param = $split[1];
305  }
306  $fileURIs = explode( ';', $param );
307  $newFileURIs = [];
308  foreach ( $fileURIs as $URI ) {
309  switch ( $val ) {
310  case "file":
311  $newURI = $URI;
312  break;
313  case "gzip":
314  $newURI = "compress.zlib://$URI";
315  break;
316  case "bzip2":
317  $newURI = "compress.bzip2://$URI";
318  break;
319  case "7zip":
320  $newURI = "mediawiki.compress.7z://$URI";
321  break;
322  default:
323  $newURI = $URI;
324  }
325  $newFileURIs[] = $newURI;
326  }
327  $val = implode( ';', $newFileURIs );
328 
329  return $val;
330  }
331 
335  public function showReport() {
336  if ( !$this->prefetch ) {
337  parent::showReport();
338 
339  return;
340  }
341 
342  if ( $this->reporting ) {
343  $now = wfTimestamp( TS_DB );
344  $nowts = microtime( true );
345  $deltaAll = $nowts - $this->startTime;
346  $deltaPart = $nowts - $this->lastTime;
347  $this->pageCountPart = $this->pageCount - $this->pageCountLast;
348  $this->revCountPart = $this->revCount - $this->revCountLast;
349 
350  if ( $deltaAll ) {
351  $portion = $this->revCount / $this->maxCount;
352  $eta = $this->startTime + $deltaAll / $portion;
353  $etats = wfTimestamp( TS_DB, intval( $eta ) );
354  if ( $this->fetchCount ) {
355  $fetchRate = 100.0 * $this->prefetchCount / $this->fetchCount;
356  } else {
357  $fetchRate = '-';
358  }
359  $pageRate = $this->pageCount / $deltaAll;
360  $revRate = $this->revCount / $deltaAll;
361  } else {
362  $pageRate = '-';
363  $revRate = '-';
364  $etats = '-';
365  $fetchRate = '-';
366  }
367  if ( $deltaPart ) {
368  if ( $this->fetchCountLast ) {
369  $fetchRatePart = 100.0 * $this->prefetchCountLast / $this->fetchCountLast;
370  } else {
371  $fetchRatePart = '-';
372  }
373  $pageRatePart = $this->pageCountPart / $deltaPart;
374  $revRatePart = $this->revCountPart / $deltaPart;
375  } else {
376  $fetchRatePart = '-';
377  $pageRatePart = '-';
378  $revRatePart = '-';
379  }
380 
381  $dbDomain = WikiMap::getCurrentWikiDbDomain()->getId();
382  $this->progress( sprintf(
383  "%s: %s (ID %d) %d pages (%0.1f|%0.1f/sec all|curr), "
384  . "%d revs (%0.1f|%0.1f/sec all|curr), %0.1f%%|%0.1f%% "
385  . "prefetched (all|curr), ETA %s [max %d]",
386  $now, $dbDomain, $this->ID, $this->pageCount, $pageRate,
387  $pageRatePart, $this->revCount, $revRate, $revRatePart,
388  $fetchRate, $fetchRatePart, $etats, $this->maxCount
389  ) );
390  $this->lastTime = $nowts;
391  $this->revCountLast = $this->revCount;
392  $this->prefetchCountLast = $this->prefetchCount;
393  $this->fetchCountLast = $this->fetchCount;
394  }
395  }
396 
397  private function setTimeExceeded() {
398  $this->timeExceeded = true;
399  }
400 
401  private function checkIfTimeExceeded() {
402  if ( $this->maxTimeAllowed
403  && ( $this->lastTime - $this->timeOfCheckpoint > $this->maxTimeAllowed )
404  ) {
405  return true;
406  }
407 
408  return false;
409  }
410 
411  private function finalOptionCheck() {
412  if ( ( $this->checkpointFiles && !$this->maxTimeAllowed )
413  || ( $this->maxTimeAllowed && !$this->checkpointFiles )
414  ) {
415  throw new MWException( "Options checkpointfile and maxtime must be specified together.\n" );
416  }
417  foreach ( $this->checkpointFiles as $checkpointFile ) {
418  $count = substr_count( $checkpointFile, "%s" );
419  if ( $count != 2 ) {
420  throw new MWException( "Option checkpointfile must contain two '%s' "
421  . "for substitution of first and last pageids, count is $count instead, "
422  . "file is $checkpointFile.\n" );
423  }
424  }
425 
426  if ( $this->checkpointFiles ) {
427  $filenameList = (array)$this->egress->getFilenames();
428  if ( count( $filenameList ) != count( $this->checkpointFiles ) ) {
429  throw new MWException( "One checkpointfile must be specified "
430  . "for each output option, if maxtime is used.\n" );
431  }
432  }
433  }
434 
440  protected function readDump( $input ) {
441  $this->buffer = "";
442  $this->openElement = false;
443  $this->atStart = true;
444  $this->state = "";
445  $this->lastName = "";
446  $this->thisPage = 0;
447  $this->thisRev = 0;
448  $this->thisRevModel = null;
449  $this->thisRevFormat = null;
450 
451  $parser = xml_parser_create( "UTF-8" );
452  xml_parser_set_option( $parser, XML_OPTION_CASE_FOLDING, false );
453 
454  xml_set_element_handler(
455  $parser,
456  [ $this, 'startElement' ],
457  [ $this, 'endElement' ]
458  );
459  xml_set_character_data_handler( $parser, [ $this, 'characterData' ] );
460 
461  $offset = 0; // for context extraction on error reporting
462  do {
463  if ( $this->checkIfTimeExceeded() ) {
464  $this->setTimeExceeded();
465  }
466  $chunk = fread( $input, $this->bufferSize );
467  if ( !xml_parse( $parser, $chunk, feof( $input ) ) ) {
468  wfDebug( "TextDumpPass::readDump encountered XML parsing error\n" );
469 
470  $byte = xml_get_current_byte_index( $parser );
471  $msg = wfMessage( 'xml-error-string',
472  'XML import parse failure',
473  xml_get_current_line_number( $parser ),
474  xml_get_current_column_number( $parser ),
475  $byte . ( is_null( $chunk ) ? null : ( '; "' . substr( $chunk, $byte - $offset, 16 ) . '"' ) ),
476  xml_error_string( xml_get_error_code( $parser ) ) )->escaped();
477 
478  xml_parser_free( $parser );
479 
480  throw new MWException( $msg );
481  }
482  $offset += strlen( $chunk );
483  } while ( $chunk !== false && !feof( $input ) );
484  if ( $this->maxTimeAllowed ) {
485  $filenameList = (array)$this->egress->getFilenames();
486  // we wrote some stuff after last checkpoint that needs renamed
487  if ( file_exists( $filenameList[0] ) ) {
488  $newFilenames = [];
489  # we might have just written the header and footer and had no
490  # pages or revisions written... perhaps they were all deleted
491  # there's no pageID 0 so we use that. the caller is responsible
492  # for deciding what to do with a file containing only the
493  # siteinfo information and the mw tags.
494  if ( !$this->firstPageWritten ) {
495  $firstPageID = str_pad( 0, 9, "0", STR_PAD_LEFT );
496  $lastPageID = str_pad( 0, 9, "0", STR_PAD_LEFT );
497  } else {
498  $firstPageID = str_pad( $this->firstPageWritten, 9, "0", STR_PAD_LEFT );
499  $lastPageID = str_pad( $this->lastPageWritten, 9, "0", STR_PAD_LEFT );
500  }
501 
502  $filenameCount = count( $filenameList );
503  for ( $i = 0; $i < $filenameCount; $i++ ) {
504  $checkpointNameFilledIn = sprintf( $this->checkpointFiles[$i], $firstPageID, $lastPageID );
505  $fileinfo = pathinfo( $filenameList[$i] );
506  $newFilenames[] = $fileinfo['dirname'] . '/' . $checkpointNameFilledIn;
507  }
508  $this->egress->closeAndRename( $newFilenames );
509  }
510  }
511  xml_parser_free( $parser );
512 
513  return true;
514  }
515 
525  private function exportTransform( $text, $model, $format = null ) {
526  try {
527  $handler = ContentHandler::getForModelID( $model );
528  $text = $handler->exportTransform( $text, $format );
529  }
530  catch ( MWException $ex ) {
531  $this->progress(
532  "Unable to apply export transformation for content model '$model': " .
533  $ex->getMessage()
534  );
535  }
536 
537  return $text;
538  }
539 
560  protected function getText( $id, $model = null, $format = null ) {
561  global $wgContentHandlerUseDB;
562 
563  $prefetchNotTried = true; // Whether or not we already tried to get the text via prefetch.
564  $text = false; // The candidate for a good text. false if no proper value.
565  $failures = 0; // The number of times, this invocation of getText already failed.
566 
567  // The number of times getText failed without yielding a good text in between.
568  static $consecutiveFailedTextRetrievals = 0;
569 
570  $this->fetchCount++;
571 
572  // To allow to simply return on success and do not have to worry about book keeping,
573  // we assume, this fetch works (possible after some retries). Nevertheless, we koop
574  // the old value, so we can restore it, if problems occur (See after the while loop).
575  $oldConsecutiveFailedTextRetrievals = $consecutiveFailedTextRetrievals;
576  $consecutiveFailedTextRetrievals = 0;
577 
578  if ( $model === null && $wgContentHandlerUseDB ) {
579  // TODO: MCR: use content table
580  $row = $this->db->selectRow(
581  'revision',
582  [ 'rev_content_model', 'rev_content_format' ],
583  [ 'rev_id' => $this->thisRev ],
584  __METHOD__
585  );
586 
587  if ( $row ) {
588  $model = $row->rev_content_model;
589  $format = $row->rev_content_format;
590  }
591  }
592 
593  if ( $model === null || $model === '' ) {
594  $model = false;
595  }
596 
597  while ( $failures < $this->maxFailures ) {
598  // As soon as we found a good text for the $id, we will return immediately.
599  // Hence, if we make it past the try catch block, we know that we did not
600  // find a good text.
601 
602  try {
603  // Step 1: Get some text (or reuse from previous iteratuon if checking
604  // for plausibility failed)
605 
606  // Trying to get prefetch, if it has not been tried before
607  if ( $text === false && isset( $this->prefetch ) && $prefetchNotTried ) {
608  $prefetchNotTried = false;
609  $tryIsPrefetch = true;
610  $text = $this->prefetch->prefetch( (int)$this->thisPage, (int)$this->thisRev );
611 
612  if ( $text === null ) {
613  $text = false;
614  }
615 
616  if ( is_string( $text ) && $model !== false ) {
617  // Apply export transformation to text coming from an old dump.
618  // The purpose of this transformation is to convert up from legacy
619  // formats, which may still be used in the older dump that is used
620  // for pre-fetching. Applying the transformation again should not
621  // interfere with content that is already in the correct form.
622  $text = $this->exportTransform( $text, $model, $format );
623  }
624  }
625 
626  if ( $text === false ) {
627  // Fallback to asking the database
628  $tryIsPrefetch = false;
629  if ( $this->spawn ) {
630  $text = $this->getTextSpawned( $id );
631  } else {
632  $text = $this->getTextDb( $id );
633  }
634 
635  if ( $text !== false && $model !== false ) {
636  // Apply export transformation to text coming from the database.
637  // Prefetched text should already have transformations applied.
638  $text = $this->exportTransform( $text, $model, $format );
639  }
640 
641  // No more checks for texts from DB for now.
642  // If we received something that is not false,
643  // We treat it as good text, regardless of whether it actually is or is not
644  if ( $text !== false ) {
645  return $text;
646  }
647  }
648 
649  if ( $text === false ) {
650  throw new MWException( "Generic error while obtaining text for id " . $id );
651  }
652 
653  // We received a good candidate for the text of $id via some method
654 
655  // Step 2: Checking for plausibility and return the text if it is
656  // plausible
657  $revID = intval( $this->thisRev );
658  if ( !isset( $this->db ) ) {
659  throw new MWException( "No database available" );
660  }
661 
662  if ( $model !== CONTENT_MODEL_WIKITEXT ) {
663  $revLength = strlen( $text );
664  } else {
665  $revLength = $this->db->selectField( 'revision', 'rev_len', [ 'rev_id' => $revID ] );
666  }
667 
668  if ( strlen( $text ) == $revLength ) {
669  if ( $tryIsPrefetch ) {
670  $this->prefetchCount++;
671  }
672 
673  return $text;
674  }
675 
676  $text = false;
677  throw new MWException( "Received text is unplausible for id " . $id );
678  } catch ( Exception $e ) {
679  $msg = "getting/checking text " . $id . " failed (" . $e->getMessage() . ")";
680  if ( $failures + 1 < $this->maxFailures ) {
681  $msg .= " (Will retry " . ( $this->maxFailures - $failures - 1 ) . " more times)";
682  }
683  $this->progress( $msg );
684  }
685 
686  // Something went wrong; we did not a text that was plausible :(
687  $failures++;
688 
689  // A failure in a prefetch hit does not warrant resetting db connection etc.
690  if ( !$tryIsPrefetch ) {
691  // After backing off for some time, we try to reboot the whole process as
692  // much as possible to not carry over failures from one part to the other
693  // parts
694  sleep( $this->failureTimeout );
695  try {
696  $this->rotateDb();
697  if ( $this->spawn ) {
698  $this->closeSpawn();
699  $this->openSpawn();
700  }
701  } catch ( Exception $e ) {
702  $this->progress( "Rebooting getText infrastructure failed (" . $e->getMessage() . ")" .
703  " Trying to continue anyways" );
704  }
705  }
706  }
707 
708  // Retirieving a good text for $id failed (at least) maxFailures times.
709  // We abort for this $id.
710 
711  // Restoring the consecutive failures, and maybe aborting, if the dump
712  // is too broken.
713  $consecutiveFailedTextRetrievals = $oldConsecutiveFailedTextRetrievals + 1;
714  if ( $consecutiveFailedTextRetrievals > $this->maxConsecutiveFailedTextRetrievals ) {
715  throw new MWException( "Graceful storage failure" );
716  }
717 
718  return "";
719  }
720 
727  private function getTextDb( $id ) {
728  $store = $this->getBlobStore();
729  $address = ( is_int( $id ) || strpos( $id, ':' ) === false )
730  ? SqlBlobStore::makeAddressFromTextId( (int)$id )
731  : $id;
732 
733  try {
734  $text = $store->getBlob( $address );
735 
736  $stripped = str_replace( "\r", "", $text );
737  $normalized = MediaWikiServices::getInstance()->getContentLanguage()
738  ->normalize( $stripped );
739 
740  return $normalized;
741  } catch ( BlobAccessException $ex ) {
742  // XXX: log a warning?
743  return false;
744  }
745  }
746 
751  private function getTextSpawned( $address ) {
752  Wikimedia\suppressWarnings();
753  if ( !$this->spawnProc ) {
754  // First time?
755  $this->openSpawn();
756  }
757  $text = $this->getTextSpawnedOnce( $address );
758  Wikimedia\restoreWarnings();
759 
760  return $text;
761  }
762 
763  protected function openSpawn() {
764  global $IP;
765 
767  if ( count( $this->php ) == 2 ) {
768  $mwscriptpath = $this->php[1];
769  } else {
770  $mwscriptpath = "$IP/../multiversion/MWScript.php";
771  }
772  if ( file_exists( $mwscriptpath ) ) {
773  $cmd = implode( " ",
774  array_map( [ Shell::class, 'escape' ],
775  [
776  $this->php[0],
777  $mwscriptpath,
778  "fetchText.php",
779  '--wiki', $wiki ] ) );
780  } else {
781  $cmd = implode( " ",
782  array_map( [ Shell::class, 'escape' ],
783  [
784  $this->php[0],
785  "$IP/maintenance/fetchText.php",
786  '--wiki', $wiki ] ) );
787  }
788  $spec = [
789  0 => [ "pipe", "r" ],
790  1 => [ "pipe", "w" ],
791  2 => [ "file", "/dev/null", "a" ] ];
792  $pipes = [];
793 
794  $this->progress( "Spawning database subprocess: $cmd" );
795  $this->spawnProc = proc_open( $cmd, $spec, $pipes );
796  if ( !$this->spawnProc ) {
797  $this->progress( "Subprocess spawn failed." );
798 
799  return false;
800  }
801  list(
802  $this->spawnWrite, // -> stdin
803  $this->spawnRead, // <- stdout
804  ) = $pipes;
805 
806  return true;
807  }
808 
809  private function closeSpawn() {
810  Wikimedia\suppressWarnings();
811  if ( $this->spawnRead ) {
812  fclose( $this->spawnRead );
813  }
814  $this->spawnRead = null;
815  if ( $this->spawnWrite ) {
816  fclose( $this->spawnWrite );
817  }
818  $this->spawnWrite = null;
819  if ( $this->spawnErr ) {
820  fclose( $this->spawnErr );
821  }
822  $this->spawnErr = false;
823  if ( $this->spawnProc ) {
824  pclose( $this->spawnProc );
825  }
826  $this->spawnProc = false;
827  Wikimedia\restoreWarnings();
828  }
829 
834  private function getTextSpawnedOnce( $address ) {
835  if ( is_int( $address ) || intval( $address ) ) {
836  $address = SqlBlobStore::makeAddressFromTextId( (int)$address );
837  }
838 
839  $ok = fwrite( $this->spawnWrite, "$address\n" );
840  // $this->progress( ">> $id" );
841  if ( !$ok ) {
842  return false;
843  }
844 
845  $ok = fflush( $this->spawnWrite );
846  // $this->progress( ">> [flush]" );
847  if ( !$ok ) {
848  return false;
849  }
850 
851  // check that the text address they are sending is the one we asked for
852  // this avoids out of sync revision text errors we have encountered in the past
853  $newAddress = fgets( $this->spawnRead );
854  if ( $newAddress === false ) {
855  return false;
856  }
857  $newAddress = trim( $newAddress );
858  if ( strpos( $newAddress, ':' ) === false ) {
859  $newAddress = SqlBlobStore::makeAddressFromTextId( intval( $newAddress ) );
860  }
861 
862  if ( $newAddress !== $address ) {
863  return false;
864  }
865 
866  $len = fgets( $this->spawnRead );
867  // $this->progress( "<< " . trim( $len ) );
868  if ( $len === false ) {
869  return false;
870  }
871 
872  $nbytes = intval( $len );
873  // actual error, not zero-length text
874  if ( $nbytes < 0 ) {
875  return false;
876  }
877 
878  $text = "";
879 
880  // Subprocess may not send everything at once, we have to loop.
881  while ( $nbytes > strlen( $text ) ) {
882  $buffer = fread( $this->spawnRead, $nbytes - strlen( $text ) );
883  if ( $buffer === false ) {
884  break;
885  }
886  $text .= $buffer;
887  }
888 
889  $gotbytes = strlen( $text );
890  if ( $gotbytes != $nbytes ) {
891  $this->progress( "Expected $nbytes bytes from database subprocess, got $gotbytes " );
892 
893  return false;
894  }
895 
896  // Do normalization in the dump thread...
897  $stripped = str_replace( "\r", "", $text );
898  $normalized = MediaWikiServices::getInstance()->getContentLanguage()->
899  normalize( $stripped );
900 
901  return $normalized;
902  }
903 
904  protected function startElement( $parser, $name, $attribs ) {
905  $this->checkpointJustWritten = false;
906 
907  $this->clearOpenElement( null );
908  $this->lastName = $name;
909 
910  if ( $name == 'revision' ) {
911  $this->state = $name;
912  $this->egress->writeOpenPage( null, $this->buffer );
913  $this->buffer = "";
914  } elseif ( $name == 'page' ) {
915  $this->state = $name;
916  if ( $this->atStart ) {
917  $this->egress->writeOpenStream( $this->buffer );
918  $this->buffer = "";
919  $this->atStart = false;
920  }
921  }
922 
923  if ( $name == "text" && isset( $attribs['id'] ) ) {
924  $id = $attribs['id'];
925  $model = trim( $this->thisRevModel );
926  $format = trim( $this->thisRevFormat );
927 
928  $model = $model === '' ? null : $model;
929  $format = $format === '' ? null : $format;
930 
931  $text = $this->getText( $id, $model, $format );
932  $this->openElement = [ $name, [ 'xml:space' => 'preserve' ] ];
933  if ( strlen( $text ) > 0 ) {
934  $this->characterData( $parser, $text );
935  }
936  } else {
937  $this->openElement = [ $name, $attribs ];
938  }
939  }
940 
941  protected function endElement( $parser, $name ) {
942  $this->checkpointJustWritten = false;
943 
944  if ( $this->openElement ) {
945  $this->clearOpenElement( "" );
946  } else {
947  $this->buffer .= "</$name>";
948  }
949 
950  if ( $name == 'revision' ) {
951  $this->egress->writeRevision( null, $this->buffer );
952  $this->buffer = "";
953  $this->thisRev = "";
954  $this->thisRevModel = null;
955  $this->thisRevFormat = null;
956  } elseif ( $name == 'page' ) {
957  if ( !$this->firstPageWritten ) {
958  $this->firstPageWritten = trim( $this->thisPage );
959  }
960  $this->lastPageWritten = trim( $this->thisPage );
961  if ( $this->timeExceeded ) {
962  $this->egress->writeClosePage( $this->buffer );
963  // nasty hack, we can't just write the chardata after the
964  // page tag, it will include leading blanks from the next line
965  $this->egress->sink->write( "\n" );
966 
967  $this->buffer = $this->xmlwriterobj->closeStream();
968  $this->egress->writeCloseStream( $this->buffer );
969 
970  $this->buffer = "";
971  $this->thisPage = "";
972  // this could be more than one file if we had more than one output arg
973 
974  $filenameList = (array)$this->egress->getFilenames();
975  $newFilenames = [];
976  $firstPageID = str_pad( $this->firstPageWritten, 9, "0", STR_PAD_LEFT );
977  $lastPageID = str_pad( $this->lastPageWritten, 9, "0", STR_PAD_LEFT );
978  $filenamesCount = count( $filenameList );
979  for ( $i = 0; $i < $filenamesCount; $i++ ) {
980  $checkpointNameFilledIn = sprintf( $this->checkpointFiles[$i], $firstPageID, $lastPageID );
981  $fileinfo = pathinfo( $filenameList[$i] );
982  $newFilenames[] = $fileinfo['dirname'] . '/' . $checkpointNameFilledIn;
983  }
984  $this->egress->closeRenameAndReopen( $newFilenames );
985  $this->buffer = $this->xmlwriterobj->openStream();
986  $this->timeExceeded = false;
987  $this->timeOfCheckpoint = $this->lastTime;
988  $this->firstPageWritten = false;
989  $this->checkpointJustWritten = true;
990  } else {
991  $this->egress->writeClosePage( $this->buffer );
992  $this->buffer = "";
993  $this->thisPage = "";
994  }
995  } elseif ( $name == 'mediawiki' ) {
996  $this->egress->writeCloseStream( $this->buffer );
997  $this->buffer = "";
998  }
999  }
1000 
1001  protected function characterData( $parser, $data ) {
1002  $this->clearOpenElement( null );
1003  if ( $this->lastName == "id" ) {
1004  if ( $this->state == "revision" ) {
1005  $this->thisRev .= $data;
1006  } elseif ( $this->state == "page" ) {
1007  $this->thisPage .= $data;
1008  }
1009  } elseif ( $this->lastName == "model" ) {
1010  $this->thisRevModel .= $data;
1011  } elseif ( $this->lastName == "format" ) {
1012  $this->thisRevFormat .= $data;
1013  }
1014 
1015  // have to skip the newline left over from closepagetag line of
1016  // end of checkpoint files. nasty hack!!
1017  if ( $this->checkpointJustWritten ) {
1018  if ( $data[0] == "\n" ) {
1019  $data = substr( $data, 1 );
1020  }
1021  $this->checkpointJustWritten = false;
1022  }
1023  $this->buffer .= htmlspecialchars( $data );
1024  }
1025 
1026  protected function clearOpenElement( $style ) {
1027  if ( $this->openElement ) {
1028  $this->buffer .= Xml::element( $this->openElement[0], $this->openElement[1], $style );
1029  $this->openElement = false;
1030  }
1031  }
1032 }
IMaintainableDatabase $db
const CONTENT_MODEL_WIKITEXT
Definition: Defines.php:215
$IP
Definition: WebStart.php:41
const WRITE_CONTENT
Output serialized revision content.
getOption( $name, $default=null)
Get an option, or return the default.
progress( $string)
characterData( $parser, $data)
resource $spawnWrite
string bool $thisRev
exportTransform( $text, $model, $format=null)
Applies applicable export transformations to $text.
static getWikiIdFromDbDomain( $domain)
Get the wiki ID of a database domain.
Definition: WikiMap.php:269
__construct( $args=null)
getText( $id, $model=null, $format=null)
Tries to load revision text.
hasOption( $name)
Checks to see if a particular option exists.
showReport()
Overridden to include prefetch ratio if enabled.
bool XmlDumpWriter $xmlwriterobj
loadWithArgv( $argv)
Load params and arguments from a given array of command-line arguments.
Exception representing a failure to access a data blob.
startElement( $parser, $name, $attribs)
clearOpenElement( $style)
resource $spawnRead
$wgContentHandlerUseDB
Set to false to disable use of the database fields introduced by the ContentHandler facility...
getTextSpawned( $address)
wfTimestamp( $outputtype=TS_UNIX, $ts=0)
Get a timestamp string in one of various formats.
if( $line===false) $args
Definition: mcc.php:124
IMaintainableDatabase null $forcedDb
The dependency-injected database to use.
dump( $history, $text=WikiExporter::TEXT)
static getForModelID( $modelId)
Returns the ContentHandler singleton for the given model ID.
string $buffer
wfDebug( $text, $dest='all', array $context=[])
Sends a line to the debug log if enabled or, optionally, to a comment in output.
addDescription( $text)
Set the description text.
Readahead helper for making large MediaWiki data dumps; reads in a previous XML dump to sequentially ...
Definition: BaseDump.php:42
bool resource $spawnProc
getTextDb( $id)
Loads the serialized content from storage.
output( $out, $channel=null)
Throw some output to the user.
string [] $checkpointFiles
string bool $thisPage
static getCurrentWikiDbDomain()
Definition: WikiMap.php:293
report( $final=false)
initProgress( $history=WikiExporter::FULL)
static element( $element, $attribs=null, $contents='', $allowShortTag=true)
Format an XML element with given attributes and, optionally, text content.
Definition: Xml.php:41
addOption( $name, $description, $required=false, $withArg=false, $shortName=false, $multiOccurrence=false)
Add a parameter to the script.
const DB_REPLICA
Definition: defines.php:25
wfMessage( $key,... $params)
This is the function for getting translated interface messages.
BaseDump $prefetch
bool resource $spawnErr
rotateDb()
Drop the database connection $this->db and try to get a new one.
getTextSpawnedOnce( $address)
endElement( $parser, $name)