server push schedule commands. #669

disksing · 2017-06-30T08:58:07Z

Another approach of #661.
Now dispatch() does not return responses, all responses transfers through heartbeatStream.

nolouch

the rest LGTM! this approach is more clear than I do.

nolouch · 2017-06-30T10:54:48Z

server/grpc_service.go

@@ -290,27 +295,13 @@ func (s *Server) RegionHeartbeat(server pdpb.PD_RegionHeartbeatServer) error {
 			continue
 		}

-		var resp *pdpb.RegionHeartbeatResponse
-		resp, err = cluster.handleRegionHeartbeat(region)
+		err = cluster.handleRegionHeartbeat(region)
 		if err != nil {
 			msg := errors.Trace(err).Error()
 			err = sendErrorRegionHeartbeatResponse(server, s.clusterID, pdpb.ErrorType_UNKNOWN, msg)


should send error by hbstreams in here?

It's ok, server can be accessed by both here and there.

function name is too long.

Btw, do we need to close the stream in the heartbeat loop?

I think no need to drop the stream. The stream itself is in good condition, only some requests are not able to respond.

siddontang · 2017-06-30T14:22:29Z

server/coordinator.go

+	for {
+		select {
+		case update := <-s.streamCh:
+			s.streams[update.storeID] = update.stream


do we need to close old one if have?

No need to. ServerStream does not have Close function, it will close automatically after eof.

siddontang · 2017-06-30T14:42:03Z

server/coordinator.go

+					delete(s.streams, storeID)
+				}
+			} else {
+				log.Debugf("heartbeat stream not found for store %v, skip send message", storeID)


should add a metric for it.

siddontang · 2017-07-01T01:13:25Z

server/coordinator.go

+			} else {
+				log.Debugf("heartbeat stream not found for store %v, skip send message", storeID)
+			}
+		case <-s.ctx.Done():


maybe we should use a sub context to avoid using the same context in different goroutines.

siddontang · 2017-07-03T01:40:33Z

PTAL @overvenus

overvenus · 2017-07-03T02:26:31Z

server/coordinator.go

+
+	if region := c.cluster.getRegion(op.GetRegionID()); region != nil {
+		if msg, _ := op.Do(region); msg != nil {
+			c.hbStreams.sendMsg(region, msg)


I think it is better to put sendMsg in dispatch.

When a scheduler calls addOperator, I want the operator be sent to tikv ASAP.

siddontang · 2017-07-03T23:50:57Z

PTAL @huachaohuang

# Conflicts: # server/cluster_worker_test.go

huachaohuang · 2017-07-04T07:08:48Z

server/grpc_service.go

+				return errors.Trace(err)
+			}
+			storeID := request.GetLeader().GetStoreId()
+			cluster.coordinator.hbStreams.bindStream(storeID, server)


Is it safe to send with server concurrently?
Seems server will still be used in this function.

disksing · 2017-07-04T08:06:56Z

PTAL @huachaohuang

huachaohuang · 2017-07-04T08:17:41Z

server/grpc_service.go

 // RegionHeartbeat implements gRPC PDServer.
 func (s *Server) RegionHeartbeat(server pdpb.PD_RegionHeartbeatServer) error {
+	syncServer := &syncHeartbeatServer{server: server}


Is the server reused?
If it is, it will be protected by different locks on each call.

I think send the error message with hbstreams same as a normal response better than use lock.

siddontang · 2017-07-04T09:14:16Z

server/coordinator.go

+					delete(s.streams, storeID)
+					regionHeartbeatCounter.WithLabelValues("down", "err")
+				} else {
+					regionHeartbeatCounter.WithLabelValues("down", "ok")


why down here?

maybe push?

+1, "up" and "down" seem strange.

push for server -> client, and what for client -> server? @huachaohuang @siddontang

server push operator to client, and client pull operator from server, make senses?

I think not pull here.

"report" and "schedule"?

report may be better
/cc @huachaohuang

I agree with "push" and "report".

siddontang · 2017-07-04T12:25:31Z

Rest LGTM

PTAL @huachaohuang @nolouch

huachaohuang · 2017-07-04T14:40:03Z

server/grpc_service.go

+	cluster := s.GetRaftCluster()
+	if cluster == nil {
+		resp := &pdpb.RegionHeartbeatResponse{
+			Header: &pdpb.ResponseHeader{


Is this the same as s.notBootstrappedHeader?

huachaohuang · 2017-07-04T14:40:41Z

server/grpc_service.go

 			storeID := request.GetLeader().GetStoreId()
-			cluster.coordinator.hbStreams.bindStream(storeID, syncServer)
+			hbStreams.bindStream(storeID, server)


Can we bind stream before the loop?

No. We don't know the storeID yet.

huachaohuang · 2017-07-04T14:57:43Z

server/coordinator.go

+					delete(s.streams, storeID)
+					regionHeartbeatCounter.WithLabelValues("down", "err")
+				} else {
+					regionHeartbeatCounter.WithLabelValues("down", "ok")


I agree with "push" and "report".

huachaohuang · 2017-07-04T14:59:49Z

server/grpc_service.go

+	cluster := s.GetRaftCluster()
+	if cluster == nil {
+		resp := &pdpb.RegionHeartbeatResponse{
+			Header: &pdpb.ResponseHeader{


Is this the same as s.notBootstrappedHeader()?

huachaohuang · 2017-07-04T15:00:15Z

server/grpc_service.go

+
+		if isNew {
+			storeID := request.GetLeader().GetStoreId()
+			hbStreams.bindStream(storeID, server)


Can we bind stream before the loop?

huachaohuang

LGTM

nolouch · 2017-07-05T03:15:43Z

LGTM

disksing added 5 commits June 30, 2017 10:10

server push schedule commands.

d62bc84

minor change.

af9d42e

check nil msg.

44654e0

update dispatch.

cf50261

fix test.

b3ff70b

disksing requested review from siddontang, overvenus and nolouch June 30, 2017 08:58

nolouch reviewed Jun 30, 2017

View reviewed changes

siddontang reviewed Jun 30, 2017

View reviewed changes

siddontang reviewed Jul 1, 2017

View reviewed changes

overvenus reviewed Jul 3, 2017

View reviewed changes

address comments.

0d3683e

Merge branch 'master' into disksing/stream

f0a8532

# Conflicts: # server/cluster_worker_test.go

huachaohuang reviewed Jul 4, 2017

View reviewed changes

address comment.

e74ebe4

huachaohuang reviewed Jul 4, 2017

View reviewed changes

siddontang reviewed Jul 4, 2017

View reviewed changes

address comments.

c5efdf3

huachaohuang reviewed Jul 4, 2017

View reviewed changes

disksing added 2 commits July 5, 2017 09:53

address comments.

711f81c

Merge branch 'master' into disksing/stream

d6e50f9

huachaohuang approved these changes Jul 5, 2017

View reviewed changes

Merge branch 'master' into disksing/stream

92c3da2

disksing merged commit 0e4b711 into master Jul 5, 2017

disksing deleted the disksing/stream branch July 5, 2017 06:10

disksing mentioned this pull request Jul 5, 2017

gRPC: support a stream protocol to notify TiKV #495

Closed

server push schedule commands. #669

server push schedule commands. #669

Conversation

disksing commented Jun 30, 2017

nolouch 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

siddontang commented Jul 3, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

siddontang commented Jul 3, 2017

Choose a reason for hiding this comment

disksing commented Jul 4, 2017

Choose a reason for hiding this comment

nolouch Jul 4, 2017 • edited Loading

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

siddontang commented Jul 4, 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

huachaohuang left a comment

Choose a reason for hiding this comment

nolouch commented Jul 5, 2017

nolouch Jul 4, 2017 •

edited

Loading