code cleaning

clientconn.go

@@ -234,7 +234,7 @@ func WithUserAgent(s string) DialOption {
 // WithKeepaliveParams returns a DialOption that specifies a user agent string for all the RPCs.
 func WithKeepaliveParams(k keepalive.Params) DialOption {
 	return func(o *dialOptions) {
-		o.copts.KParams = k
+		o.copts.KeepaliveParams = k
 	}
 }
 

keepalive/keepalive.go

@@ -13,11 +13,11 @@ type Params struct {
 	// After having pinged fot keepalive check, the client waits for a duration of keepalive_timeout before closing the transport.
 	Timeout time.Duration
 	//If true, client runs keepalive checks even with no active RPCs.
-	PermitNoStream bool
+	PermitWithoutStream bool
 }
 
-// DefaultKParams contains default values for keepalive parameters
-var DefaultKParams = Params{
+// DefaultParams contains default values for keepalive parameters
+var DefaultParams = Params{
 	Time:    time.Duration(math.MaxInt64), // default to infinite
 	Timeout: time.Duration(20 * time.Second),
 }

transport/http2_client.go

@@ -103,7 +103,7 @@ type http2Client struct {
 	// activity counter
 	activity uint64 // accessed atomically
 	// keepalive parameters
-	keepaliveParams keepalive.Params
+	kp keepalive.Params
 
 	mu            sync.Mutex     // guard the following variables
 	state         transportState // the state of underlying connection
@@ -186,9 +186,9 @@ func newHTTP2Client(ctx context.Context, addr TargetInfo, opts ConnectOptions) (
 	if opts.UserAgent != "" {
 		ua = opts.UserAgent + " " + ua
 	}
-	kp := keepalive.DefaultKParams
-	if opts.KParams != (keepalive.Params{}) {
-		kp = opts.KParams
+	kp := keepalive.DefaultParams
+	if opts.KeepaliveParams != (keepalive.Params{}) {
+		kp = opts.KeepaliveParams
 	}
 	var buf bytes.Buffer
 	t := &http2Client{
@@ -217,7 +217,7 @@ func newHTTP2Client(ctx context.Context, addr TargetInfo, opts ConnectOptions) (
 		creds:           opts.PerRPCCredentials,
 		maxStreams:      math.MaxInt32,
 		streamSendQuota: defaultWindowSize,
-		keepaliveParams: kp,
+		kp:              kp,
 	}
 	// Start the reader goroutine for incoming message. Each transport has
 	// a dedicated goroutine which reads HTTP2 frame from network. Then it
@@ -844,8 +844,6 @@ func (t *http2Client) handlePing(f *http2.PingFrame) {
 	pingAck := &ping{ack: true}
 	copy(pingAck.data[:], f.Data[:])
 	t.controlBuf.put(pingAck)
-	// activity++
-	atomic.AddUint64(&t.activity, 1)
 }
 
 func (t *http2Client) handleGoAway(f *http2.GoAwayFrame) {
@@ -1070,9 +1068,7 @@ func (t *http2Client) applySettings(ss []http2.Setting) {
 // controller running in a separate goroutine takes charge of sending control
 // frames (e.g., window update, reset stream, setting, etc.) to the server.
 func (t *http2Client) controller() {
-	// Activity value seen by timer
-	ta := atomic.LoadUint64(&t.activity)
-	timer := time.NewTimer(t.keepaliveParams.Time)
+	timer := time.NewTimer(t.kp.Time)
 	if !keepalive.Enabled() {
 		// Prevent the timer from firing, ever.
 		if !timer.Stop() {
@@ -1115,24 +1111,22 @@ func (t *http2Client) controller() {
 			t.mu.Lock()
 			ns := len(t.activeStreams)
 			t.mu.Unlock()
-			// Global activity value.
-			ga := atomic.LoadUint64(&t.activity)
-			if ga > ta || (!t.keepaliveParams.PermitNoStream && ns < 1) {
-				timer.Reset(t.keepaliveParams.Time)
+			// Get the activity counter value and reset it.
+			a := atomic.SwapUint64(&t.activity, 0)
+			if a > 0 || (!t.kp.PermitWithoutStream && ns < 1) {
+				timer.Reset(t.kp.Time)
 				isPingSent = false
 			} else {
 				if !isPingSent {
 					// send ping
 					t.controlBuf.put(keepalivePing)
 					isPingSent = true
-					timer.Reset(t.keepaliveParams.Timeout)
+					timer.Reset(t.kp.Timeout)
 				} else {
 					t.Close()
 					continue
 				}
 			}
-			// Update timer activity counter.
-			ta = ga
 		case <-t.shutdownChan:
 			return
 		}

transport/transport.go

@@ -382,7 +382,7 @@ type ConnectOptions struct {
 	// TransportCredentials stores the Authenticator required to setup a client connection.
 	TransportCredentials credentials.TransportCredentials
 	// Keepalive parameters
-	KParams keepalive.Params
+	KeepaliveParams keepalive.Params
 }
 
 // TargetInfo contains the information of the target such as network address and metadata.

transport/transport_test.go

@@ -274,7 +274,7 @@ func setUpWithOptions(t *testing.T, port int, maxStreams uint32, ht hType, copts
 	return server, ct
 }
 
-func setUpWithNoPingServer(t *testing.T, copts ConnectOptions, done chan net.Conn) *http2Client {
+func setUpWithNoPingServer(t *testing.T, copts ConnectOptions, done chan net.Conn) ClientTransport {
 	lis, err := net.Listen("tcp", "localhost:0")
 	if err != nil {
 		t.Fatalf("Failed to listen: %v", err)
@@ -290,30 +290,23 @@ func setUpWithNoPingServer(t *testing.T, copts ConnectOptions, done chan net.Con
 		}
 		done <- conn
 	}()
-	tr, err := newHTTP2Client(context.Background(), TargetInfo{Addr: lis.Addr().String()}, copts)
+	tr, err := NewClientTransport(context.Background(), TargetInfo{Addr: lis.Addr().String()}, copts)
 	if err != nil {
 		t.Fatalf("Failed to dial: %v", err)
 	}
-	cT := tr.(*http2Client)
-	// Assert client transport is healthy
-	cT.mu.Lock()
-	defer cT.mu.Unlock()
-	if cT.state != reachable {
-		t.Fatalf("Client transport not healthy")
-	}
-	return cT
+	return tr
 }
 
 func TestKeepaliveClientClosesIdleTransport(t *testing.T) {
 	keepalive.Enable()
 	defer keepalive.Disable()
 	done := make(chan net.Conn, 1)
-	cT := setUpWithNoPingServer(t, ConnectOptions{KParams: keepalive.Params{
-		Time:           2 * time.Second, // keepalive time = 2 sec
-		Timeout:        1 * time.Second, // keepalive timeout = 1 sec
-		PermitNoStream: true,            // run keepalive even with no RPCs
+	tr := setUpWithNoPingServer(t, ConnectOptions{KeepaliveParams: keepalive.Params{
+		Time:                2 * time.Second, // keepalive time = 2 sec
+		Timeout:             1 * time.Second, // keepalive timeout = 1 sec
+		PermitWithoutStream: true,            // run keepalive even with no RPCs
 	}}, done)
-	defer cT.Close()
+	defer tr.Close()
 	conn, ok := <-done
 	if !ok {
 		t.Fatalf("Server didn't return connection object")
@@ -322,9 +315,10 @@ func TestKeepaliveClientClosesIdleTransport(t *testing.T) {
 	// Sleep for keepalive to close the connection
 	time.Sleep(4 * time.Second)
 	// Assert that the connection was closed
-	cT.mu.Lock()
-	defer cT.mu.Unlock()
-	if cT.state == reachable {
+	ct := tr.(*http2Client)
+	ct.mu.Lock()
+	defer ct.mu.Unlock()
+	if ct.state == reachable {
 		t.Fatalf("Test Failed: Expected client transport to have closed.")
 	}
 }
@@ -333,12 +327,12 @@ func TestKeepaliveClientStaysHealthyOnIdleTransport(t *testing.T) {
 	keepalive.Enable()
 	defer keepalive.Disable()
 	done := make(chan net.Conn, 1)
-	cT := setUpWithNoPingServer(t, ConnectOptions{KParams: keepalive.Params{
-		Time:           2 * time.Second, // keepalive time = 2 sec
-		Timeout:        1 * time.Second, // keepalive timeout = 1 sec
-		PermitNoStream: false,           // don't run keepalive even with no RPCs
+	tr := setUpWithNoPingServer(t, ConnectOptions{KeepaliveParams: keepalive.Params{
+		Time:                2 * time.Second, // keepalive time = 2 sec
+		Timeout:             1 * time.Second, // keepalive timeout = 1 sec
+		PermitWithoutStream: false,           // don't run keepalive even with no RPCs
 	}}, done)
-	defer cT.Close()
+	defer tr.Close()
 	conn, ok := <-done
 	if !ok {
 		t.Fatalf("server didn't reutrn connection object")
@@ -347,9 +341,10 @@ func TestKeepaliveClientStaysHealthyOnIdleTransport(t *testing.T) {
 	// Give keepalive some time
 	time.Sleep(4 * time.Second)
 	// Assert that connections is still healthy
-	cT.mu.Lock()
-	defer cT.mu.Unlock()
-	if cT.state != reachable {
+	ct := tr.(*http2Client)
+	ct.mu.Lock()
+	defer ct.mu.Unlock()
+	if ct.state != reachable {
 		t.Fatalf("Test failed: Expected client transport to be healthy.")
 	}
 }
@@ -358,28 +353,29 @@ func TestKeepaliveClientClosesWithActiveStreams(t *testing.T) {
 	keepalive.Enable()
 	defer keepalive.Disable()
 	done := make(chan net.Conn, 1)
-	cT := setUpWithNoPingServer(t, ConnectOptions{KParams: keepalive.Params{
-		Time:           2 * time.Second, // keepalive time = 2 sec
-		Timeout:        1 * time.Second, // keepalive timeout = 1 sec
-		PermitNoStream: false,           // don't run keepalive even with no RPCs
+	tr := setUpWithNoPingServer(t, ConnectOptions{KeepaliveParams: keepalive.Params{
+		Time:                2 * time.Second, // keepalive time = 2 sec
+		Timeout:             1 * time.Second, // keepalive timeout = 1 sec
+		PermitWithoutStream: false,           // don't run keepalive even with no RPCs
 	}}, done)
-	defer cT.Close()
+	defer tr.Close()
 	conn, ok := <-done
 	if !ok {
 		t.Fatalf("Server didn't return connection object")
 	}
 	defer conn.Close()
 	// create a stream
-	_, err := cT.NewStream(context.Background(), &CallHdr{})
+	_, err := tr.NewStream(context.Background(), &CallHdr{})
 	if err != nil {
 		t.Fatalf("Failed to create a new stream: %v", err)
 	}
 	// Give keepalive some time
 	time.Sleep(4 * time.Second)
 	// Asser that transport was closed
-	cT.mu.Lock()
-	defer cT.mu.Unlock()
-	if cT.state == reachable {
+	ct := tr.(*http2Client)
+	ct.mu.Lock()
+	defer ct.mu.Unlock()
+	if ct.state == reachable {
 		t.Fatalf("Test failed: Expected client transport to have closed.")
 	}
 }
@@ -387,20 +383,20 @@ func TestKeepaliveClientClosesWithActiveStreams(t *testing.T) {
 func TestKeepaliveClientStaysHealthyWithResponsiveServer(t *testing.T) {
 	keepalive.Enable()
 	defer keepalive.Disable()
-	s, tr := setUpWithOptions(t, 0, math.MaxUint32, normal, ConnectOptions{KParams: keepalive.Params{
-		Time:           2 * time.Second, // keepalive time = 2 sec
-		Timeout:        1 * time.Second, // keepalive timeout = 1 sec
-		PermitNoStream: true,            // don't run keepalive even with no RPCs
+	s, tr := setUpWithOptions(t, 0, math.MaxUint32, normal, ConnectOptions{KeepaliveParams: keepalive.Params{
+		Time:                2 * time.Second, // keepalive time = 2 sec
+		Timeout:             1 * time.Second, // keepalive timeout = 1 sec
+		PermitWithoutStream: true,            // don't run keepalive even with no RPCs
 	}})
 	defer s.stop()
 	defer tr.Close()
 	// Give keep alive some time
 	time.Sleep(4 * time.Second)
 	// Assert that transport is healthy
-	cT := tr.(*http2Client)
-	cT.mu.Lock()
-	defer cT.mu.Unlock()
-	if cT.state != reachable {
+	ct := tr.(*http2Client)
+	ct.mu.Lock()
+	defer ct.mu.Unlock()
+	if ct.state != reachable {
 		t.Fatalf("Test failed: Expected client transport to be healthy.")
 	}
 }