Replace FOR UPDATE with LockManager use in LocalFile::lock()

* This avoids excess contention where inserts of rows for
  similarly named files get blocked. This often would effect
  users of UploadWizard or any bot doing mass uploads.

Change-Id: Ie7a328f7d4f03aa249770804417347a50356ea42

diff --git a/includes/filebackend/FileBackend.php b/includes/filebackend/FileBackend.php
index a9e312d..2820be2 100644 (file)
--- a/includes/filebackend/FileBackend.php
+++ b/includes/filebackend/FileBackend.php
@@ -1231,12 +1231,13 @@ abstract class FileBackend {
         *
         * @param array $paths Storage paths
         * @param int $type LockManager::LOCK_* constant
+        * @param int $timeout Timeout in seconds (0 means non-blocking) (since 1.24)
         * @return Status
         */
-       final public function lockFiles( array $paths, $type ) {
+       final public function lockFiles( array $paths, $type, $timeout = 0 ) {
                $paths = array_map( 'FileBackend::normalizeStoragePath', $paths );
 
-               return $this->lockManager->lock( $paths, $type );
+               return $this->lockManager->lock( $paths, $type, $timeout );
        }
 
        /**
@@ -1265,9 +1266,10 @@ abstract class FileBackend {
         * @param array $paths List of storage paths or map of lock types to path lists
         * @param int|string $type LockManager::LOCK_* constant or "mixed"
         * @param Status $status Status to update on lock/unlock
+        * @param int $timeout Timeout in seconds (0 means non-blocking) (since 1.24)
         * @return ScopedLock|null Returns null on failure
         */
-       final public function getScopedFileLocks( array $paths, $type, Status $status ) {
+       final public function getScopedFileLocks( array $paths, $type, Status $status, $timeout = 0 ) {
                if ( $type === 'mixed' ) {
                        foreach ( $paths as &$typePaths ) {
                                $typePaths = array_map( 'FileBackend::normalizeStoragePath', $typePaths );
@@ -1276,7 +1278,7 @@ abstract class FileBackend {
                        $paths = array_map( 'FileBackend::normalizeStoragePath', $paths );
                }
 
-               return ScopedLock::factory( $this->lockManager, $paths, $type, $status );
+               return ScopedLock::factory( $this->lockManager, $paths, $type, $status, $timeout );
        }
 
        /**

diff --git a/includes/filerepo/file/LocalFile.php b/includes/filerepo/file/LocalFile.php
index a8fa8bd..660f193 100644 (file)
--- a/includes/filerepo/file/LocalFile.php
+++ b/includes/filerepo/file/LocalFile.php
@@ -1249,7 +1249,7 @@ class LocalFile extends File {
                }
 
                if ( $timestamp === false ) {
-                       // Use FOR UPDATE in case lock()/unlock() did not start the transaction
+                       // Use FOR UPDATE to ignore any transaction snapshotting
                        $ltimestamp = $dbw->selectField( 'image', 'img_timestamp',
                                array( 'img_name' => $this->getName() ), __METHOD__, array( 'FOR UPDATE' ) );
                        $ltime = $ltimestamp ? wfTimestamp( TS_UNIX, $ltimestamp ) : false;
@@ -1836,8 +1836,8 @@ class LocalFile extends File {
        /**
         * Start a transaction and lock the image for update
         * Increments a reference counter if the lock is already held
-        * @throws MWException
-        * @return bool True if the image exists, false otherwise
+        * @throws MWException Throws an error if the lock was not acquired
+        * @return bool success
         */
        function lock() {
                $dbw = $this->repo->getMasterDB();
@@ -1849,21 +1849,22 @@ class LocalFile extends File {
                        }
                        $this->locked++;
                        // Bug 54736: use simple lock to handle when the file does not exist.
-                       // SELECT FOR UPDATE only locks records not the gaps where there are none.
-                       $cache = wfGetMainCache();
-                       $key = $this->getCacheKey();
-                       if ( !$cache->lock( $key, 5 ) ) {
+                       // SELECT FOR UPDATE prevents changes, not other SELECTs with FOR UPDATE.
+                       // Also, that would cause contention on INSERT of similarly named rows.
+                       $backend = $this->getRepo()->getBackend();
+                       $lockPaths = array( $this->getPath() ); // represents all versions of the file
+                       $status = $backend->lockFiles( $lockPaths, LockManager::LOCK_EX, 5 );
+                       if ( !$status->isGood() ) {
                                throw new MWException( "Could not acquire lock for '{$this->getName()}.'" );
                        }
-                       $dbw->onTransactionIdle( function () use ( $cache, $key ) {
-                               $cache->unlock( $key ); // release on commit
+                       $dbw->onTransactionIdle( function () use ( $backend, $lockPaths ) {
+                               $backend->unlockFiles( $lockPaths, LockManager::LOCK_EX ); // release on commit
                        } );
                }
 
                $this->markVolatile(); // file may change soon
 
-               return $dbw->selectField( 'image', '1',
-                       array( 'img_name' => $this->getName() ), __METHOD__, array( 'FOR UPDATE' ) );
+               return true;
        }
 
        /**
@@ -2394,10 +2395,14 @@ class LocalFileRestoreBatch {
                        return $this->file->repo->newGood();
                }
 
-               $exists = $this->file->lock();
+               $this->file->lock();
+
                $dbw = $this->file->repo->getMasterDB();
                $status = $this->file->repo->newGood();
 
+               $exists = (bool)$dbw->selectField( 'image', '1',
+                       array( 'img_name' => $this->file->getName() ), __METHOD__, array( 'FOR UPDATE' ) );
+
                // Fetch all or selected archived revisions for the file,
                // sorted from the most recent to the oldest.
                $conditions = array( 'fa_name' => $this->file->getName() );