AlexStocks commented on code in PR #3579:
URL: https://github.com/apache/dubbo-go/pull/3579#discussion_r3780909568
##########
protocol/jsonrpc/server.go:
##########
@@ -150,51 +132,139 @@ func (s *Server) handlePkg(conn net.Conn) {
}
reqHeader["HttpMethod"] = r.Method
- httpTimeout := s.timeout
- contentType := reqHeader["Content-Type"]
+ contentType := reqHeader[ContentTypeHeader]
mediaType, _, parseErr := mime.ParseMediaType(contentType)
- if parseErr != nil || (mediaType != "application/json" &&
mediaType != "application/json-rpc") {
- setTimeout(conn, httpTimeout)
- errMsg := "unsupported content type: " + contentType
- if errRsp := sendErrorResp(r.Header, []byte(errMsg));
errRsp != nil {
- logger.Warnf("[Jsonrpc][Server] sendErrorResp
failed, header=%v, err_msg=%v, send_err=%v",
- r.Header, errMsg, errRsp)
- }
- return
- }
-
- ctx := context.Background()
+ unsupportedContentType := parseErr != nil || (mediaType !=
"application/json" && mediaType != "application/json-rpc")
- spanCtx, err :=
opentracing.GlobalTracer().Extract(opentracing.HTTPHeaders,
- opentracing.HTTPHeadersCarrier(r.Header))
- if err == nil {
- ctx = context.WithValue(ctx,
constant.TracingRemoteSpanCtx, spanCtx)
- }
+ requestCtx, requestCancel := context.WithCancel(connectionCtx)
+ r = r.WithContext(requestCtx)
+ ctx := contextFromRequest(r)
+ var timeoutCancel context.CancelFunc
if len(reqHeader["Timeout"]) > 0 {
timeout, err := time.ParseDuration(reqHeader["Timeout"])
if err == nil {
- httpTimeout = timeout
- var cancel context.CancelFunc
- ctx, cancel = context.WithTimeout(ctx,
httpTimeout)
- defer cancel()
+ ctx, timeoutCancel = context.WithTimeout(ctx,
timeout)
}
delete(reqHeader, "Timeout")
}
- setTimeout(conn, httpTimeout)
- if err := serveRequest(ctx, reqHeader, reqBody, conn); err !=
nil {
- if errRsp := sendErrorResp(r.Header,
[]byte(perrors.WithStack(err).Error())); errRsp != nil {
- logger.Warnf("[Jsonrpc][Server] sendErrorResp
failed, header=%v, err=%v, send_err=%v",
- r.Header, perrors.WithStack(err),
errRsp)
+ requestSequence := sequence
+ sequence++
+ requestWG.Add(1)
+ go func(ctx context.Context, requestCancel, timeoutCancel
context.CancelFunc, header map[string]string, body []byte,
Review Comment:
[P1] 请限制单连接内的并发请求数和待写响应窗口。这里对每个流水线请求直接启动 goroutine，而 `writeResponsesInOrder`
会把后续完成的完整响应保存在无上限的 `pending` map 中，直到最早序号完成。固定当前 Head 的定向探针让所有调用等待取消后，在同一连接写入
256 个请求；256 个调用会在任何一个完成前全部启动，goroutine 数增加 257。若最早请求长时间未完成，持续进入的请求会同时保留
goroutine、请求 body 和完整响应字节。请增加 per-connection semaphore/有界 response
window，并通过停止读取或反压限制解析；回归测试应阻塞序号 0、发送超过上限的请求并断言在途调用和缓存保持有界。
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
