Module: Rainbows::ThreadPool

Includes:
Base
Defined in:
lib/rainbows/thread_pool.rb

Overview

Implements a worker thread pool model. This is suited for platforms like Ruby 1.9, where the cost of dynamically spawning a new thread for every new client connection is higher than with the ThreadSpawn model, but the cost of an idle thread is low (e.g. NPTL under Linux).

This model should provide a high level of compatibility with all Ruby implementations, and most libraries and applications. Applications running under this model should be thread-safe but not necessarily reentrant.

Applications using this model are required to be thread-safe. Threads are never spawned dynamically under this model.

If you’re using green threads (MRI 1.8) and need to perform DNS lookups, consider using the “resolv-replace” library which replaces parts of the core Socket package with concurrent DNS lookup capabilities.

Instance Method Summary collapse

Methods included from Base

included, #init_worker_process, #process_client, #reopen_worker_logs, #sig_receiver

Instance Method Details

#async_worker ⇒ Object

:nodoc:



47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
 
# File 'lib/rainbows/thread_pool.rb', line 47

def async_worker # :nodoc:
  begin
    # TODO: check if select() or accept() is a problem on large
    # SMP systems under Ruby 1.9.  Hundreds of native threads
    # all working off the same socket could be a thundering herd
    # problem.  On the other hand, a thundering herd may not
    # even incur as much overhead as an extra Mutex#synchronize
    ret = select(LISTENERS) and ret[0].each do |s|
      s = s.kgio_tryaccept and s.process_loop
    end
  rescue Errno::EINTR
  rescue => e
    Rainbows::Error.listen_loop(e)
  end while Rainbows.alive
end

#sync_worker ⇒ Object

:nodoc:



38
39
40
41
42
43
44
45
 
# File 'lib/rainbows/thread_pool.rb', line 38

def sync_worker # :nodoc:
  s = LISTENERS[0]
  begin
    c = s.kgio_accept and c.process_loop
  rescue => e
    Rainbows::Error.listen_loop(e)
  end while Rainbows.alive
end

#worker_loop(worker) ⇒ Object

:nodoc:



22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
 
# File 'lib/rainbows/thread_pool.rb', line 22

def worker_loop(worker) # :nodoc:
  init_worker_process(worker)
  pool = (1..worker_connections).map do
    Thread.new { LISTENERS.size == 1 ? sync_worker : async_worker }
  end

  while Rainbows.alive
    # if any worker dies, something is serious wrong, bail
    pool.each do |thr|
      Rainbows.tick or break
      thr.join(1) and Rainbows.quit!
    end
  end
  Rainbows::JoinThreads.acceptors(pool)
end