transport: remove decodeState from server to reduce allocations

internal/transport/http2_server.go

@@ -333,8 +333,6 @@ func (t *http2Server) operateHeaders(frame *http2.MetaHeadersFrame, handle func(
 		httpMethod string
 		// headerError is set if an error is encountered while parsing the headers
 		headerError bool
-		statsTrace  []byte
-		statsTags   []byte
 
 		timeoutSet bool
 		timeout    time.Duration
@@ -368,15 +366,13 @@ func (t *http2Server) operateHeaders(frame *http2.MetaHeadersFrame, handle func(
 				headerError = true
 				return
 			}
-			statsTags = v
 			mdata[hf.Name] = append(mdata[hf.Name], string(v))
 		case "grpc-trace-bin":
 			v, err := decodeBinHeader(hf.Value)
 			if err != nil {
 				headerError = true
 				return
 			}
-			statsTrace = v
 			mdata[hf.Name] = append(mdata[hf.Name], string(v))
 		default:
 			if isReservedHeader(hf.Name) && !isWhitelistedHeader(hf.Name) {
@@ -423,11 +419,11 @@ func (t *http2Server) operateHeaders(frame *http2.MetaHeadersFrame, handle func(
 	if len(mdata) > 0 {
 		s.ctx = metadata.NewIncomingContext(s.ctx, mdata)
 	}
-	if statsTags != nil {
-		s.ctx = stats.SetIncomingTags(s.ctx, statsTags)
+	if statsTags, ok := mdata["grpc-tags-bin"]; ok && len(statsTags) > 0 {
+		s.ctx = stats.SetIncomingTags(s.ctx, []byte(statsTags[0]))
 	}
-	if statsTrace != nil {
-		s.ctx = stats.SetIncomingTrace(s.ctx, statsTrace)
+	if statsTrace, ok := mdata["grpc-trace-bin"]; ok && len(statsTrace) > 0 {
+		s.ctx = stats.SetIncomingTrace(s.ctx, []byte(statsTrace[0]))
 	}
 	t.mu.Lock()
 	if t.state != reachable {
@@ -459,7 +455,7 @@ func (t *http2Server) operateHeaders(frame *http2.MetaHeadersFrame, handle func(
 	if httpMethod != http.MethodPost {
 		t.mu.Unlock()
 		if logger.V(logLevel) {
-			logger.Warningf("transport: http2Server.operateHeaders parsed a :method field: %v which should be POST", httpMethod)
+			logger.Infof("transport: http2Server.operateHeaders parsed a :method field: %v which should be POST", httpMethod)
 		}
 		t.controlBuf.put(&cleanupStream{
 			streamID: streamID,

internal/transport/http_util.go

@@ -95,43 +95,6 @@ var (
 	logger = grpclog.Component("transport")
 )
 
-type parsedHeaderData struct {
-	encoding string
-	// statusGen caches the stream status received from the trailer the server
-	// sent.  Client side only.  Do not access directly.  After all trailers are
-	// parsed, use the status method to retrieve the status.
-	statusGen *status.Status
-	// rawStatusCode and rawStatusMsg are set from the raw trailer fields and are not
-	// intended for direct access outside of parsing.
-	rawStatusCode *int
-	rawStatusMsg  string
-	httpStatus    *int
-	// Server side only fields.
-	timeoutSet bool
-	timeout    time.Duration
-	method     string
-	httpMethod string
-	// key-value metadata map from the peer.
-	mdata          map[string][]string
-	statsTags      []byte
-	statsTrace     []byte
-	contentSubtype string
-
-	// isGRPC field indicates whether the peer is speaking gRPC (otherwise HTTP).
-	//
-	// We are in gRPC mode (peer speaking gRPC) if:
-	// 	* We are client side and have already received a HEADER frame that indicates gRPC peer.
-	//  * The header contains valid  a content-type, i.e. a string starts with "application/grpc"
-	// And we should handle error specific to gRPC.
-	//
-	// Otherwise (i.e. a content-type string starts without "application/grpc", or does not exist), we
-	// are in HTTP fallback mode, and should handle error specific to HTTP.
-	isGRPC         bool
-	grpcErr        error
-	httpErr        error
-	contentTypeErr string
-}
-
 // isReservedHeader checks whether hdr belongs to HTTP2 headers
 // reserved by gRPC protocol. Any other headers are classified as the
 // user-specified metadata.

internal/transport/transport_test.go

@@ -1976,3 +1976,113 @@ func (s) TestClientHandshakeInfo(t *testing.T) {
 		t.Fatalf("received attributes %v in creds, want %v", gotAttr, wantAttr)
 	}
 }
+
+func (s) TestClientDecodeHeaderStatusErr(t *testing.T) {
+	for _, test := range []struct {
+		name string
+		// input
+		metaHeaderFrame *http2.MetaHeadersFrame
+		// output
+		wantStatus *status.Status
+	}{
+		{
+			name: "valid header",
+			metaHeaderFrame: &http2.MetaHeadersFrame{
+				Fields: []hpack.HeaderField{
+					{Name: "content-type", Value: "application/grpc"},
+					{Name: "grpc-status", Value: "0"},
+				},
+			},
+			// no error
+			wantStatus: status.New(codes.OK, ""),
+		},
+		{
+			name: "invalid grpc status header field",
+			metaHeaderFrame: &http2.MetaHeadersFrame{
+				Fields: []hpack.HeaderField{
+					{Name: "content-type", Value: "application/grpc"},
+					{Name: "grpc-status", Value: "xxxx"},
+				},
+			},
+			wantStatus: status.New(
+				codes.Internal,
+				"transport: malformed grpc-status: strconv.ParseInt: parsing \"xxxx\": invalid syntax",
+			),
+		},
+		{
+			name: "invalid http content type",
+			metaHeaderFrame: &http2.MetaHeadersFrame{
+				Fields: []hpack.HeaderField{
+					{Name: "content-type", Value: "application/json"},
+				},
+			},
+			wantStatus: status.New(
+				codes.Internal,
+				": HTTP status code 0; transport: received the unexpected content-type \"application/json\"",
+			),
+		},
+		{
+			name: "http fallback and invalid http status",
+			metaHeaderFrame: &http2.MetaHeadersFrame{
+				Fields: []hpack.HeaderField{
+					// No content type provided then fallback into handling http error.
+					{Name: ":status", Value: "xxxx"},
+				},
+			},
+			wantStatus: status.New(
+				codes.Internal,
+				"transport: malformed http-status: strconv.ParseInt: parsing \"xxxx\": invalid syntax",
+			),
+		},
+		{
+			name: "http2 frame size exceeds",
+			metaHeaderFrame: &http2.MetaHeadersFrame{
+				Fields:    nil,
+				Truncated: true,
+			},
+			wantStatus: status.New(
+				codes.Internal,
+				"peer header list size exceeded limit",
+			),
+		},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			ts := &Stream{
+				done:       make(chan struct{}),
+				headerChan: make(chan struct{}),
+				buf: &recvBuffer{
+					c:  make(chan recvMsg),
+					mu: sync.Mutex{},
+				},
+			}
+			s := &http2Client{
+				mu: sync.Mutex{},
+				activeStreams: map[uint32]*Stream{
+					0: ts,
+				},
+				controlBuf: &controlBuffer{
+					ch:   make(chan struct{}),
+					done: make(chan struct{}),
+					list: &itemList{},
+				},
+			}
+			test.metaHeaderFrame.HeadersFrame = &http2.HeadersFrame{
+				FrameHeader: http2.FrameHeader{
+					StreamID: 0,
+					Flags:    http2.FlagHeadersEndStream,
+				},
+			}
+
+			s.operateHeaders(test.metaHeaderFrame)
+
+			got := ts.status
+			want := test.wantStatus
+			if got.Code() != want.Code() {
+				t.Fatalf("operateHeaders() got code %d, want %d", got.Code(), want.Code())
+			}
+			if got.Message() != want.Message() {
+				t.Fatalf("operateHeaders() got message %s, want %s", got.Message(), want.Message())
+			}
+		})
+	}
+}