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
| | # -*- encoding: binary -*-
# Copyright (C) 2013, Eric Wong <normalperson@yhbt.net> et. al.
# License: GPLv3 or later (https://www.gnu.org/licenses/gpl-3.0.txt)
require_relative 'wbuf_common'
# This class is triggered whenever we need write buffering for clients
# reading responses slowly. Small responses which fit into kernel socket
# buffers do not trigger this. yahns will always attempt to write to kernel
# socket buffers first to avoid unnecessary copies in userspace.
#
# Thus, most data is into copied to the kernel only once, the kernel
# will perform zero-copy I/O from the page cache to the socket. The
# only time data may be copied twice is the initial write()/send()
# which triggers EAGAIN.
#
# We only buffer to the filesystem (note: likely not a disk, since this
# is short-lived). We let the sysadmin/kernel decide whether or not
# the data needs to hit disk or not.
#
# This avoids large allocations from malloc, potentially limiting
# fragmentation and keeping (common) smaller allocations fast.
# General purpose malloc implementations in the 64-bit era tend to avoid
# releasing memory back to the kernel, so large heap allocations are best
# avoided as the kernel has little chance to reclaim memory used for a
# temporary buffer.
#
# The biggest downside of this approach is it requires an FD, but yahns
# configurations are configured for many FDs anyways, so it's unlikely
# to be a scalability issue.
class Yahns::Wbuf # :nodoc:
include Yahns::WbufCommon
def initialize(body, persist, tmpdir)
@tmpio = Yahns::TmpIO.new(tmpdir)
@tmpdir = tmpdir
@sf_offset = @sf_count = 0
@wbuf_persist = persist # whether or not we keep the connection alive
@body = body
@bypass = false
end
def wbuf_write(client, buf)
# try to bypass the VFS layer if we're all caught up
case rv = client.kgio_trywrite(buf)
when String
buf = rv # retry in loop
when nil
return # yay! hopefully we don't have to buffer again
when :wait_writable, :wait_readable
@bypass = false # ugh, continue to buffering to file
end while @bypass
@sf_count += @tmpio.write(buf)
case rv = client.trysendfile(@tmpio, @sf_offset, @sf_count)
when Integer
@sf_count -= rv
@sf_offset += rv
when :wait_writable, :wait_readable
return rv
else
raise "BUG: #{rv.nil ? "EOF" : rv.inspect} on tmpio " \
"sf_offset=#@sf_offset sf_count=#@sf_count"
end while @sf_count > 0
# we're all caught up, try to prevent dirty data from getting flushed
# to disk if we can help it.
@tmpio.close
@sf_offset = 0
@tmpio = Yahns::TmpIO.new(@tmpdir)
@bypass = true
nil
end
# called by last wbuf_flush
def wbuf_close(client)
@tmpio = @tmpio.close
wbuf_close_common(client)
end
end
|