merge master

.travis.yml

@@ -1,19 +1,20 @@
 language: go
 
 go:
-  - 1.6.3
+  - 1.6.x
-  - 1.7
+  - 1.7.x
-  - 1.8
+  - 1.8.x
 
 go_import_path: google.golang.org/grpc
 
 before_install:
-  - go get github.com/golang/lint/golint
+  - if [[ $TRAVIS_GO_VERSION = 1.8* ]]; then go get -u github.com/golang/lint/golint honnef.co/go/tools/cmd/staticcheck; fi
   - go get -u golang.org/x/tools/cmd/goimports github.com/axw/gocov/gocov github.com/mattn/goveralls golang.org/x/tools/cmd/cover
 
 script:
   - '! gofmt -s -d -l . 2>&1 | read'
   - '! goimports -l . | read'
-  - '! golint ./... | grep -vE "(_mock|_string|\.pb)\.go:"'
+  - 'if [[ $TRAVIS_GO_VERSION = 1.8* ]]; then ! golint ./... | grep -vE "(_mock|_string|\.pb)\.go:"; fi'
-  - '! go tool vet -all . 2>&1 | grep -vE "constant [0-9]+ not a string in call to Errorf" | grep -vF .pb.go:' # https://github.com/golang/protobuf/issues/214
+  - 'if [[ $TRAVIS_GO_VERSION = 1.8* ]]; then ! go tool vet -all . 2>&1 | grep -vF .pb.go:; fi' # https://github.com/golang/protobuf/issues/214
   - make test testrace
+  - 'if [[ $TRAVIS_GO_VERSION = 1.8* ]]; then staticcheck -ignore google.golang.org/grpc/transport/transport_test.go:SA2002 ./...; fi' # TODO(menghanl): fix these

README.md

@@ -26,9 +26,13 @@ Documentation
 -------------
 See [API documentation](https://godoc.org/google.golang.org/grpc) for package and API descriptions and find examples in the [examples directory](examples/).
 
+Performance
+-----------
+See the current benchmarks for some of the languages supported in [this dashboard](https://performance-dot-grpc-testing.appspot.com/explore?dashboard=5652536396611584&widget=490377658&container=1286539696).
+
 Status
 ------
-GA
+General Availability [Google Cloud Platform Launch Stages](https://cloud.google.com/terms/launch-stages).
 
 FAQ
 ---

benchmark/benchmark_test.go

@@ -73,7 +73,7 @@ func runStream(b *testing.B, maxConcurrentCalls int) {
 		streamCaller(stream)
 	}
 
-	ch := make(chan int, maxConcurrentCalls*4)
+	ch := make(chan struct{}, maxConcurrentCalls*4)
 	var (
 		mu sync.Mutex
 		wg sync.WaitGroup
@@ -82,11 +82,11 @@ func runStream(b *testing.B, maxConcurrentCalls int) {
 
 	// Distribute the b.N calls over maxConcurrentCalls workers.
 	for i := 0; i < maxConcurrentCalls; i++ {
-		go func() {
 		stream, err := tc.StreamingCall(context.Background())
 		if err != nil {
 			b.Fatalf("%v.StreamingCall(_) = _, %v", tc, err)
 		}
+		go func() {
 			for range ch {
 				start := time.Now()
 				streamCaller(stream)
@@ -100,7 +100,7 @@ func runStream(b *testing.B, maxConcurrentCalls int) {
 	}
 	b.StartTimer()
 	for i := 0; i < b.N; i++ {
-		ch <- i
+		ch <- struct{}{}
 	}
 	b.StopTimer()
 	close(ch)

clientconn_test.go

@@ -295,17 +295,20 @@ func (b *emptyBalancer) Close() error {
 
 func TestNonblockingDialWithEmptyBalancer(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
-	dialDone := make(chan struct{})
+	defer cancel()
+	dialDone := make(chan error)
 	go func() {
+		dialDone <- func() error {
 			conn, err := DialContext(ctx, "Non-Existent.Server:80", WithInsecure(), WithBalancer(newEmptyBalancer()))
 			if err != nil {
-			t.Fatalf("unexpected error dialing connection: %v", err)
+				return err
 			}
-		conn.Close()
+			return conn.Close()
-		close(dialDone)
 		}()
-	<-dialDone
+	}()
-	cancel()
+	if err := <-dialDone; err != nil {
+		t.Fatalf("unexpected error dialing connection: %s", err)
+	}
 }
 
 func TestClientUpdatesParamsAfterGoAway(t *testing.T) {

credentials/oauth/oauth.go

@@ -37,6 +37,7 @@ package oauth
 import (
 	"fmt"
 	"io/ioutil"
+	"sync"
 
 	"golang.org/x/net/context"
 	"golang.org/x/oauth2"
@@ -132,20 +133,27 @@ func NewComputeEngine() credentials.PerRPCCredentials {
 
 // serviceAccount represents PerRPCCredentials via JWT signing key.
 type serviceAccount struct {
+	mu     sync.Mutex
 	config *jwt.Config
+	t      *oauth2.Token
 }
 
-func (s serviceAccount) GetRequestMetadata(ctx context.Context, uri ...string) (map[string]string, error) {
+func (s *serviceAccount) GetRequestMetadata(ctx context.Context, uri ...string) (map[string]string, error) {
-	token, err := s.config.TokenSource(ctx).Token()
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if !s.t.Valid() {
+		var err error
+		s.t, err = s.config.TokenSource(ctx).Token()
 		if err != nil {
 			return nil, err
 		}
+	}
 	return map[string]string{
-		"authorization": token.TokenType + " " + token.AccessToken,
+		"authorization": s.t.TokenType + " " + s.t.AccessToken,
 	}, nil
 }
 
-func (s serviceAccount) RequireTransportSecurity() bool {
+func (s *serviceAccount) RequireTransportSecurity() bool {
 	return true
 }
 
@@ -156,7 +164,7 @@ func NewServiceAccountFromKey(jsonKey []byte, scope ...string) (credentials.PerR
 	if err != nil {
 		return nil, err
 	}
-	return serviceAccount{config: config}, nil
+	return &serviceAccount{config: config}, nil
 }
 
 // NewServiceAccountFromFile constructs the PerRPCCredentials using the JSON key file

rpc_util.go

@@ -161,6 +161,14 @@ type CallOption interface {
 	after(*callInfo)
 }
 
+// EmptyCallOption does not alter the Call configuration.
+// It can be embedded in another structure to carry satellite data for use
+// by interceptors.
+type EmptyCallOption struct{}
+
+func (EmptyCallOption) before(*callInfo) error { return nil }
+func (EmptyCallOption) after(*callInfo)        {}
+
 type beforeCall func(c *callInfo) error
 
 func (o beforeCall) before(c *callInfo) error { return o(c) }

rpc_util_test.go

@@ -47,6 +47,8 @@ import (
 	"google.golang.org/grpc/transport"
 )
 
+var _ CallOption = EmptyCallOption{} // ensure EmptyCallOption implements the interface
+
 func TestSimpleParsing(t *testing.T) {
 	bigMsg := bytes.Repeat([]byte{'x'}, 1<<24)
 	for _, test := range []struct {

server.go

@@ -712,7 +712,6 @@ func (s *Server) processUnaryRPC(t transport.ServerTransport, stream *transport.
 		stream.SetSendCompress(s.opts.cp.Type())
 	}
 	p := &parser{r: stream}
-	for { // TODO: delete
 	pf, req, err := p.recvMsg(s.opts.maxReceiveMessageSize)
 	if err == io.EOF {
 		// The entire stream is done (for unary RPC only).
@@ -845,7 +844,6 @@ func (s *Server) processUnaryRPC(t transport.ServerTransport, stream *transport.
 	// error or allow the stats handler to see it?
 	return t.WriteStatus(stream, status.New(codes.OK, ""))
 }
-}
 
 func (s *Server) processStreamingRPC(t transport.ServerTransport, stream *transport.Stream, srv *service, sd *StreamDesc, trInfo *traceInfo) (err error) {
 	sh := s.opts.statsHandler

test/end2end_test.go

@@ -3030,8 +3030,8 @@ func testMetadataStreamingRPC(t *testing.T, e env) {
 		if err != nil || !reflect.DeepEqual(testMetadata, headerMD) {
 			t.Errorf("#2 %v.Header() = %v, %v, want %v, <nil>", stream, headerMD, err, testMetadata)
 		}
-		var index int
+		err = func() error {
-		for index < len(reqSizes) {
+			for index := 0; index < len(reqSizes); index++ {
 				respParam := []*testpb.ResponseParameters{
 					{
 						Size: proto.Int32(int32(respSizes[index])),
@@ -3040,7 +3040,7 @@ func testMetadataStreamingRPC(t *testing.T, e env) {
 
 				payload, err := newPayload(testpb.PayloadType_COMPRESSABLE, int32(reqSizes[index]))
 				if err != nil {
-				t.Fatal(err)
+					return err
 				}
 
 				req := &testpb.StreamingOutputCallRequest{
@@ -3049,13 +3049,16 @@ func testMetadataStreamingRPC(t *testing.T, e env) {
 					Payload:            payload,
 				}
 				if err := stream.Send(req); err != nil {
-				t.Errorf("%v.Send(%v) = %v, want <nil>", stream, req, err)
+					return fmt.Errorf("%v.Send(%v) = %v, want <nil>", stream, req, err)
-				return
 				}
-			index++
 			}
+			return nil
+		}()
 		// Tell the server we're done sending args.
 		stream.CloseSend()
+		if err != nil {
+			t.Error(err)
+		}
 	}()
 	for {
 		if _, err := stream.Recv(); err != nil {
@@ -3451,7 +3454,8 @@ func testStreamsQuotaRecovery(t *testing.T, e env) {
 			defer wg.Done()
 			payload, err := newPayload(testpb.PayloadType_COMPRESSABLE, 314)
 			if err != nil {
-				t.Fatal(err)
+				t.Error(err)
+				return
 			}
 			req := &testpb.SimpleRequest{
 				ResponseType: testpb.PayloadType_COMPRESSABLE.Enum(),