Point-to-point health check. Client side implementation #993

MakMukhi · 2016-11-18T21:59:15Z

No description provided.

Thomasdezeeuw · 2016-11-19T12:55:39Z

keepalive/keepalive.go

@@ -0,0 +1,30 @@
+package keepalive


Why does this needs to be another package? Can't it ben inside the grpc package?

transport package depends on this. grpc package depends on transport. Putting this in grpc will create a circular dependency.

The separate package looks weird to me as well. We may want to figure out alternatives.

Thomasdezeeuw · 2016-11-19T12:56:11Z

keepalive/keepalive.go

+// Params is used to set keepalive parameters.
+type Params struct {
+	// After a duration of this time the client pings the server to see if the transport is still alive.
+	Ktime time.Duration


Why the K prefix for all the fields?

Thomasdezeeuw · 2016-11-19T12:56:57Z

keepalive/keepalive.go

+
+// DefaultKParams contains default values for keepalive parameters
+var DefaultKParams = Params{
+	Ktime:     time.Duration(math.MaxInt64),           // default to infinite


Why not -1 as infinite?

Setting it to -1 would fire instantly.

Thomasdezeeuw · 2016-11-19T12:57:17Z

keepalive/keepalive.go

+var DefaultKParams = Params{
+	Ktime:     time.Duration(math.MaxInt64),           // default to infinite
+	Ktimeout:  time.Duration(20 * 1000 * 1000 * 1000), // default to 20 seconds
+	KNoStream: false,


This line can be removed, false is the default.

Thomasdezeeuw · 2016-11-19T12:57:46Z

keepalive/keepalive.go

+// DefaultKParams contains default values for keepalive parameters
+var DefaultKParams = Params{
+	Ktime:     time.Duration(math.MaxInt64),           // default to infinite
+	Ktimeout:  time.Duration(20 * 1000 * 1000 * 1000), // default to 20 seconds


Change this to 20 * time.Second, the is is much clearer and won't need the comment.

Thomasdezeeuw · 2016-11-19T12:59:36Z

transport/http2_client.go

@@ -98,6 +100,11 @@ type http2Client struct {

 	creds []credentials.PerRPCCredentials

+	// activity counter
+	activity *uint64


Add a comment about being atomic?

Thomasdezeeuw · 2016-11-19T13:02:10Z

transport/http2_client.go

@@ -98,6 +100,11 @@ type http2Client struct {

 	creds []credentials.PerRPCCredentials

+	// activity counter
+	activity *uint64


It's common to just use an uint64, rather then a reference. For example see http.Server code: https://github.com/golang/go/blob/7dc97d9e328edc800e2ce41d5b211ef4e0ef41d0/src/net/http/server.go#L2363-L2364.

Thomasdezeeuw · 2016-11-19T13:05:09Z

keepalive/keepalive.go

+}
+
+// Mu is a mutex to protect Enabled variable
+var Mu = sync.Mutex{}


Rather then exposing this lock, why not use functions for it. e.g.

func Enabled() bool {} func Enable() {} func Disable() {}

ejona86 · 2017-01-20T18:59:40Z

keepalive/keepalive.go

+// Params is used to set keepalive parameters.
+type Params struct {
+	// After a duration of this time the client pings the server to see if the transport is still alive.
+	Time time.Duration


Need to think about what happens when these are 0.

ejona86 · 2017-01-20T18:59:49Z

keepalive/keepalive.go

+}
+
+// DefaultParams contains default values for keepalive parameters.
+var DefaultParams = Params{


Consider moving to transport.

ejona86 · 2017-01-20T19:03:15Z

keepalive/keepalive.go

+var enable = false
+
+// Enabled exposes the value of enable variable.
+func Enabled() bool {


This seems unnecessary. I don't know a case where WithKeepaliveParams wouldn't work.

ejona86 · 2017-01-20T19:25:22Z

transport/http2_client.go

@@ -98,6 +100,11 @@ type http2Client struct {

 	creds []credentials.PerRPCCredentials

+	// Counter to keep track of activity(reading and writing on transport).


This should only be used for reading. Writing does not prevent the need for keepalive ping.

ejona86 · 2017-01-20T20:10:12Z

transport/http2_client.go

@@ -976,6 +989,7 @@ func (t *http2Client) reader() {
 	// loop to keep reading incoming messages on this transport.


Update activity after handleSettings?

ejona86 · 2017-01-20T20:22:34Z

transport/http2_client.go

+			t.mu.Unlock()
+			// Get the activity counter value and reset it.
+			a := atomic.SwapUint64(&t.activity, 0)
+			if a > 0 || (!t.kp.PermitWithoutStream && ns < 1) {


This doesn't seem to follow this part of the spec:

Since keepalive is not occurring on HTTP/2 connections without any streams, there will be a higher chance of failure for new RPCs following a long period of inactivity. To reduce the tail latency for these RPCs, it is important to not reset the keepalive time when a connection becomes active; if a new stream is created and there has been greater than keepalive time since the last read byte, then a keepalive PING should be sent (ideally before the HEADERS frame). Doing so detects the broken connection with a latency of keepalive timeout instead of keepalive time + timeout.

ejona86 · 2017-01-30T20:40:08Z

transport/http2_client.go

 		kp = opts.KeepaliveParams
+		kp.Validate()


This mutates an object from the user. Should there be some copy involved?

ejona86 · 2017-01-30T20:43:17Z

keepalive/keepalive.go

+	// Infinity is the default value of keepalive time.
+	Infinity = time.Duration(math.MaxInt64)
+	// TwentySec is the default value of timeout.
+	TwentySec = time.Duration(20 * time.Second)


This is a really poor constant to export. There is no way to change it to 30 seconds or similar in the future. One option is to export the defaults. A better option is to not export them.

Two ways to remove the export: 1) in the transport, create a zero-filled Params{} and then call Validate(), 2) move the Validate logic to transport. I have preference to 2 because it also removes the Validate export.

ejona86 · 2017-01-30T20:50:25Z

transport/http2_client.go

 		// Prevent the timer from firing, ever.
 		if !timer.Stop() {
 			<-timer.C
 		}
 	}
 	isPingSent := false
 	keepalivePing := &ping{data: [8]byte{}}
+	// select toggles between control channel and writable chanel.


nit: s/chanel/channel/

ejona86 · 2017-01-30T21:02:32Z

transport/http2_client.go

+				if atomic.LoadUint32(&t.keepaliveSkipped) == 1 {
+					// Reset the timer to 0 so that it fires.
+					if !timer.Stop() {
+						<-timer.C


Could you make a note that this is safe since the timer will always be Reset in case <-timer.C?

Otherwise it could deadlock. https://play.golang.org/p/4CHuWZJFwu

ejona86 · 2017-01-30T21:05:07Z

transport/http2_client.go

 			}
+			wchan = t.writableChan
+			cchan = nil
+			continue


This continue has no effect. Ditto in next case. Remove?

ejona86 · 2017-01-30T21:09:06Z

keepalive/keepalive.go

+		p.Time = Infinity
+	}
+	if p.Timeout == 0 {
+		p.Time = TwentySec


s/Time/Timeout/

ejona86 · 2017-01-30T21:10:04Z

transport/http2_client.go

@@ -1070,64 +1091,94 @@ func (t *http2Client) applySettings(ss []http2.Setting) {
 // frames (e.g., window update, reset stream, setting, etc.) to the server.
 func (t *http2Client) controller() {
 	timer := time.NewTimer(t.kp.Time)
-	if !keepalive.Enabled() {
+	if t.kp.Timeout == keepalive.Infinity {


s/Timeout/Time/

ejona86 · 2017-01-30T21:19:08Z

transport/http2_client.go

 		case <-timer.C:
 			t.mu.Lock()
 			ns := len(t.activeStreams)
 			t.mu.Unlock()
+			if !t.kp.PermitWithoutStream && ns < 1 {
+				timer.Reset(t.kp.Time)


Hmm... Won't this cause the timer to continue firing on an idle channel? It also makes the various states hard to follow. Could this just be reset to Inifinity?

ejona86

Pretty close

ejona86 · 2017-02-10T21:39:51Z

keepalive.go

+type KeepaliveParameters struct {
+	// After a duration of this time the client pings the server to see if the transport is still alive.
+	Time time.Duration
+	// After having pinged fot keepalive check, the client waits for a duration of keepalive_timeout before closing the transport.


ejona86 · 2017-02-10T21:40:21Z

keepalive.go

+type KeepaliveParameters struct {
+	// After a duration of this time the client pings the server to see if the transport is still alive.
+	Time time.Duration
+	// After having pinged fot keepalive check, the client waits for a duration of keepalive_timeout before closing the transport.


s/keepalive_timeout/Timeout/ ?

ejona86 · 2017-02-10T21:41:48Z

keepalive.go

+type KeepaliveParameters struct {
+	// After a duration of this time the client pings the server to see if the transport is still alive.
+	Time time.Duration
+	// After having pinged fot keepalive check, the client waits for a duration of keepalive_timeout before closing the transport.


I don't think this documentation is saying what you hope. It actually says it will always close the connection after a keepalive check. You need something saying that it only matters when nothing is received by the timeout.

ejona86 · 2017-02-10T21:42:45Z

keepalive.go

+
+// KeepaliveParameters is used to set keepalive parameters.
+type KeepaliveParameters struct {
+	// After a duration of this time the client pings the server to see if the transport is still alive.


After a duration of this time after having receiving nothing. (I'm pointing out the issue; my text still needs rewording)

ejona86 · 2017-02-10T21:45:08Z

keepalive.go

+	"time"
+)
+
+// KeepaliveParameters is used to set keepalive parameters.


How about explaining a bit more.

KeepaliveParameters configures how the ClientConn will actively probe to notice when a connection is broken and to cause activity so intermediaries are aware the connection is still in use.

ejona86 · 2017-02-10T21:50:14Z

transport/http2_client.go

+	// keepaliveSkipped = 1 means skipped
+	keepaliveSkipped uint32 // accessed atomically
+	// keepalive parameters.
+	kp           *KeepaliveParameters


Maybe get rid of the pointer? It is probably better to inline it from a locality perspective, but it would also mean validate() wouldn't need as much thought. I think it is safe today, but you are mutating a structure that is used from multiple threads but without any synchronization; it would be trivial to accidentally break that in the future.

Note this also would make defaultKeepaliveParams unnecessary.

ejona86 · 2017-02-10T22:01:58Z

transport/http2_client.go

+			if !timer.Stop() {
+				select {
+				case <-timer.C:
+				default:


So note this prevents this goroutine from blocking permanently, but it may not free the timer. The Stop() may not process immediately, in which case <-timer.C doesn't have a value yet, but it will in the future.

It's probably better to either get rid of the if t.kp.Time == infinity { optimization above or check a second time here (you could have a variable, say isTimerUsed).

Since we exit the controller function right after, wouldn't the timer be eventually gc'd. Also, since nowhere outside the function is timer.C accessed, it should be safe even if timer.C later gets a value. In fact, maybe we can get rid of the case <- timer.c: altogether?

You're forgetting there is a sender of that value. The sender could keep the timer alive and prevent GC.

Now, I happen to know that the channel is buffered, and so in this specific case the sender will be able to queue an entry and then remove the timer from its data structures and then the timer could be GCed.

ejona86 · 2017-02-10T22:22:44Z

keepalive.go

+
+// KeepaliveParameters is used to set keepalive parameters.
+type KeepaliveParameters struct {
+	// After a duration of this time the client pings the server to see if the transport is still alive.


It seems there should be some mention of the current defaults. Here and below. Try to use language that implies the defaults may change, like "the current default is" and similar.

MakMukhi · 2017-02-10T22:50:38Z

Thanks for the comments, Eric. On it!

…hecks.

tamird · 2017-02-28T19:51:40Z

keepalive/keepalive.go

+type ClientParameters struct {
+	// After a duration of this time the client pings the server to see if the transport is still alive.
+	Time time.Duration
+	// After having pinged fot keepalive check, the client waits for a duration of keepalive_timeout before closing the transport.


tamird · 2017-02-28T19:51:42Z

keepalive/keepalive.go

+	Time time.Duration
+	// 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.


missing space

tamird · 2017-02-28T19:51:52Z

keepalive/keepalive.go

+type ClientParameters struct {
+	// After a duration of this time the client pings the server to see if the transport is still alive.
+	Time time.Duration
+	// After having pinged fot keepalive check, the client waits for a duration of keepalive_timeout before closing the transport.


would be nice to wrap this.

tamird · 2017-02-28T19:52:13Z

transport/http2_client.go

@@ -182,6 +196,14 @@ func newHTTP2Client(ctx context.Context, addr TargetInfo, opts ConnectOptions) (
 	if opts.UserAgent != "" {
 		ua = opts.UserAgent + " " + ua
 	}
+	kp := opts.KeepaliveParams
+	// Validate keepalive parameters.


this is setting defaults, not validating.

tamird · 2017-02-28T19:52:53Z

transport/http2_client.go

@@ -80,6 +82,8 @@ type http2Client struct {
 	// goAway is closed to notify the upper layer (i.e., addrConn.transportMonitor)
 	// that the server sent GoAway on this transport.
 	goAway chan struct{}
+	// awakenKeepalive is used to tell keepalive goroutine to reset keepalive timer.
+	awakenKeepalive chan int


why is this int instead of struct{}? the value is never used.

tamird · 2017-02-28T19:56:40Z

transport/http2_client.go

+		return
+	}
+	p := &ping{data: [8]byte{}}
+	timer := time.NewTimer(t.kp.Time)


defer timer.Stop() below this line.

No all exit paths of this routine have a timer running. Trying to stop and empty timer.C of an already stopped timer leads to a deadlock.
Doing this in defer would mean making sure that all exit paths of the routine reset the timer(if one is not running already). Future updates to the code might miss that and get into trouble.

tamird · 2017-02-28T19:57:56Z

transport/http2_client.go

+			t.mu.Lock()
+			if len(t.activeStreams) < 1 && !t.kp.PermitWithoutStream {
+				// Make awakenKeepalive writable.
+				<-t.awakenKeepalive


why does this happen under lock? seems clearer to write:

t.mu.Lock() dormant := len(t.activeStreams) < 1 && !t.kp.PermitWithoutStream t.mu.Unlock() if dormant { ... } else { t.controlBuf.put(p) }

Emptying of the awakenKeepalive channel needs to happen inside the lock so that some other thread creating a new stream sees the channel as writable if the number of streams go form 0 to 1.

tamird · 2017-02-28T19:58:57Z

transport/http2_client.go

+			}
+
+			// By the time control gets here a ping has been sent one way or the other.
+			timer.Reset(t.kp.Timeout)


why is there so much duplicated code here?

Duplication here is by choice to keep the structure of the routine simple so that new users can understand the logic behind easily.
To reuse the outer select we'll need to have some additional book-keeping about if the ping was sent or not.

tamird · 2017-02-28T19:59:18Z

transport/transport.go

+const (
+	infinity                = time.Duration(math.MaxInt64)
+	defaultKeepaliveTime    = infinity
+	defaultKeepaliveTimeout = time.Duration(20 * time.Second)


time.Duration is not necessary here.

tamird · 2017-02-28T20:00:02Z

transport/transport_test.go

+	}
+	defer conn.Close()
+	// Sleep for keepalive to close the connection.
+	time.Sleep(4 * time.Second)


perhaps this should be a function of the settings above, rather than hardcoded.

there are 4 instances of this.

menghanl · 2017-03-03T00:51:51Z

transport/http2_client.go

@@ -1104,6 +1138,63 @@ func (t *http2Client) controller() {
 	}
 }

+// keepalive running in a separate goroutune makes sure the connection is alive by sending pings.
+func (t *http2Client) keepalive() {
+	if t.kp.Time == infinity {


Move this check before goroutine is created?

menghanl · 2017-03-03T01:27:51Z

transport/http2_client.go

+	if len(t.activeStreams) == 1 {
+		select {
+		case t.awakenKeepalive <- struct{}{}:
+			t.framer.writePing(true, false, [8]byte{})


We don't need to flush the ping, we will flush the headers later.

menghanl · 2017-03-03T01:28:11Z

transport/http2_client.go

@@ -992,6 +1025,7 @@ func (t *http2Client) reader() {
 	// loop to keep reading incoming messages on this transport.
 	for {
 		frame, err := t.framer.readFrame()
+		atomic.AddUint64(&t.activity, 1)


menghanl · 2017-03-03T21:19:43Z

transport/http2_client.go

+	for {
+		select {
+		case <-timer.C:
+			if a := atomic.SwapUint32(&t.activity, 0); a == 1 {


CAS(a, 1, 0)

menghanl

LGTM

Switch from Peter's fork of grpc to my own fork, since Peter's out of office. My fork is https://github.com/grpc/grpc-go head + Peter's one commit increasing the per-stream flow-control window. The motivation of syncing to newer gRPC is to include grpc/grpc-go#993 for helping with cockroachdb/cockroach#13989 - we will move to gRPC internal heartbeats instead of using our own connection heartbeats. Other dep updates: - the Lightstep update is a single commit that seems innocuous

Switch from Peter's fork of grpc to my own fork, since Peter's out of office. My fork is https://github.com/grpc/grpc-go head + Peter's one commit increasing the per-stream flow-control window. The motivation of syncing to newer gRPC is to include grpc/grpc-go#993 for helping with cockroachdb#13989 - we will move to gRPC internal heartbeats instead of using our own connection heartbeats in a future cockroach commit. Other dep updates: - the Lightstep update is a single commit that seems innocuous

MakMukhi added 6 commits November 17, 2016 17:50

Keepalive client-side implementation

e58450b

Added synchronization

a417b82

sending ping the right way

93111e4

refactored and rectified test failures

058907d

go tool vet rectification

45907c2

go tool vet again

97fb58a

Thomasdezeeuw reviewed Nov 19, 2016

View reviewed changes

MakMukhi added 3 commits November 21, 2016 14:18

Post review updates

901cdf6

code cleaning

eeb6f5b

Comment formatting

49d2a88

ejona86 reviewed Jan 20, 2017

View reviewed changes

MakMukhi added 5 commits January 27, 2017 11:15

intermediate change

b2448f6

conflict resolve

e78a1f3

post review updates

5fcb58f

update activity on recieveing settings

0ba0eaf

const variable name change for goimports

c8983ba

ejona86 reviewed Jan 30, 2017

View reviewed changes

post review updates

0bdf059

ejona86 reviewed Feb 10, 2017

View reviewed changes

post review update

336b4ea

ejona86 approved these changes Feb 13, 2017

View reviewed changes

MakMukhi added 2 commits February 28, 2017 11:49

Implementation changes;Using an independent goroutine for keepalive c…

bc3cca9

…hecks.

removing dead code

3ad7810

tamird reviewed Feb 28, 2017

View reviewed changes

MakMukhi added 2 commits February 28, 2017 12:41

more refactoring

e328ede

post-review updates

f220619

menghanl reviewed Mar 3, 2017

View reviewed changes

post-review updates

25f14b7

menghanl reviewed Mar 3, 2017

View reviewed changes

MakMukhi added 2 commits March 6, 2017 13:08

post-review update

9d23576

conflict resolve

661dbbc

menghanl reviewed Mar 7, 2017

View reviewed changes

MakMukhi merged commit 4eaacfe into grpc:master Mar 7, 2017

andreimatei mentioned this pull request Mar 17, 2017

distsql: tpch query 7 results in context cancelled error cockroachdb/cockroach#13989

Closed

andreimatei mentioned this pull request Mar 21, 2017

vendor: update grpc-go and update non-pinned deps cockroachdb/cockroach#14297

Merged

MakMukhi deleted the mmukhi_keepalive_client branch May 4, 2018 02:07

lock bot locked as resolved and limited conversation to collaborators Oct 31, 2018

		@@ -98,6 +100,11 @@ type http2Client struct {

		creds []credentials.PerRPCCredentials

		// Counter to keep track of activity(reading and writing on transport).

		@@ -976,6 +989,7 @@ func (t *http2Client) reader() {
		// loop to keep reading incoming messages on this transport.

Point-to-point health check. Client side implementation #993

Point-to-point health check. Client side implementation #993

Conversation

MakMukhi commented Nov 18, 2016

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ejona86 left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

MakMukhi commented Feb 10, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

menghanl left a comment

Choose a reason for hiding this comment