Initial implementation of restartable subprocess for text fetching in dumpTextPass.php

diff --git a/RELEASE-NOTES b/RELEASE-NOTES
index c6ec648..4be0590 100644 (file)
--- a/RELEASE-NOTES
+++ b/RELEASE-NOTES
@@ -71,6 +71,9 @@ it from source control: http://www.mediawiki.org/wiki/Download_from_SVN
   because it is part of a blocked range.
 * (bug 3973) Use a separate message for the email content when an account is 
   created by another user
+* dumpTextPass.php can spawn fetchText.php as a subprocess, which should restart
+  cleanly if database connections fail unpleasantly.
+
 
 === Bug fixes in 1.12 ===
 

diff --git a/maintenance/dumpTextPass.php b/maintenance/dumpTextPass.php
index 92ab4b4..bb7bd61 100644 (file)
--- a/maintenance/dumpTextPass.php
+++ b/maintenance/dumpTextPass.php
@@ -104,6 +104,13 @@ class TextPassDumper extends BackupDumper {
        var $failures = 0;
        var $maxFailures = 200;
        var $failureTimeout = 5; // Seconds to sleep after db failure
+       
+       var $php = "php";
+       var $spawn = false;
+       var $spawnProc = false;
+       var $spawnWrite = false;
+       var $spawnRead = false;
+       var $spawnErr = false;
 
        function dump() {
                # This shouldn't happen if on console... ;)
@@ -146,6 +153,12 @@ class TextPassDumper extends BackupDumper {
                case 'full':
                        $this->history = WikiExporter::FULL;
                        break;
+               case 'spawn':
+                       $this->spawn = true;
+                       if( $val ) {
+                               $this->php = $val;
+                       }
+                       break;
                }
        }
        
@@ -237,9 +250,26 @@ class TextPassDumper extends BackupDumper {
                                return $text;
                        }
                }
+               return $this->doGetText( $id );
+       }
+       
+       private function doGetText( $id ) {
+               if( $this->spawn ) {
+                       return $this->getTextSpawned( $id );
+               } else {
+                       return $this->getTextDbSafe( $id );
+               }
+       }
+       
+       /**
+        * Fetch a text revision from the database, retrying in case of failure.
+        * This may survive some transitory errors by reconnecting, but
+        * may not survive a long-term server outage.
+        */
+       private function getTextDbSafe( $id ) {
                while( true ) {
                        try {
-                               $text = $this->doGetText( $id );
+                               $text = $this->getTextDb( $id );
                                $ex = new MWException("Graceful storage failure");
                        } catch (DBQueryError $ex) {
                                $text = false;
@@ -263,7 +293,7 @@ class TextPassDumper extends BackupDumper {
        /**
         * May throw a database error if, say, the server dies during query.
         */
-       private function doGetText( $id ) {
+       private function getTextDb( $id ) {
                $id = intval( $id );
                $row = $this->db->selectRow( 'text',
                        array( 'old_text', 'old_flags' ),
@@ -277,6 +307,106 @@ class TextPassDumper extends BackupDumper {
                $normalized = UtfNormal::cleanUp( $stripped );
                return $normalized;
        }
+       
+       private function getTextSpawned( $id ) {
+               wfSuppressWarnings();
+               if( !$this->spawnProc ) {
+                       // First time?
+                       $this->openSpawn();
+               }
+               while( true ) {
+                       
+                       $text = $this->getTextSpawnedOnce( $id );
+                       if( !is_string( $text ) ) {
+                               $this->progress("Database subprocess failed. Respawning...");
+                               
+                               $this->closeSpawn();
+                               sleep( $this->failureTimeout );
+                               $this->openSpawn();
+                               
+                               continue;
+                       }
+                       wfRestoreWarnings();
+                       return $text;
+               }
+       }
+       
+       function openSpawn() {
+               global $IP, $wgDBname;
+               
+               $cmd = implode( " ",
+                       array_map( 'wfEscapeShellArg',
+                               array(
+                                       $this->php,
+                                       "$IP/maintenance/fetchText.php",
+                                       $wgDBname ) ) );
+               $spec = array(
+                       0 => array( "pipe", "r" ),
+                       1 => array( "pipe", "w" ),
+                       2 => array( "file", "/dev/null", "a" ) );
+               $pipes = array();
+               
+               $this->progress( "Spawning database subprocess: $cmd" );
+               $this->spawnProc = proc_open( $cmd, $spec, $pipes );
+               if( !$this->spawnProc ) {
+                       // shit
+                       $this->progress( "Subprocess spawn failed." );
+                       return false;
+               }
+               list(
+                       $this->spawnWrite, // -> stdin
+                       $this->spawnRead,  // <- stdout
+               ) = $pipes;
+               
+               return true;
+       }
+       
+       private function closeSpawn() {
+               if( $this->spawnRead )
+                       fclose( $this->spawnRead );
+               $this->spawnRead = false;
+               if( $this->spawnWrite )
+                       fclose( $this->spawnWrite );
+               $this->spawnWrite = false;
+               if( $this->spawnErr )
+                       fclose( $this->spawnErr );
+               $this->spawnErr = false;
+               if( $this->spawnProc )
+                       pclose( $this->spawnProc );
+               $this->spawnProc = false;
+       }
+       
+       private function getTextSpawnedOnce( $id ) {
+               $ok = fwrite( $this->spawnWrite, "$id\n" );
+               //$this->progress( ">> $id" );
+               if( !$ok ) return false;
+               
+               $ok = fflush( $this->spawnWrite );
+               //$this->progress( ">> [flush]" );
+               if( !$ok ) return false;
+               
+               $len = fgets( $this->spawnRead );
+               //$this->progress( "<< " . trim( $len ) );
+               if( $len === false ) return false;
+               
+               $nbytes = intval( $len );
+               $text = "";
+               
+               // Subprocess may not send everything at once, we have to loop.
+               while( $nbytes > strlen( $text ) ) {
+                       $buffer = fread( $this->spawnRead, $nbytes - strlen( $text ) );
+                       if( $text === false ) break;
+                       $text .= $buffer;
+               }
+               
+               $gotbytes = strlen( $text );
+               if( $gotbytes != $nbytes ) {
+                       $this->progress( "Expected $nbytes bytes from database subprocess, got $gotbytes ");
+                       return false;
+               }
+               
+               return $text;
+       }
 
        function startElement( $parser, $name, $attribs ) {
                $this->clearOpenElement( null );
@@ -371,6 +501,7 @@ Options:
               (Default: 100)
   --server=h  Force reading from MySQL server h
   --current   Base ETA on number of pages in database instead of all revisions
+  --spawn     Spawn a subprocess for loading text records
 END
 );
 }

diff --git a/maintenance/fetchText.php b/maintenance/fetchText.php
new file mode 100644 (file)
index 0000000..38ba3fb
--- /dev/null
+++ b/maintenance/fetchText.php
@@ -0,0 +1,38 @@
+<?php
+
+/**
+ * Communications protocol...
+ */
+
+require "commandLine.inc";
+
+$db = wfGetDB( DB_SLAVE );
+$stdin = fopen( "php://stdin", "rt" );
+while( !feof( $stdin ) ) {
+       $line = fgets( $stdin );
+       $textId = intval( $line );
+       $text = doGetText( $db, $textId );
+       echo strlen( $text ) . "\n";
+       echo $text;
+}
+
+/**
+ * May throw a database error if, say, the server dies during query.
+ */
+function doGetText( $db, $id ) {
+       $id = intval( $id );
+       $row = $db->selectRow( 'text',
+               array( 'old_text', 'old_flags' ),
+               array( 'old_id' => $id ),
+               'TextPassDumper::getText' );
+       $text = Revision::getRevisionText( $row );
+       if( $text === false ) {
+               return false;
+       }
+       $stripped = str_replace( "\r", "", $text );
+       $normalized = UtfNormal::cleanUp( $stripped );
+       return $normalized;
+}
+
+
+?>
\ No newline at end of file