3 * Copyright 2014 gRPC authors.
5 * Licensed under the Apache License, Version 2.0 (the "License");
6 * you may not use this file except in compliance with the License.
7 * You may obtain a copy of the License at
9 * http://www.apache.org/licenses/LICENSE-2.0
11 * Unless required by applicable law or agreed to in writing, software
12 * distributed under the License is distributed on an "AS IS" BASIS,
13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 * See the License for the specific language governing permissions and
15 * limitations under the License.
34 "github.com/golang/protobuf/proto"
35 "golang.org/x/net/http2"
36 "golang.org/x/net/http2/hpack"
37 spb "google.golang.org/genproto/googleapis/rpc/status"
38 "google.golang.org/grpc/codes"
39 "google.golang.org/grpc/status"
43 // http2MaxFrameLen specifies the max length of a HTTP2 frame.
44 http2MaxFrameLen = 16384 // 16KB frame
45 // http://http2.github.io/http2-spec/#SettingValues
46 http2InitHeaderTableSize = 4096
47 // http2IOBufSize specifies the buffer size for sending frames.
48 http2IOBufSize = 32 * 1024
52 clientPreface = []byte(http2.ClientPreface)
53 http2ErrConvTab = map[http2.ErrCode]codes.Code{
54 http2.ErrCodeNo: codes.Internal,
55 http2.ErrCodeProtocol: codes.Internal,
56 http2.ErrCodeInternal: codes.Internal,
57 http2.ErrCodeFlowControl: codes.ResourceExhausted,
58 http2.ErrCodeSettingsTimeout: codes.Internal,
59 http2.ErrCodeStreamClosed: codes.Internal,
60 http2.ErrCodeFrameSize: codes.Internal,
61 http2.ErrCodeRefusedStream: codes.Unavailable,
62 http2.ErrCodeCancel: codes.Canceled,
63 http2.ErrCodeCompression: codes.Internal,
64 http2.ErrCodeConnect: codes.Internal,
65 http2.ErrCodeEnhanceYourCalm: codes.ResourceExhausted,
66 http2.ErrCodeInadequateSecurity: codes.PermissionDenied,
67 http2.ErrCodeHTTP11Required: codes.FailedPrecondition,
69 statusCodeConvTab = map[codes.Code]http2.ErrCode{
70 codes.Internal: http2.ErrCodeInternal,
71 codes.Canceled: http2.ErrCodeCancel,
72 codes.Unavailable: http2.ErrCodeRefusedStream,
73 codes.ResourceExhausted: http2.ErrCodeEnhanceYourCalm,
74 codes.PermissionDenied: http2.ErrCodeInadequateSecurity,
76 httpStatusConvTab = map[int]codes.Code{
77 // 400 Bad Request - INTERNAL.
78 http.StatusBadRequest: codes.Internal,
79 // 401 Unauthorized - UNAUTHENTICATED.
80 http.StatusUnauthorized: codes.Unauthenticated,
81 // 403 Forbidden - PERMISSION_DENIED.
82 http.StatusForbidden: codes.PermissionDenied,
83 // 404 Not Found - UNIMPLEMENTED.
84 http.StatusNotFound: codes.Unimplemented,
85 // 429 Too Many Requests - UNAVAILABLE.
86 http.StatusTooManyRequests: codes.Unavailable,
87 // 502 Bad Gateway - UNAVAILABLE.
88 http.StatusBadGateway: codes.Unavailable,
89 // 503 Service Unavailable - UNAVAILABLE.
90 http.StatusServiceUnavailable: codes.Unavailable,
91 // 504 Gateway timeout - UNAVAILABLE.
92 http.StatusGatewayTimeout: codes.Unavailable,
96 // Records the states during HPACK decoding. Must be reset once the
97 // decoding of the entire headers are finished.
98 type decodeState struct {
100 // statusGen caches the stream status received from the trailer the server
101 // sent. Client side only. Do not access directly. After all trailers are
102 // parsed, use the status method to retrieve the status.
103 statusGen *status.Status
104 // rawStatusCode and rawStatusMsg are set from the raw trailer fields and are not
105 // intended for direct access outside of parsing.
109 // Server side only fields.
111 timeout time.Duration
113 // key-value metadata map from the peer.
114 mdata map[string][]string
117 // isReservedHeader checks whether hdr belongs to HTTP2 headers
118 // reserved by gRPC protocol. Any other headers are classified as the
119 // user-specified metadata.
120 func isReservedHeader(hdr string) bool {
121 if hdr != "" && hdr[0] == ':' {
131 "grpc-status-details-bin",
139 // isWhitelistedPseudoHeader checks whether hdr belongs to HTTP2 pseudoheaders
140 // that should be propagated into metadata visible to users.
141 func isWhitelistedPseudoHeader(hdr string) bool {
150 func validContentType(t string) bool {
151 e := "application/grpc"
152 if !strings.HasPrefix(t, e) {
155 // Support variations on the content-type
156 // (e.g. "application/grpc+blah", "application/grpc;blah").
157 if len(t) > len(e) && t[len(e)] != '+' && t[len(e)] != ';' {
163 func (d *decodeState) status() *status.Status {
164 if d.statusGen == nil {
165 // No status-details were provided; generate status using code/msg.
166 d.statusGen = status.New(codes.Code(int32(*(d.rawStatusCode))), d.rawStatusMsg)
171 const binHdrSuffix = "-bin"
173 func encodeBinHeader(v []byte) string {
174 return base64.RawStdEncoding.EncodeToString(v)
177 func decodeBinHeader(v string) ([]byte, error) {
179 // Input was padded, or padding was not necessary.
180 return base64.StdEncoding.DecodeString(v)
182 return base64.RawStdEncoding.DecodeString(v)
185 func encodeMetadataHeader(k, v string) string {
186 if strings.HasSuffix(k, binHdrSuffix) {
187 return encodeBinHeader(([]byte)(v))
192 func decodeMetadataHeader(k, v string) (string, error) {
193 if strings.HasSuffix(k, binHdrSuffix) {
194 b, err := decodeBinHeader(v)
195 return string(b), err
200 func (d *decodeState) decodeResponseHeader(frame *http2.MetaHeadersFrame) error {
201 for _, hf := range frame.Fields {
202 if err := d.processHeaderField(hf); err != nil {
207 // If grpc status exists, no need to check further.
208 if d.rawStatusCode != nil || d.statusGen != nil {
212 // If grpc status doesn't exist and http status doesn't exist,
213 // then it's a malformed header.
214 if d.httpStatus == nil {
215 return streamErrorf(codes.Internal, "malformed header: doesn't contain status(gRPC or HTTP)")
218 if *(d.httpStatus) != http.StatusOK {
219 code, ok := httpStatusConvTab[*(d.httpStatus)]
223 return streamErrorf(code, http.StatusText(*(d.httpStatus)))
226 // gRPC status doesn't exist and http status is OK.
227 // Set rawStatusCode to be unknown and return nil error.
228 // So that, if the stream has ended this Unknown status
229 // will be propogated to the user.
230 // Otherwise, it will be ignored. In which case, status from
231 // a later trailer, that has StreamEnded flag set, is propogated.
232 code := int(codes.Unknown)
233 d.rawStatusCode = &code
238 func (d *decodeState) processHeaderField(f hpack.HeaderField) error {
241 if !validContentType(f.Value) {
242 return streamErrorf(codes.FailedPrecondition, "transport: received the unexpected content-type %q", f.Value)
244 case "grpc-encoding":
247 code, err := strconv.Atoi(f.Value)
249 return streamErrorf(codes.Internal, "transport: malformed grpc-status: %v", err)
251 d.rawStatusCode = &code
253 d.rawStatusMsg = decodeGrpcMessage(f.Value)
254 case "grpc-status-details-bin":
255 v, err := decodeBinHeader(f.Value)
257 return streamErrorf(codes.Internal, "transport: malformed grpc-status-details-bin: %v", err)
260 if err := proto.Unmarshal(v, s); err != nil {
261 return streamErrorf(codes.Internal, "transport: malformed grpc-status-details-bin: %v", err)
263 d.statusGen = status.FromProto(s)
267 if d.timeout, err = decodeTimeout(f.Value); err != nil {
268 return streamErrorf(codes.Internal, "transport: malformed time-out: %v", err)
273 code, err := strconv.Atoi(f.Value)
275 return streamErrorf(codes.Internal, "transport: malformed http-status: %v", err)
279 if !isReservedHeader(f.Name) || isWhitelistedPseudoHeader(f.Name) {
281 d.mdata = make(map[string][]string)
283 v, err := decodeMetadataHeader(f.Name, f.Value)
285 errorf("Failed to decode metadata header (%q, %q): %v", f.Name, f.Value, err)
288 d.mdata[f.Name] = append(d.mdata[f.Name], v)
294 type timeoutUnit uint8
297 hour timeoutUnit = 'H'
298 minute timeoutUnit = 'M'
299 second timeoutUnit = 'S'
300 millisecond timeoutUnit = 'm'
301 microsecond timeoutUnit = 'u'
302 nanosecond timeoutUnit = 'n'
305 func timeoutUnitToDuration(u timeoutUnit) (d time.Duration, ok bool) {
308 return time.Hour, true
310 return time.Minute, true
312 return time.Second, true
314 return time.Millisecond, true
316 return time.Microsecond, true
318 return time.Nanosecond, true
324 const maxTimeoutValue int64 = 100000000 - 1
326 // div does integer division and round-up the result. Note that this is
327 // equivalent to (d+r-1)/r but has less chance to overflow.
328 func div(d, r time.Duration) int64 {
329 if m := d % r; m > 0 {
330 return int64(d/r + 1)
335 // TODO(zhaoq): It is the simplistic and not bandwidth efficient. Improve it.
336 func encodeTimeout(t time.Duration) string {
340 if d := div(t, time.Nanosecond); d <= maxTimeoutValue {
341 return strconv.FormatInt(d, 10) + "n"
343 if d := div(t, time.Microsecond); d <= maxTimeoutValue {
344 return strconv.FormatInt(d, 10) + "u"
346 if d := div(t, time.Millisecond); d <= maxTimeoutValue {
347 return strconv.FormatInt(d, 10) + "m"
349 if d := div(t, time.Second); d <= maxTimeoutValue {
350 return strconv.FormatInt(d, 10) + "S"
352 if d := div(t, time.Minute); d <= maxTimeoutValue {
353 return strconv.FormatInt(d, 10) + "M"
355 // Note that maxTimeoutValue * time.Hour > MaxInt64.
356 return strconv.FormatInt(div(t, time.Hour), 10) + "H"
359 func decodeTimeout(s string) (time.Duration, error) {
362 return 0, fmt.Errorf("transport: timeout string is too short: %q", s)
364 unit := timeoutUnit(s[size-1])
365 d, ok := timeoutUnitToDuration(unit)
367 return 0, fmt.Errorf("transport: timeout unit is not recognized: %q", s)
369 t, err := strconv.ParseInt(s[:size-1], 10, 64)
373 return d * time.Duration(t), nil
382 // encodeGrpcMessage is used to encode status code in header field
384 // It checks to see if each individual byte in msg is an
385 // allowable byte, and then either percent encoding or passing it through.
386 // When percent encoding, the byte is converted into hexadecimal notation
387 // with a '%' prepended.
388 func encodeGrpcMessage(msg string) string {
393 for i := 0; i < lenMsg; i++ {
395 if !(c >= spaceByte && c < tildaByte && c != percentByte) {
396 return encodeGrpcMessageUnchecked(msg)
402 func encodeGrpcMessageUnchecked(msg string) string {
405 for i := 0; i < lenMsg; i++ {
407 if c >= spaceByte && c < tildaByte && c != percentByte {
410 buf.WriteString(fmt.Sprintf("%%%02X", c))
416 // decodeGrpcMessage decodes the msg encoded by encodeGrpcMessage.
417 func decodeGrpcMessage(msg string) string {
422 for i := 0; i < lenMsg; i++ {
423 if msg[i] == percentByte && i+2 < lenMsg {
424 return decodeGrpcMessageUnchecked(msg)
430 func decodeGrpcMessageUnchecked(msg string) string {
433 for i := 0; i < lenMsg; i++ {
435 if c == percentByte && i+2 < lenMsg {
436 parsed, err := strconv.ParseUint(msg[i+1:i+3], 16, 8)
440 buf.WriteByte(byte(parsed))
457 func newFramer(conn net.Conn) *framer {
459 reader: bufio.NewReaderSize(conn, http2IOBufSize),
460 writer: bufio.NewWriterSize(conn, http2IOBufSize),
462 f.fr = http2.NewFramer(f.writer, f.reader)
463 // Opt-in to Frame reuse API on framer to reduce garbage.
464 // Frames aren't safe to read from after a subsequent call to ReadFrame.
465 f.fr.SetReuseFrames()
466 f.fr.ReadMetaHeaders = hpack.NewDecoder(http2InitHeaderTableSize, nil)
470 func (f *framer) adjustNumWriters(i int32) int32 {
471 return atomic.AddInt32(&f.numWriters, i)
474 // The following writeXXX functions can only be called when the caller gets
475 // unblocked from writableChan channel (i.e., owns the privilege to write).
477 func (f *framer) writeContinuation(forceFlush bool, streamID uint32, endHeaders bool, headerBlockFragment []byte) error {
478 if err := f.fr.WriteContinuation(streamID, endHeaders, headerBlockFragment); err != nil {
482 return f.writer.Flush()
487 func (f *framer) writeData(forceFlush bool, streamID uint32, endStream bool, data []byte) error {
488 if err := f.fr.WriteData(streamID, endStream, data); err != nil {
492 return f.writer.Flush()
497 func (f *framer) writeGoAway(forceFlush bool, maxStreamID uint32, code http2.ErrCode, debugData []byte) error {
498 if err := f.fr.WriteGoAway(maxStreamID, code, debugData); err != nil {
502 return f.writer.Flush()
507 func (f *framer) writeHeaders(forceFlush bool, p http2.HeadersFrameParam) error {
508 if err := f.fr.WriteHeaders(p); err != nil {
512 return f.writer.Flush()
517 func (f *framer) writePing(forceFlush, ack bool, data [8]byte) error {
518 if err := f.fr.WritePing(ack, data); err != nil {
522 return f.writer.Flush()
527 func (f *framer) writePriority(forceFlush bool, streamID uint32, p http2.PriorityParam) error {
528 if err := f.fr.WritePriority(streamID, p); err != nil {
532 return f.writer.Flush()
537 func (f *framer) writePushPromise(forceFlush bool, p http2.PushPromiseParam) error {
538 if err := f.fr.WritePushPromise(p); err != nil {
542 return f.writer.Flush()
547 func (f *framer) writeRSTStream(forceFlush bool, streamID uint32, code http2.ErrCode) error {
548 if err := f.fr.WriteRSTStream(streamID, code); err != nil {
552 return f.writer.Flush()
557 func (f *framer) writeSettings(forceFlush bool, settings ...http2.Setting) error {
558 if err := f.fr.WriteSettings(settings...); err != nil {
562 return f.writer.Flush()
567 func (f *framer) writeSettingsAck(forceFlush bool) error {
568 if err := f.fr.WriteSettingsAck(); err != nil {
572 return f.writer.Flush()
577 func (f *framer) writeWindowUpdate(forceFlush bool, streamID, incr uint32) error {
578 if err := f.fr.WriteWindowUpdate(streamID, incr); err != nil {
582 return f.writer.Flush()
587 func (f *framer) flushWrite() error {
588 return f.writer.Flush()
591 func (f *framer) readFrame() (http2.Frame, error) {
592 return f.fr.ReadFrame()
595 func (f *framer) errorDetail() error {
596 return f.fr.ErrorDetail()