From: Robert Klemme <bob.news@...>
Date: 2005-11-21T00:02:38+09:00
Subject: Pipelined Processing

------=_NextPart_000_0006_01C5EDEB.493EB130
Content-Type: text/plain;
	format=flowed;
	charset="iso-8859-1";
	reply-type=original
Content-Transfer-Encoding: 7bit


Hi,

this came up recently on IRC: question was how to chain processing so that 
each step runs concurrently to other steps.  While I don't see real benefit 
as long as there are no native threads in Ruby I played around a bit and 
this is the result (attached).  There's certainly rool for improvement.  Do 
with this whatever you like.

Kind regards

    robert

------=_NextPart_000_0006_01C5EDEB.493EB130
Content-Type: application/octet-stream;
	name="ptest.rb"
Content-Transfer-Encoding: quoted-printable
Content-Disposition: attachment;
	filename="ptest.rb"

#! /usr/bin/env ruby=0A=
require 'pipeline'=0A=
=0A=
pipe =3D Pipeline.create(=0A=
  lambda {|x| x + 1},=0A=
  lambda {|x| x * 10},=0A=
  lambda {|x| p x }=0A=
)=0A=
=0A=
pipe.feed_each 1..1000=0A=
puts "fed"=0A=
pipe.term=0A=

------=_NextPart_000_0006_01C5EDEB.493EB130
Content-Type: application/octet-stream;
	name="pipeline.rb"
Content-Transfer-Encoding: quoted-printable
Content-Disposition: attachment;
	filename="pipeline.rb"

require 'thread'=0A=
=0A=
class Pipeline=0A=
=0A=
  TERM =3D Object.new.freeze=0A=
=0A=
  class PipelineThread < Thread=0A=
    attr_reader :in, :out=0A=
=0A=
    def initialize(qin,qout,processor)=0A=
      @qin, @qout, @processor =3D qin,qout,processor=0A=
=0A=
      super() do=0A=
        until ( TERM =3D=3D (obj =3D @qin.deq) )=0A=
          @qout.enq(@processor[obj])=0A=
        end=0A=
=0A=
        @qout.enq TERM=0A=
      end=0A=
    end=0A=
  end=0A=
=0A=
  def self.create(*processors)=0A=
    self.new(processors)=0A=
  end=0A=
=0A=
  def initialize(processors)=0A=
    @procs =3D processors=0A=
    start=0A=
  end=0A=
=0A=
  def feed(x)=0A=
    @head.enq x=0A=
    self=0A=
  end=0A=
=0A=
  def feed_each(enum)=0A=
    enum.each {|x| feed(x)}=0A=
    self=0A=
  end=0A=
=0A=
  def start=0A=
    @head =3D Queue.new=0A=
    @threads =3D []=0A=
=0A=
    lq =3D @procs[0..-2].inject(@head) do |q,pr|=0A=
      out =3D Queue.new=0A=
      @threads << PipelineThread.new(q,out,pr)=0A=
      out=0A=
    end=0A=
=0A=
    @threads << Thread.new(@procs.last) do |lst|=0A=
      until ( TERM =3D=3D (obj =3D lq.deq) )=0A=
        lst[obj]=0A=
      end=0A=
    end=0A=
  end=0A=
=0A=
  def terminate=0A=
    feed TERM=0A=
    @threads.each {|th| th.join}=0A=
    @threads =3D @head =3D nil=0A=
  end=0A=
=0A=
  alias term terminate=0A=
end=0A=

------=_NextPart_000_0006_01C5EDEB.493EB130--