From: Robert <bob.news@...>
Date: 2003-03-13T17:17:51+09:00
Subject: Re: Threads and DRb

This is a multi-part message in MIME format.

------=_NextPart_000_0032_01C2E940.2CB73C60
Content-Type: text/plain;
	charset="iso-8859-1"
Content-Transfer-Encoding: 7bit


Brian,

I slept over it and came up with this pool implementation (see attachment).
The facade part is still missing, but I've no experience with drb so I guess
that's the part where you can give better input.

"Brian Candler" <B.Candler@pobox.com> schrieb im Newsbeitrag
news:20030312210421.GA52600@uk.tiscali.com...
> On Wed, Mar 12, 2003 at 06:13:46PM +0900, Robert Klemme wrote:
> > Things I don't understand:
> >
> > Why do you introduce method alive?  Are there conditions under which
> > an instance in the pool can die?  If not, you could check sanity when
> > you take back an instance and put it into the pool.
>
> Sure, I didn't see much difference in checking sanity before or after the
> call though. The instance in the pool itself can't "die", but it could get
> into a state where it is easier just to reinitialise it.
>
> Thinking about this again, maybe it makes sense to check sanity only after
> an exception has been raised. In that case, the sanity check can be
> something reasonably active: e.g.
>
>   def alive?
>     @dbh.select_one("select 123 from dual") == 123
>   end
>
> If that returns false or an exception, then you can be pretty sure that
the
> instance has a problem, and it needs re-initialising.

Sounds reasonable but not feasible with a separated pool.  My pool does it
on insertion and retrieval.  Maybe this should be changed if we assume that
instances don't change their state in the pool.  Then checking on reinsert
should be sufficient.

One could adhere to the pattern that an instance is only put back into the
pool if the method did not throw.  This is easy achieved by changing the
pattern

obj = pool.get
begin
  obj.do()
ensure
  pool.put obj
end

to

obj = pool.get
obj.do()
pool.put obj

which will omit the put() in case of an exception.

> Perhaps even cleverer would be to have a separate thread for putting new
> objects into the pool, which means you can rate-limit them to one every
> second (say)
>
> > Why do you rededine the set of methods to invoke method_missing?  This
> > happens anyway if a method is not implemented.
>
> It doesn't work with DRb. DRb validates method calls explicitly and raises
> an exception if they are not present: in drb/drb.rb see
> 'check_insecure_method', which is called from its own 'method_missing'
>
> So to keep DRb happy, the pool object has to have real methods
corresponding
> to the ones in the pool objects.
>
> You don't need to list the methods explicitly in the facade in cases where
> method_missing is sufficient though.

Ah!  I see.

> > Room for improvement:
> >
> > I'd change the design in a way that the pool and the facade are
> > separated.  This is orthogonal functionality and by extracting the
> > pool functionality to another class the design becomes cleaner and
> > more modular, hence improved reusability.
>
> Sounds reasonable, how do you suggest: (facade) has-a (pool) ?
> I can't think of a good way to express that though.

Well, just hand the pool over to the facade on construction or create it in
initialize().

> > The @poolnotempty.wait(@mutex) should be in a loop because with only
> > one element in the pool and two threads waiting and both are signalled
> > only one can fetch the element and the other has to continue waiting.
> > Normally this can't occur in your example
>
> Even if it did, wouldn't that be the correct behaviour? i.e. two threads
are
> waiting on an empty pool, the cv is signalled when one object is put back
> into the pool (which allows one of the waiting threads to take it back out
> again). When a second object is put back into the pool the cv will be
> signalled again, so the second waiting thread will be woken. Or am I
missing
> a sequence which causes this to fail?

No, as I said this can't happen in your example.  But the more general
pattern is to repeat the check.  That's not too costly and you avoid strange
errors if the condition variable is used for other signals. (I can't see
such an application yet, and maybe I'm too much influenced by the Java
notify() and notifyAll()...)

I think the situation changes with a separated pool: In that case you want
to limit the number of threads that is concurrently running.  In that case
the pool does not need a condition variable (see my example).  Instead the
facade instance needs a semaphore which is initialized to the max amount of
threads and decremented on entry and incremented on exit of a method.

> > You can look at some stuff I put here for threading purposes:
> > http://www.rubygarden.org/ruby?MultiThreading
>
> Nice introduction. I found 'ObjectPoolingAndThreading' before, but the
code
> given assumed that it was spawning fresh threads itself - which doesn't
work
> where some other process is spawning them.

Thanks!  I'm not satisfied with it yet, but I try to add something from time
to time.

Kind regards

    robert

------=_NextPart_000_0032_01C2E940.2CB73C60
Content-Type: application/octet-stream;
	name="thread-safe-instance-pool.rb"
Content-Transfer-Encoding: 7bit
Content-Disposition: attachment;
	filename="thread-safe-instance-pool.rb"


require 'thread'

=begin
=Class InstancePool

This class maintains a thread safe pool of instances.
Instances are obtained via ((<InstancePool#get>)) and
put back via ((<InstancePool#put>)).  These operations
are thread safe.

New instances are created via the factory code handed to
the constructor as block.
=end

class InstancePool
  def initialize(maxSize = nil, initialSize = 0, &factory)
    @mutex = Mutex.new
    @pool = []
    @verify = proc {|o| o}

    @max = maxSize
    @factory = factory

    initialSize.times { @pool.push @factory.call }
  end

  def verify(&v)
    @mutex.synchronize do
      @verify = v
    end

    self
  end

  def get
    @mutex.synchronize do
      obj = @pool.shift

      until @verify.call(obj) || @pool.empty? do
        obj = @pool.shift
      end

      obj or @factory.call
    end
  end


  def put(obj)
    @mutex.synchronize do
      @pool.push obj if @verify.call(obj) && (!@max || @pool.size < @max)
    end
  end

end


# demo, without threads
count = 0
pool = InstancePool.new {"s#{count+=1}"}.verify {|o| o and o =~ /^s\d*3$/}

for i in 0..10
  o = pool.get

  begin
    puts "#{i}: #{o}"
  ensure
    pool.put o
  end
end

------=_NextPart_000_0032_01C2E940.2CB73C60--