1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
| | # -*- encoding: binary -*-
module Rainbows
# This concurrency model implements a single-threaded app dispatch
# with a separate thread pool for writing responses.
#
# Unlike most \Rainbows! concurrency models, WriterThreadPool is
# designed to run behind nginx just like Unicorn is. This concurrency
# model may be useful for existing Unicorn users looking for more
# output concurrency than socket buffers can provide while still
# maintaining a single-threaded application dispatch (though if the
# response body is dynamically generated, it must be thread safe).
#
# For serving large or streaming responses, using more threads (via
# the +worker_connections+ setting) and setting "proxy_buffering off"
# in nginx is recommended. If your application does not handle
# uploads, then using any HTTP-aware proxy like haproxy is fine.
# Using a non-HTTP-aware proxy will leave you vulnerable to
# slow client denial-of-service attacks.
module WriterThreadPool
# :stopdoc:
include Base
# used to wrap a BasicSocket to use with +q+ for all writes
# this is compatible with IO.select
class QueueSocket < Struct.new(:to_io, :q) # :nodoc:
def kgio_addr
to_io.kgio_addr
end
def kgio_read(size, buf = "")
to_io.kgio_read(size, buf)
end
def kgio_read!(size, buf = "")
to_io.kgio_read!(size, buf)
end
def kgio_trywrite(buf)
to_io.kgio_trywrite(buf)
end
def timed_read(buf)
to_io.timed_read(buf)
end
def write(buf)
q << [ to_io, buf ]
end
def close
q << [ to_io, :close ]
end
def closed?
false
end
end
@@nr = 0
@@q = nil
def async_write_body(qclient, body, range)
qclient.q << [ qclient.to_io, :body, body, range ]
end
def process_client(client) # :nodoc:
@@nr += 1
super(QueueSocket.new(client, @@q[@@nr %= @@q.size]))
end
def init_worker_process(worker)
super
self.class.__send__(:alias_method, :sync_write_body, :write_body)
WriterThreadPool.__send__(:alias_method, :write_body, :async_write_body)
end
def worker_loop(worker) # :nodoc:
# we have multiple, single-thread queues since we don't want to
# interleave writes from the same client
qp = (1..worker_connections).map do |n|
QueuePool.new(1) do |response|
begin
io, arg1, arg2, arg3 = response
case arg1
when :body then sync_write_body(io, arg2, arg3)
when :close then io.close unless io.closed?
else
io.write(arg1)
end
rescue => err
Error.write(io, err)
end
end
end
@@q = qp.map { |q| q.queue }
super(worker) # accept loop from Unicorn
qp.map { |q| q.quit! }
end
# :startdoc:
end
end
|