protocol/protocol.go

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