protocol/protocol.go

   1 package protocol
   2
   3 import (
   4         "context"
   5         "sync"
   6         "time"
   7
   8         "github.com/bytom/blockchain/txdb"
   9         "github.com/bytom/errors"
  10         "github.com/bytom/protocol/bc"
  11         "github.com/bytom/protocol/bc/legacy"
  12         "github.com/bytom/protocol/state"
  13 )
  14
  15 // maxCachedValidatedTxs is the max number of validated txs to cache.
  16 const maxCachedValidatedTxs = 1000
  17
  18 var (
  19         // ErrTheDistantFuture is returned when waiting for a blockheight
  20         // too far in excess of the tip of the blockchain.
  21         ErrTheDistantFuture = errors.New("block height too far in future")
  22 )
  23
  24 // Store provides storage for blockchain data: blocks and state tree
  25 // snapshots.
  26 //
  27 // Note, this is different from a state snapshot. A state snapshot
  28 // provides access to the state at a given point in time -- outputs
  29 // and issuance memory. The Chain type uses Store to load state
  30 // from storage and persist validated data.
  31 type Store interface {
  32         BlockExist(*bc.Hash) bool
  33
  34         GetBlock(*bc.Hash) (*legacy.Block, error)
  35         GetMainchain(*bc.Hash) (map[uint64]*bc.Hash, error)
  36         GetSnapshot(*bc.Hash) (*state.Snapshot, error)
  37         GetStoreStatus() txdb.BlockStoreStateJSON
  38
  39         SaveBlock(*legacy.Block) error
  40         SaveMainchain(map[uint64]*bc.Hash, *bc.Hash) error
  41         SaveSnapshot(*state.Snapshot, *bc.Hash) error
  42         SaveStoreStatus(uint64, *bc.Hash)
  43 }
  44
  45 // OrphanManage is use to handle all the orphan block
  46 type OrphanManage struct {
  47         //TODO: add orphan cached block limit
  48         orphan     map[bc.Hash]*legacy.Block
  49         preOrphans map[bc.Hash][]*bc.Hash
  50         mtx        sync.RWMutex
  51 }
  52
  53 // NewOrphanManage return a new orphan block
  54 func NewOrphanManage() *OrphanManage {
  55         return &OrphanManage{
  56                 orphan:     make(map[bc.Hash]*legacy.Block),
  57                 preOrphans: make(map[bc.Hash][]*bc.Hash),
  58         }
  59 }
  60
  61 // BlockExist check is the block in OrphanManage
  62 func (o *OrphanManage) BlockExist(hash *bc.Hash) bool {
  63         o.mtx.RLock()
  64         _, ok := o.orphan[*hash]
  65         o.mtx.RUnlock()
  66         return ok
  67 }
  68
  69 // Add will add the block to OrphanManage
  70 func (o *OrphanManage) Add(block *legacy.Block) {
  71         blockHash := block.Hash()
  72         o.mtx.Lock()
  73         defer o.mtx.Unlock()
  74
  75         if _, ok := o.orphan[blockHash]; ok {
  76                 return
  77         }
  78
  79         o.orphan[blockHash] = block
  80         o.preOrphans[block.PreviousBlockHash] = append(o.preOrphans[block.PreviousBlockHash], &blockHash)
  81 }
  82
  83 // Delete will delelte the block from OrphanManage
  84 func (o *OrphanManage) Delete(hash *bc.Hash) {
  85         o.mtx.Lock()
  86         defer o.mtx.Unlock()
  87         block, ok := o.orphan[*hash]
  88         if !ok {
  89                 return
  90         }
  91         delete(o.orphan, *hash)
  92
  93         preOrphans, ok := o.preOrphans[block.PreviousBlockHash]
  94         if !ok || len(preOrphans) == 1 {
  95                 delete(o.preOrphans, block.PreviousBlockHash)
  96                 return
  97         }
  98
  99         for i, preOrphan := range preOrphans {
 100                 if preOrphan == hash {
 101                         o.preOrphans[block.PreviousBlockHash] = append(preOrphans[:i], preOrphans[i+1:]...)
 102                         return
 103                 }
 104         }
 105 }
 106
 107 // Get return the orphan block by hash
 108 func (o *OrphanManage) Get(hash *bc.Hash) (*legacy.Block, bool) {
 109         o.mtx.RLock()
 110         block, ok := o.orphan[*hash]
 111         o.mtx.RUnlock()
 112         return block, ok
 113 }
 114
 115 // Chain provides a complete, minimal blockchain database. It
 116 // delegates the underlying storage to other objects, and uses
 117 // validation logic from package validation to decide what
 118 // objects can be safely stored.
 119 type Chain struct {
 120         InitialBlockHash  bc.Hash
 121         MaxIssuanceWindow time.Duration // only used by generators
 122
 123         orphanManage *OrphanManage
 124         txPool       *TxPool
 125
 126         state struct {
 127                 cond      sync.Cond
 128                 block     *legacy.Block
 129                 hash      *bc.Hash
 130                 mainChain map[uint64]*bc.Hash
 131                 snapshot  *state.Snapshot
 132         }
 133         store Store
 134 }
 135
 136 // NewChain returns a new Chain using store as the underlying storage.
 137 func NewChain(initialBlockHash bc.Hash, store Store, txPool *TxPool) (*Chain, error) {
 138         c := &Chain{
 139                 InitialBlockHash: initialBlockHash,
 140                 orphanManage:     NewOrphanManage(),
 141                 store:            store,
 142                 txPool:           txPool,
 143         }
 144         c.state.cond.L = new(sync.Mutex)
 145         storeStatus := store.GetStoreStatus()
 146
 147         if storeStatus.Height == 0 {
 148                 c.state.snapshot = state.Empty()
 149                 c.state.mainChain = make(map[uint64]*bc.Hash)
 150                 return c, nil
 151         }
 152
 153         c.state.hash = storeStatus.Hash
 154         var err error
 155         if c.state.block, err = store.GetBlock(storeStatus.Hash); err != nil {
 156                 return nil, err
 157         }
 158         if c.state.snapshot, err = store.GetSnapshot(storeStatus.Hash); err != nil {
 159                 return nil, err
 160         }
 161         if c.state.mainChain, err = store.GetMainchain(storeStatus.Hash); err != nil {
 162                 return nil, err
 163         }
 164         return c, nil
 165 }
 166
 167 // Height returns the current height of the blockchain.
 168 func (c *Chain) Height() uint64 {
 169         c.state.cond.L.Lock()
 170         defer c.state.cond.L.Unlock()
 171         return c.state.block.Height
 172 }
 173
 174 func (c *Chain) BestBlockHash() *bc.Hash {
 175         c.state.cond.L.Lock()
 176         defer c.state.cond.L.Unlock()
 177         return c.state.hash
 178 }
 179
 180 func (c *Chain) inMainchain(block *legacy.Block) bool {
 181         hash, ok := c.state.mainChain[block.Height]
 182         if !ok {
 183                 return false
 184         }
 185         return *hash == block.Hash()
 186 }
 187
 188 // TimestampMS returns the latest known block timestamp.
 189 func (c *Chain) TimestampMS() uint64 {
 190         c.state.cond.L.Lock()
 191         defer c.state.cond.L.Unlock()
 192         if c.state.block == nil {
 193                 return 0
 194         }
 195         return c.state.block.TimestampMS
 196 }
 197
 198 // State returns the most recent state available. It will not be current
 199 // unless the current process is the leader. Callers should examine the
 200 // returned block header's height if they need to verify the current state.
 201 func (c *Chain) State() (*legacy.Block, *state.Snapshot) {
 202         c.state.cond.L.Lock()
 203         defer c.state.cond.L.Unlock()
 204         return c.state.block, c.state.snapshot
 205 }
 206
 207 // This function must be called with mu lock in above level
 208 func (c *Chain) setState(block *legacy.Block, s *state.Snapshot, m map[uint64]*bc.Hash) error {
 209         if block.AssetsMerkleRoot != s.Tree.RootHash() {
 210                 return ErrBadStateRoot
 211         }
 212
 213         blockHash := block.Hash()
 214         c.state.block = block
 215         c.state.hash = &blockHash
 216         c.state.snapshot = s
 217         for k, v := range m {
 218                 c.state.mainChain[k] = v
 219         }
 220
 221         if err := c.store.SaveSnapshot(c.state.snapshot, &blockHash); err != nil {
 222                 return err
 223         }
 224         if err := c.store.SaveMainchain(c.state.mainChain, &blockHash); err != nil {
 225                 return err
 226         }
 227         c.store.SaveStoreStatus(block.Height, &blockHash)
 228
 229         c.state.cond.Broadcast()
 230         return nil
 231 }
 232
 233 // BlockSoonWaiter returns a channel that
 234 // waits for the block at the given height,
 235 // but it is an error to wait for a block far in the future.
 236 // WaitForBlockSoon will timeout if the context times out.
 237 // To wait unconditionally, the caller should use WaitForBlock.
 238 func (c *Chain) BlockSoonWaiter(ctx context.Context, height uint64) <-chan error {
 239         ch := make(chan error, 1)
 240
 241         go func() {
 242                 const slop = 3
 243                 if height > c.Height()+slop {
 244                         ch <- ErrTheDistantFuture
 245                         return
 246                 }
 247
 248                 select {
 249                 case <-c.BlockWaiter(height):
 250                         ch <- nil
 251                 case <-ctx.Done():
 252                         ch <- ctx.Err()
 253                 }
 254         }()
 255
 256         return ch
 257 }
 258
 259 // BlockWaiter returns a channel that
 260 // waits for the block at the given height.
 261 func (c *Chain) BlockWaiter(height uint64) <-chan struct{} {
 262         ch := make(chan struct{}, 1)
 263         go func() {
 264                 c.state.cond.L.Lock()
 265                 defer c.state.cond.L.Unlock()
 266                 for c.state.block.Height < height {
 267                         c.state.cond.Wait()
 268                 }
 269                 ch <- struct{}{}
 270         }()
 271
 272         return ch
 273 }