ronantakizawa/codereview-bench huggingface dataset:
Repository: grpc/grpc-go
File: xds/internal/balancer/clusterimpl/clusterimpl.go (Go)

Code before:
		}
	}

	if stopOldLoadReport {
		if b.cancelLoadReport != nil {
			b.cancelLoadReport()
			b.cancelLoadReport = nil
			if !startNewLoadReport {
				// If a new LRS stream will be started later, no need to update
				// it to nil here.
				b.loadWrapper.UpdateLoadStore(nil)
			}
		}
	}
	if startNewLoadReport {
		var loadStore *load.Store
		if b.xdsClient != nil {
			loadStore, b.cancelLoadReport = b.xdsClient.ReportLoad(b.lrsServer)
		}
		b.loadWrapper.UpdateLoadStore(loadStore)
	}

	return nil
}

func (b *clusterImplBalancer) updateClientConnState(s balancer.ClientConnState) error {
	defer clientConnUpdateHook()

	if b.logger.V(2) {
		b.logger.Infof("Received configuration: %s", pretty.ToJSON(s.BalancerConfig))
	}
	newConfig, ok := s.BalancerConfig.(*LBConfig)
	if !ok {
		return fmt.Errorf("unexpected balancer config with type: %T", s.BalancerConfig)
	}

	// Need to check for potential errors at the beginning of this function, so
	// that on errors, we reject the whole config, instead of applying part of
	// it.
	bb := balancer.Get(newConfig.ChildPolicy.Name)
	if bb == nil {
		return fmt.Errorf("child policy %q not registered", newConfig.ChildPolicy.Name)
	}

	if b.xdsClient == nil {
		c := xdsclient.FromResolverState(s.ResolverState)
		if c == nil {
			return balancer.ErrBadResolverState
		}
		b.xdsClient = c
	}

Review comment (suggestion):
There is only one caller for this method now, i.e. `b.UpdateClientConnState()`. I think it makes sense to merge the two methods for better readability.

Code after:
		}
	}

	if stopOldLoadReport {
		if b.cancelLoadReport != nil {
			b.cancelLoadReport()
			b.cancelLoadReport = nil
			if !startNewLoadReport {
				// If a new LRS stream will be started later, no need to update
				// it to nil here.
				b.loadWrapper.UpdateLoadStore(nil)
			}
		}
	}
	if startNewLoadReport {
		var loadStore *load.Store
		if b.xdsClient != nil {
			loadStore, b.cancelLoadReport = b.xdsClient.ReportLoad(b.lrsServer)
		}
		b.loadWrapper.UpdateLoadStore(loadStore)
	}

	return nil
}

func (b *clusterImplBalancer) UpdateClientConnState(s balancer.ClientConnState) error {
	defer clientConnUpdateHook()

	b.mu.Lock()
	b.inhibitPickerUpdates = true
	b.mu.Unlock()
	if b.logger.V(2) {
		b.logger.Infof("Received configuration: %s", pretty.ToJSON(s.BalancerConfig))
	}
	newConfig, ok := s.BalancerConfig.(*LBConfig)
	if !ok {
		return fmt.Errorf("unexpected balancer config with type: %T", s.BalancerConfig)
	}

	// Need to check for potential errors at the beginning of this function, so
	// that on errors, we reject the whole config, instead of applying part of
	// it.
	bb := balancer.Get(newConfig.ChildPolicy.Name)
	if bb == nil {
		return fmt.Errorf("child policy %q not registered", newConfig.ChildPolicy.Name)
	}

	if b.xdsClient == nil {
		c := xdsclient.FromResolverState(s.ResolverState)
		if c == nil {
			return balancer.ErrBadResolverState
Repository: grpc/grpc-go
File: mem/buffer_slice.go (Go)

Code before:
		_, err := wt.WriteTo(w)
		return result, err
	}
	for {
		buf := pool.Get(readAllBufSize)
		// We asked for 32KiB but may have been given a bigger buffer. Use all of it if that's the case.
		*buf = (*buf)[:cap(*buf)]
		usedCap := 0
		for {
			n, err := r.Read((*buf)[usedCap:])
			usedCap += n
			if err != nil {
				if usedCap == 0 {
					// Nothing in this buf, put it back
					pool.Put(buf)
				} else {
					*buf = (*buf)[:usedCap]
					result = append(result, NewBuffer(buf, pool))
				}
				if err == io.EOF {
					err = nil
				}
				return result, err
			}
			if len(*buf) == usedCap {
				result = append(result, NewBuffer(buf, pool))
				break // grab a new buf from pool
			}
		}
	}
}


Review comment (suggestion):
Unless there's an explicit usecase where a caller of `ReadAll` would want to have access to the data on a partial read, it's probably better to have a saner default where the data is freed. This is a convention that's explicitly called out [in the docs for BufferSlice](https://github.com/grpc/grpc-go/blob/master/mem/buffer_slice.go#L36). If you want to break from the convention because there is a hard need for it, that's fine but you'd want to call this out in the docs for `ReadAll`. Otherwise, best to keep it simple


Code after:
		// This is more optimal since wt knows the size of chunks it wants to
		// write and, hence, we can allocate buffers of an optimal size to fit
		// them. E.g. might be a single big chunk, and we wouldn't chop it
		// into pieces.
		w := NewWriter(&result, pool)
		_, err := wt.WriteTo(w)
		return result, err
	}
nextBuffer:
	for {
		buf := pool.Get(readAllBufSize)
		// We asked for 32KiB but may have been given a bigger buffer.
		// Use all of it if that's the case.
		*buf = (*buf)[:cap(*buf)]
		usedCap := 0
		for {
			n, err := r.Read((*buf)[usedCap:])
			usedCap += n
			if err != nil {
				if usedCap == 0 {
					// Nothing in this buf, put it back
					pool.Put(buf)
				} else {
					*buf = (*buf)[:usedCap]
					result = append(result, NewBuffer(buf, pool))
				}
				if err == io.EOF {
					err = nil
				}
				return result, err
			}
			if len(*buf) == usedCap {
				result = append(result, NewBuffer(buf, pool))
				continue nextBuffer
			}
		}
	}
}
Repository: grpc/grpc-go
File: mem/buffer_slice_test.go (Go)

Code before:
}

func (s *stepReader) Read(buf []byte) (int, error) {
	if len(s.reads) == 0 {
		panic("unexpected Read() call")
	}
	read := s.reads[0]
	s.reads = s.reads[1:]
	_, err := rand.Read(buf[:read.n])
	if err != nil {
		panic(err)
	}
	s.read = append(s.read, buf[:read.n]...)
	return read.n, read.err
}

type testPool struct {
	allocated map[*[]byte]struct{}
}

func (t *testPool) Get(length int) *[]byte {
	buf := make([]byte, length)
	t.allocated[&buf] = struct{}{}
	return &buf
}

func (t *testPool) Put(buf *[]byte) {
	if _, ok := t.allocated[buf]; !ok {
		panic("unexpected put")
	}
	delete(t.allocated, buf)
}


Review comment (suggestion):
A short docstring would help future readers understand what this type is all about without spending time to read its implementation. Thanks.

Code after:
// stepReader implements io.Reader that reads specified amount of data and/or
// returns the specified error in specified steps.
// The read data is accumulated in the read field.
type stepReader struct {
	reads []readStep
	read  []byte
}

func (s *stepReader) Read(buf []byte) (int, error) {
	if len(s.reads) == 0 {
		panic("unexpected Read() call")
	}
	read := s.reads[0]
	s.reads = s.reads[1:]
	_, err := rand.Read(buf[:read.n])
	if err != nil {
		panic(err)
	}
	s.read = append(s.read, buf[:read.n]...)
	return read.n, read.err
}

// testPool is an implementation of BufferPool that allows to ensure that:
// - there are matching Put calls for all Get calls.
// - there are no unexpected Put calls.
type testPool struct {
	allocated map[*[]byte]struct{}
}

func (t *testPool) Get(length int) *[]byte {
	buf := make([]byte, length)
	t.allocated[&buf] = struct{}{}
	return &buf
}

func (t *testPool) Put(buf *[]byte) {
	if _, ok := t.allocated[buf]; !ok {
		panic("unexpected put")
	}
	delete(t.allocated, buf)
}
Repository: grpc/grpc-go
File: mem/buffer_slice_test.go (Go)

Code before:
		}
		if err == io.EOF {
			break
		}
		if err != nil {
			t.Fatalf("BufferSlice.Reader() failed unexpectedly: %v", err)
		}
	}
	if !bytes.Equal(gotData, wantData) {
		t.Errorf("BufferSlice.Reader() returned data %v, want %v", string(gotData), string(wantData))
	}

	// Reader should have released its references to the underlying buffers, but
	// bs still holds its reference and it should be able to read data from it.
	gotData = bs.Materialize()
	if !bytes.Equal(gotData, wantData) {
		t.Errorf("BufferSlice.Materialize() = %s, want %s", string(gotData), string(wantData))
	}
}

func (s) TestBufferSlice_ReadAll_Reads(t *testing.T) {
	testcases := []struct {
		name         string
		reads        []readStep
		expectedErr  string
		expectedBufs int
	}{
		{
			name: "EOF",
			reads: []readStep{
				{
					err: io.EOF,
				},
			},
		},
		{
			name: "data,EOF",
			reads: []readStep{
				{
					n: minReadSize,
				},
				{
					err: io.EOF,
				},
			},
			expectedBufs: 1,
		},
		{
			name: "data+EOF",
			reads: []readStep{
				{

Review comment (suggestion):
s/expected/want

Please see style guide recommendation: https://google.github.io/styleguide/go/decisions#got-before-want, where it says:

Where you would write “actual” and “expected”, prefer using the words “got” and “want”, respectively.



Code after:
			break
		}
		if err != nil {
			t.Fatalf("BufferSlice.Reader() failed unexpectedly: %v", err)
		}
	}
	if !bytes.Equal(gotData, wantData) {
		t.Errorf("BufferSlice.Reader() returned data %v, want %v", string(gotData), string(wantData))
	}

	// Reader should have released its references to the underlying buffers, but
	// bs still holds its reference and it should be able to read data from it.
	gotData = bs.Materialize()
	if !bytes.Equal(gotData, wantData) {
		t.Errorf("BufferSlice.Materialize() = %s, want %s", string(gotData), string(wantData))
	}
}

// TestBufferSlice_ReadAll_Reads exercises ReadAll by allowing it to read
// various combinations of data, empty data, EOF.
func (s) TestBufferSlice_ReadAll_Reads(t *testing.T) {
	testcases := []struct {
		name     string
		reads    []readStep
		wantErr  string
		wantBufs int
	}{
		{
			name: "EOF",
			reads: []readStep{
				{
					err: io.EOF,
				},
			},
		},
		{
			name: "data,EOF",
			reads: []readStep{
				{
					n: minReadSize,
				},
				{
					err: io.EOF,
				},
			},
			wantBufs: 1,
		},
		{
			name: "data+EOF",
			reads: []readStep{
				{
Repository: grpc/grpc-go
File: examples/features/compression/server/main.go (Go)

Code before:
/*
 *
 * Copyright 2018 gRPC authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */

// Binary server demonstrates how to install and support compressors for
// incoming RPCs.
package main

import (
	"context"
	"flag"
	"fmt"
	"log"
	"net"

	"google.golang.org/grpc"
	_ "google.golang.org/grpc/encoding/gzip" // Install the gzip compressor

	pb "google.golang.org/grpc/examples/features/proto/echo"
)

var port = flag.Int("port", 50051, "the port to serve on")

type server struct {
	pb.UnimplementedEchoServer
}

func (s *server) UnaryEcho(_ context.Context, in *pb.EchoRequest) (*pb.EchoResponse, error) {
	fmt.Printf("UnaryEcho called with message %q\n", in.GetMessage())
	return &pb.EchoResponse{Message: in.Message}, nil
}

Review comment (suggestion):
I think its not clear how compressor is supported. Perhaps we can try to be a little bit more clear how support happens. may be at line 30?

// Importing the gzip encoding registers it as an available compressor.
// gRPC will automatically negotiate and use gzip if the client supports it.

Code after:
/*
 *
 * Copyright 2018 gRPC authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */

// Binary server demonstrates how to install and support compressors for
// incoming RPCs.
package main

import (
	"context"
	"flag"
	"fmt"
	"log"
	"net"

	"google.golang.org/grpc"
	// Installing the gzip encoding registers it as an available compressor.
	// gRPC will automatically negotiate and use gzip if the client supports it.
	_ "google.golang.org/grpc/encoding/gzip"

	pb "google.golang.org/grpc/examples/features/proto/echo"
)

var port = flag.Int("port", 50051, "the port to serve on")

type server struct {
	pb.UnimplementedEchoServer
}

func (s *server) UnaryEcho(_ context.Context, in *pb.EchoRequest) (*pb.EchoResponse, error) {
	fmt.Printf("UnaryEcho called with message %q\n", in.GetMessage())
Repository: grpc/grpc-go
File: examples/features/deadline/server/main.go (Go)

Code before:
/*
 *
 * Copyright 2018 gRPC authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */

// Binary server demonstrates how to handle RPCs with deadlines and propagate
// deadlines in requests.
package main

import (
	"context"
	"flag"
	"fmt"
	"io"
	"log"
	"net"
	"strings"
	"time"

	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/credentials/insecure"
	"google.golang.org/grpc/status"

	pb "google.golang.org/grpc/examples/features/proto/echo"
)

var port = flag.Int("port", 50052, "port number")

// server is used to implement EchoServer.
type server struct {

Review comment (suggestion):
Same here.

// Binary server demonstrates how to handle RPCs with deadlines and how to propagate
// deadlines in requests.

Code after:
/*
 *
 * Copyright 2018 gRPC authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */

// Binary server demonstrates how to handle RPCs with deadlines and how to
// propagate deadlines in requests.
package main

import (
	"context"
	"flag"
	"fmt"
	"io"
	"log"
	"net"
	"strings"
	"time"

	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/credentials/insecure"
	"google.golang.org/grpc/status"

	pb "google.golang.org/grpc/examples/features/proto/echo"
)

var port = flag.Int("port", 50052, "port number")

// server is used to implement EchoServer.
type server struct {
Repository: grpc/grpc-go
File: examples/features/name_resolving/client/main.go (Go)

Code before:
/*
 *
 * Copyright 2018 gRPC authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */

// Binary client demonstrates how to use different name resolvers to connect
// to a server.
package main

import (
	"context"
	"fmt"
	"log"
	"time"

	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
	ecpb "google.golang.org/grpc/examples/features/proto/echo"
	"google.golang.org/grpc/resolver"
)

const (
	exampleScheme      = "example"
	exampleServiceName = "resolver.example.grpc.io"

	backendAddr = "localhost:50051"
)

func callUnaryEcho(c ecpb.EchoClient, message string) {
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()

Review comment (suggestion):
Nit: instead of `connect to a server`, use something like `resolver server backend addresses`.

Also, the core of this example is an implementation of a custom name resolver. That needs to be mentioned in this comment.

Code after:
/*
 *
 * Copyright 2018 gRPC authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */

// Binary client demonstrates how to use custom name resolvers to resolve
// server backend addresses.
package main

import (
	"context"
	"fmt"
	"log"
	"time"

	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
	ecpb "google.golang.org/grpc/examples/features/proto/echo"
	"google.golang.org/grpc/resolver"
)

const (
	exampleScheme      = "example"
	exampleServiceName = "resolver.example.grpc.io"

	backendAddr = "localhost:50051"
)

func callUnaryEcho(c ecpb.EchoClient, message string) {
	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
	defer cancel()
Repository: grpc/grpc-go
File: balancer_wrapper.go (Go)

Code before:
}

func (acbw *acBalancerWrapper) GetOrBuildProducer(pb balancer.ProducerBuilder) (balancer.Producer, func()) {
	acbw.producersMu.Lock()
	defer acbw.producersMu.Unlock()

	// Look up existing producer from this builder.
	pData := acbw.producers[pb]
	if pData == nil {
		// Not found; create a new one and add it to the producers map.
		p, closeFn := pb.Build(acbw)
		pData = &refCountedProducer{producer: p, close: closeFn}
		acbw.producers[pb] = pData
	}
	// Account for this new reference.
	pData.refs++

	// Return a cleanup function wrapped in a OnceFunc to remove this reference
	// and delete the refCountedProducer from the map if the total reference
	// count goes to zero.
	unref := func() {
		acbw.producersMu.Lock()
		pData.refs--
		if pData.refs == 0 {
			defer pData.close() // Run outside the acbw mutex
			delete(acbw.producers, pb)
		}
		acbw.producersMu.Unlock()
	}
	return pData.producer, grpcsync.OnceFunc(unref)
}

func (acbw *acBalancerWrapper) closeProducers() {
	acbw.producersMu.Lock()
	defer acbw.producersMu.Unlock()
	for pb, pData := range acbw.producers {
		pData.refs = 0
		pData.close()
		delete(acbw.producers, pb)
	}
}


Review comment (suggestion):
Could this become negative if the LB policy calls `close()` after the subchannel state has changed? 

We can check if `acbw.producers[pb] != pData` and exit early to avoid this.

Code after:
}

func (acbw *acBalancerWrapper) GetOrBuildProducer(pb balancer.ProducerBuilder) (balancer.Producer, func()) {
	acbw.producersMu.Lock()
	defer acbw.producersMu.Unlock()

	// Look up existing producer from this builder.
	pData := acbw.producers[pb]
	if pData == nil {
		// Not found; create a new one and add it to the producers map.
		p, closeFn := pb.Build(acbw)
		pData = &refCountedProducer{producer: p, close: closeFn}
		acbw.producers[pb] = pData
	}
	// Account for this new reference.
	pData.refs++

	// Return a cleanup function wrapped in a OnceFunc to remove this reference
	// and delete the refCountedProducer from the map if the total reference
	// count goes to zero.
	unref := func() {
		acbw.producersMu.Lock()
		// If closeProducers has already closed this producer instance, refs is
		// set to 0, so the check after decrementing will never pass, and the
		// producer will not be double-closed.
		pData.refs--
		if pData.refs == 0 {
			defer pData.close() // Run outside the acbw mutex
			delete(acbw.producers, pb)
		}
		acbw.producersMu.Unlock()
	}
	return pData.producer, grpcsync.OnceFunc(unref)
}

func (acbw *acBalancerWrapper) closeProducers() {
	acbw.producersMu.Lock()
	defer acbw.producersMu.Unlock()
	for pb, pData := range acbw.producers {
		pData.refs = 0
		pData.close()
		delete(acbw.producers, pb)
	}
}
Repository: grpc/grpc-go
File: xds/internal/server/listener_wrapper_test.go (Go)

Code before:
		cc, err := grpc.NewClient(lw.Addr().String(), grpc.WithTransportCredentials(insecure.NewCredentials()))
		if err != nil {
			t.Fatalf("grpc.NewClient failed with err: %v", err)
		}
		client := testgrpc.NewTestServiceClient(cc)
		if _, err := client.EmptyCall(ctx, &testpb.Empty{}, grpc.WaitForReady(true)); err != nil {
			t.Fatalf("client.EmptyCall() failed: %v", err)
		}
		cc.Close()
	}

	lisWrapper := lw.(*listenerWrapper)
	// Eventually when the server processes the connection shutdowns, the
	// listener wrapper should clear its references to the wrapped connections.
	var lenConns int
	for ; ctx.Err() == nil; <-time.After(time.Millisecond) {
		lisWrapper.mu.Lock()
		if lenConns = len(lisWrapper.conns); lenConns == 0 {
			lisWrapper.mu.Unlock()
			break
		}
		lisWrapper.mu.Unlock()
	}
	if ctx.Err() != nil {
		t.Fatalf("timeout waiting for lis wrapper conns to clear, size: %v", lenConns)
	}

	server.Stop()
	wg.Wait()
}


Review comment (suggestion):
Maybe a simplification:

```suggestion
	lenConns := 1
	for ; ctx.Err() == nil && lenConns > 0; <-time.After(time.Millisecond) {
		lisWrapper.mu.Lock()
		lenConns = len(lisWrapper.conns)
		lisWrapper.mu.Unlock()
	}
	if lenConns > 0 {
		t.Fatalf("timeout waiting for lis wrapper conns to clear, size: %v", lenConns)
	}
```


Code after:
			t.Fatalf("grpc.NewClient failed with err: %v", err)
		}
		client := testgrpc.NewTestServiceClient(cc)
		if _, err := client.EmptyCall(ctx, &testpb.Empty{}); err != nil {
			t.Fatalf("client.EmptyCall() failed: %v", err)
		}
		cc.Close()
	}

	lisWrapper := lw.(*listenerWrapper)
	// Eventually when the server processes the connection shutdowns, the
	// listener wrapper should clear its references to the wrapped connections.
	lenConns := 1
	for ; ctx.Err() == nil && lenConns > 0; <-time.After(time.Millisecond) {
		lisWrapper.mu.Lock()
		lenConns = len(lisWrapper.conns)
		lisWrapper.mu.Unlock()
	}
	if lenConns > 0 {
		t.Fatalf("timeout waiting for lis wrapper conns to clear, size: %v", lenConns)
	}

	server.Stop()
	wg.Wait()
}