Merge pull request #4577 from magento-thunder/MC-18477

Fixed issues:
 - MC-18477: Deadlocks when consumers run from cron

app/code/Magento/MessageQueue/Console/StartConsumerCommand.php

@@ -11,7 +11,7 @@
 use Symfony\Component\Console\Input\InputOption;
 use Symfony\Component\Console\Output\OutputInterface;
 use Magento\Framework\MessageQueue\ConsumerFactory;
-use Magento\MessageQueue\Model\Cron\ConsumersRunner\PidConsumerManager;
+use Magento\Framework\Lock\LockManagerInterface;
 
 /**
  * Command for starting MessageQueue consumers.
@@ -22,6 +22,7 @@ class StartConsumerCommand extends Command
     const OPTION_NUMBER_OF_MESSAGES = 'max-messages';
     const OPTION_BATCH_SIZE = 'batch-size';
     const OPTION_AREACODE = 'area-code';
+    const OPTION_SINGLE_THREAD = 'single-thread';
     const PID_FILE_PATH = 'pid-file-path';
     const COMMAND_QUEUE_CONSUMERS_START = 'queue:consumers:start';
 
@@ -36,9 +37,9 @@ class StartConsumerCommand extends Command
     private $appState;
 
     /**
-     * @var PidConsumerManager
+     * @var LockManagerInterface
      */
-    private $pidConsumerManager;
+    private $lockManager;
 
     /**
      * StartConsumerCommand constructor.
@@ -47,54 +48,60 @@ class StartConsumerCommand extends Command
      * @param \Magento\Framework\App\State $appState
      * @param ConsumerFactory $consumerFactory
      * @param string $name
-     * @param PidConsumerManager $pidConsumerManager
+     * @param LockManagerInterface $lockManager
      */
     public function __construct(
         \Magento\Framework\App\State $appState,
         ConsumerFactory $consumerFactory,
         $name = null,
-        PidConsumerManager $pidConsumerManager = null
+        LockManagerInterface $lockManager = null
     ) {
         $this->appState = $appState;
         $this->consumerFactory = $consumerFactory;
-        $this->pidConsumerManager = $pidConsumerManager ?: \Magento\Framework\App\ObjectManager::getInstance()
-            ->get(PidConsumerManager::class);
+        $this->lockManager = $lockManager ?: \Magento\Framework\App\ObjectManager::getInstance()
+            ->get(LockManagerInterface::class);
         parent::__construct($name);
     }
 
     /**
-     * {@inheritdoc}
+     * @inheritdoc
      */
     protected function execute(InputInterface $input, OutputInterface $output)
     {
         $consumerName = $input->getArgument(self::ARGUMENT_CONSUMER);
         $numberOfMessages = $input->getOption(self::OPTION_NUMBER_OF_MESSAGES);
         $batchSize = (int)$input->getOption(self::OPTION_BATCH_SIZE);
         $areaCode = $input->getOption(self::OPTION_AREACODE);
-        $pidFilePath = $input->getOption(self::PID_FILE_PATH);
 
-        if ($pidFilePath && $this->pidConsumerManager->isRun($pidFilePath)) {
-            $output->writeln('<error>Consumer with the same PID is running</error>');
-            return \Magento\Framework\Console\Cli::RETURN_FAILURE;
+        if ($input->getOption(self::PID_FILE_PATH)) {
+            $input->setOption(self::OPTION_SINGLE_THREAD, true);
         }
 
-        if ($pidFilePath) {
-            $this->pidConsumerManager->savePid($pidFilePath);
+        $singleThread = $input->getOption(self::OPTION_SINGLE_THREAD);
+
+        if ($singleThread && $this->lockManager->isLocked(md5($consumerName))) { //phpcs:ignore
+            $output->writeln('<error>Consumer with the same name is running</error>');
+            return \Magento\Framework\Console\Cli::RETURN_FAILURE;
         }
 
-        if ($areaCode !== null) {
-            $this->appState->setAreaCode($areaCode);
-        } else {
-            $this->appState->setAreaCode('global');
+        if ($singleThread) {
+            $this->lockManager->lock(md5($consumerName)); //phpcs:ignore
         }
 
+        $this->appState->setAreaCode($areaCode ?? 'global');
+
         $consumer = $this->consumerFactory->get($consumerName, $batchSize);
         $consumer->process($numberOfMessages);
+
+        if ($singleThread) {
+            $this->lockManager->unlock(md5($consumerName)); //phpcs:ignore
+        }
+
         return \Magento\Framework\Console\Cli::RETURN_SUCCESS;
     }
 
     /**
-     * {@inheritdoc}
+     * @inheritdoc
      */
     protected function configure()
     {
@@ -125,11 +132,17 @@ protected function configure()
             'The preferred area (global, adminhtml, etc...) '
             . 'default is global.'
         );
+        $this->addOption(
+            self::OPTION_SINGLE_THREAD,
+            null,
+            InputOption::VALUE_NONE,
+            'This option prevents running multiple copies of one consumer simultaneously.'
+        );
         $this->addOption(
             self::PID_FILE_PATH,
             null,
             InputOption::VALUE_REQUIRED,
-            'The file path for saving PID'
+            'The file path for saving PID (This option is deprecated, use --single-thread instead)'
         );
         $this->setHelp(
             <<<HELP
@@ -150,8 +163,12 @@ protected function configure()
 To specify the preferred area:
 
     <comment>%command.full_name% someConsumer --area-code='adminhtml'</comment>
+    
+To do not run multiple copies of one consumer simultaneously:
+
+    <comment>%command.full_name% someConsumer --single-thread'</comment>
 
-To save PID enter path:
+To save PID enter path (This option is deprecated, use --single-thread instead):
 
     <comment>%command.full_name% someConsumer --pid-file-path='/var/someConsumer.pid'</comment>
 HELP

app/code/Magento/MessageQueue/Model/Cron/ConsumersRunner.php

@@ -13,18 +13,13 @@
 use Magento\Framework\App\DeploymentConfig;
 use Psr\Log\LoggerInterface;
 use Symfony\Component\Process\PhpExecutableFinder;
-use Magento\MessageQueue\Model\Cron\ConsumersRunner\PidConsumerManager;
+use Magento\Framework\Lock\LockManagerInterface;
 
 /**
  * Class for running consumers processes by cron
  */
 class ConsumersRunner
 {
-    /**
-     * Extension of PID file
-     */
-    const PID_FILE_EXT = '.pid';
-
     /**
      * Shell command line wrapper for executing command in background
      *
@@ -53,13 +48,6 @@ class ConsumersRunner
      */
     private $phpExecutableFinder;
 
-    /**
-     * The class for checking status of process by PID
-     *
-     * @var PidConsumerManager
-     */
-    private $pidConsumerManager;
-
     /**
      * @var ConnectionTypeResolver
      */
@@ -70,13 +58,20 @@ class ConsumersRunner
      */
     private $logger;
 
+    /**
+     * Lock Manager
+     *
+     * @var LockManagerInterface
+     */
+    private $lockManager;
+
     /**
      * @param PhpExecutableFinder $phpExecutableFinder The executable finder specifically designed
      *        for the PHP executable
      * @param ConsumerConfigInterface $consumerConfig The consumer config provider
      * @param DeploymentConfig $deploymentConfig The application deployment configuration
      * @param ShellInterface $shellBackground The shell command line wrapper for executing command in background
-     * @param PidConsumerManager $pidConsumerManager The class for checking status of process by PID
+     * @param LockManagerInterface $lockManager The lock manager
      * @param ConnectionTypeResolver $mqConnectionTypeResolver Consumer connection resolver
      * @param LoggerInterface $logger Logger
      */
@@ -85,15 +80,15 @@ public function __construct(
         ConsumerConfigInterface $consumerConfig,
         DeploymentConfig $deploymentConfig,
         ShellInterface $shellBackground,
-        PidConsumerManager $pidConsumerManager,
+        LockManagerInterface $lockManager,
         ConnectionTypeResolver $mqConnectionTypeResolver = null,
         LoggerInterface $logger = null
     ) {
         $this->phpExecutableFinder = $phpExecutableFinder;
         $this->consumerConfig = $consumerConfig;
         $this->deploymentConfig = $deploymentConfig;
         $this->shellBackground = $shellBackground;
-        $this->pidConsumerManager = $pidConsumerManager;
+        $this->lockManager = $lockManager;
         $this->mqConnectionTypeResolver = $mqConnectionTypeResolver
             ?: ObjectManager::getInstance()->get(ConnectionTypeResolver::class);
         $this->logger = $logger
@@ -120,11 +115,9 @@ public function run()
                 continue;
             }
 
-            $consumerName = $consumer->getName();
-
             $arguments = [
-                $consumerName,
-                '--pid-file-path=' . $this->getPidFilePath($consumerName),
+                $consumer->getName(),
+                '--single-thread'
             ];
 
             if ($maxMessages) {
@@ -154,36 +147,25 @@ private function canBeRun(ConsumerConfigItemInterface $consumerConfig, array $al
             return false;
         }
 
-        if ($this->pidConsumerManager->isRun($this->getPidFilePath($consumerName))) {
+        if ($this->lockManager->isLocked(md5($consumerName))) { //phpcs:ignore
             return false;
         }
 
         $connectionName = $consumerConfig->getConnection();
         try {
             $this->mqConnectionTypeResolver->getConnectionType($connectionName);
         } catch (\LogicException $e) {
-            $this->logger->info(sprintf(
-                'Consumer "%s" skipped as required connection "%s" is not configured. %s',
-                $consumerName,
-                $connectionName,
-                $e->getMessage()
-            ));
+            $this->logger->info(
+                sprintf(
+                    'Consumer "%s" skipped as required connection "%s" is not configured. %s',
+                    $consumerName,
+                    $connectionName,
+                    $e->getMessage()
+                )
+            );
             return false;
         }
 
         return true;
     }
-
-    /**
-     * Returns default path to file with PID by consumers name
-     *
-     * @param string $consumerName The consumers name
-     * @return string The path to file with PID
-     */
-    private function getPidFilePath($consumerName)
-    {
-        $sanitizedHostname = preg_replace('/[^a-z0-9]/i', '', gethostname());
-
-        return $consumerName . '-' . $sanitizedHostname . static::PID_FILE_EXT;
-    }
 }