<commit_before><?php

namespace Maghead\Sharding\QueryMapper\Pthread;

use SQLBuilder\Universal\Query\CreateDatabaseQuery;
use SQLBuilder\Universal\Query\SelectQuery;
use SQLBuilder\ArgumentArray;
use SQLBuilder\Driver\PDOMySQLDriver;
use Maghead\Manager\ConnectionManager;
use Maghead\Sharding\QueryMapper\QueryMapper;

class PthreadQueryMapper implements QueryMapper
{
    protected $connectionManager;

    public function __construct(ConnectionManager $connectionManager)
    {
        $this->connectionManager = $connectionManager;
    }

    public function map(array $shards, string $repoClass, SelectQuery $query)
    {
        $nodeIds = $this->getNodeIds($shards);
        $workers = $this->start($nodeIds);
        $jobs = $this->send($workers, $query);
        $this->wait($workers);
        return $this->mergeJobsResults($jobs);
    }

    protected function start(array $nodeIds)
    {
        $workers = [];
        foreach ($nodeIds as $nodeId) {
            $ds = $this->connectionManager->getDataSource($nodeId);
            $workers[$nodeId] = $w = new PthreadQueryWorker($ds['dsn'], $ds['user'], $ds['pass'], $ds['connection_options']);
            $w->start();
        }
        return $workers;
    }

    protected function send(array $workers, $query)
    {
        $jobs = [];
        foreach ($workers as $nodeId => $worker) {
            // For different connection, we have different query driver to build the sql statement.
            $conn = $this->connectionManager->getConnection($nodeId);
            $args = new ArgumentArray;
            $sql = $query->toSql($conn->getQueryDriver(), $args);
            $jobs[$nodeId] = $job = new PthreadQueryJob($sql, serialize($args->toArray()));
            $worker->stack($job);
        }
        return $jobs;
    }

    protected function wait(array $workers)
    {
        foreach ($workers as $worker) {
            $worker->join();
        }
    }

    protected function getNodeIds(array $shards)
    {
        $nodeIds = [];
        foreach ($shards as $shard) {
            $nodeIds[] = $shard->getReadNode();
        }
        return $nodeIds;
    }

    protected function mergeJobsResults(array $jobs)
    {
        $results = [];
        foreach ($jobs as $nodeId => $job) {
            $results[$nodeId] = $job->getRows();
        }
        return $results;
    }
}<commit_msg>use selectNodeIds instead of getNodeIds<commit_after><?php

namespace Maghead\Sharding\QueryMapper\Pthread;

use SQLBuilder\Universal\Query\CreateDatabaseQuery;
use SQLBuilder\Universal\Query\SelectQuery;
use SQLBuilder\ArgumentArray;
use SQLBuilder\Driver\PDOMySQLDriver;
use Maghead\Manager\ConnectionManager;
use Maghead\Sharding\QueryMapper\QueryMapper;

class PthreadQueryMapper implements QueryMapper
{
    protected $connectionManager;

    public function __construct(ConnectionManager $connectionManager)
    {
        $this->connectionManager = $connectionManager;
    }

    public function map(array $shards, string $repoClass, SelectQuery $query)
    {
        $nodeIds = $this->selectNodes($shards);
        $workers = $this->start($nodeIds);
        $jobs = $this->send($workers, $query);
        $this->wait($workers);
        return $this->mergeJobsResults($jobs);
    }

    protected function start(array $nodeIds)
    {
        $workers = [];
        foreach ($nodeIds as $nodeId) {
            $ds = $this->connectionManager->getDataSource($nodeId);
            $workers[$nodeId] = $w = new PthreadQueryWorker($ds['dsn'], $ds['user'], $ds['pass'], $ds['connection_options']);
            $w->start();
        }
        return $workers;
    }

    protected function send(array $workers, $query)
    {
        $jobs = [];
        foreach ($workers as $nodeId => $worker) {
            // For different connection, we have different query driver to build the sql statement.
            $conn = $this->connectionManager->getConnection($nodeId);
            $args = new ArgumentArray;
            $sql = $query->toSql($conn->getQueryDriver(), $args);
            $jobs[$nodeId] = $job = new PthreadQueryJob($sql, serialize($args->toArray()));
            $worker->stack($job);
        }
        return $jobs;
    }

    protected function wait(array $workers)
    {
        foreach ($workers as $worker) {
            $worker->join();
        }
    }

    protected function selectNodes(array $shards)
    {
        $nodeIds = [];
        foreach ($shards as $shard) {
            $nodeIds[] = $shard->getReadNode();
        }
        return $nodeIds;
    }

    protected function mergeJobsResults(array $jobs)
    {
        $results = [];
        foreach ($jobs as $nodeId => $job) {
            $results[$nodeId] = $job->getRows();
        }
        return $results;
    }
}
{'fraction_non_alphanumeric': 0.1357142857142857, 'fraction_numerical': 0.0, 'mean_word_length': 2.571642910727682, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 43, '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': '1787677', 'n_tokens_mistral': 1356, 'n_tokens_neox': 1325, 'n_words': 288}