reduce pointer fields

Author: arjan-bal

internal/transport/client_stream.go

@@ -29,7 +29,7 @@ import (
 
 // ClientStream implements streaming functionality for a gRPC client.
 type ClientStream struct {
-	*Stream // Embed for common stream functionality.
+	Stream // Embed for common stream functionality.
 
 	ct       *http2Client
 	done     chan struct{} // closed at the end of stream to unblock writers.

internal/transport/flowcontrol.go

@@ -28,6 +28,7 @@ import (
 // writeQuota is a soft limit on the amount of data a stream can
 // schedule before some of it is written out.
 type writeQuota struct {
+	_     noCopy
 	quota int32
 	// get waits on read from when quota goes less than or equal to zero.
 	// replenish writes on it when quota goes positive again.
@@ -40,14 +41,13 @@ type writeQuota struct {
 	replenish func(n int)
 }
 
-func newWriteQuota(sz int32, done <-chan struct{}) *writeQuota {
-	w := &writeQuota{
+func initWriteQuota(s *Stream, sz int32, done <-chan struct{}) {
+	s.wq = writeQuota{
 		quota: sz,
 		ch:    make(chan struct{}, 1),
 		done:  done,
 	}
-	w.replenish = w.realReplenish
-	return w
+	s.wq.replenish = s.wq.realReplenish
 }
 
 func (w *writeQuota) get(sz int32) error {

internal/transport/handler_server.go

@@ -411,11 +411,10 @@ func (ht *serverHandlerTransport) HandleStreams(ctx context.Context, startStream
 	ctx = metadata.NewIncomingContext(ctx, ht.headerMD)
 	req := ht.req
 	s := &ServerStream{
-		Stream: &Stream{
+		Stream: Stream{
 			id:             0, // irrelevant
 			ctx:            ctx,
 			requestRead:    func(int) {},
-			buf:            newRecvBuffer(),
 			method:         req.URL.Path,
 			recvCompress:   req.Header.Get("grpc-encoding"),
 			contentSubtype: ht.contentSubtype,
@@ -424,8 +423,9 @@ func (ht *serverHandlerTransport) HandleStreams(ctx context.Context, startStream
 		st:               ht,
 		headerWireLength: 0, // won't have access to header wire length until golang/go#18997.
 	}
-	s.trReader = &transportReader{
-		reader:        &recvBufferReader{ctx: s.ctx, ctxDone: s.ctx.Done(), recv: s.buf},
+	initRecvBuffer(&s.Stream)
+	s.trReader = transportReader{
+		reader:        &recvBufferReader{ctx: s.ctx, ctxDone: s.ctx.Done(), recv: &s.buf},
 		windowHandler: func(int) {},
 	}
 

internal/transport/http2_client.go

@@ -481,30 +481,30 @@ func NewHTTP2Client(connectCtx, ctx context.Context, addr resolver.Address, opts
 func (t *http2Client) newStream(ctx context.Context, callHdr *CallHdr) *ClientStream {
 	// TODO(zhaoq): Handle uint32 overflow of Stream.id.
 	s := &ClientStream{
-		Stream: &Stream{
+		Stream: Stream{
 			method:         callHdr.Method,
 			sendCompress:   callHdr.SendCompress,
-			buf:            newRecvBuffer(),
 			contentSubtype: callHdr.ContentSubtype,
 		},
 		ct:         t,
 		done:       make(chan struct{}),
 		headerChan: make(chan struct{}),
 		doneFunc:   callHdr.DoneFunc,
 	}
-	s.wq = newWriteQuota(defaultWriteQuota, s.done)
+	initRecvBuffer(&s.Stream)
+	initWriteQuota(&s.Stream, defaultWriteQuota, s.done)
 	s.requestRead = func(n int) {
 		t.adjustWindow(s, uint32(n))
 	}
 	// The client side stream context should have exactly the same life cycle with the user provided context.
 	// That means, s.ctx should be read-only. And s.ctx is done iff ctx is done.
 	// So we use the original context here instead of creating a copy.
 	s.ctx = ctx
-	s.trReader = &transportReader{
+	s.trReader = transportReader{
 		reader: &recvBufferReader{
 			ctx:     s.ctx,
 			ctxDone: s.ctx.Done(),
-			recv:    s.buf,
+			recv:    &s.buf,
 			closeStream: func(err error) {
 				s.Close(err)
 			},
@@ -823,7 +823,7 @@ func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (*ClientS
 			return nil
 		},
 		onOrphaned: cleanup,
-		wq:         s.wq,
+		wq:         &s.wq,
 	}
 	firstTry := true
 	var ch chan struct{}
@@ -854,7 +854,7 @@ func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (*ClientS
 		transportDrainRequired = t.nextID > MaxStreamID
 
 		s.id = hdr.streamID
-		s.fc = &inFlow{limit: uint32(t.initialWindowSize)}
+		s.fc = inFlow{limit: uint32(t.initialWindowSize)}
 		t.activeStreams[s.id] = s
 		t.mu.Unlock()
 

internal/transport/http2_server.go

@@ -390,16 +390,15 @@ func (t *http2Server) operateHeaders(ctx context.Context, frame *http2.MetaHeade
 	}
 	t.maxStreamID = streamID
 
-	buf := newRecvBuffer()
 	s := &ServerStream{
-		Stream: &Stream{
-			id:  streamID,
-			buf: buf,
-			fc:  &inFlow{limit: uint32(t.initialWindowSize)},
+		Stream: Stream{
+			id: streamID,
+			fc: inFlow{limit: uint32(t.initialWindowSize)},
 		},
 		st:               t,
 		headerWireLength: int(frame.Header().Length),
 	}
+	initRecvBuffer(&s.Stream)
 	var (
 		// if false, content-type was missing or invalid
 		isGRPC      = false
@@ -644,12 +643,12 @@ func (t *http2Server) operateHeaders(ctx context.Context, frame *http2.MetaHeade
 		t.adjustWindow(s, uint32(n))
 	}
 	s.ctxDone = s.ctx.Done()
-	s.wq = newWriteQuota(defaultWriteQuota, s.ctxDone)
-	s.trReader = &transportReader{
+	initWriteQuota(&s.Stream, defaultWriteQuota, s.ctxDone)
+	s.trReader = transportReader{
 		reader: &recvBufferReader{
 			ctx:     s.ctx,
 			ctxDone: s.ctxDone,
-			recv:    s.buf,
+			recv:    &s.buf,
 		},
 		windowHandler: func(n int) {
 			t.updateWindow(s, uint32(n))
@@ -658,7 +657,7 @@ func (t *http2Server) operateHeaders(ctx context.Context, frame *http2.MetaHeade
 	// Register the stream with loopy.
 	t.controlBuf.put(&registerStream{
 		streamID: s.id,
-		wq:       s.wq,
+		wq:       &s.wq,
 	})
 	handle(s)
 	return nil

internal/transport/server_stream.go

@@ -32,7 +32,7 @@ import (
 
 // ServerStream implements streaming functionality for a gRPC server.
 type ServerStream struct {
-	*Stream // Embed for common stream functionality.
+	Stream // Embed for common stream functionality.
 
 	st      internalServerTransport
 	ctxDone <-chan struct{} // closed at the end of stream.  Cache of ctx.Done() (for performance)

internal/transport/transport.go

@@ -68,11 +68,10 @@ type recvBuffer struct {
 	err     error
 }
 
-func newRecvBuffer() *recvBuffer {
-	b := &recvBuffer{
+func initRecvBuffer(s *Stream) {
+	s.buf = recvBuffer{
 		c: make(chan recvMsg, 1),
 	}
-	return b
 }
 
 func (b *recvBuffer) put(r recvMsg) {
@@ -290,10 +289,10 @@ type Stream struct {
 	method       string          // the associated RPC method of the stream
 	recvCompress string
 	sendCompress string
-	buf          *recvBuffer
-	trReader     *transportReader
-	fc           *inFlow
-	wq           *writeQuota
+	buf          recvBuffer
+	trReader     transportReader
+	fc           inFlow
+	wq           writeQuota
 
 	// Callback to state application's intentions to read data. This
 	// is used to adjust flow control, if needed.
@@ -401,11 +400,23 @@ func (s *Stream) read(n int) (data mem.BufferSlice, err error) {
 	return data, nil
 }
 
+// noCopy may be embedded into structs which must not be copied
+// after the first use.
+//
+// See https://golang.org/issues/8005#issuecomment-190753527
+// for details.
+type noCopy struct {
+}
+
+func (*noCopy) Lock()   {}
+func (*noCopy) Unlock() {}
+
 // transportReader reads all the data available for this Stream from the transport and
 // passes them into the decoder, which converts them into a gRPC message stream.
 // The error is io.EOF when the stream is done or another non-nil error if
 // the stream broke.
 type transportReader struct {
+	_      noCopy
 	reader *recvBufferReader
 	// The handler to control the window update procedure for both this
 	// particular stream and the associated transport.

internal/transport/transport_test.go

@@ -88,9 +88,6 @@ func (s *Stream) readTo(p []byte) (int, error) {
 	}
 
 	if data.Len() != len(p) {
-		if err == nil {
-			err = io.ErrUnexpectedEOF
-		}
 		return 0, err
 	}
 
@@ -1856,17 +1853,16 @@ func waitWhileTrue(t *testing.T, condition func() (bool, error)) {
 func (s) TestReadGivesSameErrorAfterAnyErrorOccurs(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
 	defer cancel()
-	testRecvBuffer := newRecvBuffer()
 	s := &Stream{
 		ctx:         ctx,
-		buf:         testRecvBuffer,
 		requestRead: func(int) {},
 	}
-	s.trReader = &transportReader{
+	initRecvBuffer(s)
+	s.trReader = transportReader{
 		reader: &recvBufferReader{
 			ctx:     s.ctx,
 			ctxDone: s.ctx.Done(),
-			recv:    s.buf,
+			recv:    &s.buf,
 		},
 		windowHandler: func(int) {},
 	}
@@ -2591,8 +2587,8 @@ func (s) TestClientHandshakeInfoDialer(t *testing.T) {
 func (s) TestClientDecodeHeaderStatusErr(t *testing.T) {
 	testStream := func() *ClientStream {
 		return &ClientStream{
-			Stream: &Stream{
-				buf: &recvBuffer{
+			Stream: Stream{
+				buf: recvBuffer{
 					c:  make(chan recvMsg),
 					mu: sync.Mutex{},
 				},
@@ -3071,22 +3067,24 @@ func (s) TestCloseSetsConnectionDeadlines(t *testing.T) {
 // number of bytes read for flow control is correct.
 func (s) TestReadMessageHeaderMultipleBuffers(t *testing.T) {
 	headerLen := 5
-	recvBuffer := newRecvBuffer()
-	recvBuffer.put(recvMsg{buffer: make(mem.SliceBuffer, 3)})
-	recvBuffer.put(recvMsg{buffer: make(mem.SliceBuffer, headerLen-3)})
 	bytesRead := 0
 	s := Stream{
 		requestRead: func(int) {},
-		trReader: &transportReader{
-			reader: &recvBufferReader{
-				recv: recvBuffer,
-			},
-			windowHandler: func(i int) {
-				bytesRead += i
-			},
+	}
+	initRecvBuffer(&s)
+	recvBuffer := &s.buf
+	s.trReader = transportReader{
+		reader: &recvBufferReader{
+			recv: recvBuffer,
+		},
+		windowHandler: func(i int) {
+			bytesRead += i
 		},
 	}
 
+	recvBuffer.put(recvMsg{buffer: make(mem.SliceBuffer, 3)})
+	recvBuffer.put(recvMsg{buffer: make(mem.SliceBuffer, headerLen-3)})
+
 	header := make([]byte, headerLen)
 	err := s.ReadMessageHeader(header)
 	if err != nil {
@@ -3189,8 +3187,8 @@ func (s) TestServerSendsRSTAfterDeadlineToMisbehavedClient(t *testing.T) {
 func (s) TestClientTransport_Handle1xxHeaders(t *testing.T) {
 	testStream := func() *ClientStream {
 		return &ClientStream{
-			Stream: &Stream{
-				buf: &recvBuffer{
+			Stream: Stream{
+				buf: recvBuffer{
 					c:  make(chan recvMsg),
 					mu: sync.Mutex{},
 				},