protocol/protocol.go

   1 package protocol
   2
   3 import (
   4         "context"
   5         "sync"
   6         "time"
   7
   8         "github.com/golang/groupcache/lru"
   9
  10         "github.com/bytom/errors"
  11         "github.com/bytom/log"
  12         "github.com/bytom/protocol/bc"
  13         "github.com/bytom/protocol/bc/legacy"
  14         "github.com/bytom/protocol/state"
  15 )
  16
  17 // maxCachedValidatedTxs is the max number of validated txs to cache.
  18 const maxCachedValidatedTxs = 1000
  19
  20 var (
  21         // ErrTheDistantFuture is returned when waiting for a blockheight
  22         // too far in excess of the tip of the blockchain.
  23         ErrTheDistantFuture = errors.New("block height too far in future")
  24 )
  25
  26 // Store provides storage for blockchain data: blocks and state tree
  27 // snapshots.
  28 //
  29 // Note, this is different from a state snapshot. A state snapshot
  30 // provides access to the state at a given point in time -- outputs
  31 // and issuance memory. The Chain type uses Store to load state
  32 // from storage and persist validated data.
  33 type Store interface {
  34         Height() uint64
  35         GetBlock(uint64) (*legacy.Block, error)
  36         LatestSnapshot(context.Context) (*state.Snapshot, uint64, error)
  37
  38         SaveBlock(*legacy.Block) error
  39         FinalizeBlock(context.Context, uint64) error
  40         SaveSnapshot(context.Context, uint64, *state.Snapshot) error
  41 }
  42
  43 // Chain provides a complete, minimal blockchain database. It
  44 // delegates the underlying storage to other objects, and uses
  45 // validation logic from package validation to decide what
  46 // objects can be safely stored.
  47 type Chain struct {
  48         InitialBlockHash  bc.Hash
  49         MaxIssuanceWindow time.Duration // only used by generators
  50
  51         state struct {
  52                 cond     sync.Cond // protects height, block, snapshot
  53                 height   uint64
  54                 block    *legacy.Block   // current only if leader
  55                 snapshot *state.Snapshot // current only if leader
  56         }
  57         store Store
  58
  59         lastQueuedSnapshot time.Time
  60         pendingSnapshots   chan pendingSnapshot
  61
  62         prevalidated prevalidatedTxsCache
  63 }
  64
  65 type pendingSnapshot struct {
  66         height   uint64
  67         snapshot *state.Snapshot
  68 }
  69
  70 // NewChain returns a new Chain using store as the underlying storage.
  71 func NewChain(ctx context.Context, initialBlockHash bc.Hash, store Store, heights <-chan uint64) (*Chain, error) {
  72         c := &Chain{
  73                 InitialBlockHash: initialBlockHash,
  74                 store:            store,
  75                 pendingSnapshots: make(chan pendingSnapshot, 1),
  76                 prevalidated: prevalidatedTxsCache{
  77                         lru: lru.New(maxCachedValidatedTxs),
  78                 },
  79         }
  80         c.state.cond.L = new(sync.Mutex)
  81
  82         c.state.height = store.Height()
  83
  84         // Note that c.height.n may still be zero here.
  85         if heights != nil {
  86                 go func() {
  87                         for h := range heights {
  88                                 c.setHeight(h)
  89                         }
  90                 }()
  91         }
  92
  93         go func() {
  94                 for {
  95                         select {
  96                         case <-ctx.Done():
  97                                 return
  98                         case ps := <-c.pendingSnapshots:
  99                                 err := store.SaveSnapshot(ctx, ps.height, ps.snapshot)
 100                                 if err != nil {
 101                                         log.Error(ctx, err, "at", "saving snapshot")
 102                                 }
 103                         }
 104                 }
 105         }()
 106
 107         return c, nil
 108 }
 109
 110 func (c *Chain) GetStore() *Store {
 111         return &(c.store)
 112 }
 113
 114 // Height returns the current height of the blockchain.
 115 func (c *Chain) Height() uint64 {
 116         c.state.cond.L.Lock()
 117         defer c.state.cond.L.Unlock()
 118         return c.state.height
 119 }
 120
 121 // TimestampMS returns the latest known block timestamp.
 122 func (c *Chain) TimestampMS() uint64 {
 123         c.state.cond.L.Lock()
 124         defer c.state.cond.L.Unlock()
 125         if c.state.block == nil {
 126                 return 0
 127         }
 128         return c.state.block.TimestampMS
 129 }
 130
 131 // State returns the most recent state available. It will not be current
 132 // unless the current process is the leader. Callers should examine the
 133 // returned block header's height if they need to verify the current state.
 134 func (c *Chain) State() (*legacy.Block, *state.Snapshot) {
 135         c.state.cond.L.Lock()
 136         defer c.state.cond.L.Unlock()
 137         return c.state.block, c.state.snapshot
 138 }
 139
 140 func (c *Chain) setState(b *legacy.Block, s *state.Snapshot) {
 141         c.state.cond.L.Lock()
 142         defer c.state.cond.L.Unlock()
 143         c.state.block = b
 144         c.state.snapshot = s
 145         if b != nil && b.Height > c.state.height {
 146                 c.state.height = b.Height
 147                 c.state.cond.Broadcast()
 148         }
 149 }
 150
 151 // BlockSoonWaiter returns a channel that
 152 // waits for the block at the given height,
 153 // but it is an error to wait for a block far in the future.
 154 // WaitForBlockSoon will timeout if the context times out.
 155 // To wait unconditionally, the caller should use WaitForBlock.
 156 func (c *Chain) BlockSoonWaiter(ctx context.Context, height uint64) <-chan error {
 157         ch := make(chan error, 1)
 158
 159         go func() {
 160                 const slop = 3
 161                 if height > c.Height()+slop {
 162                         ch <- ErrTheDistantFuture
 163                         return
 164                 }
 165
 166                 select {
 167                 case <-c.BlockWaiter(height):
 168                         ch <- nil
 169                 case <-ctx.Done():
 170                         ch <- ctx.Err()
 171                 }
 172         }()
 173
 174         return ch
 175 }
 176
 177 // BlockWaiter returns a channel that
 178 // waits for the block at the given height.
 179 func (c *Chain) BlockWaiter(height uint64) <-chan struct{} {
 180         ch := make(chan struct{}, 1)
 181         go func() {
 182                 c.state.cond.L.Lock()
 183                 defer c.state.cond.L.Unlock()
 184                 for c.state.height < height {
 185                         c.state.cond.Wait()
 186                 }
 187                 ch <- struct{}{}
 188         }()
 189
 190         return ch
 191 }