<commit_before><?php

namespace Amp\Parallel\Context\Internal;

use Amp\Deferred;
use Amp\Loop;
use Amp\Parallel\Context\ContextException;
use Amp\Parallel\Sync\ChannelledSocket;
use Amp\Promise;
use Amp\TimeoutException;
use function Amp\call;

class ProcessHub
{
    const PROCESS_START_TIMEOUT = 5000;
    const KEY_RECEIVE_TIMEOUT = 1000;

    /** @var resource|null */
    private $server;

    /** @var string|null */
    private $uri;

    /** @var int[] */
    private $keys;

    /** @var string|null */
    private $watcher;

    /** @var Deferred[] */
    private $acceptor = [];

    public function __construct()
    {
        $isWindows = \strncasecmp(\PHP_OS, "WIN", 3) === 0;

        if ($isWindows) {
            $this->uri = "tcp://localhost:0";
        } else {
            $this->uri = "unix://" . \tempnam(\sys_get_temp_dir(), "amp-parallel-ipc-") . ".sock";
        }

        $this->server = \stream_socket_server($this->uri, $errno, $errstr, \STREAM_SERVER_BIND | \STREAM_SERVER_LISTEN);

        if (!$this->server) {
            throw new \RuntimeException(\sprintf("Could not create IPC server: (Errno: %d) %s", $errno, $errstr));
        }

        if ($isWindows) {
            $name = \stream_socket_get_name($this->server, false);
            $port = \substr($name, \strrpos($name, ":") + 1);
            $this->uri = "tcp://localhost:" . $port;
        }

        $keys = &$this->keys;
        $acceptor = &$this->acceptor;
        $this->watcher = Loop::onReadable($this->server, static function (string $watcher, $server) use (&$keys, &$acceptor) {
            // Error reporting suppressed since stream_socket_accept() emits E_WARNING on client accept failure.
            if (!$client = @\stream_socket_accept($server, 0)) {  // Timeout of 0 to be non-blocking.
                return; // Accepting client failed.
            }

            $channel = new ChannelledSocket($client, $client);

            try {
                $received = yield Promise\timeout($channel->receive(), self::KEY_RECEIVE_TIMEOUT);
            } catch (TimeoutException $exception) {
                return; // Ignore possible foreign connection attempt.
            }<commit_msg>Use 127.0.0.1 instead of localhost<commit_after><?php

namespace Amp\Parallel\Context\Internal;

use Amp\Deferred;
use Amp\Loop;
use Amp\Parallel\Context\ContextException;
use Amp\Parallel\Sync\ChannelledSocket;
use Amp\Promise;
use Amp\TimeoutException;
use function Amp\call;

class ProcessHub
{
    const PROCESS_START_TIMEOUT = 5000;
    const KEY_RECEIVE_TIMEOUT = 1000;

    /** @var resource|null */
    private $server;

    /** @var string|null */
    private $uri;

    /** @var int[] */
    private $keys;

    /** @var string|null */
    private $watcher;

    /** @var Deferred[] */
    private $acceptor = [];

    public function __construct()
    {
        $isWindows = \strncasecmp(\PHP_OS, "WIN", 3) === 0;

        if ($isWindows) {
            $this->uri = "tcp://127.0.0.1:0";
        } else {
            $this->uri = "unix://" . \tempnam(\sys_get_temp_dir(), "amp-parallel-ipc-") . ".sock";
        }

        $this->server = \stream_socket_server($this->uri, $errno, $errstr, \STREAM_SERVER_BIND | \STREAM_SERVER_LISTEN);

        if (!$this->server) {
            throw new \RuntimeException(\sprintf("Could not create IPC server: (Errno: %d) %s", $errno, $errstr));
        }

        if ($isWindows) {
            $name = \stream_socket_get_name($this->server, false);
            $port = \substr($name, \strrpos($name, ":") + 1);
            $this->uri = "tcp://127.0.0.1:" . $port;
        }

        $keys = &$this->keys;
        $acceptor = &$this->acceptor;
        $this->watcher = Loop::onReadable($this->server, static function (string $watcher, $server) use (&$keys, &$acceptor) {
            // Error reporting suppressed since stream_socket_accept() emits E_WARNING on client accept failure.
            if (!$client = @\stream_socket_accept($server, 0)) {  // Timeout of 0 to be non-blocking.
                return; // Accepting client failed.
            }

            $channel = new ChannelledSocket($client, $client);

            try {
                $received = yield Promise\timeout($channel->receive(), self::KEY_RECEIVE_TIMEOUT);
            } catch (TimeoutException $exception) {
                return; // Ignore possible foreign connection attempt.
            }
{'fraction_non_alphanumeric': 0.1525731126728633, 'fraction_numerical': 0.010428474268873271, 'mean_word_length': 2.3273001508295628, 'pattern_counts': {'":': 2, '<': 5, '<?xml version=': 0, '>': 27, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 3, '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': '2119272', 'n_tokens_mistral': 1407, 'n_tokens_neox': 1341, 'n_words': 310}