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