Prevent running again already running cron group

app/code/Magento/Cron/Observer/ProcessCronQueueObserver.php

@@ -56,6 +56,11 @@ class ProcessCronQueueObserver implements ObserverInterface
      */
     const SECONDS_IN_MINUTE = 60;
 
+    /**
+     * How long to wait for cron group to become unlocked
+     */
+    const LOCK_TIMEOUT = 5;
+
     /**
      * @var \Magento\Cron\Model\ResourceModel\Schedule\Collection
      */
@@ -116,6 +121,11 @@ class ProcessCronQueueObserver implements ObserverInterface
      */
     private $state;
 
+    /**
+     * @var \Magento\Framework\Lock\LockManagerInterface
+     */
+    private $lockManager;
+
     /**
      * @var array
      */
@@ -138,6 +148,7 @@ class ProcessCronQueueObserver implements ObserverInterface
      * @param \Magento\Framework\Process\PhpExecutableFinderFactory $phpExecutableFinderFactory
      * @param \Psr\Log\LoggerInterface $logger
      * @param \Magento\Framework\App\State $state
+     * @param \Magento\Framework\Lock\LockManagerInterface $lockManager
      * @SuppressWarnings(PHPMD.ExcessiveParameterList)
      */
     public function __construct(
@@ -151,7 +162,8 @@ public function __construct(
         \Magento\Framework\Stdlib\DateTime\DateTime $dateTime,
         \Magento\Framework\Process\PhpExecutableFinderFactory $phpExecutableFinderFactory,
         \Psr\Log\LoggerInterface $logger,
-        \Magento\Framework\App\State $state
+        \Magento\Framework\App\State $state,
+        \Magento\Framework\Lock\LockManagerInterface $lockManager
     ) {
         $this->_objectManager = $objectManager;
         $this->_scheduleFactory = $scheduleFactory;
@@ -164,6 +176,7 @@ public function __construct(
         $this->phpExecutableFinder = $phpExecutableFinderFactory->create();
         $this->logger = $logger;
         $this->state = $state;
+        $this->lockManager = $lockManager;
     }
 
     /**
@@ -186,8 +199,6 @@ public function execute(\Magento\Framework\Event\Observer $observer)
         $phpPath = $this->phpExecutableFinder->find() ?: 'php';
 
         foreach ($jobGroupsRoot as $groupId => $jobsRoot) {
-            $this->_cleanup($groupId);
-            $this->_generate($groupId);
             if ($this->_request->getParam('group') !== null
                 && $this->_request->getParam('group') !== '\'' . ($groupId) . '\''
                 && $this->_request->getParam('group') !== $groupId
@@ -211,6 +222,19 @@ public function execute(\Magento\Framework\Event\Observer $observer)
                 continue;
             }
 
+            if (!$this->lockManager->setLock($groupId, self::LOCK_TIMEOUT)) {
+                $this->logger->warning(
+                    sprintf(
+                        "Could not acquire lock for cron group: %s, skipping run",
+                        $groupId
+                    )
+                );
+                continue;
+            }
+
+            $this->_cleanup($groupId);
+            $this->_generate($groupId);
+
             /** @var \Magento\Cron\Model\Schedule $schedule */
             foreach ($pendingJobs as $schedule) {
                 $jobConfig = isset($jobsRoot[$schedule->getJobCode()]) ? $jobsRoot[$schedule->getJobCode()] : null;
@@ -247,6 +271,8 @@ public function execute(\Magento\Framework\Event\Observer $observer)
                 }
                 $schedule->save();
             }
+
+            $this->lockManager->releaseLock($groupId);
         }
     }
 

app/etc/di.xml

@@ -37,6 +37,7 @@
     <preference for="Magento\Framework\Locale\ListsInterface" type="Magento\Framework\Locale\TranslatedLists" />
     <preference for="Magento\Framework\Locale\AvailableLocalesInterface" type="Magento\Framework\Locale\Deployed\Codes" />
     <preference for="Magento\Framework\Locale\OptionInterface" type="Magento\Framework\Locale\Deployed\Options" />
+    <preference for="Magento\Framework\Lock\LockManagerInterface" type="Magento\Framework\Lock\Backend\Database" />
     <preference for="Magento\Framework\Api\AttributeTypeResolverInterface" type="Magento\Framework\Reflection\AttributeTypeResolver" />
     <preference for="Magento\Framework\Api\Search\SearchResultInterface" type="Magento\Framework\Api\Search\SearchResult" />
     <preference for="Magento\Framework\Api\Search\SearchCriteriaInterface" type="Magento\Framework\Api\Search\SearchCriteria"/>

lib/internal/Magento/Framework/Lock/Backend/Database.php

@@ -0,0 +1,54 @@
+<?php
+/**
+ * Copyright © Magento, Inc. All rights reserved.
+ * See COPYING.txt for license details.
+ */
+
+namespace Magento\Framework\lock\Backend;
+
+class Database implements \Magento\Framework\Lock\LockManagerInterface
+{
+    /** @var \Magento\Framework\App\ResourceConnection */
+    private $resource;
+
+    public function __construct(
+        \Magento\Framework\App\ResourceConnection $resource
+    )
+    {
+        $this->resource = $resource;
+    }
+
+    /**
+     * Sets a lock for name
+     *
+     * @param string $name lock name
+     * @return bool
+     */
+    public function setLock($name, $timeout = -1)
+    {
+        return (bool)$this->resource->getConnection()->query("SELECT GET_LOCK(?, ?);", array((string)$name, (int)$timeout))
+            ->fetchColumn();
+    }
+
+    /**
+     * Releases a lock for name
+     *
+     * @param string $name lock name
+     * @return bool
+     */
+    public function releaseLock($name)
+    {
+        return (bool)$this->resource->getConnection()->query("SELECT RELEASE_LOCK(?);", array((string)$name))->fetchColumn();
+    }
+
+    /**
+     * Tests of lock is set for name
+     *
+     * @param string $name lock name
+     * @return bool
+     */
+    public function isLocked($name)
+    {
+        return (bool)$this->resource->getConnection()->query("SELECT IS_USED_LOCK(?);", array((string)$name))->fetchColumn();
+    }
+}

lib/internal/Magento/Framework/Lock/LockManagerInterface.php

@@ -0,0 +1,41 @@
+<?php
+/**
+ * Copyright © Magento, Inc. All rights reserved.
+ * See COPYING.txt for license details.
+ */
+namespace Magento\Framework\Lock;
+
+/**
+ * Interface of a lock manager
+ *
+ * @api
+ */
+interface LockManagerInterface
+{
+    /**
+     * Sets a lock for name
+     *
+     * @param string $name lock name
+     * @param int $timeout Timeout in seconds, negative value means infinite timeout
+     * @return bool
+     */
+    public function setLock($name, $timeout = -1);
+
+    /**
+     * Releases a lock for name
+     *
+     * @param string $name lock name
+     * @return bool
+     * @api
+     */
+    public function releaseLock($name);
+
+    /**
+     * Tests of lock is set for name
+     *
+     * @param string $name lock name
+     * @return bool
+     * @api
+     */
+    public function isLocked($name);
+}