Merge pull request #24863 from owncloud/propagator-batching

Propagator batching for the file scanner

lib/private/Files/Cache/Propagator.php

@@ -29,6 +29,10 @@ use OCP\IDBConnection;
  * Propagate etags and mtimes within the storage
  */
 class Propagator implements IPropagator {
+	private $inBatch = false;
+
+	private $batch = [];
+
 	/**
 	 * @var \OC\Files\Storage\Storage
 	 */
@@ -59,6 +63,13 @@ class Propagator implements IPropagator {
 
 		$parents = $this->getParents($internalPath);
 
+		if ($this->inBatch) {
+			foreach ($parents as $parent) {
+				$this->addToBatch($parent, $time, $sizeDifference);
+			}
+			return;
+		}
+
 		$parentHashes = array_map('md5', $parents);
 		$etag = uniqid(); // since we give all folders the same etag we don't ask the storage for the etag
 
@@ -68,7 +79,7 @@ class Propagator implements IPropagator {
 		}, $parentHashes);
 
 		$builder->update('filecache')
-			->set('mtime', $builder->createFunction('GREATEST(`mtime`, ' . $builder->createNamedParameter($time) . ')'))
+			->set('mtime', $builder->createFunction('GREATEST(`mtime`, ' . $builder->createNamedParameter($time, IQueryBuilder::PARAM_INT) . ')'))
 			->set('etag', $builder->createNamedParameter($etag, IQueryBuilder::PARAM_STR))
 			->where($builder->expr()->eq('storage', $builder->createNamedParameter($storageId, IQueryBuilder::PARAM_INT)))
 			->andWhere($builder->expr()->in('path_hash', $hashParams));
@@ -98,4 +109,79 @@ class Propagator implements IPropagator {
 		}
 		return $parents;
 	}
+
+	/**
+	 * Mark the beginning of a propagation batch
+	 *
+	 * Note that not all cache setups support propagation in which case this will be a noop
+	 *
+	 * Batching for cache setups that do support it has to be explicit since the cache state is not fully consistent
+	 * before the batch is committed.
+	 */
+	public function beginBatch() {
+		$this->inBatch = true;
+	}
+
+	private function addToBatch($internalPath, $time, $sizeDifference) {
+		if (!isset($this->batch[$internalPath])) {
+			$this->batch[$internalPath] = [
+				'hash' => md5($internalPath),
+				'time' => $time,
+				'size' => $sizeDifference
+			];
+		} else {
+			$this->batch[$internalPath]['size'] += $sizeDifference;
+			if ($time > $this->batch[$internalPath]['time']) {
+				$this->batch[$internalPath]['time'] = $time;
+			}
+		}
+	}
+
+	/**
+	 * Commit the active propagation batch
+	 */
+	public function commitBatch() {
+		if (!$this->inBatch) {
+			throw new \BadMethodCallException('Not in batch');
+		}
+		$this->inBatch = false;
+
+		$this->connection->beginTransaction();
+
+		$query = $this->connection->getQueryBuilder();
+		$storageId = (int)$this->storage->getStorageCache()->getNumericId();
+
+		$query->update('filecache')
+			->set('mtime', $query->createFunction('GREATEST(`mtime`, ' . $query->createParameter('time') . ')'))
+			->set('etag', $query->expr()->literal(uniqid()))
+			->where($query->expr()->eq('storage', $query->expr()->literal($storageId, IQueryBuilder::PARAM_INT)))
+			->andWhere($query->expr()->eq('path_hash', $query->createParameter('hash')));
+
+		$sizeQuery = $this->connection->getQueryBuilder();
+		$sizeQuery->update('filecache')
+			->set('size', $sizeQuery->createFunction('`size` + ' . $sizeQuery->createParameter('size')))
+			->where($query->expr()->eq('storage', $query->expr()->literal($storageId, IQueryBuilder::PARAM_INT)))
+			->andWhere($query->expr()->eq('path_hash', $query->createParameter('hash')))
+			->andWhere($sizeQuery->expr()->gt('size', $sizeQuery->expr()->literal(-1, IQueryBuilder::PARAM_INT)));
+
+		foreach ($this->batch as $item) {
+			$query->setParameter('time', $item['time'], IQueryBuilder::PARAM_INT);
+			$query->setParameter('hash', $item['hash']);
+
+			$query->execute();
+
+			if ($item['size']) {
+				$sizeQuery->setParameter('size', $item['size'], IQueryBuilder::PARAM_INT);
+				$sizeQuery->setParameter('hash', $item['hash']);
+
+				$sizeQuery->execute();
+			}
+		}
+
+		$this->batch = [];
+
+		$this->connection->commit();
+	}
+
+
 }

lib/private/Files/Utils/Scanner.php

@@ -138,7 +138,9 @@ class Scanner extends PublicEmitter {
 				$this->triggerPropagator($storage, $path);
 			});
 
+			$storage->getPropagator()->beginBatch();
 			$scanner->backgroundScan();
+			$storage->getPropagator()->commitBatch();
 		}
 	}
 
@@ -187,12 +189,14 @@ class Scanner extends PublicEmitter {
 				$this->db->beginTransaction();
 			}
 			try {
+				$storage->getPropagator()->beginBatch();
 				$scanner->scan($relativePath, \OC\Files\Cache\Scanner::SCAN_RECURSIVE, \OC\Files\Cache\Scanner::REUSE_ETAG | \OC\Files\Cache\Scanner::REUSE_SIZE);
 				$cache = $storage->getCache();
 				if ($cache instanceof Cache) {
 					// only re-calculate for the root folder we scanned, anything below that is taken care of by the scanner
 					$cache->correctFolderSize($relativePath);
 				}
+				$storage->getPropagator()->commitBatch();
 			} catch (StorageNotAvailableException $e) {
 				$this->logger->error('Storage ' . $storage->getId() . ' not available');
 				$this->logger->logException($e);

lib/public/Files/Cache/IPropagator.php

@@ -27,6 +27,25 @@ namespace OCP\Files\Cache;
  * @since 9.0.0
  */
 interface IPropagator {
+	/**
+	 * Mark the beginning of a propagation batch
+	 *
+	 * Note that not all cache setups support propagation in which case this will be a noop
+	 *
+	 * Batching for cache setups that do support it has to be explicit since the cache state is not fully consistent
+	 * before the batch is committed.
+	 *
+	 * @since 9.1.0
+	 */
+	public function beginBatch();
+
+	/**
+	 * Commit the active propagation batch
+	 *
+	 * @since 9.1.0
+	 */
+	public function commitBatch();
+
 	/**
 	 * @param string $internalPath
 	 * @param int $time

tests/lib/Files/Cache/PropagatorTest.php

@@ -0,0 +1,125 @@
+<?php
+/**
+ * Copyright (c) 2012 Robin Appelman <icewind@owncloud.com>
+ * This file is licensed under the Affero General Public License version 3 or
+ * later.
+ * See the COPYING-README file.
+ */
+
+namespace Test\Files\Cache;
+
+use OC\Files\Storage\Temporary;
+use OCP\Files\Cache\ICacheEntry;
+use OCP\Files\Storage\IStorage;
+use Test\TestCase;
+
+/**
+ * @group DB
+ */
+class PropagatorTest extends TestCase {
+	/** @var  IStorage */
+	private $storage;
+
+	public function setUp() {
+		parent::setUp();
+		$this->storage = new Temporary();
+		$this->storage->mkdir('foo/bar');
+		$this->storage->file_put_contents('foo/bar/file.txt', 'bar');
+		$this->storage->getScanner()->scan('');
+	}
+
+	/**
+	 * @param $paths
+	 * @return ICacheEntry[]
+	 */
+	private function getFileInfos($paths) {
+		$values = array_map(function ($path) {
+			return $this->storage->getCache()->get($path);
+		}, $paths);
+		return array_combine($paths, $values);
+	}
+
+	public function testEtagPropagation() {
+		$paths = ['', 'foo', 'foo/bar'];
+		$oldInfos = $this->getFileInfos($paths);
+		$this->storage->getPropagator()->propagateChange('foo/bar/file.txt', time());
+		$newInfos = $this->getFileInfos($paths);
+
+		foreach ($oldInfos as $i => $oldInfo) {
+			$this->assertNotEquals($oldInfo->getEtag(), $newInfos[$i]->getEtag());
+		}
+	}
+
+	public function testTimePropagation() {
+		$paths = ['', 'foo', 'foo/bar'];
+		$oldTime = time() - 200;
+		$targetTime = time() - 100;
+		$now = time();
+		$cache = $this->storage->getCache();
+		$cache->put('', ['mtime' => $now]);
+		$cache->put('foo', ['mtime' => $now]);
+		$cache->put('foo/bar', ['mtime' => $oldTime]);
+		$cache->put('foo/bar/file.txt', ['mtime' => $oldTime]);
+		$this->storage->getPropagator()->propagateChange('foo/bar/file.txt', $targetTime);
+		$newInfos = $this->getFileInfos($paths);
+
+		$this->assertEquals($targetTime, $newInfos['foo/bar']->getMTime());
+
+		// dont lower mtimes
+		$this->assertEquals($now, $newInfos['foo']->getMTime());
+		$this->assertEquals($now, $newInfos['']->getMTime());
+	}
+
+	public function testSizePropagation() {
+		$paths = ['', 'foo', 'foo/bar'];
+		$oldInfos = $this->getFileInfos($paths);
+		$this->storage->getPropagator()->propagateChange('foo/bar/file.txt', time(), 10);
+		$newInfos = $this->getFileInfos($paths);
+
+		foreach ($oldInfos as $i => $oldInfo) {
+			$this->assertEquals($oldInfo->getSize() + 10, $newInfos[$i]->getSize());
+		}
+	}
+
+	public function testBatchedPropagation() {
+		$this->storage->mkdir('foo/baz');
+		$this->storage->mkdir('asd');
+		$this->storage->file_put_contents('asd/file.txt', 'bar');
+		$this->storage->file_put_contents('foo/baz/file.txt', 'bar');
+		$this->storage->getScanner()->scan('');
+
+		$paths = ['', 'foo', 'foo/bar', 'asd', 'foo/baz'];
+
+		$oldInfos = $this->getFileInfos($paths);
+		$propagator = $this->storage->getPropagator();
+
+		$propagator->beginBatch();
+		$propagator->propagateChange('asd/file.txt', time(), 10);
+		$propagator->propagateChange('foo/bar/file.txt', time(), 2);
+
+		$newInfos = $this->getFileInfos($paths);
+
+		// no changes until we finish the batch
+		foreach ($oldInfos as $i => $oldInfo) {
+			$this->assertEquals($oldInfo->getSize(), $newInfos[$i]->getSize());
+			$this->assertEquals($oldInfo->getEtag(), $newInfos[$i]->getEtag());
+			$this->assertEquals($oldInfo->getMTime(), $newInfos[$i]->getMTime());
+		}
+
+		$propagator->commitBatch();
+
+		$newInfos = $this->getFileInfos($paths);
+
+		foreach ($oldInfos as $i => $oldInfo) {
+			if ($oldInfo->getPath() !== 'foo/baz') {
+				$this->assertNotEquals($oldInfo->getEtag(), $newInfos[$i]->getEtag());
+			}
+		}
+
+		$this->assertEquals($oldInfos['']->getSize() + 12, $newInfos['']->getSize());
+		$this->assertEquals($oldInfos['asd']->getSize() + 10, $newInfos['asd']->getSize());
+		$this->assertEquals($oldInfos['foo']->getSize() + 2, $newInfos['foo']->getSize());
+		$this->assertEquals($oldInfos['foo/bar']->getSize() + 2, $newInfos['foo/bar']->getSize());
+		$this->assertEquals($oldInfos['foo/baz']->getSize(), $newInfos['foo/baz']->getSize());
+	}
+}