<commit_before><?php

namespace TLE\ElfBot\Console\Command;

use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Output\OutputInterface;
use TLE\ElfBot\Utility;

class RunCommand extends AbstractLoggerAwareCommand
{
    protected function configure()
    {
        $this
            ->setName('run')
            ->setDescription('Run the queue worker')
        ;
    }

    protected function execute(InputInterface $input, OutputInterface $output)
    {
        $application = $this->getApplication();

        // tasks

        $tasks = [
            'core.display_dialog' => Utility::createService($application, 'TLE\ElfBot\Tasks\Core\DisplayDialog\Manifest'),
            'core.log_message' => Utility::createService($application, 'TLE\ElfBot\Tasks\Core\LogMessage\Manifest'),
            'core.self_terminate' => Utility::createService($application, 'TLE\ElfBot\Tasks\Core\SelfTerminate\Manifest'),
        ];

        foreach ($application->getConfig()['tasks'] as $name => $task) {
            $tasks[$name] = Utility::createService(
                $application,
                $task['class'],
                isset($task['options']) ? $task['options'] : []
            );
        }

        // signals

        $wantsExit = false;
        $isWorking = false;
        $timeoutAlarm = time() + 20;

        $logger = $this->logger;

        $signals = function ($signal) use (&$wantsExit, &$timeoutAlarm, &$isWorking, $logger) {
            if ($signal == SIGALRM) {
                if ($timeoutAlarm < time()) {
                    $logger->error('missed a SIGALRM');

                    if (!$isWorking) {
                        $logger->info('suicide since no job is being worked');

                        posix_kill(getmypid(), SIGKILL);
                    }
                } else {
                    $timeoutAlarm = time() + 20;

                    pcntl_alarm(16);

                    return;
                }
            }

            $logger->info('received signal to exit');

            if (!$isWorking) {
                $logger->warn('waiting for queue polling to finish');
            } else {
                $logger->warn('waiting for task to finish');
            }

            $wantsExit = true;
        };

        declare(ticks = 1);

        pcntl_signal(SIGINT, $signals);
        pcntl_signal(SIGTERM, $signals);
        pcntl_signal(SIGALRM, $signals);

        pcntl_alarm(16);

        // ready, set, go

        $this->logger->debug('resolving queue url');

        $queueClient = $application->getQueueClient();

        $queueUrl = $queueClient->getQueueUrl([
            'QueueName' => $application->getConfig()['queue']['aws_sqs']['queue_name'],
        ])['QueueUrl'];

        $this->logger->debug($queueUrl);

        $this->logger->info('ready');

        while (!$wantsExit) {
            $msgpack = $queueClient->receiveMessage(
                [
                    'QueueUrl' => $queueUrl,
                    'WaitTimeSeconds' => 20,
                ]
            );

            if (!isset($msgpack['Messages'])) {
                $this->logger->debug('no messages received');
<commit_msg>Add extra safeguard for half-open TCP connections<commit_after><?php

namespace TLE\ElfBot\Console\Command;

use Symfony\Component\Console\Input\InputInterface;
use Symfony\Component\Console\Output\OutputInterface;
use TLE\ElfBot\Utility;

class RunCommand extends AbstractLoggerAwareCommand
{
    protected function configure()
    {
        $this
            ->setName('run')
            ->setDescription('Run the queue worker')
        ;
    }

    protected function execute(InputInterface $input, OutputInterface $output)
    {
        $application = $this->getApplication();

        // tasks

        $tasks = [
            'core.display_dialog' => Utility::createService($application, 'TLE\ElfBot\Tasks\Core\DisplayDialog\Manifest'),
            'core.log_message' => Utility::createService($application, 'TLE\ElfBot\Tasks\Core\LogMessage\Manifest'),
            'core.self_terminate' => Utility::createService($application, 'TLE\ElfBot\Tasks\Core\SelfTerminate\Manifest'),
        ];

        foreach ($application->getConfig()['tasks'] as $name => $task) {
            $tasks[$name] = Utility::createService(
                $application,
                $task['class'],
                isset($task['options']) ? $task['options'] : []
            );
        }

        // signals

        $wantsExit = false;
        $isWorking = false;
        $timeoutQueue = null;
        $timeoutAlarm = time() + 20;

        $logger = $this->logger;

        $signals = function ($signal) use (&$wantsExit, &$timeoutAlarm, &$timeoutQueue, &$isWorking, $logger) {
            if ($signal == SIGALRM) {
                $now = time();

                if ($timeoutAlarm < $now) {
                    $logger->error('missed a SIGALRM');

                    if (!$isWorking) {
                        $logger->info('suicide since no job is being worked');

                        posix_kill(getmypid(), SIGKILL);
                    }
                } elseif ((null !== $timeoutQueue) && ($timeoutQueue < $now)) {
                    if ($wantsExit) {
                        // assume we've already been here
                        $logger->info('suicide since queue polling timed out');

                        posix_kill(getmypid(), SIGKILL);
                    }

                    $logger->error('queue polling is timing out');

                    posix_kill(getmypid(), SIGTERM);
                } else {
                    $timeoutAlarm = $now + 20;

                    pcntl_alarm(16);

                    return;
                }
            }

            $logger->info('received signal to exit');

            if (!$isWorking) {
                $logger->warn('waiting for queue polling to finish');
            } else {
                $logger->warn('waiting for task to finish');
            }

            $wantsExit = true;
        };

        declare(ticks = 1);

        pcntl_signal(SIGINT, $signals);
        pcntl_signal(SIGTERM, $signals);
        pcntl_signal(SIGALRM, $signals);

        pcntl_alarm(16);

        // ready, set, go

        $this->logger->debug('resolving queue url');

        $queueClient = $application->getQueueClient();

        $queueUrl = $queueClient->getQueueUrl([
            'QueueName' => $application->getConfig()['queue']['aws_sqs']['queue_name'],
        ])['QueueUrl'];

        $this->logger->debug($queueUrl);

        $this->logger->info('ready');

        while (!$wantsExit) {
            $timeoutQueue = time() + 60;

            $msgpack = $queueClient->receiveMessage(
                [
                    'QueueUrl' => $queueUrl,
                    'WaitTimeSeconds' => 20,
                ]
            );

            $timeoutQueue = null;

            if (!isset($msgpack['Messages'])) {
                $this->logger->debug('no messages received');

{'fraction_non_alphanumeric': 0.137615988579586, 'fraction_numerical': 0.003426124197002141, 'mean_word_length': 1.7218337218337219, 'pattern_counts': {'":': 0, '<': 8, '<?xml version=': 0, '>': 63, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '968995', 'n_tokens_mistral': 1983, 'n_tokens_neox': 1833, 'n_words': 349}