<commit_before>module RDF::Raptor
  ##
  # A command-line interface to Raptor's `rapper` utility.
  module CLI
    ##
    # Reader implementation.
    module Reader
      ##
      # @param  [IO, File, RDF::URI, String] input
      # @param  [Hash{Symbol => Object}]     options
      # @option (options) [String, #to_s]    :base_uri ("file:///dev/stdin")
      # @yield  [reader]
      # @yieldparam [RDF::Reader] reader
      def initialize(input = $stdin, options = {}, &block)
        raise RDF::ReaderError.new("`rapper` binary not found") unless RDF::Raptor.available?

        format = self.class.format.rapper_format
        case input
          when RDF::URI, %r(^(file|http|https|ftp)://)
            @command = "#{RAPPER} -q -i #{format} -o ntriples '#{input}'"
            @command << " '#{options[:base_uri]}'" if options.has_key?(:base_uri)
            @rapper  = IO.popen(@command, 'rb')
          when File
            @command = "#{RAPPER} -q -i #{format} -o ntriples '#{File.expand_path(input.path)}'"
            @command << " '#{options[:base_uri]}'" if options.has_key?(:base_uri)
            @rapper  = IO.popen(@command, 'rb')
          else # IO, String
            @command = "#{RAPPER} -q -i #{format} -o ntriples file:///dev/stdin"
            @command << " '#{options[:base_uri]}'" if options.has_key?(:base_uri)
            @rapper  = IO.popen(@command, 'rb+')
            pid = fork do
              begin
                if input.respond_to?(:read)
                  buf = String.new
                  while input.read(8192, buf)
                    @rapper.write(buf)
                  end
                else
                  @rapper.write(input.to_s)
                end
                @rapper.close_write
              ensure
                exit
              end
            end
            Process.detach(pid)
            @rapper.close_write
        end
        @reader = RDF::NTriples::Reader.new(@rapper, options, &block)
      end

      protected

      ##
      # @return [Array]
      def read_triple
        @reader.read_triple
      end
    end

    ##
    # Writer implementation.
    module Writer
      ##
      # @param  [IO, File]               output
      # @param  [Hash{Symbol => Object}] options
      # @yield  [writer]
      # @yieldparam [RDF::Writer] writer
      def initialize(output = $stdout, options = {}, &block)
        raise RDF::WriterError.new("`rapper` binary not found") unless RDF::Raptor.available?

        format = self.class.format.rapper_format
        case output
          when File, IO, StringIO
            @command = "#{RAPPER} -q -i ntriples -o #{format} file:///dev/stdin"
            @command << " '#{options[:base_uri]}'" if options.has_key?(:base_uri)
            @rapper  = IO.popen(@command, 'rb+')
          else
            raise ArgumentError.new("unsupported output type: #{output.inspect}")
        end
        @writer = RDF::NTriples::Writer.new(@rapper, options)
        super(output, options, &block)
      end

      protected

      ##
      # @return [void]
      def write_prologue
        super
      end

      ##
      # @param  [RDF::Resource] subject
      # @param  [RDF::URI]      predicate
      # @param  [RDF::Value]    object
      # @return [void]
      def write_triple(subject, predicate, object)
        @writer.write_triple(subject, predicate, object)
      end

      ##
      # @return [void]
      def write_epilogue
        @rapper.close_write
        begin
          chunk_size = @options[:chunk_size] || 4096 # bytes
          loop do
            @output.write(@rapper.readpartial(chunk_size))
          end
        rescue EOFError => e
          # we're all done
        end
      end
    end
  end
end<commit_msg>Fix rapper zombies and writer deadlocks.<commit_after>require 'tempfile'

module RDF::Raptor
  ##
  # A command-line interface to Raptor's `rapper` utility.
  module CLI
    ##
    # Reader implementation.
    module Reader
      ##
      # @param  [IO, File, RDF::URI, String] input
      # @param  [Hash{Symbol => Object}]     options
      # @option (options) [String, #to_s]    :base_uri ("file:///dev/stdin")
      # @yield  [reader]
      # @yieldparam [RDF::Reader] reader
      def initialize(input = $stdin, options = {}, &block)
        raise RDF::ReaderError.new("`rapper` binary not found") unless RDF::Raptor.available?

        format = self.class.format.rapper_format
        case input
          when RDF::URI, %r(^(file|http|https|ftp)://)
            @command = "#{RAPPER} -q -i #{format} -o ntriples '#{input}'"
            @command << " '#{options[:base_uri]}'" if options.has_key?(:base_uri)
            @rapper  = IO.popen(@command, 'rb')
          when File, Tempfile
            @command = "#{RAPPER} -q -i #{format} -o ntriples '#{File.expand_path(input.path)}'"
            @command << " '#{options[:base_uri]}'" if options.has_key?(:base_uri)
            @rapper  = IO.popen(@command, 'rb')
          else # IO, String
            @command = "#{RAPPER} -q -i #{format} -o ntriples file:///dev/stdin"
            @command << " '#{options[:base_uri]}'" if options.has_key?(:base_uri)
            @rapper  = IO.popen(@command, 'rb+')
            pid = fork do
              # process to feed rapper
              begin
                @rapper.close_read
                if input.respond_to?(:read)
                  buf = String.new
                  while input.read(8192, buf)
                    @rapper.write(buf)
                  end
                else
                  @rapper.write(input.to_s)
                end
                @rapper.close_write
              ensure
                Process.exit
              end
            end
            Process.detach(pid)
            @rapper.close_write
        end
        @reader = RDF::NTriples::Reader.new(@rapper, options, &block)
      end

      protected

      ##
      # @return [Array]
      def read_triple
        raise EOFError if @rapper.closed?
        begin
          triple = @reader.read_triple
        rescue EOFError
          @rapper.close
          raise
        end
        triple
      end

    end

    ##
    # Writer implementation.
    module Writer
      ##
      # @param  [IO, File]               output
      # @param  [Hash{Symbol => Object}] options
      # @yield  [writer]
      # @yieldparam [RDF::Writer] writer
      def initialize(output = $stdout, options = {}, &block)
        raise RDF::WriterError.new("`rapper` binary not found") unless RDF::Raptor.available?

        format = self.class.format.rapper_format
        case output
          when File, IO, StringIO, Tempfile
            @command = "#{RAPPER} -q -i ntriples -o #{format} file:///dev/stdin"
            @command << " '#{options[:base_uri]}'" if options.has_key?(:base_uri)
            @rapper  = IO.popen(@command, 'rb+')
          else
            raise ArgumentError.new("unsupported output type: #{output.inspect}")
        end
        @writer = RDF::NTriples::Writer.new(@rapper, options)
        super(output, options, &block)
      end

      protected

      ##
      # @return [void]
      def write_prologue
        super
      end

      ##
      # @param  [RDF::Resource] subject
      # @param  [RDF::URI]      predicate
      # @param  [RDF::Value]    object
      # @return [void]
      def write_triple(subject, predicate, object)
        output_transit(false)
        @writer.write_triple(subject, predicate, object)
        output_transit(false)
      end

      ##
      # @return [void]
      def write_epilogue
        @rapper.close_write unless @rapper.closed?
        output_transit(true)
      end

      ##
      # Feed any available rapper output to the destination.
      # @return [void]
      def output_transit(block)
        unless @rapper.closed?
          chunk_size = @options[:chunk_size] || 4096 # bytes
          begin
            loop do
              @output.write(block ? @rapper.readpartial(chunk_size) : @rapper.read_nonblock(chunk_size))
            end
          rescue EOFError => e
            @rapper.close
          rescue Errno::EAGAIN, Errno::EINTR
            # eat
          end
        end
      end

    end
  end
end
{'fraction_non_alphanumeric': 0.12670731707317073, 'fraction_numerical': 0.001951219512195122, 'mean_word_length': 1.7400601403274307, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 9, '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': '2843136', 'n_tokens_mistral': 2509, 'n_tokens_neox': 2439, 'n_words': 568}