<commit_before># --------------------------------------------------------------------------
# Copyright 2002-2020, OpenNebula Project, OpenNebula Systems
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# --------------------------------------------------------------------------

require 'pp'
require 'open3'
require 'timeout'
require 'base64'

# Generic command executor that holds the code shared by all the command
# executors.
#
# Properties:
#
# * +code+: integer holding the exit code. Read-only
# * +stdout+: string of the standard output. Read-only
# * +stderr+: string of the standard error. Read-only
# * +command+: command to execute. Read-only
#
# The protocol for scripts to log is as follows:
#
# * Log messages will be sent to STDOUT
# * The script will return 0 if it succeded or any other value
#   if there was a failure
# * In case of failure the cause of the error will be written to STDERR
#   wrapped by start and end marks as follows:
#
#     ERROR MESSAGE --8<------
#     error message for the failure
#     ERROR MESSAGE ------>8--


class GenericCommand
    ERROR_OPEN  = "ERROR MESSAGE --8<------"
    ERROR_CLOSE = "ERROR MESSAGE ------>8--"

    attr_reader :code, :stdout, :stderr, :command

    # Creates a command and runs it
    def self.run(command, logger=nil, stdin=nil, timeout=nil)
        cmd = self.new(command, logger, stdin, timeout)
        cmd.run
        cmd
    end

    # Creates the new command:
    # +command+: string with the command to be executed
    # +logger+: proc that takes a message parameter and logs it
    def initialize(command, logger=nil, stdin=nil, timeout=nil)
        @command = command
        @logger  = logger
        @stdin   = stdin
        @timeout = timeout
    end

    # Sends a log message to the logger proc
    def log(message, all=true)
        @logger.call(message, all) if @logger
    end

    # Runs the command
    def run
        begin
            @stdout, @stderr, status = execute

            if status && status.exited?
                @code = status.exitstatus
            else
                @code = 255
            end

            if @code != 0
                log("Command execution failed (exit code: #{@code}): #{command}")
                log(@stderr)
            end
        rescue Exception => e
            if e.is_a?(Timeout::Error)
                error_message = "Timeout executing #{command}"
            else
                error_message = "Internal error #{e}"
            end

            log(error_message)
            @stderr = ERROR_OPEN + "\n" + error_message + "\n" + ERROR_CLOSE
            @code = 255
        end

        return @code
    end

    # Parses error message from +stderr+ output
    def get_error_message
        tmp=@stderr.scan(/^#{ERROR_OPEN}\n(.*?)#{ERROR_CLOSE}$/m)
        return "-" if !tmp[0]
        tmp[0].join(' ').strip
    end

    def to_xml
        stdout = @stdout.nil? ? '' : @stdout
        stderr = @stderr.nil? ? '' : @stderr

        '<EXECUTION_RESULT>' \
            "<COMMAND>#{@command}</COMMAND>" \
            "<STDOUT>#{Base64.encode64(stdout)}</STDOUT>" \
            "<STDERR>#{Base64.encode64(stderr)}</STDERR>" \
            "<CODE>#{@code}</CODE>" \
        '</EXECUTION_RESULT>'
    end

private

    # Low level command execution. This method has to be redefined
    # for each kind of command execution. Returns an array with
    # +stdout+, +stderr+ and +status+ of the command execution.
    def execute
        puts "About to execute \"#{@command}\""
        ['', '', nil]
    end

    # modified Open3.capture with terminator thread
    # to deal with timeouts
    def capture3_timeout(*cmd)
        if Hash === cmd.last
            opts = cmd.pop.dup
        else
            opts = {}
        end

        stdin_data = opts.delete(:stdin_data) || ''
        binmode = opts.delete(:binmode)

        Open3.popen3(*cmd, opts) {|i, o, e, t|
            if binmode
                i.binmode
                o.binmode
                e.binmode
            end

            terminator_e = nil
            mutex = Mutex.new

            out_reader = Thread.new { o.read }
            err_reader = Thread.new { e.read }
            terminator = Thread.new {
                if @timeout and @timeout>0
                    begin
                        pid = Process.getpgid(t.pid) * -1
                    rescue
                        pid = t.pid
                    end

                    if pid
                        begin
                            sleep @timeout

                            mutex.synchronize do
                                terminator_e = Timeout::Error
                            end
                        ensure
                        end

                        Process.kill('TERM', pid)
                    end
                end
            }

            i.write stdin_data
            i.close

            out = [out_reader.value, err_reader.value, t.value]

            mutex.lock
            terminator.kill
            raise terminator_e if terminator_e

            out
        }
    end

end

# Executes commands in the machine where it is called. See documentation
# for GenericCommand
class LocalCommand < GenericCommand
private
<commit_msg>B #4867: CommandManager terminate reading threads (#4886)<commit_after># --------------------------------------------------------------------------
# Copyright 2002-2020, OpenNebula Project, OpenNebula Systems
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# --------------------------------------------------------------------------

require 'pp'
require 'open3'
require 'timeout'
require 'base64'

# Generic command executor that holds the code shared by all the command
# executors.
#
# Properties:
#
# * +code+: integer holding the exit code. Read-only
# * +stdout+: string of the standard output. Read-only
# * +stderr+: string of the standard error. Read-only
# * +command+: command to execute. Read-only
#
# The protocol for scripts to log is as follows:
#
# * Log messages will be sent to STDOUT
# * The script will return 0 if it succeded or any other value
#   if there was a failure
# * In case of failure the cause of the error will be written to STDERR
#   wrapped by start and end marks as follows:
#
#     ERROR MESSAGE --8<------
#     error message for the failure
#     ERROR MESSAGE ------>8--


class GenericCommand
    ERROR_OPEN  = "ERROR MESSAGE --8<------"
    ERROR_CLOSE = "ERROR MESSAGE ------>8--"

    attr_reader :code, :stdout, :stderr, :command

    # Creates a command and runs it
    def self.run(command, logger=nil, stdin=nil, timeout=nil)
        cmd = self.new(command, logger, stdin, timeout)
        cmd.run
        cmd
    end

    # Creates the new command:
    # +command+: string with the command to be executed
    # +logger+: proc that takes a message parameter and logs it
    def initialize(command, logger=nil, stdin=nil, timeout=nil)
        @command = command
        @logger  = logger
        @stdin   = stdin
        @timeout = timeout
    end

    # Sends a log message to the logger proc
    def log(message, all=true)
        @logger.call(message, all) if @logger
    end

    # Runs the command
    def run
        begin
            @stdout, @stderr, status = execute

            if status && status.exited?
                @code = status.exitstatus
            else
                @code = 255
            end

            if @code != 0
                log("Command execution failed (exit code: #{@code}): #{command}")
                log(@stderr)
            end
        rescue Exception => e
            if e.is_a?(Timeout::Error)
                error_message = "Timeout executing #{command}"
            else
                error_message = "Internal error #{e}"
            end

            log(error_message)
            @stderr = ERROR_OPEN + "\n" + error_message + "\n" + ERROR_CLOSE
            @code = 255
        end

        return @code
    end

    # Parses error message from +stderr+ output
    def get_error_message
        tmp=@stderr.scan(/^#{ERROR_OPEN}\n(.*?)#{ERROR_CLOSE}$/m)
        return "-" if !tmp[0]
        tmp[0].join(' ').strip
    end

    def to_xml
        stdout = @stdout.nil? ? '' : @stdout
        stderr = @stderr.nil? ? '' : @stderr

        '<EXECUTION_RESULT>' \
            "<COMMAND>#{@command}</COMMAND>" \
            "<STDOUT>#{Base64.encode64(stdout)}</STDOUT>" \
            "<STDERR>#{Base64.encode64(stderr)}</STDERR>" \
            "<CODE>#{@code}</CODE>" \
        '</EXECUTION_RESULT>'
    end

private

    # Low level command execution. This method has to be redefined
    # for each kind of command execution. Returns an array with
    # +stdout+, +stderr+ and +status+ of the command execution.
    def execute
        puts "About to execute \"#{@command}\""
        ['', '', nil]
    end

    # modified Open3.capture with terminator thread
    # to deal with timeouts
    def capture3_timeout(*cmd)
        if Hash === cmd.last
            opts = cmd.pop.dup
        else
            opts = {}
        end

        stdin_data = opts.delete(:stdin_data) || ''
        binmode = opts.delete(:binmode)

        Open3.popen3(*cmd, opts) {|i, o, e, t|
            if binmode
                i.binmode
                o.binmode
                e.binmode
            end

            terminator_e = nil
            mutex = Mutex.new

            out_reader = Thread.new { o.read }
            err_reader = Thread.new { e.read }
            terminator = Thread.new {
                if @timeout and @timeout>0
                    begin
                        pid = Process.getpgid(t.pid) * -1
                    rescue
                        pid = t.pid
                    end

                    if pid
                        begin
                            sleep @timeout

                            mutex.synchronize do
                                terminator_e = Timeout::Error
                            end
                        ensure
                        end

                        Process.kill('TERM', pid)
                    end
                end
            }

            i.write stdin_data
            i.close

            # blocking wait for process termination
            t.value

            # if reader threads are not dead yet, kill them
            [out_reader, err_reader].each do |reader|
                next unless reader.status

                reader.join(0.1)
                reader.kill
            end

            mutex.lock
            terminator.kill
            raise terminator_e if terminator_e

            # return values
            [out_reader.value, err_reader.value, t.value]
        }
    end

end

# Executes commands in the machine where it is called. See documentation
# for GenericCommand
class LocalCommand < GenericCommand
private

{'fraction_non_alphanumeric': 0.11643200404687631, 'fraction_numerical': 0.008093752634685102, 'mean_word_length': 1.560880829015544, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 31, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 2}, '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': '3493169', 'n_tokens_mistral': 3243, 'n_tokens_neox': 3097, 'n_words': 1071}