pipeline支持

[lesismal]

尝试了下http client pipeline，同一个write发送两个http request的数据，代码如下：

package main

import (
	"fmt"
	"net"
)

func main() {
	addr := "localhost:9000"
	conn, err := net.Dial("tcp", addr)
	if err != nil {
		panic(err)
	}

	// 一个请求的数据
	reqData := []byte(fmt.Sprintf("POST / HTTP/1.1\r\nHost: %v\r\nContent-Length: 5\r\n\r\n\r\nhello", addr))
	// 两个请求一起发送
	reqData = append(reqData, reqData...)

	fmt.Println("---------------------")
	n, err := conn.Write(reqData)
	if err != nil {
		panic(err)
	}
	fmt.Println("write:\n", string(reqData))
	fmt.Println("---------------------")
	defer fmt.Println("---------------------")
	resData := make([]byte, 1024*8)
	fmt.Println("read:", n)
	for {
		n, err = conn.Read(resData)
		if err != nil {
			fmt.Println("read failed:", n, err)
			return
		}
		fmt.Println(string(resData[:n]))
	}
}

output:

root@k8s:~/workflow/benchmark# go run ./client.go 
---------------------
write:
 GET / HTTP/1.1
Host: localhost:9000

GET / HTTP/1.1
Host: localhost:9000


---------------------
read:
HTTP/1.1 200 OK
Date: Sun, 15 May 2022 06:35:29 UTC
Content-Type: text/plain; charset=UTF-8
Content-Length: 32
Connection: Keep-Alive

g\g,u(VC:E!Qa\nygJ$&g.H)-(RJi+f>
read failed: 0 EOF

返回了一个响应后连接就被关闭了。

[lesismal]

用golang标准库server则能正常返回并且连接没有被关闭：

package main

import (
	"net/http"
	"sync/atomic"
	"time"
)

var (
	cnt = int64(0)
)

type HttpHandler struct{}

func (h *HttpHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	ret := ""
	if atomic.AddInt64(&cnt, 1)%2 == 1 {
		ret = "aaaaa"
		time.Sleep(time.Second)
	} else {
		ret = "bbbbb"
	}
	w.Header().Add("Date", time.Now().Format(time.RFC1123)) //"Mon, 02 Jan 2006 15:04:05 MST"
	w.Header().Add("Content-Type", "text/plain; charset=UTF-8")
	w.Write([]byte(ret))
}

func main() {

	server := http.Server{
		Addr:    "localhost:9000",
		Handler: &HttpHandler{},
	}
	server.ListenAndServe()

}

output:

root@k8s:~/workflow/benchmark# go run ./client.go 
---------------------
write:
 GET / HTTP/1.1
Host: localhost:9000

GET / HTTP/1.1
Host: localhost:9000


---------------------
read:
HTTP/1.1 200 OK
Content-Type: text/plain; charset=UTF-8
Date: Sun, 15 May 2022 06:39:04 UTC
Content-Length: 5

aaaaaHTTP/1.1 200 OK
Content-Type: text/plain; charset=UTF-8
Date: Sun, 15 May 2022 06:39:04 UTC
Content-Length: 5

bbbbb

client连接挂起、并未被关闭

[lesismal]

pipeline应该算是http1.1 server的一项标准功能，主流浏览器默认未开启，但是server端还是支持上比较好：
https://developer.mozilla.org/zh-CN/docs/Web/HTTP/Connection_management_in_HTTP_1.x

另外，这种单个连接上的pipeline msg，在需要有序返回的场景，也依赖于msg派发给逻辑线程池后的执行顺序，我今天读了workflow一点代码，逻辑线程池本身是mux+cond调度的，msgqueue派发时应该是无法指定线程的。所以如果支持pipeline并且按照msg的粒度put到msgqueue里就没法保证单个连接上的多个msg有序，但是如果msgqueue里是连接、连接派发给逻辑线程执行时再由逻辑线程判断当前连接上的msg队列，则能保证有序，并且msg加入到连接自己的队列时还可以判断当前msg是否为head，如果是head才需要去触发mux+cond唤醒并取得逻辑线程，如果不是head则无需触发。
workflow派发相关的逻辑我还没读太懂，没找到是按msg还是按连接去派发给逻辑线程的。我自己的框架里是这样子的：
https://github.com/lesismal/nbio/blob/master/conn.go#L70

http1.x这种每个请求不包括本次请求的session id所以依赖于回包顺序的比较恼火，server、client端的实现都比较秀发不友好

[Barenboim]

pipeline是我们一开始就放弃的功能。支持pipeline不如支持streaming。
因为我们不是http server（你可以看一下我们的WFRedisServer，这个是完备的，但也不支持pipeline），所以pipeline的话，不能只为http/1.1服务。http/1.1对pipeline的定义是，server端并行执行请求，按顺序返回。由client来保证幂等性，也就是说两个POST或POST+GET不应该发pipeline。
但是，像redis和thrift协议，就要求server端严格串行执行，否则肯定是错误。
然后，brpc的pipeline又可以并行执行，乱序返回。另外有些协议pipeline又是完全没意义的，最好直接错误。可以说pipeline的模式没有一个general定义。
如果我们同时支持这些，server和client的配置都会非常复杂。如果我们只支持最保守的redis模式，那么，http client的pipeline请求，性能就会受影响，因为一个pipeline上的GET被串行执行了。
最关键的是，http/1.1的pineline基本上只是一个用来跑benchmark的，实际应用中有很多问题，后端应用里基本没用。为了支持它，牺牲正常http性能，没有这个必要。
另外，我们支持client发pipeline请求，例如这个项目，就是workflow的pipeline redis client：
https://github.com/kedixa/workflow-extra

[lesismal]

http/1.1对pipeline的定义是，server端并行执行请求，按顺序返回

这个我不太同意，我觉得应该不只是server端的实现，而是首先，client方的发送能力，比如mdn上的描述：

或者

其次，server端的回包有序。因为具体到http1.1上，每个请求没有自身的请求id、如果回的包无序则client无法区分当前响应的是哪个请求，所以只能依赖server回包有序。

我open issue时候的描述可能不够清楚，可以分成两件事来看：

1. http1.1的pipeline，这与其他协议无关，只是http1.1
2. 我提到有序是因为http1.1这种需要server支持pipeline而目前workflow的http实现没有支持、并且逻辑线程池是乱序执行的，如果想实现可能需要如何改造。

至于抛开http1.1、单就单个Conn的pipeline，无需等待上一个请求返回就能够发送其他请求的能力，这个应该是已经支持了的吧？至少通常的网络库，作为4层数据收发的处理，是不会太去影响7层协议如何编解码和考虑消息顺序的。只是workflow本身是4层做了、7层也做了并且直接向用户提供了7层的方法。比如像libevent那些，暴露的是4层的消息或者数据给用户，用户想怎么用这些事件、数据来parsing和处理顺序都是可以的。

[lesismal]

对于静态资源这种需求，pipeline不给力，http2.0 streaming也不行，pipeline解决了client端发送时必须等上一个返回的这个RTT太久的空窗期问题，2.0则允许server端乱序响应、进一步提高了单个tcp信道的利用率，然后瓶颈就在tcp信道本身了，单个连接的拥塞控制以及tcp stream本身的数据顺序保障又跟pipeline以前的一个等一个类似、成为了新的障碍。

所以才会有3.0，真正解锁了c/s两端的并发无序能力以及单一tcp信道的限制，剩下的就交给3层了。

对于非静态资源的接口类请求，pipeline还是能提升性能的，而且万一人家client开启了、server却不支持，就尴尬了呢。。

[lesismal]

nginx的c代码看起来更累，直接搜了一段：

nginx的处理就是相当于我前面说的以连接为单位去处理请求而不是以msg为单位

[lesismal]

我主要是觉得，pipeline是http1.1的一部分，如果不支持，相当于是功能缺失。

另外一个问题就是，如果不支持pipeline，http1.1 client端想提高吞吐时则只能通过更多连接数来解决，然后我们又来到了#906 ，需要继续优化连接数多了性能下降的问题:joy:

[Barenboim]

哈哈哈。其实至今为止，你是第一个提出http pipeline需求的。而且，也不是业务中遇到，主要还是讨论一个http协议支持完整性的问题。其实也说明了实际业务中不太用到😂
HTTP协议里有关于client什么情况下可以发送pipeline请求的描述。例如，PUT /A 和GET /A这两个请求，client不可以pipeline发送，否则行为可能不正确。但HTTP协议不限制server端并发处理一个pipeline上的请求。server当然也可以串行执行(redis模式。也是你提出的分发到一个线程上执行），但是性能就影响严重了。在后端业务了，其实这是一个坑。某些情况就必须要求client关闭pipeline。
workflow底层网络模块，不限制请求发送时序，可以全双工。Communicator层简化为req，resp一来一回。但也可改造为全双工，已经有用户自己的魔改版。
至于连接管理，这是workflow里实现得最好的。其实不存在connection过多的的问题，workflow的client会非常好的复用持久连接，实际业务中，几万qps经常也就是几百个连接，不会出现上万连接的情况。
当然，几万连接下的性能优化，是另一个问题 ：）

[lesismal]

golang标准库底层非阻塞但net.Conn提供的是阻塞读，读一个处理一个，所以自然就支持了完整的pipeline。
redis逻辑单线程，所以redis server也基本是自然就支持了。
网络库的部分本身就是支持读写或者事件本身、跟pipeline没有直接关系。只是workflow的http实现不支持pipeline，但改起来的话估计也不会有性能下降，因为可用减少cond的操作，说不定还能提升一点

[lesismal]

哈哈哈。其实至今为止，你是第一个提出http pipeline需求的。而且，也不是业务中遇到，主要还是讨论一个http协议支持完整性的问题。其实也说明了实际业务中不太用到😂

不支持问题不大，只是毕竟不太标准。。

[Barenboim]

改可以改，但有几个问题：
1、如果单线程处理，性能慢，不如让用户直接多连接。
2、如果并行处理，顺序返回，有许多行为定义无法完美。比如pipeline中间一个请求要求短连接，这个怎么处理？代码没有现在的干净。
3、现在相当于是一个状态明确的自动机。支持pipeline的话，性能没有办法做到无损失。现在的代码极端精简。
4、workflow要考虑其他协议……不能只为http服务。其实我们用户还比较多，也是因为我们选择性放弃一些功能，让用户上手简单。然后，有留给用户魔改的空间。

[lesismal]

server当然也可以串行执行(redis模式。也是你提出的分发到一个线程上执行），但是性能就影响严重了

其实框架层的消息分发与应用层的处理顺序，是两个不同的层，框架层默认无序了，是限制了自己的场景能力，因为应用层难再去让消息有序处理（应用层按消息id排序之类的成本更高），这相当于阉割了业务姿势。
但如果框架层本身提供的就是顺序的消息，应用层仍然可以选择用更多的业务线程池去处理，有序和无序都是用户自己可定制的，这能够支持更全面的业务模式。

之所以考虑这个，也是因为我在实现nbio的时候，最初也是无序，当我实现http，需要有序。甚至当我实现了http的有序后在其基础之上实现websocket时仍然是先无序地把websocket message无序地丢给用户，但是有时候，这样确实不好，因为限制了用户。
比如流媒体，乱序丢给应用层，已经处理完了后面一帧才轮到处理前一帧，那么我丢弃前面这帧吗？

单个连接的数据有序到达、框架对其msg的有序派发应该是框架提供给应用层的基础保障，其他的无序与并发让应用层自己根据需要来做。

[lesismal]

1、如果单线程处理，性能慢，不如让用户直接多连接。

有序不意味着整个进程的逻辑单线程，只是单个连接的有序，多个连接仍然是均衡地从线程池中拿活跃线程来执行各自连接上的消息队列，比如我之前提到的我这里的实现：conn.Execute ，而且对于workflow的用cond调度的线程池来说，单个连接只有当前消息是head时才需要触发这个cond的操作，如果单个连接同时有10个msg、第一个尚未处理完，则其他9个msg入队列都不需要触发cond的操作，单个conn上的一个链表或者数组的O(1)操作+head判断，相比于cond，反倒可能节约一点性能

[lesismal]

2、如果并行处理，顺序返回，有许多行为定义无法完美。比如pipeline中间一个请求要求短连接，这个怎么处理？代码没有现在的干净。

按照我上一楼说的方式，其实单个连接上的消息队列仍然是串行处理，但是每个消息丢给应用层时，应用层可以自行选择是否要把这个消息放到另外的线程池去异步处理从而并发

[lesismal]

3、现在相当于是一个状态明确的自动机。支持pipeline的话，性能没有办法做到无损失。现在的代码极端精简。

不会复杂太多的，只是每个Conn上挂个队列，每个msg解析出来判断是不是队列head，如果是就照常丢给msgqueue cond去给线程池里的一个线程处理，改动不会太大的

我没调试，cpp的读起来不太确认，这里应该是解析到一个完整消息后回调、丢给msgqueue然后逻辑线程池里去处理吧：
https://github.com/sogou/workflow/blob/master/src/kernel/poller.c#L370
如果是这样的话，大概只需在单个conn对应的struct/class上加个队列，然后把这里改下丢到conn自己的队列里去判断是否为head，是head才丢给逻辑线程池去循环处理

[lesismal]

4、workflow要考虑其他协议……不能只为http服务。其实我们用户还比较多，也是因为我们选择性放弃一些功能，让用户上手简单。然后，有留给用户魔改的空间。

如我前面解释的几条，真的不冲突，而是提供了更全面的基础能力，性能甚至不会下降反而因为减少了cond操作而提升，但数万连接时的性能下降我估计不是这个修改能提升上去的、可能还是有其他原因需要优化:joy:

[Barenboim]

4、workflow要考虑其他协议……不能只为http服务。其实我们用户还比较多，也是因为我们选择性放弃一些功能，让用户上手简单。然后，有留给用户魔改的空间。

如我前面解释的几条，真的不冲突，而是提供了更全面的基础能力，性能甚至不会下降反而提升😂

你说的都没错啊，我们可以提供这些选择，但问题是如果带来用户接口上的复杂性，结果反而损失更多的普通用户。这年头，用法复杂比功能欠缺更严重。除非我们能设计成对用户完全无感😂好像也不太容易。
另外，根据我的经验，后端高性能业务，我们现有模式是最好的。后端一般都是持久链接，不太担心连接次数的问题。

[lesismal]

你说的都没错啊，我们可以提供这些选择，但问题是如果带来用户接口上的复杂性，结果反而损失更多的普通用户。这年头，用法复杂比功能欠缺更严重。除非我们能设计成对用户完全无感😂好像也不太容易。

单就这里讨论的实现上来讲，改动只限于框架层就可以了，用户接口不需要改，因为workflow已经自带了逻辑线程池这些，只是msgqueue派发到逻辑线程的时候的依据改变一下而已。我的nbio改造有序的时候，应用层接口也都没变，甚至，我直接是兼容标准库的，所以用户用我的框架只需要改下server启动的地方，那些hadler都不需要变，甚至那些基于标准库的主流框架也是只需要改下server启动的地方用我的就可以了。。。

这些都是分层的，每一层处理好自己的逻辑就可以了

[gnblao]

#873

hi，现在尝试给加的channel功能，可以满足你pipline的想法，不过现在用户协议不完善😂～～

[lesismal]

#873

hi，现在尝试给加的channel功能，可以满足你pipline的想法，不过现在用户协议不完善😂～～

等搞好了试试看效果

[lesismal]

这个也先关掉了，要不要框架层支持有序主要看各位取舍了