<commit_before>require 'open3'
require 'timeout'

module Shexec

  class PipeExecutor

    PAGE_SIZE = 4096

    def initialize(stdin, stdout, stderr)
      @stdin, @stdout, @stderr = stdin, stdout, stderr
    end

    def run(cmd, *args)
      assert_untainted_command(cmd, *args)
      threads = []
      Open3.popen3([cmd, cmd], *args) do |stdin, stdout, stderr, wait_thr|
        threads << pusher(@stdin, stdin)
        threads << drainer(stdout, @stdout)
        threads << drainer(stderr, @stderr)

        yield wait_thr if block_given?

        wait_thr.value.tap { threads.each { |t| t.join } }
      end
    end

    def timeout(seconds, exception = Timeout::Error)
      TimeoutDelegator.new(self, seconds, exception)
    end

    private

      def pusher(input, output)
        Thread.new do
          stream(input, output)
          output.close unless output.closed?
        end
      end

      def drainer(input, output)
        Thread.new do
          stream(input, output)
          output.flush
        end
      end

      def stream(input, output)
        until input.closed? or input.eof?
          output.write input.read(PAGE_SIZE)
        end
      end

      def assert_untainted_command(*components)
        components.each do |component|
          raise SecurityError.new("refusing to construct a command line from tainted component #{component}") if component.tainted?
        end
      end

      class TimeoutDelegator

        def initialize(executor, timeout, exception)
          @executor, @timeout, @exception = executor, timeout, exception
        end

        def run(cmd, *args)
          @executor.run(cmd, *args) do |t|
            begin
              Timeout.timeout(@timeout, @exception) do
                if block_given?
                  yield t
                else
                  sleep(0.001) while t.alive?
                end
              end
            rescue @exception => e
              begin
                Process.kill("TERM", t.pid)
              rescue Errno::ESRCH
              ensure
                raise e
              end
            end
          end
        end

      end

  end

end<commit_msg>Sleep peacefully<commit_after>require 'open3'
require 'timeout'

module Shexec

  class PipeExecutor

    PAGE_SIZE = 4096

    def initialize(stdin, stdout, stderr)
      @stdin, @stdout, @stderr = stdin, stdout, stderr
    end

    def run(cmd, *args)
      assert_untainted_command(cmd, *args)
      threads = []
      Open3.popen3([cmd, cmd], *args) do |stdin, stdout, stderr, wait_thr|
        threads << pusher(@stdin, stdin)
        threads << drainer(stdout, @stdout)
        threads << drainer(stderr, @stderr)

        yield wait_thr if block_given?

        wait_thr.value.tap { threads.each { |t| t.join } }
      end
    end

    def timeout(seconds, exception = Timeout::Error)
      TimeoutDelegator.new(self, seconds, exception)
    end

    private

      def pusher(input, output)
        Thread.new do
          stream(input, output)
          output.close unless output.closed?
        end
      end

      def drainer(input, output)
        Thread.new do
          stream(input, output)
          output.flush
        end
      end

      def stream(input, output)
        until input.closed? or input.eof?
          output.write input.read(PAGE_SIZE)
        end
      end

      def assert_untainted_command(*components)
        components.each do |component|
          raise SecurityError.new("refusing to construct a command line from tainted component #{component}") if component.tainted?
        end
      end

      class TimeoutDelegator

        def initialize(executor, timeout, exception)
          @executor, @timeout, @exception = executor, timeout, exception
        end

        def run(cmd, *args)
          @executor.run(cmd, *args) do |t|
            begin
              Timeout.timeout(@timeout, @exception) do
                yield t if block_given?
                t.value
              end
            rescue @exception => e
              begin
                Process.kill("TERM", t.pid)
              rescue Errno::ESRCH
              ensure
                raise e
              end
            end
          end
        end

      end

  end

end
{'fraction_non_alphanumeric': 0.08208255159474671, 'fraction_numerical': 0.004221388367729831, 'mean_word_length': 1.6976597090449084, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 5, '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': '6858324', 'n_tokens_mistral': 1244, 'n_tokens_neox': 1185, 'n_words': 332}