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
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
| | # Copyright (C) 2013, Eric Wong <normalperson@yhbt.net> and all contributors
# License: GPLv3 or later (https://www.gnu.org/licenses/gpl-3.0.txt)
require_relative 'server_helper'
class TestServer < Testcase
parallelize_me!
include ServerHelper
alias setup server_helper_setup
alias teardown server_helper_teardown
def test_single_process
err = @err
cfg = Yahns::Config.new
host, port = @srv.addr[3], @srv.addr[1]
cfg.instance_eval do
ru = lambda { |_| [ 200, {'Content-Length'=>'2'}, ['HI'] ] }
GTL.synchronize { app(:rack, ru) { listen "#{host}:#{port}" } }
logger(Logger.new(err.path))
end
srv = Yahns::Server.new(cfg)
pid = fork do
ENV["YAHNS_FD"] = @srv.fileno.to_s
srv.start.join
end
run_client(host, port) { |res| assert_equal "HI", res.body }
c = TCPSocket.new(host, port)
# test pipelining
r = "GET / HTTP/1.1\r\nHost: example.com\r\n\r\n"
c.write(r + r)
buf = ""
Timeout.timeout(10) do
until buf =~ /HI.+HI/m
buf << c.readpartial(4096)
end
end
# trickle pipelining
c.write(r + "GET ")
buf = ""
Timeout.timeout(10) do
until buf =~ /HI\z/
buf << c.readpartial(4096)
end
end
c.write("/ HTTP/1.1\r\nHost: example.com\r\n\r\n")
Timeout.timeout(10) do
until buf =~ /HI.+HI/m
buf << c.readpartial(4096)
end
end
Process.kill(:QUIT, pid)
"GET / HTTP/1.1\r\nHost: example.com\r\n\r\n".each_byte do |x|
sleep(0.01)
c.write(x.chr)
end
buf = Timeout.timeout(30) { c.read }
assert_match(/Connection: close/, buf)
_, status = Timeout.timeout(10) { Process.waitpid2(pid) }
assert status.success?, status.inspect
c.close
rescue => e
Yahns::Log.exception(Logger.new($stderr), "test", e)
raise
end
def test_input_body_true; input_body(true); end
def test_input_body_false; input_body(false); end
def test_input_body_lazy; input_body(:lazy); end
def input_body(btype)
err = @err
cfg = Yahns::Config.new
host, port = @srv.addr[3], @srv.addr[1]
cfg.instance_eval do
ru = lambda {|e|[ 200, {'Content-Length'=>'2'},[e["rack.input"].read]]}
GTL.synchronize do
app(:rack, ru) do
listen "#{host}:#{port}"
input_buffering btype
end
end
logger(Logger.new(err.path))
end
srv = Yahns::Server.new(cfg)
pid = fork do
ENV["YAHNS_FD"] = @srv.fileno.to_s
srv.start.join
end
c = TCPSocket.new(host, port)
buf = "PUT / HTTP/1.0\r\nContent-Length: 2\r\n\r\nHI"
c.write(buf)
IO.select([c], nil, nil, 5)
rv = c.read(666)
head, body = rv.split(/\r\n\r\n/)
assert_match(%r{^Content-Length: 2\r\n}, head)
assert_equal "HI", body, "#{rv.inspect} - #{btype.inspect}"
c.close
# pipelined oneshot
buf = "PUT / HTTP/1.1\r\nContent-Length: 2\r\n\r\nHI"
c = TCPSocket.new(host, port)
c.write(buf + buf)
buf = ""
Timeout.timeout(10) do
until buf =~ /HI.+HI/m
buf << c.readpartial(4096)
end
end
assert buf.gsub!(/Date:[^\r\n]+\r\n/, ""), "kill differing Date"
rv = buf.sub!(/\A(HTTP.+?\r\n\r\nHI)/m, "")
first = $1
assert rv
assert_equal first, buf
# pipelined trickle
buf = "PUT / HTTP/1.1\r\nContent-Length: 5\r\n\r\nHIBYE"
(buf + buf).each_byte do |b|
c.write(b.chr)
sleep(0.01) if b.chr == ":"
Thread.pass
end
buf = ""
Timeout.timeout(10) do
until buf =~ /HIBYE.+HIBYE/m
buf << c.readpartial(4096)
end
end
assert buf.gsub!(/Date:[^\r\n]+\r\n/, ""), "kill differing Date"
rv = buf.sub!(/\A(HTTP.+?\r\n\r\nHIBYE)/m, "")
first = $1
assert rv
assert_equal first, buf
rescue => e
Yahns::Log.exception(Logger.new($stderr), "test", e)
raise
ensure
c.close if c
quit_wait(pid)
end
def test_trailer_true; trailer(true); end
def test_trailer_false; trailer(false); end
def test_trailer_lazy; trailer(:lazy); end
def test_slow_trailer_true; trailer(true, 0.02); end
def test_slow_trailer_false; trailer(false, 0.02); end
def test_slow_trailer_lazy; trailer(:lazy, 0.02); end
def trailer(btype, delay = false)
err = @err
cfg = Yahns::Config.new
host, port = @srv.addr[3], @srv.addr[1]
cfg.instance_eval do
ru = lambda do |e|
body = e["rack.input"].read
s = e["HTTP_XBT"] + "\n" + body
[ 200, {'Content-Length'=>s.size.to_s}, [ s ] ]
end
GTL.synchronize do
app(:rack, ru) do
listen "#{host}:#{port}"
input_buffering btype
end
end
logger(Logger.new(err.path))
end
srv = Yahns::Server.new(cfg)
pid = fork do
ENV["YAHNS_FD"] = @srv.fileno.to_s
srv.start.join
end
c = TCPSocket.new(host, port)
buf = "PUT / HTTP/1.0\r\nTrailer:xbt\r\nTransfer-Encoding: chunked\r\n\r\n"
c.write(buf)
xbt = btype.to_s
sleep(delay) if delay
c.write(sprintf("%x\r\n", xbt.size))
sleep(delay) if delay
c.write(xbt)
sleep(delay) if delay
c.write("\r\n")
sleep(delay) if delay
c.write("0\r\nXBT: ")
sleep(delay) if delay
c.write("#{xbt}\r\n\r\n")
IO.select([c], nil, nil, 5000) or raise "timed out"
rv = c.read(666)
_, body = rv.split(/\r\n\r\n/)
a, b = body.split(/\n/)
assert_equal xbt, a
assert_equal xbt, b
ensure
c.close if c
quit_wait(pid)
end
def test_check_client_connection
msgs = %w(ZZ zz)
err = @err
cfg = Yahns::Config.new
bpipe = IO.pipe
host, port = @srv.addr[3], @srv.addr[1]
cfg.instance_eval do
ru = lambda { |e|
case e['PATH_INFO']
when '/sleep'
a = Object.new
a.instance_variable_set(:@bpipe, bpipe)
a.instance_variable_set(:@msgs, msgs)
def a.each
@msgs.each do |msg|
yield @bpipe[0].read(msg.size)
end
end
when '/cccfail'
# we should not get here if check_client_connection worked
abort "CCCFAIL"
else
a = %w(HI)
end
[ 200, {'Content-Length'=>'2'}, a ]
}
GTL.synchronize {
app(:rack, ru) {
listen "#{host}:#{port}"
check_client_connection true
# needed to avoid concurrency with check_client_connection
queue { worker_threads 1 }
output_buffering false
}
}
logger(Logger.new(err.path))
end
srv = Yahns::Server.new(cfg)
# ensure we set worker_threads correctly
eggs = srv.instance_variable_get(:@config).qeggs
assert_equal 1, eggs.size
assert_equal 1, eggs[:default].instance_variable_get(:@worker_threads)
pid = fork do
bpipe[1].close
ENV["YAHNS_FD"] = @srv.fileno.to_s
srv.start.join
end
bpipe[0].close
a = TCPSocket.new(host, port)
b = TCPSocket.new(host, port)
a.write("GET /sleep HTTP/1.0\r\n\r\n")
r = IO.select([a], nil, nil, 4)
assert r, "nothing ready"
assert_equal a, r[0][0]
buf = a.read(8)
assert_equal "HTTP/1.1", buf
# hope the kernel sees this before it sees the bpipe ping-ponging below
b.write("GET /cccfail HTTP/1.0\r\n\r\n")
b.shutdown
b.close
# ping-pong a bit to stall the server
msgs.each do |msg|
bpipe[1].write(msg)
Timeout.timeout(10) { buf << a.readpartial(10) until buf =~ /#{msg}/ }
end
bpipe[1].close
assert_equal msgs.join, buf.split(/\r\n\r\n/)[1]
# do things still work?
run_client(host, port) { |res| assert_equal "HI", res.body }
a.close
ensure
quit_wait(pid)
end
def test_mp
pid, host, port = new_mp_server
wpid = nil
run_client(host, port) do |res|
wpid ||= res.body.to_i
end
ensure
quit_wait(pid)
if wpid
assert_raises(Errno::ESRCH) { Process.kill(:KILL, wpid) }
assert_raises(Errno::ECHILD) { Process.waitpid2(wpid) }
end
end
# Linux blocking accept() has fair behavior between multiple tasks
def test_mp_balance
skip("this fails occasionally on Linux, still...")
skip("linux-only test") unless RUBY_PLATFORM =~ /linux/
pid, host, port = new_mp_server(2)
seen = {}
# wait for both processes to spin up
Timeout.timeout(10) do
run_client(host, port) { |res| seen[res.body] = 1 } until seen.size == 2
end
prev = nil
req = Net::HTTP::Get.new("/")
# we should bounce new connections between 2 processes
4.times do
Net::HTTP.start(host, port) do |http|
res = http.request(req)
assert_equal 200, res.code.to_i
assert_equal "keep-alive", res["Connection"]
refute_equal prev, res.body, "same PID accepted twice"
prev = res.body.dup
seen[prev] += 1
666.times { Thread.pass } # have the other acceptor to wake up
end
end
assert_equal 2, seen.size
ensure
quit_wait(pid)
end
def test_mp_worker_die
pid, host, port = new_mp_server
wpid1 = wpid2 = nil
run_client(host, port) do |res|
wpid1 ||= res.body.to_i
end
Process.kill(:QUIT, wpid1)
poke_until_dead(wpid1)
run_client(host, port) do |res|
wpid2 ||= res.body.to_i
end
refute_equal wpid2, wpid1
ensure
quit_wait(pid)
assert_raises(Errno::ESRCH) { Process.kill(:KILL, wpid2) } if wpid2
end
def test_mp_dead_parent
pid, host, port = new_mp_server
wpid = nil
run_client(host, port) do |res|
wpid ||= res.body.to_i
end
Process.kill(:KILL, pid)
_, status = Process.waitpid2(pid)
assert status.signaled?, status.inspect
poke_until_dead(wpid)
end
def run_client(host, port)
c = TCPSocket.new(host, port)
Net::HTTP.start(host, port) do |http|
res = http.request(Net::HTTP::Get.new("/"))
assert_equal 200, res.code.to_i
assert_equal "keep-alive", res["Connection"]
yield res
res = http.request(Net::HTTP::Get.new("/"))
assert_equal 200, res.code.to_i
assert_equal "keep-alive", res["Connection"]
yield res
end
c.write "GET / HTTP/1.0\r\n\r\n"
res = Timeout.timeout(10) { c.read }
head, _ = res.split(/\r\n\r\n/)
head = head.split(/\r\n/)
assert_equal "HTTP/1.1 200 OK", head[0]
assert_equal "Connection: close", head[-1]
c.close
end
def new_mp_server(nr = 1)
ru = @ru = tmpfile(%w(config .ru))
@ru.puts('a = $$.to_s')
@ru.puts('run lambda { |_| [ 200, {"Content-Length"=>a.size.to_s},[a]]}')
err = @err
cfg = Yahns::Config.new
host, port = @srv.addr[3], @srv.addr[1]
cfg.instance_eval do
worker_processes 2
GTL.synchronize { app(:rack, ru.path) { listen "#{host}:#{port}" } }
logger(Logger.new(File.open(err.path, "a")))
end
srv = Yahns::Server.new(cfg)
pid = fork do
ENV["YAHNS_FD"] = @srv.fileno.to_s
srv.start.join
end
[ pid, host, port ]
end
def test_nonpersistent
err = @err
cfg = Yahns::Config.new
host, port = @srv.addr[3], @srv.addr[1]
cfg.instance_eval do
ru = lambda { |_| [ 200, {'Content-Length'=>'2'}, ['HI'] ] }
GTL.synchronize {
app(:rack, ru) {
listen "#{host}:#{port}"
persistent_connections false
}
}
logger(Logger.new(err.path))
end
srv = Yahns::Server.new(cfg)
pid = fork do
ENV["YAHNS_FD"] = @srv.fileno.to_s
srv.start.join
end
c = TCPSocket.new(host, port)
c.write("GET / HTTP/1.1\r\nHost: example.com\r\n\r\n")
buf = Timeout.timeout(10) { c.read }
assert_match(/Connection: close/, buf)
c.close
ensure
quit_wait(pid)
end
end
|