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

davidhenrygao · 2024-10-10T12:16:07Z

davidhenrygao
Oct 10, 2024

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

stream.connected 赋值为 nil 和 false 的目的是什么？stream_length 和 stream_read 两函数的处理似乎不太一致？
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 = tonu
8000
mber(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 · 2024-10-10T12:29:21Z

davidhenrygao
Oct 10, 2024
Author

另外，因 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

3 replies

davidhenrygao Oct 11, 2024
Author

我自己写了 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 Oct 12, 2024
Maintainer

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

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

davidhenrygao Oct 12, 2024
Author

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

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

cloudwu · 2024-10-10T13:13:34Z

cloudwu
Oct 10, 2024
Maintainer

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

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

15 replies

cloudwu Oct 12, 2024
Maintainer

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

cloudwu Oct 12, 2024
Maintainer

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

davidhenrygao Oct 12, 2024
Author

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

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

davidhenrygao Oct 12, 2024
Author

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

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

那这个还要改吗？

davidhenrygao Oct 12, 2024
Author

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

cloudwu · 2024-10-10T13:19:06Z

cloudwu
Oct 10, 2024
Maintainer

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

0 replies

cloudwu · 2024-10-12T09:05:50Z

cloudwu
Oct 12, 2024
Maintainer

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

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

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

2 replies

cloudwu Oct 12, 2024
Maintainer

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

davidhenrygao Oct 12, 2024
Author

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

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

Uh oh!

Uh oh!

davidhenrygao Oct 10, 2024

Replies: 4 comments · 20 replies

Uh oh!

davidhenrygao Oct 10, 2024 Author

Uh oh!

davidhenrygao Oct 11, 2024 Author

Uh oh!

cloudwu Oct 12, 2024 Maintainer

Uh oh!

davidhenrygao Oct 12, 2024 Author

Uh oh!

cloudwu Oct 10, 2024 Maintainer

Uh oh!

cloudwu Oct 12, 2024 Maintainer

Uh oh!

cloudwu Oct 12, 2024 Maintainer

Uh oh!

davidhenrygao Oct 12, 2024 Author

Uh oh!

davidhenrygao Oct 12, 2024 Author

Uh oh!

davidhenrygao Oct 12, 2024 Author

Uh oh!

cloudwu Oct 10, 2024 Maintainer

Uh oh!

cloudwu Oct 12, 2024 Maintainer

Uh oh!

cloudwu Oct 12, 2024 Maintainer

Uh oh!

Uh oh!

davidhenrygao Oct 12, 2024 Author

davidhenrygao
Oct 10, 2024

Replies: 4 comments 20 replies

davidhenrygao
Oct 10, 2024
Author

davidhenrygao Oct 11, 2024
Author

cloudwu Oct 12, 2024
Maintainer

davidhenrygao Oct 12, 2024
Author

cloudwu
Oct 10, 2024
Maintainer

cloudwu Oct 12, 2024
Maintainer

cloudwu Oct 12, 2024
Maintainer

davidhenrygao Oct 12, 2024
Author

davidhenrygao Oct 12, 2024
Author

davidhenrygao Oct 12, 2024
Author

cloudwu
Oct 10, 2024
Maintainer

cloudwu
Oct 12, 2024
Maintainer

cloudwu Oct 12, 2024
Maintainer

davidhenrygao Oct 12, 2024
Author