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
  81         if c.state.height < 1 {
  82                 c.state.snapshot = state.Empty()
  83         } else {
  84                 c.state.block, _ = store.GetBlock(c.state.height)
  85                 c.state.snapshot, _, _ = store.LatestSnapshot(ctx)
  86         }
  87
  88         // Note that c.height.n may still be zero here.
  89         if heights != nil {
  90                 go func() {
  91                         for h := range heights {
  92                                 c.setHeight(h)
  93                         }
  94                 }()
  95         }
  96
  97         go func() {
  98                 for {
  99                         select {
 100                         case <-ctx.Done():
 101                                 return
 102                         case ps := <-c.pendingSnapshots:
 103                                 err := store.SaveSnapshot(ctx, ps.height, ps.snapshot)
 104                                 if err != nil {
 105                                         log.Error(ctx, err, "at", "saving snapshot")
 106                                 }
 107                         }
 108                 }
 109         }()
 110
 111         return c, nil
 112 }
 113
 114 func (c *Chain) GetStore() *Store {
 115         return &(c.store)
 116 }
 117
 118 // Height returns the current height of the blockchain.
 119 func (c *Chain) Height() uint64 {
 120         c.state.cond.L.Lock()
 121         defer c.state.cond.L.Unlock()
 122         return c.state.height
 123 }
 124
 125 // TimestampMS returns the latest known block timestamp.
 126 func (c *Chain) TimestampMS() uint64 {
 127         c.state.cond.L.Lock()
 128         defer c.state.cond.L.Unlock()
 129         if c.state.block == nil {
 130                 return 0
 131         }
 132         return c.state.block.TimestampMS
 133 }
 134
 135 // State returns the most recent state available. It will not be current
 136 // unless the current process is the leader. Callers should examine the
 137 // returned block header's height if they need to verify the current state.
 138 func (c *Chain) State() (*legacy.Block, *state.Snapshot) {
 139         c.state.cond.L.Lock()
 140         defer c.state.cond.L.Unlock()
 141         return c.state.block, c.state.snapshot
 142 }
 143
 144 func (c *Chain) setState(b *legacy.Block, s *state.Snapshot) {
 145         c.state.cond.L.Lock()
 146         defer c.state.cond.L.Unlock()
 147         c.state.block = b
 148         c.state.snapshot = s
 149         if b != nil && b.Height > c.state.height {
 150                 c.state.height = b.Height
 151                 c.state.cond.Broadcast()
 152         }
 153 }
 154
 155 // BlockSoonWaiter returns a channel that
 156 // waits for the block at the given height,
 157 // but it is an error to wait for a block far in the future.
 158 // WaitForBlockSoon will timeout if the context times out.
 159 // To wait unconditionally, the caller should use WaitForBlock.
 160 func (c *Chain) BlockSoonWaiter(ctx context.Context, height uint64) <-chan error {
 161         ch := make(chan error, 1)
 162
 163         go func() {
 164                 const slop = 3
 165                 if height > c.Height()+slop {
 166                         ch <- ErrTheDistantFuture
 167                         return
 168                 }
 169
 170                 select {
 171                 case <-c.BlockWaiter(height):
 172                         ch <- nil
 173                 case <-ctx.Done():
 174                         ch <- ctx.Err()
 175                 }
 176         }()
 177
 178         return ch
 179 }
 180
 181 // BlockWaiter returns a channel that
 182 // waits for the block at the given height.
 183 func (c *Chain) BlockWaiter(height uint64) <-chan struct{} {
 184         ch := make(chan struct{}, 1)
 185         go func() {
 186                 c.state.cond.L.Lock()
 187                 defer c.state.cond.L.Unlock()
 188                 for c.state.height < height {
 189                         c.state.cond.Wait()
 190                 }
 191                 ch <- struct{}{}
 192         }()
 193
 194         return ch
 195 }