From d7ee083b500388d671b1b891126b3244c5dc5697 Mon Sep 17 00:00:00 2001 From: Qi Xiao Date: Thu, 31 Jan 2019 19:15:49 +0800 Subject: [PATCH 1/3] Refactor observer to pull mode --- cmd/cql-observer/config_test.go | 2 + cmd/cql-observer/main.go | 10 +- cmd/cql-observer/node.go | 24 +-- cmd/cql-observer/observation_test.go | 7 +- cmd/cql-observer/observer.go | 16 +- cmd/cql-observer/service.go | 133 +++----------- cmd/cql-observer/worker.go | 162 +++++++++++++++++ route/acl.go | 18 +- route/acl_test.go | 2 +- sqlchain/blockindex.go | 15 ++ sqlchain/chain.go | 148 +++++---------- sqlchain/observer.go | 257 --------------------------- worker/dbms.go | 69 ------- worker/dbms_rpc.go | 40 +---- worker/dbms_test.go | 10 +- worker/observer.go | 98 ++++++++++ 16 files changed, 370 insertions(+), 641 deletions(-) create mode 100644 cmd/cql-observer/worker.go delete mode 100644 sqlchain/observer.go create mode 100644 worker/observer.go diff --git a/cmd/cql-observer/config_test.go b/cmd/cql-observer/config_test.go index e542af247..5c3583bfd 100644 --- a/cmd/cql-observer/config_test.go +++ b/cmd/cql-observer/config_test.go @@ -1,3 +1,5 @@ +// +build !testbinary + /* * Copyright 2018 The CovenantSQL Authors. * diff --git a/cmd/cql-observer/main.go b/cmd/cql-observer/main.go index 34162cbc3..d1d6c9dcf 100644 --- a/cmd/cql-observer/main.go +++ b/cmd/cql-observer/main.go @@ -30,7 +30,6 @@ import ( "github.com/CovenantSQL/CovenantSQL/crypto/asymmetric" "github.com/CovenantSQL/CovenantSQL/crypto/kms" "github.com/CovenantSQL/CovenantSQL/proto" - "github.com/CovenantSQL/CovenantSQL/rpc" "github.com/CovenantSQL/CovenantSQL/utils" "github.com/CovenantSQL/CovenantSQL/utils/log" ) @@ -85,15 +84,14 @@ func main() { kms.InitBP() - // start rpc - var server *rpc.Server - if server, err = initNode(); err != nil { + // init node + if err = initNode(); err != nil { log.WithError(err).Fatal("init node failed") } // start service var service *Service - if service, err = startService(server); err != nil { + if service, err = startService(); err != nil { log.WithError(err).Fatal("start observation failed") } @@ -144,7 +142,7 @@ func main() { } // stop subscriptions - if err = stopService(service, server); err != nil { + if err = stopService(service); err != nil { log.WithError(err).Fatal("stop service failed") } diff --git a/cmd/cql-observer/node.go b/cmd/cql-observer/node.go index 49bb83513..29b9c5351 100644 --- a/cmd/cql-observer/node.go +++ b/cmd/cql-observer/node.go @@ -18,18 +18,16 @@ package main import ( "fmt" - "os" "syscall" "github.com/CovenantSQL/CovenantSQL/conf" "github.com/CovenantSQL/CovenantSQL/crypto/kms" "github.com/CovenantSQL/CovenantSQL/route" - "github.com/CovenantSQL/CovenantSQL/rpc" "github.com/CovenantSQL/CovenantSQL/utils/log" "golang.org/x/crypto/ssh/terminal" ) -func initNode() (server *rpc.Server, err error) { +func initNode() (err error) { var masterKey []byte if !conf.GConf.IsTestMode { fmt.Print("Type in Master key to continue:") @@ -50,25 +48,5 @@ func initNode() (server *rpc.Server, err error) { // init kms routing route.InitKMS(conf.GConf.PubKeyStoreFile) - // init server - if server, err = createServer( - conf.GConf.PrivateKeyFile, conf.GConf.PubKeyStoreFile, masterKey, conf.GConf.ListenAddr); err != nil { - log.WithError(err).Error("create server failed") - return - } - - return -} - -func createServer(privateKeyPath, pubKeyStorePath string, masterKey []byte, listenAddr string) (server *rpc.Server, err error) { - os.Remove(pubKeyStorePath) - - server = rpc.NewServer() - if err != nil { - return - } - - err = server.InitRPCServer(listenAddr, privateKeyPath, masterKey) - return } diff --git a/cmd/cql-observer/observation_test.go b/cmd/cql-observer/observation_test.go index bd3b9e620..532672371 100644 --- a/cmd/cql-observer/observation_test.go +++ b/cmd/cql-observer/observation_test.go @@ -500,7 +500,7 @@ func TestFullProcess(t *testing.T) { observerCmd.Cmd.Wait() }() - // wait for the observer to collect blocks, two periods is enough + // wait for the observer to collect blocks time.Sleep(conf.GConf.SQLChainPeriod * 5) // test get genesis block by height @@ -686,11 +686,14 @@ func TestFullProcess(t *testing.T) { }) So(err, ShouldBeNil) + // wait for the observer to be enabled query by miner, and collect blocks + time.Sleep(conf.GConf.SQLChainPeriod * 5) + // test get genesis block by height res, err = getJSON("v3/head/%v", dbID2) So(err, ShouldBeNil) So(ensureSuccess(res.Interface("block")), ShouldNotBeNil) - So(ensureSuccess(res.Int("block", "height")), ShouldEqual, 0) + So(ensureSuccess(res.Int("block", "height")), ShouldBeGreaterThanOrEqualTo, 0) log.Info(err, res) err = client.Drop(dsn) diff --git a/cmd/cql-observer/observer.go b/cmd/cql-observer/observer.go index ff7ed1b22..2190f170e 100644 --- a/cmd/cql-observer/observer.go +++ b/cmd/cql-observer/observer.go @@ -20,7 +20,6 @@ import ( "github.com/CovenantSQL/CovenantSQL/conf" "github.com/CovenantSQL/CovenantSQL/crypto/kms" "github.com/CovenantSQL/CovenantSQL/proto" - "github.com/CovenantSQL/CovenantSQL/route" "github.com/CovenantSQL/CovenantSQL/rpc" ) @@ -41,33 +40,22 @@ func registerNode() (err error) { return } -func startService(server *rpc.Server) (service *Service, err error) { +func startService() (service *Service, err error) { // register observer service to rpc server service, err = NewService() if err != nil { return } - if err = server.RegisterService(route.ObserverRPCName, service); err != nil { - return - } - - // start service rpc, observer acts as client role but listen to - go server.Serve() - // start observer service service.start() return } -func stopService(service *Service, server *rpc.Server) (err error) { +func stopService(service *Service) (err error) { // stop subscription service.stop() - // stop rpc service - server.Listener.Close() - server.Stop() - return } diff --git a/cmd/cql-observer/service.go b/cmd/cql-observer/service.go index a2465a368..c49138860 100644 --- a/cmd/cql-observer/service.go +++ b/cmd/cql-observer/service.go @@ -30,11 +30,9 @@ import ( "github.com/CovenantSQL/CovenantSQL/proto" "github.com/CovenantSQL/CovenantSQL/route" "github.com/CovenantSQL/CovenantSQL/rpc" - "github.com/CovenantSQL/CovenantSQL/sqlchain" "github.com/CovenantSQL/CovenantSQL/types" "github.com/CovenantSQL/CovenantSQL/utils" "github.com/CovenantSQL/CovenantSQL/utils/log" - "github.com/CovenantSQL/CovenantSQL/worker" bolt "github.com/coreos/bbolt" ) @@ -95,9 +93,8 @@ var ( // Service defines the observer service structure. type Service struct { - lock sync.Mutex - subscription map[proto.DatabaseID]int32 - upstreamServers sync.Map + subscription sync.Map // map[proto.DatabaseID]*subscribeWorker + upstreamServers sync.Map // map[proto.DatabaseID]*types.ServiceInstance db *bolt.DB caller *rpc.Caller @@ -147,17 +144,16 @@ func NewService() (service *Service, err error) { // init service service = &Service{ - subscription: make(map[proto.DatabaseID]int32), - db: db, - caller: rpc.NewCaller(), + db: db, + caller: rpc.NewCaller(), } // load previous subscriptions if err = db.View(func(tx *bolt.Tx) error { - return tx.Bucket(subscriptionBucket).ForEach(func(rawDBID, rawHeight []byte) (err error) { + return tx.Bucket(subscriptionBucket).ForEach(func(rawDBID, rawCount []byte) (err error) { dbID := proto.DatabaseID(string(rawDBID)) - h := bytesToInt32(rawHeight) - service.subscription[dbID] = h + count := bytesToInt32(rawCount) + service.subscription.Store(dbID, newSubscribeWorker(dbID, count, service)) return }) }); err != nil { @@ -182,10 +178,6 @@ func (s *Service) subscribe(dbID proto.DatabaseID, resetSubscribePosition string return ErrStopped } - s.lock.Lock() - - shouldStartSubscribe := false - if resetSubscribePosition != "" { var fromPos int32 @@ -198,46 +190,25 @@ func (s *Service) subscribe(dbID proto.DatabaseID, resetSubscribePosition string fromPos = types.ReplicateFromNewest } - s.subscription[dbID] = fromPos - - // send start subscription request - // TODO(leventeliu): should also clean up obsolete data in db file! - shouldStartSubscribe = true + unpackWorker(s.subscription.LoadOrStore(dbID, + newSubscribeWorker(dbID, fromPos, s))).reset(fromPos) } else { // not resetting - if _, exists := s.subscription[dbID]; !exists { - s.subscription[dbID] = types.ReplicateFromNewest - shouldStartSubscribe = true - } - } - - s.lock.Unlock() - - if shouldStartSubscribe { - return s.startSubscribe(dbID) + unpackWorker(s.subscription.LoadOrStore(dbID, + newSubscribeWorker(dbID, types.ReplicateFromNewest, s))).start() } return } -// AdviseNewBlock handles block replication request from the remote database chain service. -func (s *Service) AdviseNewBlock(req *sqlchain.MuxAdviseNewBlockReq, resp *sqlchain.MuxAdviseNewBlockResp) (err error) { - if atomic.LoadInt32(&s.stopped) == 1 { - // stopped - return ErrStopped - } - - if req.Block == nil { - log.WithField("node", req.GetNodeID().String()).Warning("received empty block") +func unpackWorker(actual interface{}, _ ...interface{}) (worker *subscribeWorker) { + if actual == nil { return } - log.WithFields(log.Fields{ - "node": req.GetNodeID().String(), - "block": req.Block.BlockHash(), - }).Debug("received block") + worker, _ = actual.(*subscribeWorker) - return s.addBlock(req.DatabaseID, req.Count, req.Block) + return } func (s *Service) start() (err error) { @@ -246,54 +217,14 @@ func (s *Service) start() (err error) { return ErrStopped } - s.lock.Lock() - dbs := make([]proto.DatabaseID, len(s.subscription)) - for dbID := range s.subscription { - dbs = append(dbs, dbID) - } - s.lock.Unlock() - - for _, dbID := range dbs { - if err = s.startSubscribe(dbID); err != nil { - log.WithField("db", dbID).WithError(err).Warning("start subscription failed") - } - } + s.subscription.Range(func(_, rawWorker interface{}) bool { + unpackWorker(rawWorker).start() + return true + }) return nil } -func (s *Service) startSubscribe(dbID proto.DatabaseID) (err error) { - if atomic.LoadInt32(&s.stopped) == 1 { - // stopped - return ErrStopped - } - - s.lock.Lock() - defer s.lock.Unlock() - - // start subscribe on first node of each sqlchain server peers - log.WithField("db", dbID).Info("start subscribing transactions") - - instance, err := s.getUpstream(dbID) - if err != nil { - return - } - - // store the genesis block - if err = s.addBlock(dbID, 0, instance.GenesisBlock); err != nil { - return - } - - req := &worker.SubscribeTransactionsReq{} - resp := &worker.SubscribeTransactionsResp{} - req.Height = s.subscription[dbID] - req.DatabaseID = dbID - - err = s.minerRequest(dbID, route.DBSSubscribeTransactions.String(), req, resp) - - return -} - func (s *Service) addAck(dbID proto.DatabaseID, height int32, offset int32, ack *types.SignedAckHeader) (err error) { log.WithFields(log.Fields{ "height": height, @@ -306,9 +237,6 @@ func (s *Service) addAck(dbID proto.DatabaseID, height int32, offset int32, ack return ErrStopped } - s.lock.Lock() - defer s.lock.Unlock() - if err = ack.Verify(); err != nil { return } @@ -335,9 +263,6 @@ func (s *Service) addQueryTracker(dbID proto.DatabaseID, height int32, offset in return ErrStopped } - s.lock.Lock() - defer s.lock.Unlock() - if err = qt.Request.Verify(); err != nil { return } @@ -433,23 +358,13 @@ func (s *Service) stop() (err error) { return ErrStopped } - s.lock.Lock() - defer s.lock.Unlock() - // send cancel subscription to all databases log.Info("stop subscribing all databases") - for dbID := range s.subscription { - // send cancel subscription rpc - req := &worker.CancelSubscriptionReq{} - resp := &worker.CancelSubscriptionResp{} - req.DatabaseID = dbID - - if err = s.minerRequest(dbID, route.DBSCancelSubscription.String(), req, resp); err != nil { - // cancel subscription failed - log.WithField("db", dbID).WithError(err).Warning("cancel subscription") - } - } + s.subscription.Range(func(_, rawWorker interface{}) bool { + unpackWorker(rawWorker).stop() + return true + }) // close the subscription database s.db.Close() @@ -467,7 +382,7 @@ func (s *Service) minerRequest(dbID proto.DatabaseID, method string, request int } func (s *Service) getUpstream(dbID proto.DatabaseID) (instance *types.ServiceInstance, err error) { - log.WithField("db", dbID).Info("get peers info for database") + log.WithField("db", dbID).Debug("get peers info for database") if iInstance, exists := s.upstreamServers.Load(dbID); exists { instance = iInstance.(*types.ServiceInstance) diff --git a/cmd/cql-observer/worker.go b/cmd/cql-observer/worker.go new file mode 100644 index 000000000..23f12e24d --- /dev/null +++ b/cmd/cql-observer/worker.go @@ -0,0 +1,162 @@ +/* + * Copyright 2019 The CovenantSQL 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. + */ + +package main + +import ( + "errors" + "sync" + "sync/atomic" + "time" + + "github.com/CovenantSQL/CovenantSQL/conf" + "github.com/CovenantSQL/CovenantSQL/proto" + "github.com/CovenantSQL/CovenantSQL/route" + "github.com/CovenantSQL/CovenantSQL/utils/log" + "github.com/CovenantSQL/CovenantSQL/worker" +) + +type subscribeWorker struct { + l sync.Mutex + s *Service + dbID proto.DatabaseID + head int32 + wg *sync.WaitGroup + stopCh chan struct{} +} + +func newSubscribeWorker(dbID proto.DatabaseID, head int32, s *Service) *subscribeWorker { + return &subscribeWorker{ + dbID: dbID, + head: head, + s: s, + } +} + +func (w *subscribeWorker) run() { + defer w.wg.Done() + + // calc next tick + var nextTick time.Duration + + for { + + select { + case <-w.stopCh: + return + case <-time.After(nextTick): + if err := w.pull(atomic.LoadInt32(&w.head)); err != nil { + // calc next tick + nextTick = conf.GConf.SQLChainPeriod + } else { + nextTick /= 10 + } + } + } +} + +func (w *subscribeWorker) pull(count int32) (err error) { + var ( + req = new(worker.ObserverFetchBlockReq) + resp = new(worker.ObserverFetchBlockResp) + next int32 + ) + + defer func() { + lf := log.WithFields(log.Fields{ + "req_count": count, + "count": resp.Count, + }) + + if err != nil { + lf.WithError(err).Debug("sync block failed") + } else { + if resp.Block != nil { + lf = lf.WithField("block", resp.Block.BlockHash()) + } else { + lf = lf.WithField("block", nil) + } + lf.WithField("next", next).Debug("sync block success") + } + }() + + req.DatabaseID = w.dbID + req.Count = count + + if err = w.s.minerRequest(w.dbID, route.DBSObserverFetchBlock.String(), req, resp); err != nil { + return + } + + if resp.Block == nil { + err = errors.New("nil block, try later") + return + } + + if err = w.s.addBlock(w.dbID, count, resp.Block); err != nil { + return + } + + if count < 0 { + next = resp.Count + 1 + } else { + next = count + 1 + } + + atomic.CompareAndSwapInt32(&w.head, count, next) + + return +} + +func (w *subscribeWorker) reset(head int32) { + atomic.StoreInt32(&w.head, head) + w.start() +} + +func (w *subscribeWorker) start() { + w.l.Lock() + defer w.l.Unlock() + + if w.isStopped() { + w.stopCh = make(chan struct{}) + w.wg = new(sync.WaitGroup) + w.wg.Add(1) + go w.run() + } +} + +func (w *subscribeWorker) stop() { + w.l.Lock() + defer w.l.Unlock() + + if !w.isStopped() { + // stop + close(w.stopCh) + w.wg.Wait() + } +} + +func (w *subscribeWorker) isStopped() bool { + if w.stopCh == nil { + return true + } + + select { + case <-w.stopCh: + return true + default: + return false + } +} diff --git a/route/acl.go b/route/acl.go index d6a42b3f5..032aff54e 100644 --- a/route/acl.go +++ b/route/acl.go @@ -77,10 +77,8 @@ const ( DBSAck // DBSDeploy is used by BP to create/drop/update database DBSDeploy - // DBSSubscribeTransactions is used by dbms to handle observer subscription request - DBSSubscribeTransactions - // DBSCancelSubscription is used by dbms to handle observer subscription cancellation request - DBSCancelSubscription + // DBSObserverFetchBlock is used by observer to fetch block. + DBSObserverFetchBlock // DBCCall is used by Miner for data consistency DBCCall // SQLCAdviseNewBlock is used by sqlchain to advise new block between adjacent node @@ -95,8 +93,6 @@ const ( SQLCSignBilling // SQLCLaunchBilling is used by blockproducer to trigger the billing process in sqlchain SQLCLaunchBilling - // OBSAdviseNewBlock is used by sqlchain to push new block to observers - OBSAdviseNewBlock // MCCAdviseNewBlock is used by block producer to push block to adjacent nodes MCCAdviseNewBlock // MCCAdviseTxBilling is used by block producer to push billing transaction to adjacent nodes @@ -131,8 +127,6 @@ const ( SQLChainRPCName = "SQLC" // DBRPCName defines the sql chain db service rpc name DBRPCName = "DBS" - // ObserverRPCName defines the observer node service rpc name - ObserverRPCName = "OBS" ) // String returns the RemoteFunc string. @@ -154,10 +148,8 @@ func (s RemoteFunc) String() string { return "DBS.Ack" case DBSDeploy: return "DBS.Deploy" - case DBSSubscribeTransactions: - return "DBS.SubscribeTransactions" - case DBSCancelSubscription: - return "DBS.CancelSubscription" + case DBSObserverFetchBlock: + return "DBS.ObserverFetchBlock" case DBCCall: return "DBC.Call" case SQLCAdviseNewBlock: @@ -172,8 +164,6 @@ func (s RemoteFunc) String() string { return "SQLC.SignBilling" case SQLCLaunchBilling: return "SQLC.LaunchBilling" - case OBSAdviseNewBlock: - return "OBS.AdviseNewBlock" case MCCAdviseNewBlock: return "MCC.AdviseNewBlock" case MCCAdviseTxBilling: diff --git a/route/acl_test.go b/route/acl_test.go index c59e85811..60732f031 100644 --- a/route/acl_test.go +++ b/route/acl_test.go @@ -59,7 +59,7 @@ func TestIsPermitted(t *testing.T) { }) Convey("string RemoteFunc", t, func() { - for i := DHTPing; i <= OBSAdviseNewBlock; i++ { + for i := DHTPing; i <= MCCQueryTxState; i++ { So(fmt.Sprintf("%s", RemoteFunc(i)), ShouldContainSubstring, ".") } So(fmt.Sprintf("%s", RemoteFunc(9999)), ShouldContainSubstring, "Unknown") diff --git a/sqlchain/blockindex.go b/sqlchain/blockindex.go index 04bf7d97f..4256b675f 100644 --- a/sqlchain/blockindex.go +++ b/sqlchain/blockindex.go @@ -77,6 +77,21 @@ func (n *blockNode) ancestor(height int32) (ancestor *blockNode) { return } +func (n *blockNode) ancestorByCount(count int32) (ancestor *blockNode) { + if count < 0 || count > n.count { + return nil + } + + for ancestor = n; ancestor != nil && ancestor.count > count; ancestor = ancestor.parent { + } + + if ancestor != nil && ancestor.count < count { + ancestor = nil + } + + return +} + func (n *blockNode) indexKey() (key []byte) { key = make([]byte, hash.HashSize+4) binary.BigEndian.PutUint32(key[0:4], uint32(n.height)) diff --git a/sqlchain/chain.go b/sqlchain/chain.go index 2ee56f765..dae801222 100644 --- a/sqlchain/chain.go +++ b/sqlchain/chain.go @@ -119,15 +119,6 @@ type Chain struct { gasPrice uint64 updatePeriod uint64 - // observerLock defines the lock of observer update operations. - observerLock sync.Mutex - // observers defines the observer nodes of current chain. - observers map[proto.NodeID]int32 - // observerReplicators defines the observer states of current chain. - observerReplicators map[proto.NodeID]*observerReplicator - // replCh defines the replication trigger channel for replication check. - replCh chan struct{} - // Cached fileds, may need to renew some of this fields later. // // pk is the private key of the local miner. @@ -215,11 +206,6 @@ func NewChainWithContext(ctx context.Context, c *Config) (chain *Chain, err erro updatePeriod: c.UpdatePeriod, databaseID: c.DatabaseID, - // Observer related - observers: make(map[proto.NodeID]int32), - observerReplicators: make(map[proto.NodeID]*observerReplicator), - replCh: make(chan struct{}), - pk: pk, addr: &addr, } @@ -295,11 +281,6 @@ func LoadChainWithContext(ctx context.Context, c *Config) (chain *Chain, err err updatePeriod: c.UpdatePeriod, databaseID: c.DatabaseID, - // Observer related - observers: make(map[proto.NodeID]int32), - observerReplicators: make(map[proto.NodeID]*observerReplicator), - replCh: make(chan struct{}), - pk: pk, addr: &addr, } @@ -657,8 +638,6 @@ func (c *Chain) produceBlock(now time.Time) (err error) { } wg.Wait() - // fire replication to observers - c.startStopReplication(c.rt.ctx) return } @@ -939,8 +918,6 @@ func (c *Chain) processBlocks(ctx context.Context) { } } } - // fire replication to observers - c.startStopReplication(c.rt.ctx) case <-ctx.Done(): return } @@ -955,7 +932,6 @@ func (c *Chain) Start() (err error) { c.rt.goFunc(c.processBlocks) c.rt.goFunc(c.mainCycle) - c.rt.goFunc(c.replicationCycle) c.rt.startService(c) return } @@ -1007,21 +983,53 @@ func (c *Chain) Stop() (err error) { // FetchBlock fetches the block at specified height from local cache. func (c *Chain) FetchBlock(height int32) (b *types.Block, err error) { if n := c.rt.getHead().node.ancestor(height); n != nil { - k := utils.ConcatAll(metaBlockIndex[:], n.indexKey()) - var v []byte - v, err = c.bdb.Get(k, nil) + b, err = c.fetchBlockByIndexKey(n.indexKey()) if err != nil { - err = errors.Wrapf(err, "fetch block %s", string(k)) return } + } + + return +} - b = &types.Block{} - statBlock(b) - err = utils.DecodeMsgPack(v, b) +// FetchBlockByCount fetches the block at specified count from local cache. +func (c *Chain) FetchBlockByCount(count int32) (b *types.Block, realCount int32, height int32, err error) { + var n *blockNode + + if count < 0 { + n = c.rt.getHead().node + } else { + n = c.rt.getHead().node.ancestorByCount(count) + } + + if n != nil { + b, err = c.fetchBlockByIndexKey(n.indexKey()) if err != nil { - err = errors.Wrapf(err, "fetch block %s", string(k)) return } + + height = n.height + realCount = n.count + } + + return +} + +func (c *Chain) fetchBlockByIndexKey(indexKey []byte) (b *types.Block, err error) { + k := utils.ConcatAll(metaBlockIndex[:], indexKey) + var v []byte + v, err = c.bdb.Get(k, nil) + if err != nil { + err = errors.Wrapf(err, "fetch block %s", string(k)) + return + } + + b = &types.Block{} + statBlock(b) + err = utils.DecodeMsgPack(v, b) + if err != nil { + err = errors.Wrapf(err, "fetch block %s", string(k)) + return } return @@ -1121,82 +1129,6 @@ func (c *Chain) UpdatePeers(peers *proto.Peers) error { return c.rt.updatePeers(peers) } -// AddSubscription is used by dbms to add an observer. -func (c *Chain) AddSubscription(nodeID proto.NodeID, startHeight int32) (err error) { - // send previous height and transactions using AdviseAckedQuery/AdviseNewBlock RPC method - // add node to subscriber list - c.observerLock.Lock() - defer c.observerLock.Unlock() - c.observers[nodeID] = startHeight - c.startStopReplication(c.rt.ctx) - return -} - -// CancelSubscription is used by dbms to cancel an observer. -func (c *Chain) CancelSubscription(nodeID proto.NodeID) (err error) { - // remove node from subscription list - c.observerLock.Lock() - defer c.observerLock.Unlock() - delete(c.observers, nodeID) - c.startStopReplication(c.rt.ctx) - return -} - -func (c *Chain) startStopReplication(ctx context.Context) { - if c.replCh != nil { - select { - case c.replCh <- struct{}{}: - case <-ctx.Done(): - default: - } - } -} - -func (c *Chain) populateObservers() { - c.observerLock.Lock() - defer c.observerLock.Unlock() - - // handle replication threads - for nodeID, startHeight := range c.observers { - if replicator, exists := c.observerReplicators[nodeID]; exists { - // already started - if startHeight >= 0 { - replicator.setNewHeight(startHeight) - c.observers[nodeID] = int32(-1) - } - } else { - // start new replication routine - replicator := newObserverReplicator(nodeID, startHeight, c) - c.observerReplicators[nodeID] = replicator - c.rt.goFunc(replicator.run) - } - } - - // stop replicators - for nodeID, replicator := range c.observerReplicators { - if _, exists := c.observers[nodeID]; !exists { - replicator.stop() - delete(c.observerReplicators, nodeID) - } - } -} - -func (c *Chain) replicationCycle(ctx context.Context) { - for { - select { - case <-c.replCh: - // populateObservers - c.populateObservers() - // send triggers to replicators - for _, replicator := range c.observerReplicators { - replicator.tick() - } - case <-ctx.Done(): - return - } - } -} - // Query queries req from local chain state and returns the query results in resp. func (c *Chain) Query( req *types.Request, isLeader bool) (tracker *x.QueryTracker, resp *types.Response, err error, diff --git a/sqlchain/observer.go b/sqlchain/observer.go deleted file mode 100644 index 169b4892d..000000000 --- a/sqlchain/observer.go +++ /dev/null @@ -1,257 +0,0 @@ -/* - * Copyright 2018 The CovenantSQL 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. - */ - -package sqlchain - -import ( - "context" - "sync" - - "github.com/CovenantSQL/CovenantSQL/proto" - "github.com/CovenantSQL/CovenantSQL/route" - "github.com/CovenantSQL/CovenantSQL/types" - "github.com/CovenantSQL/CovenantSQL/utils/log" -) - -/* -Observer implements method AdviseNewBlock to receive blocks from sqlchain node. -Request/Response entity from sqlchain api is re-used for simplicity. - -type Observer interface { - AdviseNewBlock(*MuxAdviseNewBlockReq, *MuxAdviseNewBlockResp) error -} -*/ - -// observerReplicator defines observer replication state. -type observerReplicator struct { - nodeID proto.NodeID - height int32 - triggerCh chan struct{} - stopOnce sync.Once - stopCh chan struct{} - replLock sync.Mutex - c *Chain -} - -// newObserverReplicator creates new observer. -func newObserverReplicator(nodeID proto.NodeID, startHeight int32, c *Chain) *observerReplicator { - return &observerReplicator{ - nodeID: nodeID, - height: startHeight, - triggerCh: make(chan struct{}, 1), - stopCh: make(chan struct{}, 1), - c: c, - } -} - -func (r *observerReplicator) setNewHeight(newHeight int32) { - r.replLock.Lock() - defer r.replLock.Unlock() - r.height = newHeight -} - -func (r *observerReplicator) stop() { - r.stopOnce.Do(func() { - select { - case <-r.stopCh: - default: - close(r.stopCh) - } - }) -} - -func (r *observerReplicator) replicate() { - r.replLock.Lock() - defer r.replLock.Unlock() - - var err error - - defer func() { - if err != nil { - // TODO(xq262144), add backoff logic to prevent sqlchain node from flooding the observer - } - }() - - curHeight := r.c.rt.getHead().Height - - if r.height == types.ReplicateFromNewest { - log.WithFields(log.Fields{ - "node": r.nodeID, - "height": curHeight, - }).Warning("observer being set to read from the newest block") - r.height = curHeight - } else if r.height > curHeight+1 { - log.WithFields(log.Fields{ - "node": r.nodeID, - "height": r.height, - }).Warning("observer subscribes to height not yet produced") - log.WithFields(log.Fields{ - "node": r.nodeID, - "height": curHeight + 1, - }).Warning("reset observer to height") - r.height = curHeight + 1 - } else if r.height == curHeight+1 { - // wait for next block - log.WithField("node", r.nodeID).Info("no more blocks for observer to read") - return - } - - log.WithFields(log.Fields{ - "node": r.nodeID, - "height": r.height, - }).Debug("try replicating block for observer") - - // replicate one record - var block *types.Block - if block, err = r.c.FetchBlock(r.height); err != nil { - // fetch block failed - log.WithField("height", r.height).WithError(err).Warning("fetch block with height failed") - return - } else if block == nil { - log.WithFields(log.Fields{ - "node": r.nodeID, - "height": r.height, - }).Debug("no block of height for observer") - - // black hole in chain? - // find last available block - log.Debug("start block height hole detection") - - var lastBlock, nextBlock *types.Block - var lastHeight, nextHeight int32 - - for h := r.height - 1; h >= 0; h-- { - if lastBlock, err = r.c.FetchBlock(h); err == nil && lastBlock != nil { - lastHeight = h - log.WithFields(log.Fields{ - "block": lastBlock.BlockHash().String(), - "height": lastHeight, - }).Debug("found last available block of height") - break - } - } - - if lastBlock == nil { - // could not find last available block, this should be a fatal issue - log.Warning("could not found last available block during hole detection") - return - } - - // find next available block - for h := r.height + 1; h <= curHeight; h++ { - if nextBlock, err = r.c.FetchBlock(h); err == nil && nextBlock != nil { - if !nextBlock.ParentHash().IsEqual(lastBlock.BlockHash()) { - // inconsistency - log.WithFields(log.Fields{ - "lastHeight": lastHeight, - "lastHash": lastBlock.BlockHash().String(), - "nextHeight": h, - "nextHash": nextBlock.BlockHash().String(), - "actualParentHash": nextBlock.ParentHash().String(), - }).Warning("inconsistency detected during hole detection") - - return - } - - nextHeight = h - log.WithFields(log.Fields{ - "block": nextBlock.BlockHash().String(), - "height": nextHeight, - }).Debug("found next available block of height") - break - } - } - - if nextBlock == nil { - // could not find next available block, try next time - log.Debug("could not found next available block during hole detection") - return - } - - // successfully found a hole in chain - log.WithFields(log.Fields{ - "fromBlock": lastBlock.BlockHash().String(), - "fromHeight": lastHeight, - "toBlock": nextBlock.BlockHash().String(), - "toHeight": nextHeight, - "skipped": nextHeight - lastHeight - 1, - }).Debug("found a hole in chain, skipping") - - r.height = nextHeight - block = nextBlock - - log.WithFields(log.Fields{ - "block": block.BlockHash().String(), - "height": r.height, - }).Debug("finish block height hole detection, skipping") - } - - // send block - req := &MuxAdviseNewBlockReq{ - Envelope: proto.Envelope{}, - DatabaseID: r.c.databaseID, - AdviseNewBlockReq: AdviseNewBlockReq{ - Block: block, - Count: func() int32 { - if nd := r.c.bi.lookupNode(block.BlockHash()); nd != nil { - return nd.count - } - if pn := r.c.bi.lookupNode(block.ParentHash()); pn != nil { - return pn.count + 1 - } - return -1 - }(), - }, - } - resp := &MuxAdviseNewBlockResp{} - err = r.c.cl.CallNode(r.nodeID, route.OBSAdviseNewBlock.String(), req, resp) - if err != nil { - log.WithFields(log.Fields{ - "node": r.nodeID, - "height": r.height, - }).WithError(err).Warning("send block advise to observer failed") - return - } - - // advance to next height - r.height++ - - if r.height <= r.c.rt.getHead().Height { - // send ticks to myself - r.tick() - } -} - -func (r *observerReplicator) tick() { - select { - case r.triggerCh <- struct{}{}: - default: - } -} -func (r *observerReplicator) run(ctx context.Context) { - for { - select { - case <-r.triggerCh: - // replication - r.replicate() - case <-ctx.Done(): - r.stop() - return - case <-r.stopCh: - return - } - } -} diff --git a/worker/dbms.go b/worker/dbms.go index 6aa7834f9..d608281af 100644 --- a/worker/dbms.go +++ b/worker/dbms.go @@ -572,75 +572,6 @@ func (dbms *DBMS) checkPermission(addr proto.AccountAddress, return } -func (dbms *DBMS) addTxSubscription(dbID proto.DatabaseID, nodeID proto.NodeID, startHeight int32) (err error) { - // check permission - pubkey, err := kms.GetPublicKey(nodeID) - if err != nil { - log.WithFields(log.Fields{ - "databaseID": dbID, - "nodeID": nodeID, - }).WithError(err).Warning("get public key failed in addTxSubscription") - return - } - addr, err := crypto.PubKeyHash(pubkey) - if err != nil { - log.WithFields(log.Fields{ - "databaseID": dbID, - "nodeID": nodeID, - }).WithError(err).Warning("generate addr failed in addTxSubscription") - return - } - - log.WithFields(log.Fields{ - "dbID": dbID, - "nodeID": nodeID, - "addr": addr.String(), - "startHeight": startHeight, - }).Debugf("addTxSubscription") - - err = dbms.checkPermission(addr, dbID, types.ReadQuery, nil) - if err != nil { - log.WithFields(log.Fields{"databaseID": dbID, "addr": addr}).WithError(err).Warning("permission deny") - return - } - - rawDB, ok := dbms.dbMap.Load(dbID) - if !ok { - err = ErrNotExists - log.WithFields(log.Fields{ - "databaseID": dbID, - "nodeID": nodeID, - "startHeight": startHeight, - }).WithError(err).Warning("unexpected error in addTxSubscription") - return - } - db := rawDB.(*Database) - err = db.chain.AddSubscription(nodeID, startHeight) - return -} - -func (dbms *DBMS) cancelTxSubscription(dbID proto.DatabaseID, nodeID proto.NodeID) (err error) { - rawDB, ok := dbms.dbMap.Load(dbID) - if !ok { - err = ErrNotExists - log.WithFields(log.Fields{ - "databaseID": dbID, - "nodeID": nodeID, - }).WithError(err).Warning("unexpected error in cancelTxSubscription") - return - } - db := rawDB.(*Database) - err = db.chain.CancelSubscription(nodeID) - if err != nil { - log.WithFields(log.Fields{ - "databaseID": dbID, - "nodeID": nodeID, - }).WithError(err).Warning("unexpected error in cancelTxSubscription") - return - } - return -} - // Shutdown defines dbms shutdown logic. func (dbms *DBMS) Shutdown() (err error) { dbms.dbMap.Range(func(_, rawDB interface{}) bool { diff --git a/worker/dbms_rpc.go b/worker/dbms_rpc.go index e3d5dda3b..b2289c7f0 100644 --- a/worker/dbms_rpc.go +++ b/worker/dbms_rpc.go @@ -30,27 +30,17 @@ var ( dbQueryFailCounter metrics.Meter ) -// SubscribeTransactionsReq defines a request of SubscribeTransaction RPC method. -type SubscribeTransactionsReq struct { +// ObserverFetchBlockReq defines the request for observer to fetch block. +type ObserverFetchBlockReq struct { proto.Envelope - DatabaseID proto.DatabaseID - Height int32 + proto.DatabaseID + Count int32 } -// SubscribeTransactionsResp defines a response of SubscribeTransaction RPC method. -type SubscribeTransactionsResp struct { - proto.Envelope -} - -// CancelSubscriptionReq defines a request of CancelSubscription RPC method. -type CancelSubscriptionReq struct { - proto.Envelope - DatabaseID proto.DatabaseID -} - -// CancelSubscriptionResp defines a response of CancelSubscription RPC method. -type CancelSubscriptionResp struct { - proto.Envelope +// ObserverFetchBlockResp defines the response for observer to fetch block. +type ObserverFetchBlockResp struct { + Count int32 + Block *types.Block } // DBMSRPCService is the rpc endpoint of database management. @@ -143,17 +133,3 @@ func (rpc *DBMSRPCService) Deploy(req *types.UpdateService, _ *types.UpdateServi return } - -// SubscribeTransactions is the RPC method to fetch subscribe new packed and confirmed transactions from the target server. -func (rpc *DBMSRPCService) SubscribeTransactions(req *SubscribeTransactionsReq, resp *SubscribeTransactionsResp) (err error) { - subscribeID := req.GetNodeID().ToNodeID() - err = rpc.dbms.addTxSubscription(req.DatabaseID, subscribeID, req.Height) - return -} - -// CancelSubscription is the RPC method to cancel subscription in the target server. -func (rpc *DBMSRPCService) CancelSubscription(req *CancelSubscriptionReq, _ *CancelSubscriptionResp) (err error) { - nodeID := req.GetNodeID().ToNodeID() - err = rpc.dbms.cancelTxSubscription(req.DatabaseID, nodeID) - return -} diff --git a/worker/dbms_test.go b/worker/dbms_test.go index 5984bb6c9..db29cead8 100644 --- a/worker/dbms_test.go +++ b/worker/dbms_test.go @@ -192,11 +192,9 @@ func TestDBMS(t *testing.T) { err = testRequest(route.DBSAck, ack, &ackRes) So(err, ShouldBeNil) - err = dbms.addTxSubscription(dbID2, nodeID, 1) + _, _, err = dbms.observerFetchBlock(dbID2, nodeID, 1) So(err.Error(), ShouldContainSubstring, ErrPermissionDeny.Error()) - err = dbms.addTxSubscription(dbID, nodeID, 1) - So(err, ShouldBeNil) - err = dbms.cancelTxSubscription(dbID, nodeID) + _, _, err = dbms.observerFetchBlock(dbID, nodeID, 1) So(err, ShouldBeNil) // revoke write permission @@ -235,7 +233,7 @@ func TestDBMS(t *testing.T) { err = testRequest(route.DBSQuery, readQuery, &queryRes) So(err, ShouldBeNil) - err = dbms.addTxSubscription(dbID, nodeID, 1) + _, _, err = dbms.observerFetchBlock(dbID, nodeID, 1) So(err, ShouldBeNil) }) @@ -315,7 +313,7 @@ func TestDBMS(t *testing.T) { err = testRequest(route.DBSQuery, readQuery, &queryRes) So(err.Error(), ShouldContainSubstring, ErrPermissionDeny.Error()) - err = dbms.addTxSubscription(dbID, nodeID, 1) + _, _, err = dbms.observerFetchBlock(dbID, nodeID, 1) So(err.Error(), ShouldContainSubstring, ErrPermissionDeny.Error()) }) diff --git a/worker/observer.go b/worker/observer.go new file mode 100644 index 000000000..37e580c38 --- /dev/null +++ b/worker/observer.go @@ -0,0 +1,98 @@ +/* + * Copyright 2019 The CovenantSQL 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. + */ + +package worker + +import ( + "github.com/CovenantSQL/CovenantSQL/crypto" + "github.com/CovenantSQL/CovenantSQL/crypto/asymmetric" + "github.com/CovenantSQL/CovenantSQL/crypto/kms" + "github.com/CovenantSQL/CovenantSQL/proto" + "github.com/CovenantSQL/CovenantSQL/types" + "github.com/CovenantSQL/CovenantSQL/utils/log" +) + +// ObserverFetchBlock handles observer fetch block logic. +func (rpc *DBMSRPCService) ObserverFetchBlock(req *ObserverFetchBlockReq, resp *ObserverFetchBlockResp) (err error) { + subscriberID := req.GetNodeID().ToNodeID() + resp.Block, resp.Count, err = rpc.dbms.observerFetchBlock(req.DatabaseID, subscriberID, req.Count) + return +} + +func (dbms *DBMS) observerFetchBlock(dbID proto.DatabaseID, nodeID proto.NodeID, count int32) ( + block *types.Block, realCount int32, err error) { + var ( + pubKey *asymmetric.PublicKey + addr proto.AccountAddress + height int32 + ) + + // node parameters + pubKey, err = kms.GetPublicKey(nodeID) + if err != nil { + log.WithFields(log.Fields{ + "databaseID": dbID, + "nodeID": nodeID, + }).WithError(err).Warning("get public key failed in observerFetchBlock") + return + } + + addr, err = crypto.PubKeyHash(pubKey) + if err != nil { + log.WithFields(log.Fields{ + "databaseID": dbID, + "nodeID": nodeID, + }).WithError(err).Warning("generate addr failed in observerFetchBlock") + return + } + + defer func() { + lf := log.WithFields(log.Fields{ + "dbID": dbID, + "nodeID": nodeID, + "addr": addr.String(), + "count": count, + }) + + if err != nil { + lf.WithError(err).Debug("observer fetch block") + } else { + if block != nil { + lf = lf.WithField("block", block.BlockHash()) + } + lf.WithField("height", height).Debug("observer fetch block") + } + }() + + // check permission + err = dbms.checkPermission(addr, dbID, types.ReadQuery, nil) + if err != nil { + log.WithFields(log.Fields{ + "databaseID": dbID, + "addr": addr, + }).WithError(err).Warning("permission deny") + return + } + + rawDB, ok := dbms.dbMap.Load(dbID) + if !ok { + err = ErrNotExists + return + } + db := rawDB.(*Database) + block, realCount, height, err = db.chain.FetchBlockByCount(count) + return +} From baf34342b24f99d94a9f13a3bbfaf5e2c7702b71 Mon Sep 17 00:00:00 2001 From: Qi Xiao Date: Mon, 18 Feb 2019 17:16:49 +0800 Subject: [PATCH 2/3] Fix bug of observer restart recover --- cmd/cql-observer/api.go | 11 +++++++++++ cmd/cql-observer/observation_test.go | 26 ++++++++++++++++++++++++++ cmd/cql-observer/service.go | 24 ++++++++++++++++++++++++ cmd/cql-observer/worker.go | 14 ++++++++++++-- 4 files changed, 73 insertions(+), 2 deletions(-) diff --git a/cmd/cql-observer/api.go b/cmd/cql-observer/api.go index 4026c3a89..b22a0eac4 100644 --- a/cmd/cql-observer/api.go +++ b/cmd/cql-observer/api.go @@ -83,6 +83,16 @@ func newPaginationFromReq(r *http.Request) (op *paginationOps) { return } +func (a *explorerAPI) GetAllSubscriptions(rw http.ResponseWriter, r *http.Request) { + subscriptions, err := a.service.getAllSubscriptions() + if err != nil { + sendResponse(500, false, err, nil, rw) + return + } + + sendResponse(200, true, "", subscriptions, rw) +} + func (a *explorerAPI) GetAck(rw http.ResponseWriter, r *http.Request) { vars := mux.Vars(r) @@ -677,6 +687,7 @@ func startAPI(service *Service, listenAddr string) (server *http.Server, err err v3Router.HandleFunc("/count/{db}/{count:[0-9]+}", api.GetBlockByCountV3).Methods("GET") v3Router.HandleFunc("/height/{db}/{height:[0-9]+}", api.GetBlockByHeightV3).Methods("GET") v3Router.HandleFunc("/head/{db}", api.GetHighestBlockV3).Methods("GET") + v3Router.HandleFunc("/subscriptions", api.GetAllSubscriptions).Methods("GET") server = &http.Server{ Addr: listenAddr, diff --git a/cmd/cql-observer/observation_test.go b/cmd/cql-observer/observation_test.go index 532672371..a0d0d7d87 100644 --- a/cmd/cql-observer/observation_test.go +++ b/cmd/cql-observer/observation_test.go @@ -701,6 +701,32 @@ func TestFullProcess(t *testing.T) { err = client.Drop(dsn2) So(err, ShouldBeNil) + + observerCmd.Cmd.Process.Signal(os.Interrupt) + observerCmd.Cmd.Wait() + + // start observer again + observerCmd, err = utils.RunCommandNB( + FJ(baseDir, "./bin/cql-observer.test"), + []string{"-config", FJ(testWorkingDir, "./observation/node_observer/config.yaml"), + "-database", string(dbID), "-reset", "oldest", + "-test.coverprofile", FJ(baseDir, "./cmd/cql-observer/observer.cover.out"), + }, + "observer", testWorkingDir, logDir, false, + ) + So(err, ShouldBeNil) + + // call observer subscription status + // wait for observer to start + time.Sleep(time.Second * 3) + + res, err = getJSON("v3/subscriptions") + So(err, ShouldBeNil) + subscriptions, err := res.Object() + So(subscriptions, ShouldContainKey, string(dbID)) + So(subscriptions, ShouldContainKey, string(dbID2)) + So(subscriptions[string(dbID)], ShouldBeGreaterThanOrEqualTo, 1) + So(subscriptions[string(dbID2)], ShouldBeGreaterThanOrEqualTo, 0) }) } diff --git a/cmd/cql-observer/service.go b/cmd/cql-observer/service.go index c49138860..fc6379606 100644 --- a/cmd/cql-observer/service.go +++ b/cmd/cql-observer/service.go @@ -225,6 +225,19 @@ func (s *Service) start() (err error) { return nil } +func (s *Service) saveSubscriptionStatus(dbID proto.DatabaseID, count int32) (err error) { + log.WithFields(log.Fields{}).Debug("save subscription status") + + if atomic.LoadInt32(&s.stopped) == 1 { + // stopped + return ErrStopped + } + + return s.db.Update(func(tx *bolt.Tx) error { + return tx.Bucket(subscriptionBucket).Put([]byte(dbID), int32ToBytes(count)) + }) +} + func (s *Service) addAck(dbID proto.DatabaseID, height int32, offset int32, ack *types.SignedAckHeader) (err error) { log.WithFields(log.Fields{ "height": height, @@ -773,3 +786,14 @@ func (s *Service) getBlock(dbID proto.DatabaseID, h *hash.Hash) (count int32, he return } + +func (s *Service) getAllSubscriptions() (subscriptions map[proto.DatabaseID]int32, err error) { + subscriptions = map[proto.DatabaseID]int32{} + s.subscription.Range(func(_, rawWorker interface{}) bool { + worker := unpackWorker(rawWorker) + subscriptions[worker.dbID] = worker.getHead() + return true + }) + + return +} diff --git a/cmd/cql-observer/worker.go b/cmd/cql-observer/worker.go index 23f12e24d..fdc8e6920 100644 --- a/cmd/cql-observer/worker.go +++ b/cmd/cql-observer/worker.go @@ -58,7 +58,7 @@ func (w *subscribeWorker) run() { case <-w.stopCh: return case <-time.After(nextTick): - if err := w.pull(atomic.LoadInt32(&w.head)); err != nil { + if err := w.pull(w.getHead()); err != nil { // calc next tick nextTick = conf.GConf.SQLChainPeriod } else { @@ -115,7 +115,10 @@ func (w *subscribeWorker) pull(count int32) (err error) { next = count + 1 } - atomic.CompareAndSwapInt32(&w.head, count, next) + if atomic.CompareAndSwapInt32(&w.head, count, next) { + // update subscription status to database + _ = w.s.saveSubscriptionStatus(w.dbID, next) + } return } @@ -129,6 +132,9 @@ func (w *subscribeWorker) start() { w.l.Lock() defer w.l.Unlock() + // update subscription status to database + _ = w.s.saveSubscriptionStatus(w.dbID, w.getHead()) + if w.isStopped() { w.stopCh = make(chan struct{}) w.wg = new(sync.WaitGroup) @@ -137,6 +143,10 @@ func (w *subscribeWorker) start() { } } +func (w *subscribeWorker) getHead() int32 { + return atomic.LoadInt32(&w.head) +} + func (w *subscribeWorker) stop() { w.l.Lock() defer w.l.Unlock() From 64f76b57140cb899c6ac0188e61bf5a79edc57fa Mon Sep 17 00:00:00 2001 From: Qi Xiao Date: Mon, 18 Feb 2019 17:25:32 +0800 Subject: [PATCH 3/3] Add comments for count in ObserverFetchBlock rpc --- worker/dbms_rpc.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/worker/dbms_rpc.go b/worker/dbms_rpc.go index b2289c7f0..e0cff5b8c 100644 --- a/worker/dbms_rpc.go +++ b/worker/dbms_rpc.go @@ -34,12 +34,12 @@ var ( type ObserverFetchBlockReq struct { proto.Envelope proto.DatabaseID - Count int32 + Count int32 // sqlchain block serial number since genesis block (0) } // ObserverFetchBlockResp defines the response for observer to fetch block. type ObserverFetchBlockResp struct { - Count int32 + Count int32 // sqlchain block serial number since genesis block (0) Block *types.Block }