lib/rainbows/thread_spawn.rb

   1 # -*- encoding: binary -*-
   2 require 'thread'
   3 module Rainbows
   4
   5   # Spawns a new thread for every client connection we accept().  This
   6   # model is recommended for platforms like Ruby 1.8 where spawning new
   7   # threads is inexpensive.
   8   #
   9   # This model should provide a high level of compatibility with all
  10   # Ruby implementations, and most libraries and applications.
  11   # Applications running under this model should be thread-safe
  12   # but not necessarily reentrant.
  13   #
  14   # If you're connecting to external services and need to perform DNS
  15   # lookups, consider using the "resolv-replace" library which replaces
  16   # parts of the core Socket package with concurrent DNS lookup
  17   # capabilities
  18
  19   module ThreadSpawn
  20
  21     include Base
  22
  23     def accept_loop(klass)
  24       lock = Mutex.new
  25       limit = worker_connections
  26       LISTENERS.each do |l|
  27         klass.new(l) do |l|
  28           begin
  29             if lock.synchronize { G.cur >= limit }
  30               # Sleep if we're busy, another less busy worker process may
  31               # take it for us if we sleep. This is gross but other options
  32               # still suck because they require expensive/complicated
  33               # synchronization primitives for _every_ case, not just this
  34               # unlikely one.  Since this case is (or should be) uncommon,
  35               # just busy wait when we have to.
  36               sleep(0.01)
  37             elsif c = Rainbows.sync_accept(l)
  38               klass.new(c) do |c|
  39                 begin
  40                   lock.synchronize { G.cur += 1 }
  41                   process_client(c)
  42                 ensure
  43                   lock.synchronize { G.cur -= 1 }
  44                 end
  45               end
  46             end
  47           rescue => e
  48             Error.listen_loop(e)
  49           end while G.alive
  50         end
  51       end
  52       sleep 1 while G.tick || lock.synchronize { G.cur > 0 }
  53     end
  54
  55     def worker_loop(worker)
  56       init_worker_process(worker)
  57       accept_loop(Thread)
  58     end
  59   end
  60 end