Multi concurrent processes update

lib/Command/Generate.php

@@ -36,6 +36,7 @@
 use OCP\Files\NotFoundException;
 use OCP\Files\StorageNotAvailableException;
 use OCP\IConfig;
+use OCP\IDBConnection;
 use OCP\IPreview;
 use OCP\IUser;
 use OCP\IUserManager;
@@ -48,19 +49,38 @@
 use Symfony\Component\Console\Output\OutputInterface;
 
 class Generate extends Command {
+
+	/** @var ?GlobalStoragesService */
 	protected ?GlobalStoragesService $globalService;
+
+	/** @var array */
 	protected array $specifications;
+	/** @var IUserManager */
 	protected IUserManager $userManager;
+
+	/** @var IRootFolder */
 	protected IRootFolder $rootFolder;
+
+	/** @var IPreview */
 	protected IPreview $previewGenerator;
+
+	/** @var IConfig */
 	protected IConfig $config;
+
+	/** @var IDBConnection */
+	protected $connection;
+
+	/** @var OutputInterface */
 	protected OutputInterface $output;
+
+	/** @var IManager */
 	protected IManager $encryptionManager;
 
 	public function __construct(IRootFolder $rootFolder,
 		IUserManager $userManager,
 		IPreview $previewGenerator,
 		IConfig $config,
+		IDBConnection $connection,
 		IManager $encryptionManager,
 		ContainerInterface $container) {
 		parent::__construct();
@@ -69,6 +89,7 @@ public function __construct(IRootFolder $rootFolder,
 		$this->rootFolder = $rootFolder;
 		$this->previewGenerator = $previewGenerator;
 		$this->config = $config;
+		$this->connection = $connection;
 		$this->encryptionManager = $encryptionManager;
 
 		try {
@@ -178,7 +199,7 @@ private function generatePathPreviews(IUser $user, string $path): void {
 		}
 		$pathFolder = $userFolder->get($relativePath);
 		$noPreviewMountPaths = $this->getNoPreviewMountPaths($user);
-		$this->parseFolder($pathFolder, $noPreviewMountPaths);
+		$this->parseFolder($pathFolder, $noPreviewMountPaths, $user);
 	}
 
 	private function generateUserPreviews(IUser $user): void {
@@ -187,10 +208,10 @@ private function generateUserPreviews(IUser $user): void {
 
 		$userFolder = $this->rootFolder->getUserFolder($user->getUID());
 		$noPreviewMountPaths = $this->getNoPreviewMountPaths($user);
-		$this->parseFolder($userFolder, $noPreviewMountPaths);
+		$this->parseFolder($userFolder, $noPreviewMountPaths, $user);
 	}
 
-	private function parseFolder(Folder $folder, array $noPreviewMountPaths): void {
+	private function parseFolder(Folder $folder, array $noPreviewMountPaths, IUser $user): void {
 		try {
 			$folderPath = $folder->getPath();
 
@@ -208,8 +229,44 @@ private function parseFolder(Folder $folder, array $noPreviewMountPaths): void {
 			foreach ($nodes as $node) {
 				if ($node instanceof Folder) {
 					$this->parseFolder($node, $noPreviewMountPaths);
-				} elseif ($node instanceof File) {
-					$this->parseFile($node);
+				} else if ($node instanceof File) {
+					$is_locked = false;
+					$qb = $this->connection->getQueryBuilder();
+					$row = $qb->select('*')
+						->from('preview_generation')
+						->where($qb->expr()->eq('file_id', $qb->createNamedParameter($node->getId())))
+						->setMaxResults(1)
+						->execute()
+						->fetch();
+					if ($row !== false) {
+						if ($row['locked'] == 1) {
+							// already being processed
+							$is_locked = true;
+						} else {
+							$qb->update('preview_generation')
+								->where($qb->expr()->eq('file_id', $qb->createNamedParameter($node->getId())))
+								->set('locked', $qb->createNamedParameter(true))
+								->execute();
+						}
+					} else {
+						$qb->insert('preview_generation')
+							->values([
+								'uid'     => $qb->createNamedParameter($user->getUID()),
+								'file_id' => $qb->createNamedParameter($node->getId()),
+								'locked'  => $qb->createNamedParameter(true),
+							])
+							->execute();
+					}
+
+					if ($is_locked === false) {
+						try {
+							$this->parseFile($node);
+						} finally {
+							$qb->delete('preview_generation')
+								->where($qb->expr()->eq('file_id', $qb->createNamedParameter($node->getId())))
+								->execute();
+						}
+					}
 				}
 			}
 		} catch (StorageNotAvailableException $e) {

lib/Command/PreGenerate.php

@@ -96,13 +96,15 @@ protected function execute(InputInterface $input, OutputInterface $output): int
 			$output->writeln('Encryption is enabled. Aborted.');
 			return 1;
 		}
-
+		/*
+		this locks it to only be run once
 		if ($this->checkAlreadyRunning()) {
 			$output->writeln('Command is already running.');
 			return 2;
 		}
 
 		$this->setPID();
+		*/
 
 		// Set timestamp output
 		$formatter = new TimestampFormatter($this->config, $output->getFormatter());
@@ -112,37 +114,41 @@ protected function execute(InputInterface $input, OutputInterface $output): int
 		$this->sizes = SizeHelper::calculateSizes($this->config);
 		$this->startProcessing();
 
+		/*
 		$this->clearPID();
+		*/
 
 		return 0;
 	}
 
 	private function startProcessing(): void {
+		// random sleep between 0 and 50ms to avoid collision between 2 processes
+		usleep(rand(0,50000));
+
 		while (true) {
 			$qb = $this->connection->getQueryBuilder();
-			$qb->select('*')
+			$row = $qb->select('*')
 				->from('preview_generation')
 				->orderBy('id')
-				->setMaxResults(1000);
-			$cursor = $qb->execute();
-			$rows = $cursor->fetchAll();
-			$cursor->closeCursor();
+				->where($qb->expr()->eq('locked', $qb->createNamedParameter(false)))
+				->setMaxResults(1)
+				->execute()
+				->fetch();
 
-			if ($rows === []) {
+			if ($row === false) {
 				break;
 			}
 
-			foreach ($rows as $row) {
-				/*
-				 * First delete the row so that if preview generation fails for some reason
-				 * the next run can just continue
-				 */
-				$qb = $this->connection->getQueryBuilder();
-				$qb->delete('preview_generation')
-					->where($qb->expr()->eq('id', $qb->createNamedParameter($row['id'])));
-				$qb->execute();
-
+			$qb->update('preview_generation')
+			   ->where($qb->expr()->eq('id', $qb->createNamedParameter($row['id'])))
+			   ->set('locked', $qb->createNamedParameter(true))
+			   ->execute();
+			try {
 				$this->processRow($row);
+			} finally {
+				$qb->delete('preview_generation')
+					->where($qb->expr()->eq('id', $qb->createNamedParameter($row['id'])))
+				    ->execute();
 			}
 		}
 	}
@@ -204,7 +210,10 @@ private function processFile(File $file): void {
 				);
 				$this->previewGenerator->generatePreviews($file, $specifications);
 			} catch (NotFoundException $e) {
-				// Maybe log that previews could not be generated?
+                            if ($this->output->getVerbosity() > OutputInterface::VERBOSITY_VERBOSE) {
+				$error = $e->getMessage();
+				$this->output->writeln("<error>${error} " . $file->getPath() . " not found.</error>");
+                            }
 			} catch (\InvalidArgumentException $e) {
 				$error = $e->getMessage();
 				$this->output->writeln("<error>{$error}</error>");

lib/Migration/Version020200Date20190608205303.php

@@ -0,0 +1,53 @@
+<?php
+declare(strict_types=1);
+/**
+ * @copyright Copyleft (c) 2019, Ignacio Nunez <[email protected]>
+ *
+ * @author Ignacio Nunez <[email protected]>
+ *
+ * @license GNU AGPL version 3 or any later version
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ */
+
+namespace OCA\PreviewGenerator\Migration;
+
+use OCP\DB\ISchemaWrapper;
+use OCP\Migration\SimpleMigrationStep;
+use OCP\Migration\IOutput;
+use Doctrine\DBAL\Types\Type;
+
+class Version020200Date20190608205303 extends SimpleMigrationStep {
+
+	/**
+	 * @param IOutput $output
+	 * @param \Closure $schemaClosure The `\Closure` returns a `ISchemaWrapper`
+	 * @param array $options
+	 * @return null|ISchemaWrapper
+	 */
+	public function changeSchema(IOutput $output, \Closure $schemaClosure, array $options) {
+		/** @var ISchemaWrapper $schema */
+		$schema = $schemaClosure();
+		$table = $schema->getTable('preview_generation');
+
+		if (!$table->hasColumn('locked')) {
+			$table->addColumn('locked', Type::BOOLEAN, [
+				'notnull' => true,
+				'default' => 0,
+			]);
+		}
+		return $schema;
+	}
+}