对 skynet http stream 代码的一些疑问，希望大家指点和探讨一下，谢谢

[davidhenrygao]

新项目引入 etcd 组件，watch 监听需要使用 skynet http stream，由于 wiki 上没有相关说明，所以详细看了相关代码，有两点疑问想请教一下：

1. stream.connected 赋值为 nil 和 false 的目的是什么？stream_length 和 stream_read 两函数的处理似乎不太一致？
2. stream_all 使用 stream_read 进行 padding 读取，而不是使用 stream._interface.readall 进行一次性读取的原因？

这里的说明 #1463 (comment) 结合代码，个人理解 stream.connected = nil 时应该是表示正常连接关闭的情况，而 stream.connected = false 时表示出错的情况。

对于 stream._interface.read 函数，对应 http/sockethelper.lua 中 sockethelper.readfunc return 的闭包函数，连接正常关闭时，stream._interface.read()->readbytes()->socket.read() 会返回 false, ret ，根据代码，应该会直接 error 报错，则 stream_all 最后的读取会报错。不知道我理解是否有误？

response_stream代码如下，mode == "chunked" 会使用 stream_chunked 分片读取，其他情况，若有 "content-length" 头字段表明数据长度，使用 stream_length(length) 读取指定长度数据，否则再根据情况不读取 stream_nobody 或读取所有剩余数据 stream_all。

function M.response_stream(interface, code, body, header)
	local mode = header["transfer-encoding"]
	if mode then
		if mode ~= "identity" and mode ~= "chunked" then
			error ("Unsupport transfer-encoding")
		end
	end


	local read_func


	if mode == "chunked" then
		read_func = stream_chunked
	else
		-- identity mode
		local length = header["content-length"]
		if length then
			length = tonumber(length)
		end
		if length then
			read_func = stream_length(length)
		elseif code == 204 or code == 304 or code < 200 then
			read_func = stream_nobody
		else
			read_func = stream_all
		end
	end


	-- todo: timeout


	return setmetatable({
		status = code,
		_body = body,
		_interface = interface,
		_reading = read_func,
		header = header,
		connected = true,
	}, stream)
end

stream_length 、stream_all 和 stream_read 代码如下：

local function stream_length(length)
	return function(stream)
		local body = stream._body
		if body == nil then
			local ret, padding = stream._interface.read()
			if not ret then
				-- disconnected
				body = padding
				stream.connected = false
			else
				body = ret
			end
		end
		local n = #body
		if n >= length then
			stream._reading = stream.close
			stream.connected = nil
			return (body:sub(1,length))
		else
			length = length - n
			stream._body = nil
			if not stream.connected then
				stream._reading = stream.close
			end
			return body
		end
	end
end


local function stream_read(stream)
	local ret, padding = stream._interface.read()
	if ret == "" or not ret then
		stream.connected = nil
		stream:close()
		if padding == "" then
			return
		end
		return padding
	end
	return ret
end


local function stream_all(stream)
	local body = stream._body
	stream._body = nil
	stream._reading = stream_read
	return body
end

sockethelper.readfunc 代码如下：

function sockethelper.readfunc(fd, pre)
	if pre then
		return preread(fd, pre)
	end
	return function (sz)
		local ret = readbytes(fd, sz)
		if ret then
			return ret
		else
			error(socket_error("read failed fd = " .. fd))
		end
	end
end

@cloudwu 不清楚云大是否有时间指点一下？
或者其他熟悉相关代码的同学，若我理解有误，也恳请指点和指正一下，谢谢！

[davidhenrygao]

另外，因 http stream 的疑问查看相关代码，产生了另外一个疑问：当调用 socket.readall 时，s.read_required = true 被设置，那么在 SKYNET_SOCKET_TYPE_DATA 事件被处理时，假如数据大小超过 BUFFER_LIMIT，根据代码，应该会 pause_socket 屏蔽了底层的 EPOLLIN 事件，那么在连接断开前，都不会被唤醒了？我的理解是这样，不知道是否正确？

function socket.readall(id)
	local s = socket_pool[id]
	assert(s)
	if not s.connected then
		local r = driver.readall(s.buffer, s.pool)
		return r ~= "" and r
	end
	assert(not s.read_required)
	s.read_required = true
	suspend(s)
	assert(s.connected == false)
	return driver.readall(s.buffer, s.pool)
end

-- SKYNET_SOCKET_TYPE_DATA = 1
socket_message[1] = function(id, size, data)
	local s = socket_pool[id]
	if s == nil then
		skynet.error("socket: drop package from " .. id)
		driver.drop(data, size)
		return
	end


	local sz = driver.push(s.buffer, s.pool, data, size)
	local rr = s.read_required
	local rrt = type(rr)
	if rrt == "number" then
		-- read size
		if sz >= rr then
			s.read_required = nil
			if sz > BUFFER_LIMIT then
				pause_socket(s, sz)
			end
			wakeup(s)
		end
	else
		if s.buffer_limit and sz > s.buffer_limit then
			skynet.error(string.format("socket buffer overflow: fd=%d size=%d", id , sz))
			driver.close(id)
			return
		end
		if rrt == "string" then
			-- read line
			if driver.readline(s.buffer,nil,rr) then
				s.read_required = nil
				if sz > BUFFER_LIMIT then
					pause_socket(s, sz)
				end
				wakeup(s)
			end
		elseif sz > BUFFER_LIMIT and not s.pause then
			pause_socket(s, sz)
		end
	end
end

[davidhenrygao]

我自己写了 client 和 server 服务测试了一下我的疑问，代码就先不贴了，免得又说不清楚。

大概是 client 连接上 server 后，发送一个整数 n 给 server，然后使用 socket.readall(id) 进行阻塞读取； server 收到请求后，返回 n KB字节数据给 client。当 client 发送 n = 1024 时，打印显示会进入 pause_socket 函数，底层会屏蔽了 EPOLLIN 事件。server 发送完数据后，socket.close 关闭连接，tcpdump 显示有发送 FIN 报文到 client，但此时 client 会一直阻塞，因为不会有 EPOLLIN 事件，也不会有 EPOLLHUP 事件。

@cloudwu 云大能不能顺道解答一下？这里调用 pause_socket 的目的是什么？

skynet/lualib/skynet/socket.lua

Lines 99 to 101 in 8900360

	elseif sz > BUFFER_LIMIT and not s.pause then
	pause_socket(s, sz)
	end

[cloudwu]

这里 BUFFER_LIMIT 其实是一个防呆设定，来源于早期对 readall 设计的考虑不周。readall 早于 pause 机制实现。

因为 readall 的语义是读 fd 直到对端关闭，中间没有干涉的手段。如果对端不友好，那么只要一直发送数据而不关闭，最终接收方内存一定会耗尽。这会是一个非常简单的攻击手段。这也是为什么前面提到不建议使用 readall 的原因。

[davidhenrygao]

readall 设计上确实欠佳，但目前使用 pause_socket 进行约束，感觉对尚存的使用者不太友好，因为 pause 后就无法恢复了。

在目前 readall 函数向前兼容并存在的情况下，我的一个想法是，增加一个可选参数 limit，默认值为 BUFFER_LIMIIT，超过 limit 值就直接报错，因为目前 pause 后其实也无法恢复。

[cloudwu]

我觉得这里给的 interface 是个未实现的功能组，应该是依赖数量越少越好。如果可以不依赖 readall 就不应该依赖。readall 本身也是一个不太正常的功能，我认为应该回避使用。

另外 steam 既然是流处理，也不应该把整个流的数据全部读完 (all) 再处理。

[cloudwu]

我指 http 服务端发送完数据就断开，可能产生 timewait ；反之 http server 不主动断开，而是 client 认为自己获得了足够数据后断开，则不会。所以我倾向于避免使用连接断开作为明确的流结束的信号。

[cloudwu]

现在已经几乎不存在还在使用 http 1.0 协议的场所了，即内容长度是必填项。所以我觉得这里的问题不是什么大问题。

[davidhenrygao]

当然，如果接受报错分支正确处理 error ，我想也可以把事情做完。

这是指在 else 分支再判断是否报 error ？还是使用者外层报错处理？使用者无法处理吧？

[davidhenrygao]

现在已经几乎不存在还在使用 http 1.0 协议的场所了，即内容长度是必填项。所以我觉得这里的问题不是什么大问题。

的确是一些几乎不会遇到的问题，一开始就是代码里面看到了，感觉没有内容长度的时候，是会报错的，所以在这里问问看。

那这个还要改吗？

[davidhenrygao]

那其实是不是就不要兼容 http1.0 没有长度头的情况了？不要 stream_all ，直接 close 连接，不读 body 算了。

[cloudwu]

stream.connected 为 nil 时，这个 http 流是正常结束的，false 指某个环节的数据不正常。

[cloudwu]

我想了一下，目前解决这个问题伤害最小的改法还是加一个 interface.readany ，然后让 stream api 使用它。但以后在别的模块使用 readany 应该斟酌。

这样可以最小程度影响已有代码的行为，同时解决这个指定问题。

如果想兼容再好一些，当 readany 不存在时，转发到 read 更好。

[cloudwu]

在处理这些非常见情况的问题时，我倾向于增加一些代码处理特例，避免修改已有代码的行为。

[davidhenrygao]

确实，刚刚在 tls 那边看到有不带 sz 的调用，判断 sz 为 nil 的改法会有影响。