lib/process_factory/factory.rb
def run
    exception = nil

    if debug?
      loop do
        break if exception
        begin
          instr = @parent.send(@options[:pre], @options)
          break unless instr.is_a?(Hash)
          @parent.send(@options[:post], @parent.send(@options[:worker], instr))
        rescue Interrupt # ctrl+c
          puts "\nEnding threads/processes..."
          exit
        rescue StandardError => e
          exception = ProcessFactory::ErrDebug.new(e)
        end
      end
    else
      threaded(@options[:count]) do |t|
        begin
          loop do
            break if exception

            begin
              opts = @mutex.synchronize { @parent.send(@options[:pre], @options.merge(thread: t)) }
              break unless opts.is_a?(Hash)

              if processes?
                begin
                  # write/read IO.pipe
                  Marshal.dump(opts, @workers[t].write)
                  opts = Marshal.load(@workers[t].read)
                rescue IOError, Errno::EPIPE # process died
                  # TODO: retry up to 3 times?
                  opts = ProcessFactory::ErrDebug.new("Process #{t} died...")
                end
              elsif threads?
                opts = @parent.send(@options[:worker], opts)
              end

              if opts.is_a?(ProcessFactory::ErrDebug)
                exception = opts
              else
                @mutex.synchronize { @parent.send(@options[:post], opts) }
              end

            rescue StandardError => e
              exception = ProcessFactory::ErrDebug.new(e)
            end

          end
        ensure
          @workers[t].close && @workers[t].wait if processes?
        end
      end
    end

    raise exception if exception
  end
#-------------------------------------------------------------------------------------#
# main method to start synchronized calculation
main method to start synchronized calculation
main method to start synchronized calculation
def run
    exception = nil
    if debug?
      loop do
        break if exception
        begin
          instr = @parent.send(@options[:pre], @options)
          break unless instr.is_a?(Hash)
          @parent.send(@options[:post], @parent.send(@options[:worker], instr))
        rescue Interrupt 
          puts "\nEnding threads/processes..."
          exit
        rescue StandardError => e
          exception = ProcessFactory::ErrDebug.new(e)
        end
      end
    else
      threaded(@options[:count]) do |t|
        begin
          loop do
            break if exception
            begin
              opts = @mutex.synchronize { @parent.send(@options[:pre], @options.merge(thread: t)) }
              break unless opts.is_a?(Hash)
              if processes?
                begin
                  Marshal.dump(opts, @workers[t].write)
                  opts = Marshal.load(@workers[t].read)
                rescue IOError, Errno::EPIPE 
                  opts = ProcessFactory::ErrDebug.new("Process #{t} died...")
                end
              elsif threads?
                opts = @parent.send(@options[:worker], opts)
              end
              if opts.is_a?(ProcessFactory::ErrDebug)
                exception = opts
              else
                @mutex.synchronize { @parent.send(@options[:post], opts) }
              end
            rescue StandardError => e
              exception = ProcessFactory::ErrDebug.new(e)
            end
          end
        ensure
          @workers[t].close && @workers[t].wait if processes?
        end
      end
    end
    raise exception if exception
  end
def run exception = nil if debug? loop do break if exception begin instr = @parent . send ( @options [ :pre ] , @options ) break unless instr . is_a? ( Hash ) @parent . send ( @options [ :post ] , @parent . send ( @options [ :worker ] , instr ) ) rescue Interrupt puts "\nEnding threads/processes..." exit rescue StandardError => e exception = ProcessFactory :: ErrDebug . new ( e ) end end else threaded ( @options [ :count ] ) do | t | begin loop do break if exception begin opts = @mutex . synchronize { @parent . send ( @options [ :pre ] , @options . merge ( thread : t ) ) } break unless opts . is_a? ( Hash ) if processes? begin Marshal . dump ( opts , @workers [ t ] . write ) opts = Marshal . load ( @workers [ t ] . read ) rescue IOError , Errno :: EPIPE opts = ProcessFactory :: ErrDebug . new ( "Process #{t} died..." ) end elsif threads? opts = @parent . send ( @options [ :worker ] , opts ) end if opts . is_a? ( ProcessFactory :: ErrDebug ) exception = opts else @mutex . synchronize { @parent . send ( @options [ :post ] , opts ) } end rescue StandardError => e exception = ProcessFactory :: ErrDebug . new ( e ) end end ensure @workers [ t ] . close && @workers [ t ] . wait if processes? end end end raise exception if exception end
main method to start synchronized calculation
main method to start synchronized calculation
# ctrl+c # write/read IO.pipe # process died # TODO: retry up to 3 times?