Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Use the same hpack encoder on a transport and share it between RPCs. #1536

Merged
merged 4 commits into from
Sep 28, 2017
Merged

Use the same hpack encoder on a transport and share it between RPCs. #1536

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
567b1ea
Use the same hpack encoder on a transport and share it between RPCs.
MakMukhi Sep 23, 2017
f7f1e8e
Remove WireLength field from OutHeader.
MakMukhi Sep 25, 2017
2d53de8
Post-review updates
MakMukhi Sep 25, 2017
bc6999b
mend
MakMukhi Sep 28, 2017
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 0 additions & 2 deletions stats/stats.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -135,8 +135,6 @@ func (s *OutPayload) isRPCStats() {}
type OutHeader struct {
// Client is true if this OutHeader is from client side.
Client bool
// WireLength is the wire length of header.
WireLength int

// The following fields are valid only if Client is true.
// FullMethod is the full RPC method string, i.e., /package.service/method.
Expand Down
19 changes: 1 addition & 18 deletions stats/stats_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -444,10 +444,6 @@ func checkInHeader(t *testing.T, d *gotData, e *expectedData) {
if d.ctx == nil {
t.Fatalf("d.ctx = nil, want <non-nil>")
}
// TODO check real length, not just > 0.
if st.WireLength <= 0 {
t.Fatalf("st.Lenght = 0, want > 0")
}
if !d.client {
if st.FullMethod != e.method {
t.Fatalf("st.FullMethod = %s, want %v", st.FullMethod, e.method)
Expand Down Expand Up @@ -530,18 +526,13 @@ func checkInPayload(t *testing.T, d *gotData, e *expectedData) {
func checkInTrailer(t *testing.T, d *gotData, e *expectedData) {
var (
ok bool
st *stats.InTrailer
)
if st, ok = d.s.(*stats.InTrailer); !ok {
if _, ok = d.s.(*stats.InTrailer); !ok {
t.Fatalf("got %T, want InTrailer", d.s)
}
if d.ctx == nil {
t.Fatalf("d.ctx = nil, want <non-nil>")
}
// TODO check real length, not just > 0.
if st.WireLength <= 0 {
t.Fatalf("st.Lenght = 0, want > 0")
}
}

func checkOutHeader(t *testing.T, d *gotData, e *expectedData) {
Expand All @@ -555,10 +546,6 @@ func checkOutHeader(t *testing.T, d *gotData, e *expectedData) {
if d.ctx == nil {
t.Fatalf("d.ctx = nil, want <non-nil>")
}
// TODO check real length, not just > 0.
if st.WireLength <= 0 {
t.Fatalf("st.Lenght = 0, want > 0")
}
if d.client {
if st.FullMethod != e.method {
t.Fatalf("st.FullMethod = %s, want %v", st.FullMethod, e.method)
Expand Down Expand Up @@ -642,10 +629,6 @@ func checkOutTrailer(t *testing.T, d *gotData, e *expectedData) {
if st.Client {
t.Fatalf("st IsClient = true, want false")
}
// TODO check real length, not just > 0.
if st.WireLength <= 0 {
t.Fatalf("st.Lenght = 0, want > 0")
}
}

func checkEnd(t *testing.T, d *gotData, e *expectedData) {
Expand Down
5 changes: 4 additions & 1 deletion transport/control.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@ import (
"time"

"golang.org/x/net/http2"
"golang.org/x/net/http2/hpack"
)

const (
Expand Down Expand Up @@ -56,7 +57,9 @@ const (
// control tasks, e.g., flow control, settings, streaming resetting, etc.

type headerFrame struct {
p http2.HeadersFrameParam
streamID uint32
hf []hpack.HeaderField
endStream bool
}

func (*headerFrame) item() {}
Expand Down
123 changes: 65 additions & 58 deletions transport/http2_client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -193,6 +193,7 @@ func newHTTP2Client(ctx context.Context, addr TargetInfo, opts ConnectOptions) (
icwz = opts.InitialConnWindowSize
dynamicWindow = false
}
var buf bytes.Buffer
t := &http2Client{
ctx: ctx,
target: addr.Addr,
Expand All @@ -209,6 +210,8 @@ func newHTTP2Client(ctx context.Context, addr TargetInfo, opts ConnectOptions) (
goAway: make(chan struct{}),
awakenKeepalive: make(chan struct{}, 1),
framer: newFramer(conn),
hBuf: &buf,
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

hBuf: &bytes.Buffer{}

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I know we talked about this but now I realize that hEnc needs the same buffer.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, that should have been obvious, sorry!

hEnc: hpack.NewEncoder(&buf),
controlBuf: newControlBuffer(),
fc: &inFlow{limit: uint32(icwz)},
sendQuotaPool: newQuotaPool(defaultWindowSize),
Expand Down Expand Up @@ -361,13 +364,13 @@ func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (_ *Strea
authData[k] = v
}
}
callAuthData := make(map[string]string)
callAuthData := map[string]string{}
// Check if credentials.PerRPCCredentials were provided via call options.
// Note: if these credentials are provided both via dial options and call
// options, then both sets of credentials will be applied.
if callCreds := callHdr.Creds; callCreds != nil {
if !t.isSecure && callCreds.RequireTransportSecurity() {
return nil, streamErrorf(codes.Unauthenticated, "transport: cannot send secure credentials on an insecure connection")
return nil, streamErrorf(codes.Unauthenticated, "transport: cannot send secure credentials on an insecure conneciton")
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

connection

}
data, err := callCreds.GetRequestMetadata(ctx, audience)
if err != nil {
Expand Down Expand Up @@ -401,40 +404,38 @@ func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (_ *Strea
if sq > 1 {
t.streamsQuota.add(sq - 1)
}
// HPACK encodes various headers.
hBuf := bytes.NewBuffer([]byte{})
hEnc := hpack.NewEncoder(hBuf)
hEnc.WriteField(hpack.HeaderField{Name: ":method", Value: "POST"})
hEnc.WriteField(hpack.HeaderField{Name: ":scheme", Value: t.scheme})
hEnc.WriteField(hpack.HeaderField{Name: ":path", Value: callHdr.Method})
hEnc.WriteField(hpack.HeaderField{Name: ":authority", Value: callHdr.Host})
hEnc.WriteField(hpack.HeaderField{Name: "content-type", Value: "application/grpc"})
hEnc.WriteField(hpack.HeaderField{Name: "user-agent", Value: t.userAgent})
hEnc.WriteField(hpack.HeaderField{Name: "te", Value: "trailers"})
// Make the slice of certain predictable size to reduce allocations made by append.
hfLen := 7 // :method, :scheme, :path, :authority, content-type, user-agent, te
hfLen += len(authData) + len(callAuthData)
headerFields := make([]hpack.HeaderField, 0, hfLen)
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This will result in reallocations if there is any user-provided metadata.

It probably would be a win to count it first.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I thought of that but we'll need to loop through the and see if there are any reserved headers in there or not twice (one to count and one to actually append in the slice). And that holds true for other headers too. So I ended up counting the ones that are easily countable. And since append allocates twice the current size. I'm guessing there won't be too many allocations happening.
Thoughts?

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could you add some benchmarks to cover this so we can decide? We don't have anything that includes metadata. If the number of entries and size of each entry can be tweaked through settings, that would be ideal. Ideally we could independently configure client headers, server headers, and server trailers. It would be interesting to compare the performance of both approaches in the following scenarios:

  1. no metadata
  2. with a small amount of metadata (~2)
  3. with a large amount of metadata (~20)

headerFields = append(headerFields, hpack.HeaderField{Name: ":method", Value: "POST"})
headerFields = append(headerFields, hpack.HeaderField{Name: ":scheme", Value: t.scheme})
headerFields = append(headerFields, hpack.HeaderField{Name: ":path", Value: callHdr.Method})
headerFields = append(headerFields, hpack.HeaderField{Name: ":authority", Value: callHdr.Host})
headerFields = append(headerFields, hpack.HeaderField{Name: "content-type", Value: "application/grpc"})
headerFields = append(headerFields, hpack.HeaderField{Name: "user-agent", Value: t.userAgent})
headerFields = append(headerFields, hpack.HeaderField{Name: "te", Value: "trailers"})

if callHdr.SendCompress != "" {
hEnc.WriteField(hpack.HeaderField{Name: "grpc-encoding", Value: callHdr.SendCompress})
headerFields = append(headerFields, hpack.HeaderField{Name: "grpc-encoding", Value: callHdr.SendCompress})
}
if dl, ok := ctx.Deadline(); ok {
// Send out timeout regardless its value. The server can detect timeout context by itself.
// TODO(mmukhi): Perhaps this field should be updated when actually writing out to the wire.
timeout := dl.Sub(time.Now())
hEnc.WriteField(hpack.HeaderField{Name: "grpc-timeout", Value: encodeTimeout(timeout)})
headerFields = append(headerFields, hpack.HeaderField{Name: "grpc-timeout", Value: encodeTimeout(timeout)})
}

for k, v := range authData {
hEnc.WriteField(hpack.HeaderField{Name: k, Value: encodeMetadataHeader(k, v)})
headerFields = append(headerFields, hpack.HeaderField{Name: k, Value: encodeMetadataHeader(k, v)})
}
for k, v := range callAuthData {
hEnc.WriteField(hpack.HeaderField{Name: k, Value: encodeMetadataHeader(k, v)})
headerFields = append(headerFields, hpack.HeaderField{Name: k, Value: encodeMetadataHeader(k, v)})
}
var (
endHeaders bool
)
if b := stats.OutgoingTags(ctx); b != nil {
hEnc.WriteField(hpack.HeaderField{Name: "grpc-tags-bin", Value: encodeBinHeader(b)})
headerFields = append(headerFields, hpack.HeaderField{Name: "grpc-tags-bin", Value: encodeBinHeader(b)})
}
if b := stats.OutgoingTrace(ctx); b != nil {
hEnc.WriteField(hpack.HeaderField{Name: "grpc-trace-bin", Value: encodeBinHeader(b)})
headerFields = append(headerFields, hpack.HeaderField{Name: "grpc-trace-bin", Value: encodeBinHeader(b)})
}
if md, ok := metadata.FromOutgoingContext(ctx); ok {
for k, vv := range md {
Expand All @@ -443,7 +444,7 @@ func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (_ *Strea
continue
}
for _, v := range vv {
hEnc.WriteField(hpack.HeaderField{Name: k, Value: encodeMetadataHeader(k, v)})
headerFields = append(headerFields, hpack.HeaderField{Name: k, Value: encodeMetadataHeader(k, v)})
}
}
}
Expand All @@ -453,7 +454,7 @@ func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (_ *Strea
continue
}
for _, v := range vv {
hEnc.WriteField(hpack.HeaderField{Name: k, Value: encodeMetadataHeader(k, v)})
headerFields = append(headerFields, hpack.HeaderField{Name: k, Value: encodeMetadataHeader(k, v)})
}
}
}
Expand Down Expand Up @@ -482,34 +483,11 @@ func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (_ *Strea
default:
}
}
first := true
bufLen := hBuf.Len()
// Sends the headers in a single batch even when they span multiple frames.
for !endHeaders {
size := hBuf.Len()
if size > http2MaxFrameLen {
size = http2MaxFrameLen
} else {
endHeaders = true
}
if first {
// Sends a HeadersFrame to server to start a new stream.
p := http2.HeadersFrameParam{
StreamID: s.id,
BlockFragment: hBuf.Next(size),
EndStream: false,
EndHeaders: endHeaders,
}
// Do a force flush for the buffered frames iff it is the last headers frame
// and there is header metadata to be sent. Otherwise, there is flushing until
// the corresponding data frame is written.
t.controlBuf.put(&headerFrame{p})
first = false
} else {
// Sends Continuation frames for the leftover headers.
t.controlBuf.put(&continuationFrame{streamID: s.id, endHeaders: endHeaders, headerBlockFragment: hBuf.Next(size)})
}
}
t.controlBuf.put(&headerFrame{
streamID: s.id,
hf: headerFields,
endStream: false,
})
t.mu.Unlock()

s.mu.Lock()
Expand All @@ -518,8 +496,8 @@ func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (_ *Strea

if t.statsHandler != nil {
outHeader := &stats.OutHeader{
Client: true,
WireLength: bufLen,
Client: true,
//WireLength: // TODO(mmukhi): Revisit this if needed.
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Delete

FullMethod: callHdr.Method,
RemoteAddr: t.remoteAddr,
LocalAddr: t.localAddr,
Expand Down Expand Up @@ -961,7 +939,7 @@ func (t *http2Client) handleGoAway(f *http2.GoAwayFrame) {
t.notifyError(connectionErrorf(true, nil, "received illegal http2 GOAWAY frame: stream ID %d is even", f.LastStreamID))
return
}
// A client can receive multiple GoAways from server (look at https://github.com/grpc/grpc-go/issues/1387).
// A client can recieve multiple GoAways from server (look at https://github.com/grpc/grpc-go/issues/1387).
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

except after "c"

// The idea is that the first GoAway will be sent with an ID of MaxInt32 and the second GoAway will be sent after an RTT delay
// with the ID of the last stream the server will process.
// Therefore, when we get the first GoAway we don't really close any streams. While in case of second GoAway we
Expand Down Expand Up @@ -1214,9 +1192,38 @@ func (t *http2Client) itemHandler(i item) error {
i.f()
}
case *headerFrame:
err = t.framer.fr.WriteHeaders(i.p)
case *continuationFrame:
err = t.framer.fr.WriteContinuation(i.streamID, i.endHeaders, i.headerBlockFragment)
t.hBuf.Reset()
for _, f := range i.hf {
t.hEnc.WriteField(f)
}
endHeaders := false
first := true
for !endHeaders {
size := t.hBuf.Len()
if size > http2MaxFrameLen {
size = http2MaxFrameLen
} else {
endHeaders = true
}
if first {
first = false
err = t.framer.fr.WriteHeaders(http2.HeadersFrameParam{
StreamID: i.streamID,
BlockFragment: t.hBuf.Next(size),
EndStream: i.endStream,
EndHeaders: endHeaders,
})
} else {
err = t.framer.fr.WriteContinuation(
i.streamID,
endHeaders,
t.hBuf.Next(size),
)
}
if err != nil {
return err
}
}
case *windowUpdate:
err = t.framer.fr.WriteWindowUpdate(i.streamID, i.increment)
case *settings:
Expand Down
Loading