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
| | # -*- encoding: binary -*-
require 'thread'
# Soft timeout middleware for thread-based concurrency models in \Rainbows!
# This timeout only includes application dispatch, and will not take into
# account the (rare) response bodies that are dynamically generated while
# they are being written out to the client.
#
# In your rackup config file (config.ru), the following line will
# cause execution to timeout in 1.5 seconds.
#
# use Rainbows::ThreadTimeout, :timeout => 1.5
# run MyApplication.new
#
# You may also specify a threshold, so the timeout does not take
# effect until there are enough active clients. It does not make
# sense to set a +:threshold+ higher or equal to the
# +worker_connections+ \Rainbows! configuration parameter.
# You may specify a negative threshold to be an absolute
# value relative to the +worker_connections+ parameter, thus
# if you specify a threshold of -1, and have 100 worker_connections,
# ThreadTimeout will only activate when there are 99 active requests.
#
# use Rainbows::ThreadTimeout, :timeout => 1.5, :threshold => -1
# run MyApplication.new
#
# This middleware only affects elements below it in the stack, so
# it can be configured to ignore certain endpoints or middlewares.
#
# Timed-out requests will cause this middleware to return with a
# "408 Request Timeout" response.
class Rainbows::ThreadTimeout
# :stopdoc:
class ExecutionExpired < Exception
end
def initialize(app, opts)
@timeout = opts[:timeout]
Numeric === @timeout or
raise TypeError, "timeout=#{@timeout.inspect} is not numeric"
if @threshold = opts[:threshold]
Integer === @threshold or
raise TypeError, "threshold=#{@threshold.inspect} is not an integer"
@threshold == 0 and
raise ArgumentError, "threshold=0 does not make sense"
@threshold < 0 and
@threshold += Rainbows.server.worker_connections
end
@app = app
@active = {}
@lock = Mutex.new
end
def call(env)
@lock.synchronize do
start_watchdog unless @watchdog
@active[Thread.current] = Time.now + @timeout
end
begin
@app.call(env)
ensure
@lock.synchronize { @active.delete(Thread.current) }
end
rescue ExecutionExpired
[ 408, { 'Content-Type' => 'text/plain', 'Content-Length' => '0' }, [] ]
end
def start_watchdog
@watchdog = Thread.new do
begin
if next_wake = @lock.synchronize { @active.values }.min
next_wake -= Time.now
# because of the lack of GVL-releasing syscalls in this branch
# of the thread loop, we need Thread.pass to ensure other threads
# get scheduled appropriately under 1.9. This is likely a threading
# bug in 1.9 that warrants further investigation when we're in a
# better mood.
next_wake > 0 ? sleep(next_wake) : Thread.pass
else
sleep(@timeout)
end
# "active.size" is atomic in MRI 1.8 and 1.9
next if @threshold && @active.size < @threshold
now = Time.now
@lock.synchronize do
@active.delete_if do |thread, time|
now >= time and thread.raise(ExecutionExpired).nil?
end
end
end while true
end
end
# :startdoc:
end
|