Method thread

src/DataStreams/ParallelInputsProcessor.h:182–265

    }

    void thread(ThreadGroupStatusPtr thread_group, size_t thread_num)
    {
        std::exception_ptr exception;
        CurrentMetrics::Increment metric_increment{CurrentMetrics::QueryThread};

        try
        {
            setThreadName("ParalInputsProc");
            if (thread_group)
                CurrentThread::attachTo(thread_group);

            while (!finish)
            {
                InputData unprepared_input;
                {
                    std::lock_guard lock(unprepared_inputs_mutex);

                    if (unprepared_inputs.empty())
                        break;

                    unprepared_input = unprepared_inputs.front();
                    unprepared_inputs.pop();
                }

                unprepared_input.in->readPrefix();

                {
                    std::lock_guard lock(available_inputs_mutex);
                    available_inputs.push(unprepared_input);
                }
            }

            loop(thread_num);

            handler.onFinishThread(thread_num);
        }
        catch (...)
        {
            exception = std::current_exception();
        }

        if (exception)
        {
            handler.onException(exception, thread_num);
        }

        /// The last thread on the output indicates that there is no more data.
        if (0 == --active_threads)
        {
            /// And then it processes an additional source, if there is one.
            if (additional_input_at_end)
            {
                try
                {
                    additional_input_at_end->readPrefix();
                    while (Block block = additional_input_at_end->read())
                        publishPayload(block, thread_num);
                }