<commit_before>
require 'pp'
require 'open3'
require 'thread'


class SSHCommand
    attr_accessor :value, :code, :stdout, :stderr, :command
    attr_accessor :callback
    
    def initialize(command)
        @command=command
        @callback=nil
    end
    
    def run(host)
        (@stdout, @stderr)=execute(host, @command)
        @code=get_exit_code(@stderr)
    end
    
    def exec_callback(num)
        @callback.call(self, num) if @callback
    end
    
    # Gets exit code from STDERR
    def get_exit_code(str)
        tmp=str.scan(/^ExitCode: (\d*)$/)
        return nil if !tmp[0]
        tmp[0][0].to_i
    end
    
    # Executes a command in a remote machine
    def execute(host, command)
        std=exec_remote_command(host, command)
        [std[1].read, std[2].read]
    end
    
    # Low level local command execution
    def exec_local_command(command)
        std=Open3.popen3(
            "#{command} ;"+
            " echo ExitCode: $? 1>&2")
        [std[1].read, std[2].read]
    end
    
    # Low level remote command execution
    def exec_remote_command(host, command)
        Open3.popen3(
            "ssh -n #{host} #{command} ;"+
            " echo ExitCode: $? 1>&2")
    end
end

class SSHCommandList < Array
    def clone_actions
        new_array=Array.new
        self.each {|s|
            new_array << s.clone
        }
        new_array
    end
end

class SSHAction
    attr_accessor :callback
    
    def initialize(number, host, actions)
        @number=number
        @host=host
        if actions.is_a? SSHCommandList
            @actions=clone_actions(actions)
        else
            @actions=SSHCommandList.new
            @actions<<actions
            # Really needed
            @actions=clone_actions(@actions)
        end
        @finished=false
        @callback=nil
    end
    
    def finished
        @finished
    end
    
    def run
        @thread=Thread.new {
            run_actions
            @finished=true
        }   
    end
    
    def run_actions
        @actions.each {|a|
            a.run(@host)
        }
    end
    
    def exec_callbacks
        @actions.each {|a|
            a.exec_callback(@number)
        }
        
        @callback.call(@actions, @number) if @callback
    end
    
    
    # Clone an array of sensors
    def clone_actions(actions)
        actions.collect{|a| a.clone }
    end
end

module SSHActionController
    
    def init_actions
        @actions=Array.new
        @action_mutex=Mutex.new
        start_action_thread
    end
    
    def send_ssh_action(number, host, action)
        @action_mutex.lock
        action.run
        @actions << action
        @action_mutex.unlock
    end
    
    def start_action_thread
        @action_thread=Thread.new {
            while true
                @action_mutex.lock
                done=@actions.select{|a| a.finished }
                @action_mutex.unlock

                if done.length>0
                    done.each{|a|
                        #STDOUT.puts a.get_result
                        #STDOUT.flush
                        a.exec_callbacks
                    }

                    @action_mutex.lock
                    @actions-=done
                    @action_mutex.unlock
                end
                sleep(1)
            end
        }
    end
end

=begin<commit_msg>better thread synchronization<commit_after>
require 'pp'
require 'open3'
require 'thread'


class SSHCommand
    attr_accessor :value, :code, :stdout, :stderr, :command
    attr_accessor :callback
    
    def initialize(command)
        @command=command
        @callback=nil
    end
    
    def run(host)
        (@stdout, @stderr)=execute(host, @command)
        @code=get_exit_code(@stderr)
    end
    
    def exec_callback(num)
        @callback.call(self, num) if @callback
    end
    
    # Gets exit code from STDERR
    def get_exit_code(str)
        tmp=str.scan(/^ExitCode: (\d*)$/)
        return nil if !tmp[0]
        tmp[0][0].to_i
    end
    
    # Executes a command in a remote machine
    def execute(host, command)
        std=exec_remote_command(host, command)
        [std[1].read, std[2].read]
    end
    
    # Low level local command execution
    def exec_local_command(command)
        std=Open3.popen3(
            "#{command} ;"+
            " echo ExitCode: $? 1>&2")
        [std[1].read, std[2].read]
    end
    
    # Low level remote command execution
    def exec_remote_command(host, command)
        Open3.popen3(
            "ssh -n #{host} #{command} ;"+
            " echo ExitCode: $? 1>&2")
    end
end

class SSHCommandList < Array
    def clone_actions
        new_array=Array.new
        self.each {|s|
            new_array << s.clone
        }
        new_array
    end
end

class SSHAction
    attr_accessor :callback
    
    def initialize(number, host, actions)
        @number=number
        @host=host
        if actions.is_a? SSHCommandList
            @actions=clone_actions(actions)
        else
            @actions=SSHCommandList.new
            @actions<<actions
            # Really needed
            @actions=clone_actions(@actions)
        end
        @finished=false
        @callback=nil
    end
    
    def finished
        @finished
    end
    
    def run
        @thread=Thread.new {
            run_actions
            @finished=true
        }   
    end
    
    def run_actions
        @actions.each {|a|
            a.run(@host)
        }
    end
    
    def exec_callbacks
        @actions.each {|a|
            a.exec_callback(@number)
        }
        
        @callback.call(@actions, @number) if @callback
    end
    
    
    # Clone an array of sensors
    def clone_actions(actions)
        actions.collect{|a| a.clone }
    end
end

module SSHActionController
    
    def init_actions
        @actions=Array.new
        @action_mutex=Mutex.new
        @action_cond=ConditionVariable.new
        start_action_thread
    end
    
    def send_ssh_action(number, host, action)
        @action_mutex.synchronize {
        	action.run
        	@actions << action
        	@action_cond.signal
        }
    end
    
    def start_action_thread
        @action_thread=Thread.new {
            while true
                @action_mutex.synchronize {
                	@action_cond.wait(mutex)                	
                	done=@actions.delete_if{|a| a.finished }
                }
                if done
                	done.exec_callbacks
                end
            end
        }
    end
end

=begin
{'fraction_non_alphanumeric': 0.08689024390243902, 'fraction_numerical': 0.005182926829268293, 'mean_word_length': 1.5420379697791553, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 8, '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': '3793026', 'n_tokens_mistral': 1986, 'n_tokens_neox': 1936, 'n_words': 426}