Fix #238 timeout to execute read/write operations (#239)

Author: zhangxu19830126

Makefile

@@ -26,8 +26,8 @@ example-http: http; $(info ======== compiled matrixcube http example:)
 
 .PHONY: test
 test: ; $(info ======== test matrixcube)
-	go test $(RACE) -count=1 -v -timeout 600s $(PKGNAME)/storage
-	go test $(RACE) -count=1 -v -timeout 600s $(PKGNAME)/raftstore
+	go test $(RACE) -count=1 -v -timeout 300s $(PKGNAME)/storage
+	go test $(RACE) -count=1 -v -timeout 300s $(PKGNAME)/raftstore
 
 .PHONY: race-test
 race-test: override RACE=-race

components/prophet/client.go

@@ -734,9 +734,10 @@ func (c *asyncClient) initLeaderConn(conn goetty.IOSession, timeout time.Duratio
 func (c *asyncClient) maybeRegisterContainer() {
 	if c.containerID > 0 {
 		req := &rpcpb.Request{}
+		req.ID = c.nextID()
 		req.Type = rpcpb.TypeRegisterContainer
 		req.ContainerID = c.containerID
-		c.asyncDo(req, nil)
+		c.doWrite(newAsyncCtx(req, nil))
 	}
 }
 

components/prophet/join/join.go

@@ -128,7 +128,13 @@ func PrepareJoinCluster(ctx context.Context, cfg *config.Config) (*clientv3.Clie
 	defer client.Close()
 
 	for {
+		util.GetLogger().Infof("%s(%s) begin to check etcd members",
+			cfg.Name,
+			cfg.DataDir)
 		checkMembers(client, cfg)
+		util.GetLogger().Infof("%s(%s) end to check etcd members",
+			cfg.Name,
+			cfg.DataDir)
 
 		var prophets []string
 		// - A new Prophet joins an existing cluster.
@@ -138,12 +144,18 @@ func PrepareJoinCluster(ctx context.Context, cfg *config.Config) (*clientv3.Clie
 				// First adds member through the API
 				resp, err := util.AddEtcdMember(client, []string{cfg.EmbedEtcd.AdvertisePeerUrls})
 				if err != nil {
-					util.GetLogger().Errorf("add member to embed etcd failed with %+v, retry later", err)
+					util.GetLogger().Errorf("%s(%s) add member to embed etcd failed with %+v, retry later",
+						cfg.Name,
+						cfg.DataDir,
+						err)
 					time.Sleep(time.Millisecond * 500)
 					continue
 				}
 
-				util.GetLogger().Infof("%s added into embed etcd cluster with resp %+v", cfg.Name, resp)
+				util.GetLogger().Infof("%s(%s) added into embed etcd cluster with resp %+v",
+					cfg.Name,
+					cfg.DataDir,
+					resp)
 
 				for _, m := range resp.Members {
 					if m.Name != "" {
@@ -163,6 +175,10 @@ func PrepareJoinCluster(ctx context.Context, cfg *config.Config) (*clientv3.Clie
 
 		c, e, err := startEmbedEtcd(ctx, cfg)
 		if err != nil && strings.Contains(err.Error(), "member count is unequal") {
+			util.GetLogger().Infof("%s(%s) start embed etcd failed with %+v, retry",
+				cfg.Name,
+				cfg.DataDir,
+				err)
 			continue
 		}
 		if err != nil {
@@ -186,7 +202,10 @@ func PrepareJoinCluster(ctx context.Context, cfg *config.Config) (*clientv3.Clie
 			util.GetLogger().Fatalf("write data path failed with %+v",
 				err)
 		}
-
+		util.GetLogger().Errorf("%s(%s) save InitialCluster with %+v",
+			cfg.Name,
+			cfg.DataDir,
+			cfg.EmbedEtcd.InitialCluster)
 		return c, e, nil
 	}
 }
@@ -196,7 +215,9 @@ OUTER:
 	for {
 		listResp, err := util.ListEtcdMembers(client)
 		if err != nil {
-			util.GetLogger().Errorf("list embed etcd members failed with %+v, retry later",
+			util.GetLogger().Errorf("%s(%s) list embed etcd members failed with %+v, retry later",
+				cfg.Name,
+				cfg.DataDir,
 				err)
 			time.Sleep(time.Second)
 			continue
@@ -205,7 +226,11 @@ OUTER:
 		for _, m := range listResp.Members {
 			if len(m.Name) == 0 {
 				// A new member added, but not started
-				util.GetLogger().Warningf("there is a member that has not joined successfully")
+				util.GetLogger().Warningf("%s(%s) there is a member that has not joined successfully, member %+v, self %+v",
+					cfg.Name,
+					cfg.DataDir,
+					m.PeerURLs,
+					cfg.EmbedEtcd.PeerUrls)
 				time.Sleep(time.Second)
 				continue OUTER
 			}

components/prophet/prophet_leader.go

@@ -79,6 +79,7 @@ func (p *defaultProphet) createEventNotifer() {
 func (p *defaultProphet) stopEventNotifer() {
 	if p.wn != nil {
 		p.wn.stop()
+		p.wn = nil
 	}
 }
 

components/prophet/prophet_watcher.go

@@ -14,6 +14,7 @@
 package prophet
 
 import (
+	"fmt"
 	"sync"
 	"sync/atomic"
 
@@ -43,13 +44,16 @@ func (wt *watcherSession) notify(evt rpcpb.EventNotify) error {
 }
 
 type watcherNotifier struct {
-	watchers sync.Map
+	sync.Mutex
+
+	watchers map[uint64]*watcherSession
 	cluster  *cluster.RaftCluster
 }
 
 func newWatcherNotifier(cluster *cluster.RaftCluster) *watcherNotifier {
 	return &watcherNotifier{
-		cluster: cluster,
+		cluster:  cluster,
+		watchers: make(map[uint64]*watcherSession),
 	}
 }
 
@@ -84,21 +88,46 @@ func (wn *watcherNotifier) handleCreateWatcher(req *rpcpb.Request, resp *rpcpb.R
 			resp.Event.InitEvent = rsp
 		}
 
-		wn.watchers.Store(session.ID(), &watcherSession{
-			flag:    req.CreateWatcher.Flag,
-			session: session,
-		})
+		return wn.addWatcher(req.CreateWatcher.Flag, session)
+	}
+
+	return nil
+}
+
+func (wn *watcherNotifier) addWatcher(flag uint32, session goetty.IOSession) error {
+	wn.Lock()
+	defer wn.Unlock()
+
+	if wn.watchers == nil {
+		return fmt.Errorf("watcher notifier stopped")
 	}
 
+	wn.watchers[session.ID()] = &watcherSession{
+		flag:    flag,
+		session: session,
+	}
 	return nil
 }
 
-func (wn *watcherNotifier) clearWatcher(w *watcherSession) {
-	wn.watchers.Delete(w.session.ID())
+func (wn *watcherNotifier) doClearWatcherLocked(w *watcherSession) {
+	delete(wn.watchers, w.session.ID())
+	w.session.Close()
 	util.GetLogger().Infof("watcher %s removed",
 		w.session.RemoteAddr())
 }
 
+func (wn *watcherNotifier) doNotify(evt rpcpb.EventNotify) {
+	wn.Lock()
+	defer wn.Unlock()
+
+	for _, wt := range wn.watchers {
+		err := wt.notify(evt)
+		if err != nil {
+			wn.doClearWatcherLocked(wt)
+		}
+	}
+}
+
 func (wn *watcherNotifier) start() {
 	go func() {
 		defer func() {
@@ -108,40 +137,25 @@ func (wn *watcherNotifier) start() {
 			}
 		}()
 
-		var closed []*watcherSession
 		for {
 			evt, ok := <-wn.cluster.ChangedEventNotifier()
 			if !ok {
 				util.GetLogger().Infof("watcher notifer exited")
 				return
 			}
 
-			if len(closed) > 0 {
-				closed = closed[:0]
-			}
-
-			wn.watchers.Range(func(key, value interface{}) bool {
-				wt := value.(*watcherSession)
-				err := wt.notify(evt)
-				if err != nil {
-					closed = append(closed, wt)
-				}
-				return true
-			})
-
-			for _, w := range closed {
-				wn.clearWatcher(w)
-			}
+			wn.doNotify(evt)
 		}
 	}()
 }
 
 func (wn *watcherNotifier) stop() {
-	wn.watchers.Range(func(key, value interface{}) bool {
-		wn.watchers.Delete(key)
-		value.(*watcherSession).session.Close()
-		return true
-	})
+	wn.Lock()
+	defer wn.Unlock()
 
+	for _, wt := range wn.watchers {
+		wn.doClearWatcherLocked(wt)
+	}
+	wn.watchers = nil
 	util.GetLogger().Infof("watcher notifier stopped")
 }

components/prophet/storage/storage.go

@@ -510,7 +510,7 @@ func (s *storage) PutBootstrapped(container metadata.Container, resources ...met
 		batch.SaveValues = append(batch.SaveValues, string(v))
 	}
 
-	ok, _, err := s.kv.SaveIfNotExists(s.clusterPath, string(format.UInt64ToString(clusterID)), batch)
+	ok, _, err := s.kv.SaveIfNotExists(s.clusterPath, format.Uint64ToString(clusterID), batch)
 	return ok, err
 }
 
@@ -547,9 +547,9 @@ func (s *storage) containerWeightPath(id uint64, typ string) string {
 }
 
 func (s *storage) jobKey(jobType metapb.JobType) string {
-	return path.Join(s.jobPath, string(format.UInt64ToString(uint64(jobType))))
+	return path.Join(s.jobPath, format.Uint64ToString(uint64(jobType)))
 }
 
 func (s *storage) jobDataKey(jobType metapb.JobType) string {
-	return path.Join(s.jobDataPath, string(format.UInt64ToString(uint64(jobType))))
+	return path.Join(s.jobDataPath, format.Uint64ToString(uint64(jobType)))
 }

components/prophet/util/keyutil/util.go

@@ -29,5 +29,5 @@ func BuildKeyRangeKey(group uint64, startKey, endKey []byte) string {
 
 // GetGroupFromRangeKey return group from range key
 func GetGroupFromRangeKey(key string) uint64 {
-	return format.MustParseStrUInt64(strings.Split(key, "-")[0])
+	return format.MustParseStringUint64(strings.Split(key, "-")[0])
 }

go.mod

@@ -11,9 +11,9 @@ require (
 	github.com/cockroachdb/pebble v0.0.0-20210503173641-1387689d3d7c
 	github.com/coreos/go-semver v0.3.0
 	github.com/docker/go-units v0.4.0
-	github.com/fagongzi/goetty v1.11.0
+	github.com/fagongzi/goetty v1.11.1
 	github.com/fagongzi/log v0.0.0-20191122063922-293b75312445
-	github.com/fagongzi/util v0.0.0-20210409031311-a10fdf8fbd7a
+	github.com/fagongzi/util v0.0.0-20210923134909-bccc37b5040d
 	github.com/go-ole/go-ole v1.2.4 // indirect
 	github.com/gogo/protobuf v1.3.2
 	github.com/google/btree v1.0.1

go.sum

@@ -118,13 +118,13 @@ github.com/envoyproxy/go-control-plane v0.9.9-0.20201210154907-fd9021fe5dad/go.m
 github.com/envoyproxy/go-control-plane v0.9.9-0.20210217033140-668b12f5399d/go.mod h1:cXg6YxExXjJnVBQHBLXeUAgxn2UodCpnH306RInaBQk=
 github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
 github.com/etcd-io/bbolt v1.3.3/go.mod h1:ZF2nL25h33cCyBtcyWeZ2/I3HQOfTP+0PIEvHjkjCrw=
-github.com/fagongzi/goetty v1.11.0 h1:Ld65qimH2AnMX0O7KEIssiLGbFNNCRK5+m3pBFOMmCM=
-github.com/fagongzi/goetty v1.11.0/go.mod h1:42WH1Shi0NYzyop19JaY5oabuk1OzB9AWNDqbF/NdP4=
+github.com/fagongzi/goetty v1.11.1 h1:Y2jnmkpTXdBmpQ7dchfUgPLcZOdQHvfVhOjNapuYNFA=
+github.com/fagongzi/goetty v1.11.1/go.mod h1:42WH1Shi0NYzyop19JaY5oabuk1OzB9AWNDqbF/NdP4=
 github.com/fagongzi/log v0.0.0-20191122063922-293b75312445 h1:XhTJpo5oMYnjgQPq28pufQ/XDGoQXi1x8qa87A4Fsb8=
 github.com/fagongzi/log v0.0.0-20191122063922-293b75312445/go.mod h1:1aUifpZCjAdZK7L3vRRl2cpSktbKasAU5vOSyLSCODs=
 github.com/fagongzi/util v0.0.0-20201116094402-221cc40c4593/go.mod h1:jYDIbpaqHXCCQ7QIDXRVfsQYAGKSNNb6N8BPTgdpcdE=
-github.com/fagongzi/util v0.0.0-20210409031311-a10fdf8fbd7a h1:TaySCZ6YAnc3dhVc2tzfw6Pu1Nufee98gkFBCI/AxKY=
-github.com/fagongzi/util v0.0.0-20210409031311-a10fdf8fbd7a/go.mod h1:5cqSns2zMRcJeVGvAqeTrbXFqh5AqBFr5uVKP9T2kiE=
+github.com/fagongzi/util v0.0.0-20210923134909-bccc37b5040d h1:1pILVCatHj3eVo9i52dZyY4BwjTmSIeN+/hoJh8rD0Y=
+github.com/fagongzi/util v0.0.0-20210923134909-bccc37b5040d/go.mod h1:5cqSns2zMRcJeVGvAqeTrbXFqh5AqBFr5uVKP9T2kiE=
 github.com/fasthttp-contrib/websocket v0.0.0-20160511215533-1f3b11f56072/go.mod h1:duJ4Jxv5lDcvg4QuQr0oowTf7dz4/CR8NtyCooz9HL8=
 github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4=
 github.com/fatih/structs v1.1.0/go.mod h1:9NiDSp5zOcgEDl+j00MP/WkGVPOlPRLejGD8Ga6PJ7M=

raftstore/peer_apply.go

@@ -93,7 +93,9 @@ func (pr *peerReplica) doCompactRaftLog(shardID, startIndex, endIndex uint64) er
 }
 
 func (pr *peerReplica) doApplyingSnapshotJob() error {
-	logger.Infof("shard %d begin apply snapshot data", pr.shardID)
+	logger.Infof("shard %d peer %d begin apply snapshot data",
+		pr.shardID,
+		pr.peer.ID)
 	localState, err := pr.ps.loadShardLocalState(pr.ps.applySnapJob)
 	if err != nil {
 		logger.Fatalf("shard %d apply snap load local state failed with %+v",
@@ -132,8 +134,9 @@ func (pr *peerReplica) doApplyingSnapshotJob() error {
 		pr.store.aware.SnapshotApplied(pr.ps.shard)
 	}
 	pr.stopRaftTick = false
-	logger.Infof("shard %d apply snapshot data complete, %+v",
+	logger.Infof("shard %d peer %d apply snapshot data complete, %+v",
 		pr.shardID,
+		pr.peer.ID,
 		pr.ps.raftLocalState.HardState)
 	return nil
 }