AlexStocks commented on code in PR #3579:
URL: https://github.com/apache/dubbo-go/pull/3579#discussion_r3764623584
##########
protocol/jsonrpc/server.go:
##########
@@ -151,48 +145,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,
httpTimeout)
}
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 后，前面的 `setTimeout(conn, httpTimeout)` 仍按每个请求调用
`conn.SetDeadline`，所以后续请求会覆盖整条连接的 deadline。负向探针在同一连接先发送 `Timeout: 5s` 的阻塞请求，再发送
`Timeout: 50ms` 的请求；50ms 后读循环报 `i/o timeout`，并通过 `connectionCancel`
提前取消了第一个请求。请把业务请求 timeout 只保留在各自的 `context.WithTimeout` 中，不要用它更新共享连接
deadline；连接级 read/idle timeout 应单独管理，并增加不同 timeout 的 pipelining 回归测试。
--
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]
