protocol/block.go

   1 package protocol
   2
   3 import (
   4         "context"
   5         "time"
   6
   7         "github.com/bytom/errors"
   8         "github.com/bytom/log"
   9         "github.com/bytom/protocol/bc/legacy"
  10         "github.com/bytom/protocol/state"
  11         "github.com/bytom/protocol/validation"
  12 )
  13
  14 // maxBlockTxs limits the number of transactions
  15 // included in each block.
  16 const maxBlockTxs = 10000
  17
  18 // saveSnapshotFrequency stores how often to save a state
  19 // snapshot to the Store.
  20 const saveSnapshotFrequency = time.Hour
  21
  22 var (
  23         // ErrBadBlock is returned when a block is invalid.
  24         ErrBadBlock = errors.New("invalid block")
  25
  26         // ErrStaleState is returned when the Chain does not have a current
  27         // blockchain state.
  28         ErrStaleState = errors.New("stale blockchain state")
  29
  30         // ErrBadStateRoot is returned when the computed assets merkle root
  31         // disagrees with the one declared in a block header.
  32         ErrBadStateRoot = errors.New("invalid state merkle root")
  33 )
  34
  35 // GetBlock returns the block at the given height, if there is one,
  36 // otherwise it returns an error.
  37 func (c *Chain) GetBlock(height uint64) (*legacy.Block, error) {
  38         return c.store.GetBlock(height)
  39 }
  40
  41 // ValidateBlock validates an incoming block in advance of applying it
  42 // to a snapshot (with ApplyValidBlock) and committing it to the
  43 // blockchain (with CommitAppliedBlock).
  44 func (c *Chain) ValidateBlock(block, prev *legacy.Block) error {
  45         blockEnts := legacy.MapBlock(block)
  46         prevEnts := legacy.MapBlock(prev)
  47         err := validation.ValidateBlock(blockEnts, prevEnts)
  48         if err != nil {
  49                 return errors.Sub(ErrBadBlock, err)
  50         }
  51         return errors.Sub(ErrBadBlock, err)
  52 }
  53
  54 // ApplyValidBlock creates an updated snapshot without validating the
  55 // block.
  56 func (c *Chain) ApplyValidBlock(block *legacy.Block) (*state.Snapshot, error) {
  57         newSnapshot := state.Copy(c.state.snapshot)
  58         err := newSnapshot.ApplyBlock(legacy.MapBlock(block))
  59         if err != nil {
  60                 return nil, err
  61         }
  62         //fmt.Printf("want %v, ger %v \n", block.BlockHeader.AssetsMerkleRoot, newSnapshot.Tree.RootHash())
  63         if block.AssetsMerkleRoot != newSnapshot.Tree.RootHash() {
  64                 return nil, ErrBadStateRoot
  65         }
  66         return newSnapshot, nil
  67 }
  68
  69 // CommitBlock commits a block to the blockchain. The block
  70 // must already have been applied with ApplyValidBlock or
  71 // ApplyNewBlock, which will have produced the new snapshot that's
  72 // required here.
  73 //
  74 // This function saves the block to the store and sometimes (not more
  75 // often than saveSnapshotFrequency) saves the state tree to the
  76 // store. New-block callbacks (via asynchronous block-processor pins)
  77 // are triggered.
  78 //
  79 // TODO(bobg): rename to CommitAppliedBlock for clarity (deferred from https://github.com/chain/chain/pull/788)
  80 func (c *Chain) CommitAppliedBlock(ctx context.Context, block *legacy.Block, snapshot *state.Snapshot) error {
  81         // SaveBlock is the linearization point. Once the block is committed
  82         // to persistent storage, the block has been applied and everything
  83         // else can be derived from that block.
  84         err := c.store.SaveBlock(block)
  85         if err != nil {
  86                 return errors.Wrap(err, "storing block")
  87         }
  88         if block.Time().After(c.lastQueuedSnapshot.Add(saveSnapshotFrequency)) {
  89                 c.queueSnapshot(ctx, block.Height, block.Time(), snapshot)
  90         }
  91
  92         err = c.store.FinalizeBlock(ctx, block.Height)
  93         if err != nil {
  94                 return errors.Wrap(err, "finalizing block")
  95         }
  96
  97         // c.setState will update the local blockchain state and height.
  98         // When c.store is a txdb.Store, and c has been initialized with a
  99         // channel from txdb.ListenBlocks, then the above call to
 100         // c.store.FinalizeBlock will have done a postgresql NOTIFY and
 101         // that will wake up the goroutine in NewChain, which also calls
 102         // setHeight.  But duplicate calls with the same blockheight are
 103         // harmless; and the following call is required in the cases where
 104         // it's not redundant.
 105         c.setState(block, snapshot)
 106
 107         go c.SetAssetsAmount(block)
 108
 109         return nil
 110 }
 111
 112 func (c *Chain) SetAssetsAmount(block *legacy.Block){
 113         assets_amount := c.assets_utxo.assets_amount
 114
 115         if(block.Transactions != nil){
 116                 c.assets_utxo.cond.L.Lock()
 117                 for _,item := range block.Transactions[1:]{
 118                         if (item.Outputs != nil){
 119                                 for _,utxo := range item.Outputs{
 120                                         if _,ok := assets_amount[utxo.AssetId.String()]; ok {
 121                                                 assets_amount[utxo.AssetId.String()] += utxo.Amount
 122                                         }else{
 123                                                 assets_amount[utxo.AssetId.String()] = utxo.Amount
 124                                         }
 125
 126                                 }
 127                         }
 128                 }
 129                 c.assets_utxo.cond.L.Unlock()
 130         }
 131 }
 132
 133 func (c *Chain) GetAssetsAmount() ([]interface{}) {
 134         var result = make([]interface{},0)
 135
 136         c.assets_utxo.cond.L.Lock()
 137         defer c.assets_utxo.cond.L.Unlock()
 138
 139         if(len(c.assets_utxo.assets_amount)>0) {
 140                 result = append(result,c.assets_utxo.assets_amount)
 141         }
 142
 143         return result
 144 }
 145
 146 func (c *Chain) AddBlock(ctx context.Context, block *legacy.Block) error {
 147         currentBlock, _ := c.State()
 148         if err := c.ValidateBlock(block, currentBlock); err != nil {
 149                 return err
 150         }
 151
 152         newSnap, err := c.ApplyValidBlock(block)
 153         if err != nil {
 154                 return err
 155         }
 156
 157         if err := c.CommitAppliedBlock(ctx, block, newSnap); err != nil {
 158                 return err
 159         }
 160
 161         for _, tx := range block.Transactions {
 162                 c.txPool.RemoveTransaction(&tx.Tx.ID)
 163         }
 164         return nil
 165 }
 166
 167 func (c *Chain) queueSnapshot(ctx context.Context, height uint64, timestamp time.Time, s *state.Snapshot) {
 168         // Non-blockingly queue the snapshot for storage.
 169         ps := pendingSnapshot{height: height, snapshot: s}
 170         select {
 171         case c.pendingSnapshots <- ps:
 172                 c.lastQueuedSnapshot = timestamp
 173         default:
 174                 // Skip it; saving snapshots is taking longer than the snapshotting period.
 175                 log.Printf(ctx, "snapshot storage is taking too long; last queued at %s",
 176                         c.lastQueuedSnapshot)
 177         }
 178 }
 179
 180 func (c *Chain) setHeight(h uint64) {
 181         // We call setHeight from two places independently:
 182         // CommitBlock and the Postgres LISTEN goroutine.
 183         // This means we can get here twice for each block,
 184         // and any of them might be arbitrarily delayed,
 185         // which means h might be from the past.
 186         // Detect and discard these duplicate calls.
 187
 188         c.state.cond.L.Lock()
 189         defer c.state.cond.L.Unlock()
 190
 191         if h <= c.state.height {
 192                 return
 193         }
 194         c.state.height = h
 195         c.state.cond.Broadcast()
 196 }