<commit_before>module DeadlySerious
  module Engine
    class Pipeline
      include DeadlySerious::Engine::Commands

      attr_reader :data_dir, :pipe_dir, :pids

      def initialize(data_dir: './data',
                     pipe_dir: "/tmp/deadly_serious/#{Process.pid}",
                     preserve_pipe_dir: false,
                     &block)
        @data_dir = data_dir
        @pipe_dir = pipe_dir
        @block = block
        @pids = []
        Channel.config(data_dir, pipe_dir, preserve_pipe_dir)
      end

      def run
        Channel.setup
        @block.call(self)
        wait_children
      rescue => e
        kill_children
        raise e
      ensure
        Channel.teardown
      end

      # Wait all sub processes to finish before
      # continue the pipeline.
      #
      # Always prefer to use {DeadlySerious::Engine::Commands#spawn_capacitor}
      # if possible.
      def wait_processes!
        wait_children
      end

      # Spawn a  class in a separated process.
      #
      # This is a basic command, use it only if you have
      # more than one input or output pipe. Otherwise
      # prefer the simpler {DeadlySerious::Engine::Commands#spawn_class} or
      # the {DeadlySerious::Engine::Commands#spawn} methods.
      def spawn_process(class_or_object, *args, process_name: nil, readers: [last_pipe], writers: [next_pipe])
        # TODO if we have no readers, alarm! (how about data sources???)
        # TODO if we have no readers, and this is the first process, read from STDIN
        # TODO if we have no writers, alarm! (how about data sinks???)
        # TODO if we have no writers, and this is the last process, write to STDOUT
        process_name ||= class_or_object.respond_to?(:name) ? class_or_object.name : class_or_object.to_s
        writers.each { |writer| create_pipe(writer) }
        @pids << fork do
          begin
            set_process_name(process_name, readers, writers)
            # TODO Change this to not modify "a_class", so we can pass instances too
            the_object = Class === class_or_object ? class_or_object.new : class_or_object
            append_open_io_if_needed(the_object)<commit_msg>A little better process naming<commit_after>module DeadlySerious
  module Engine
    class Pipeline
      include DeadlySerious::Engine::Commands

      attr_reader :data_dir, :pipe_dir, :pids

      def initialize(data_dir: './data',
                     pipe_dir: "/tmp/deadly_serious/#{Process.pid}",
                     preserve_pipe_dir: false,
                     &block)
        @data_dir = data_dir
        @pipe_dir = pipe_dir
        @block = block
        @pids = []
        Channel.config(data_dir, pipe_dir, preserve_pipe_dir)
      end

      def run
        Channel.setup
        @block.call(self)
        wait_children
      rescue => e
        kill_children
        raise e
      ensure
        Channel.teardown
      end

      # Wait all sub processes to finish before
      # continue the pipeline.
      #
      # Always prefer to use {DeadlySerious::Engine::Commands#spawn_capacitor}
      # if possible.
      def wait_processes!
        wait_children
      end

      # Spawn a  class in a separated process.
      #
      # This is a basic command, use it only if you have
      # more than one input or output pipe. Otherwise
      # prefer the simpler {DeadlySerious::Engine::Commands#spawn_class} or
      # the {DeadlySerious::Engine::Commands#spawn} methods.
      def spawn_process(class_or_object, *args, process_name: nil, readers: [last_pipe], writers: [next_pipe])
        # TODO if we have no readers, alarm! (how about data sources???)
        # TODO if we have no readers, and this is the first process, read from STDIN
        # TODO if we have no writers, alarm! (how about data sinks???)
        # TODO if we have no writers, and this is the last process, write to STDOUT
        the_object = Class === class_or_object ? class_or_object.new : class_or_object
        process_name ||= the_object.respond_to?(:name) ? the_object.name : the_object.to_s
        writers.each { |writer| create_pipe(writer) }
        @pids << fork do
          begin
            set_process_name(process_name, readers, writers)
            append_open_io_if_needed(the_object)
{'fraction_non_alphanumeric': 0.07895351553375379, 'fraction_numerical': 0.0, 'mean_word_length': 2.301464919043948, 'pattern_counts': {'":': 0, '<': 7, '<?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': '5756501', 'n_tokens_mistral': 1232, 'n_tokens_neox': 1217, 'n_words': 376}