Re: ruby 1.8.6, threadpooling and blocking sockets - advice/help

On 20.10.2009 02:31, Daniel Bush wrote:
> Robert Klemme wrote:
>> On 10/19/2009 02:51 PM, Daniel Bush wrote:
>>>       puts "#{Thread.current} received #{data}... "
>>>     end
>>>     client.close
>>>
>>> on @socket which was set up like this:
>>>
>>>     @socket = Socket.new( AF_INET, SOCK_STREAM, 0 )
>>>     @sockaddr = Socket.pack_sockaddr_in( 2200, 'localhost')
>>>     @socket.bind( @sockaddr )
>>>     @socket.listen( 100 )
>> This won't work.  You can have only 1 acceptor thread per server socket.
>>   Typically you dispatch processing *after* the accept to a thread
>> (either newly created or taken from a pool).
>>
>> I have no idea what the interpreter is going to do if you have multiple
>> threads trying to accept from the same socket.  In the best case #accept
>> is synchronized and only one thread gets to enter it.  In worse
>> scenarios anything bad may happen.
> 
> Ok, I wasn't sure if it was appropriate having >1 thread per socket 
> instance.  It *appears* to work ok on ruby 1.9 up to about 100 socket 
> connections - not that that means anything when it comes to testing 
> stuff with threads.  Maybe if I do 100,000+ I might elicit some type of 
> error.
> 
> I was intending to process the result of accept in another pool but I 
> was toying with the idea of having 2-3 threads waiting on #accept 
> assuming no synchronisation issues.  I didn't know if it really mattered 
> or not.  It might make a difference if you have a large number of 
> connections coming in depending on what the acceptor is doing in 
> addition; I wasn't sure.
> 
> I guess I'll have to scupper that idea or exhaustively test it to prove 
> it works and has benefit - both of which are questionable at this point.

Frankly, I wouldn't invest that effort: every example in all programming 
languages I have seen has just a single acceptor thread.  Accepting 
socket connections is not an expensive operation so as long as you 
refrain from further processing a single thread is completely sufficient 
for handling accepts.

>>> I wanted to create a barrage of requests so next I
>>> create a pool of requester threads which each run
>>> something like this:
>>>
>>>   socket = Socket.new( AF_INET, SOCK_STREAM, 0 )
>>>   sockaddr = Socket.pack_sockaddr_in( 2200, 'localhost' )
>>>   socket.connect( sockaddr )
>>>   socket.puts "request #{i}"
>>>   socket.close
>> Btw, why don't you use TCPServer and TCPSocket?
> 
> yeah I was going to, I was just going off some examples in the 
> documentation and trying to cut my teeth on them and writing some tests. 
> But I was heading that way.
> 
>>> queue.
>> See above.
>>
>>> I haven't submitted the full code because it uses
>>> a threadpool library I'm still building/reviewing.
>> I would rather do something like this (sketeched):
>>
>> require 'thread'
>> queue = Queue.new
>> workers = (1..10).map do
>>    Thread.new queue do |q|
>>      until (cl = q.deq).equal? q
>>        # process data from / for client cl
>>        begin
>>          data = cl.gets.chomp
>>          @mutex.synchronize do
>>            puts "#{Thread.current} received #{data}..."
>>          end
>>        ensure
>>          cl.close
>>        end
>>      end
>>    end
>> end
>>
>> server = TCPServer.new ...
>>
>> while client = server.accept
>>    queue.enq client
>> end
>>
>> # elsewhere
>>
>>
>> TCPSocket.open do |sock|
>>     sock.puts "request"
>> end
> 
> Thanks for the example.
> I am scratching my head a little with this line:
>   until (cl = q.deq).equal? q
> 
> I'm familiar with Queue and its behaviour.

That's the worker thread termination code which basically works by 
checking whether the item fetched from the Queue is the Queue instance 
itself.  Actually I omitted the other half of the code (the place which 
puts all those q instances in itself) because I didn't want to make the 
code more complex and also termination condition was unknown (may be a 
signal, a number of handled connections etc.).

If you want to make termination more readable you can also do something 
like this

QueueTermination = Object.new
...
until QueueTermination.equal?(cl = q.deq)
   ...
end

or

until QueueTermination == (cl = q.deq)
   ...
end

or

until QueueTermination === (cl = q.deq)
   ...
end

The basic idea is to stuff something in the queue which is unambiguously 
identifiable as non work content.

Kind regards

	robert

-- 
remember.guy do |as, often| as.you_can - without end
http://blog.rubybestpractices.com/