Skip to content
Permalink
Browse files

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

Fixed issues:
 - MC-18477: Deadlocks when consumers run from cron
  • Loading branch information...
arhiopterecs committed Aug 3, 2019
2 parents e50fe6f + a842b18 commit 1d9e07b218c7c8ad1f05706828cb2dd47d2d2d58
@@ -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
@@ -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;
}
}

0 comments on commit 1d9e07b

Please sign in to comment.
You can’t perform that action at this time.