lib/rainbows/thread_spawn.rb

   1 # -*- encoding: binary -*-
   2 module Rainbows
   3
   4   # Spawns a new thread for every client connection we accept().  This
   5   # model is recommended for platforms like Ruby 1.8 where spawning new
   6   # threads is inexpensive.
   7   #
   8   # This model should provide a high level of compatibility with all
   9   # Ruby implementations, and most libraries and applications.
  10   # Applications running under this model should be thread-safe
  11   # but not necessarily reentrant.
  12   #
  13   # If you're connecting to external services and need to perform DNS
  14   # lookups, consider using the "resolv-replace" library which replaces
  15   # parts of the core Socket package with concurrent DNS lookup
  16   # capabilities
  17
  18   module ThreadSpawn
  19
  20     include Base
  21
  22     def worker_loop(worker)
  23       init_worker_process(worker)
  24       RACK_DEFAULTS["rack.multithread"] = true
  25       threads = ThreadGroup.new
  26       alive = worker.tmp
  27       m = 0
  28       limit = worker_connections
  29
  30       begin
  31         ret = begin
  32           alive.chmod(m = 0 == m ? 1 : 0)
  33           IO.select(LISTENERS, nil, nil, timeout) or next
  34         rescue Errno::EINTR
  35           retry
  36         rescue Errno::EBADF, TypeError
  37           break
  38         end
  39         alive.chmod(m = 0 == m ? 1 : 0)
  40
  41         ret.first.each do |l|
  42           # Sleep if we're busy, another less busy worker process may
  43           # take it for us if we sleep. This is gross but other options
  44           # still suck because they require expensive/complicated
  45           # synchronization primitives for _every_ case, not just this
  46           # unlikely one.  Since this case is (or should be) uncommon,
  47           # just busy wait when we have to.
  48           while threads.list.size > limit # unlikely
  49             sleep(0.1) # hope another process took it
  50             break # back to IO.select
  51           end
  52           begin
  53             threads.add(Thread.new(l.accept_nonblock) {|c| process_client(c) })
  54           rescue Errno::EAGAIN, Errno::ECONNABORTED
  55           end
  56         end
  57       rescue Object => e
  58         listen_loop_error(e)
  59       end while LISTENERS.first && master_pid == Process.ppid
  60       join_threads(threads.list, worker)
  61     end
  62
  63   end
  64 end