Merge d6ed2f67a8 into 2688dab48c

.travis.yml

@@ -126,7 +126,7 @@ matrix:
 
     # This builder does the Android Maven and Azure uploads
     - os: linux
-      dist: precise # Needed for the android tools
+      dist: trusty
       addons:
         apt:
           packages:
@@ -146,7 +146,7 @@ matrix:
       git:
         submodules: false # avoid cloning ethereum/tests
       before_install:
-        - curl https://storage.googleapis.com/golang/go1.10.1.linux-amd64.tar.gz | tar -xz
+        - curl https://storage.googleapis.com/golang/go1.10.2.linux-amd64.tar.gz | tar -xz
         - export PATH=`pwd`/go/bin:$PATH
         - export GOROOT=`pwd`/go
         - export GOPATH=$HOME/go

VERSION

@@ -1 +1 @@
-1.8.8
+1.8.9

accounts/abi/argument.go

@@ -111,9 +111,14 @@ func (arguments Arguments) unpackTuple(v interface{}, marshalledValues []interfa
 	if err := requireUnpackKind(value, typ, kind, arguments); err != nil {
 		return err
 	}
-	// If the output interface is a struct, make sure names don't collide
+
+	// If the interface is a struct, get of abi->struct_field mapping
+
+	var abi2struct map[string]string
 	if kind == reflect.Struct {
-		if err := requireUniqueStructFieldNames(arguments); err != nil {
+		var err error
+		abi2struct, err = mapAbiToStructFields(arguments, value)
+		if err != nil {
 			return err
 		}
 	}
@@ -123,9 +128,10 @@ func (arguments Arguments) unpackTuple(v interface{}, marshalledValues []interfa
 
 		switch kind {
 		case reflect.Struct:
-			err := unpackStruct(value, reflectValue, arg)
-			if err != nil {
-				return err
+			if structField, ok := abi2struct[arg.Name]; ok {
+				if err := set(value.FieldByName(structField), reflectValue, arg); err != nil {
+					return err
+				}
 			}
 		case reflect.Slice, reflect.Array:
 			if value.Len() < i {
@@ -151,17 +157,22 @@ func (arguments Arguments) unpackAtomic(v interface{}, marshalledValues []interf
 	if len(marshalledValues) != 1 {
 		return fmt.Errorf("abi: wrong length, expected single value, got %d", len(marshalledValues))
 	}
+
 	elem := reflect.ValueOf(v).Elem()
 	kind := elem.Kind()
 	reflectValue := reflect.ValueOf(marshalledValues[0])
 
+	var abi2struct map[string]string
 	if kind == reflect.Struct {
-		//make sure names don't collide
-		if err := requireUniqueStructFieldNames(arguments); err != nil {
+		var err error
+		if abi2struct, err = mapAbiToStructFields(arguments, elem); err != nil {
 			return err
 		}
-
-		return unpackStruct(elem, reflectValue, arguments[0])
+		arg := arguments.NonIndexed()[0]
+		if structField, ok := abi2struct[arg.Name]; ok {
+			return set(elem.FieldByName(structField), reflectValue, arg)
+		}
+		return nil
 	}
 
 	return set(elem, reflectValue, arguments.NonIndexed()[0])
@@ -277,18 +288,3 @@ func capitalise(input string) string {
 	}
 	return strings.ToUpper(input[:1]) + input[1:]
 }
-
-//unpackStruct extracts each argument into its corresponding struct field
-func unpackStruct(value, reflectValue reflect.Value, arg Argument) error {
-	name := capitalise(arg.Name)
-	typ := value.Type()
-	for j := 0; j < typ.NumField(); j++ {
-		// TODO read tags: `abi:"fieldName"`
-		if typ.Field(j).Name == name {
-			if err := set(value.Field(j), reflectValue, arg); err != nil {
-				return err
-			}
-		}
-	}
-	return nil
-}

accounts/abi/bind/backends/simulated.go

@@ -454,7 +454,7 @@ func (fb *filterBackend) GetLogs(ctx context.Context, hash common.Hash) ([][]*ty
 	return logs, nil
 }
 
-func (fb *filterBackend) SubscribeTxPreEvent(ch chan<- core.TxPreEvent) event.Subscription {
+func (fb *filterBackend) SubscribeNewTxsEvent(ch chan<- core.NewTxsEvent) event.Subscription {
 	return event.NewSubscription(func(quit <-chan struct{}) error {
 		<-quit
 		return nil

accounts/abi/event_test.go

@@ -58,12 +58,28 @@ var jsonEventPledge = []byte(`{
   "type": "event"
 }`)
 
+var jsonEventMixedCase = []byte(`{
+	"anonymous": false,
+	"inputs": [{
+		"indexed": false, "name": "value", "type": "uint256"
+	  }, {
+		"indexed": false, "name": "_value", "type": "uint256"
+	  }, {
+		"indexed": false, "name": "Value", "type": "uint256"
+	}],
+	"name": "MixedCase",
+	"type": "event"
+  }`)
+
 // 1000000
 var transferData1 = "00000000000000000000000000000000000000000000000000000000000f4240"
 
 // "0x00Ce0d46d924CC8437c806721496599FC3FFA268", 2218516807680, "usd"
 var pledgeData1 = "00000000000000000000000000ce0d46d924cc8437c806721496599fc3ffa2680000000000000000000000000000000000000000000000000000020489e800007573640000000000000000000000000000000000000000000000000000000000"
 
+// 1000000,2218516807680,1000001
+var mixedCaseData1 = "00000000000000000000000000000000000000000000000000000000000f42400000000000000000000000000000000000000000000000000000020489e8000000000000000000000000000000000000000000000000000000000000000f4241"
+
 func TestEventId(t *testing.T) {
 	var table = []struct {
 		definition   string
@@ -121,6 +137,27 @@ func TestEventTupleUnpack(t *testing.T) {
 		Value *big.Int
 	}
 
+	type EventTransferWithTag struct {
+		// this is valid because `value` is not exportable,
+		// so value is only unmarshalled into `Value1`.
+		value  *big.Int
+		Value1 *big.Int `abi:"value"`
+	}
+
+	type BadEventTransferWithSameFieldAndTag struct {
+		Value  *big.Int
+		Value1 *big.Int `abi:"value"`
+	}
+
+	type BadEventTransferWithDuplicatedTag struct {
+		Value1 *big.Int `abi:"value"`
+		Value2 *big.Int `abi:"value"`
+	}
+
+	type BadEventTransferWithEmptyTag struct {
+		Value *big.Int `abi:""`
+	}
+
 	type EventPledge struct {
 		Who      common.Address
 		Wad      *big.Int
@@ -133,9 +170,16 @@ func TestEventTupleUnpack(t *testing.T) {
 		Currency [3]byte
 	}
 
+	type EventMixedCase struct {
+		Value1 *big.Int `abi:"value"`
+		Value2 *big.Int `abi:"_value"`
+		Value3 *big.Int `abi:"Value"`
+	}
+
 	bigint := new(big.Int)
 	bigintExpected := big.NewInt(1000000)
 	bigintExpected2 := big.NewInt(2218516807680)
+	bigintExpected3 := big.NewInt(1000001)
 	addr := common.HexToAddress("0x00Ce0d46d924CC8437c806721496599FC3FFA268")
 	var testCases = []struct {
 		data     string
@@ -158,6 +202,34 @@ func TestEventTupleUnpack(t *testing.T) {
 		jsonEventTransfer,
 		"",
 		"Can unpack ERC20 Transfer event into slice",
+	}, {
+		transferData1,
+		&EventTransferWithTag{},
+		&EventTransferWithTag{Value1: bigintExpected},
+		jsonEventTransfer,
+		"",
+		"Can unpack ERC20 Transfer event into structure with abi: tag",
+	}, {
+		transferData1,
+		&BadEventTransferWithDuplicatedTag{},
+		&BadEventTransferWithDuplicatedTag{},
+		jsonEventTransfer,
+		"struct: abi tag in 'Value2' already mapped",
+		"Can not unpack ERC20 Transfer event with duplicated abi tag",
+	}, {
+		transferData1,
+		&BadEventTransferWithSameFieldAndTag{},
+		&BadEventTransferWithSameFieldAndTag{},
+		jsonEventTransfer,
+		"abi: multiple variables maps to the same abi field 'value'",
+		"Can not unpack ERC20 Transfer event with a field and a tag mapping to the same abi variable",
+	}, {
+		transferData1,
+		&BadEventTransferWithEmptyTag{},
+		&BadEventTransferWithEmptyTag{},
+		jsonEventTransfer,
+		"struct: abi tag in 'Value' is empty",
+		"Can not unpack ERC20 Transfer event with an empty tag",
 	}, {
 		pledgeData1,
 		&EventPledge{},
@@ -216,6 +288,13 @@ func TestEventTupleUnpack(t *testing.T) {
 		jsonEventPledge,
 		"abi: cannot unmarshal tuple into map[string]interface {}",
 		"Can not unpack Pledge event into map",
+	}, {
+		mixedCaseData1,
+		&EventMixedCase{},
+		&EventMixedCase{Value1: bigintExpected, Value2: bigintExpected2, Value3: bigintExpected3},
+		jsonEventMixedCase,
+		"",
+		"Can unpack abi variables with mixed case",
 	}}
 
 	for _, tc := range testCases {
@@ -227,7 +306,7 @@ func TestEventTupleUnpack(t *testing.T) {
 				assert.Nil(err, "Should be able to unpack event data.")
 				assert.Equal(tc.expected, tc.dest, tc.name)
 			} else {
-				assert.EqualError(err, tc.error)
+				assert.EqualError(err, tc.error, tc.name)
 			}
 		})
 	}

accounts/abi/reflect.go

@@ -19,6 +19,7 @@ package abi
 import (
 	"fmt"
 	"reflect"
+	"strings"
 )
 
 // indirect recursively dereferences the value until it either gets the value
@@ -111,18 +112,101 @@ func requireUnpackKind(v reflect.Value, t reflect.Type, k reflect.Kind,
 	return nil
 }
 
-// requireUniqueStructFieldNames makes sure field names don't collide
-func requireUniqueStructFieldNames(args Arguments) error {
-	exists := make(map[string]bool)
-	for _, arg := range args {
-		field := capitalise(arg.Name)
-		if field == "" {
-			return fmt.Errorf("abi: purely underscored output cannot unpack to struct")
+// mapAbiToStringField maps abi to struct fields.
+// first round: for each Exportable field that contains a `abi:""` tag
+//   and this field name exists in the arguments, pair them together.
+// second round: for each argument field that has not been already linked,
+//   find what variable is expected to be mapped into, if it exists and has not been
+//   used, pair them.
+func mapAbiToStructFields(args Arguments, value reflect.Value) (map[string]string, error) {
+
+	typ := value.Type()
+
+	abi2struct := make(map[string]string)
+	struct2abi := make(map[string]string)
+
+	// first round ~~~
+	for i := 0; i < typ.NumField(); i++ {
+		structFieldName := typ.Field(i).Name
+
+		// skip private struct fields.
+		if structFieldName[:1] != strings.ToUpper(structFieldName[:1]) {
+			continue
 		}
-		if exists[field] {
-			return fmt.Errorf("abi: multiple outputs mapping to the same struct field '%s'", field)
+
+		// skip fields that have no abi:"" tag.
+		var ok bool
+		var tagName string
+		if tagName, ok = typ.Field(i).Tag.Lookup("abi"); !ok {
+			continue
 		}
-		exists[field] = true
+
+		// check if tag is empty.
+		if tagName == "" {
+			return nil, fmt.Errorf("struct: abi tag in '%s' is empty", structFieldName)
+		}
+
+		// check which argument field matches with the abi tag.
+		found := false
+		for _, abiField := range args.NonIndexed() {
+			if abiField.Name == tagName {
+				if abi2struct[abiField.Name] != "" {
+					return nil, fmt.Errorf("struct: abi tag in '%s' already mapped", structFieldName)
+				}
+				// pair them
+				abi2struct[abiField.Name] = structFieldName
+				struct2abi[structFieldName] = abiField.Name
+				found = true
+			}
+		}
+
+		// check if this tag has been mapped.
+		if !found {
+			return nil, fmt.Errorf("struct: abi tag '%s' defined but not found in abi", tagName)
+		}
+
 	}
-	return nil
+
+	// second round ~~~
+	for _, arg := range args {
+
+		abiFieldName := arg.Name
+		structFieldName := capitalise(abiFieldName)
+
+		if structFieldName == "" {
+			return nil, fmt.Errorf("abi: purely underscored output cannot unpack to struct")
+		}
+
+		// this abi has already been paired, skip it... unless there exists another, yet unassigned
+		// struct field with the same field name. If so, raise an error:
+		//    abi: [ { "name": "value" } ]
+		//    struct { Value  *big.Int , Value1 *big.Int `abi:"value"`}
+		if abi2struct[abiFieldName] != "" {
+			if abi2struct[abiFieldName] != structFieldName &&
+				struct2abi[structFieldName] == "" &&
+				value.FieldByName(structFieldName).IsValid() {
+				return nil, fmt.Errorf("abi: multiple variables maps to the same abi field '%s'", abiFieldName)
+			}
+			continue
+		}
+
+		// return an error if this struct field has already been paired.
+		if struct2abi[structFieldName] != "" {
+			return nil, fmt.Errorf("abi: multiple outputs mapping to the same struct field '%s'", structFieldName)
+		}
+
+		if value.FieldByName(structFieldName).IsValid() {
+			// pair them
+			abi2struct[abiFieldName] = structFieldName
+			struct2abi[structFieldName] = abiFieldName
+		} else {
+			// not paired, but annotate as used, to detect cases like
+			//   abi : [ { "name": "value" }, { "name": "_value" } ]
+			//   struct { Value *big.Int }
+			struct2abi[structFieldName] = abiFieldName
+		}
+
+	}
+
+	return abi2struct, nil
 }

appveyor.yml

@@ -23,8 +23,8 @@ environment:
 install:
   - git submodule update --init
   - rmdir C:\go /s /q
-  - appveyor DownloadFile https://storage.googleapis.com/golang/go1.10.1.windows-%GETH_ARCH%.zip
-  - 7z x go1.10.1.windows-%GETH_ARCH%.zip -y -oC:\ > NUL
+  - appveyor DownloadFile https://storage.googleapis.com/golang/go1.10.2.windows-%GETH_ARCH%.zip
+  - 7z x go1.10.2.windows-%GETH_ARCH%.zip -y -oC:\ > NUL
   - go version
   - gcc --version
 

consensus/clique/clique.go

@@ -383,7 +383,7 @@ func (c *Clique) snapshot(chain consensus.ChainReader, number uint64, hash commo
 		// If an on-disk checkpoint snapshot can be found, use that
 		if number%checkpointInterval == 0 {
 			if s, err := loadSnapshot(c.config, c.signatures, c.db, hash); err == nil {
-				log.Trace("Loaded voting snapshot form disk", "number", number, "hash", hash)
+				log.Trace("Loaded voting snapshot from disk", "number", number, "hash", hash)
 				snap = s
 				break
 			}

core/events.go

@@ -21,8 +21,8 @@ import (
 	"github.com/ethereum/go-ethereum/core/types"
 )
 
-// TxPreEvent is posted when a transaction enters the transaction pool.
-type TxPreEvent struct{ Tx *types.Transaction }
+// NewTxsEvent is posted when a batch of transactions enter the transaction pool.
+type NewTxsEvent struct{ Txs []*types.Transaction }
 
 // PendingLogsEvent is posted pre mining and notifies of pending logs.
 type PendingLogsEvent struct {
@@ -35,9 +35,6 @@ type PendingStateEvent struct{}
 // NewMinedBlockEvent is posted when a block has been imported.
 type NewMinedBlockEvent struct{ Block *types.Block }
 
-// RemovedTransactionEvent is posted when a reorg happens
-type RemovedTransactionEvent struct{ Txs types.Transactions }
-
 // RemovedLogsEvent is posted when a reorg happens
 type RemovedLogsEvent struct{ Logs []*types.Log }
 

core/state/statedb.go

@@ -358,7 +358,7 @@ func (self *StateDB) deleteStateObject(stateObject *stateObject) {
 	self.setError(self.trie.TryDelete(addr[:]))
 }
 
-// Retrieve a state object given my the address. Returns nil if not found.
+// Retrieve a state object given by the address. Returns nil if not found.
 func (self *StateDB) getStateObject(addr common.Address) (stateObject *stateObject) {
 	// Prefer 'live' objects.
 	if obj := self.stateObjects[addr]; obj != nil {

core/tx_journal.go

@@ -56,7 +56,7 @@ func newTxJournal(path string) *txJournal {
 
 // load parses a transaction journal dump from disk, loading its contents into
 // the specified pool.
-func (journal *txJournal) load(add func(*types.Transaction) error) error {
+func (journal *txJournal) load(add func([]*types.Transaction) []error) error {
 	// Skip the parsing if the journal file doens't exist at all
 	if _, err := os.Stat(journal.path); os.IsNotExist(err) {
 		return nil
@@ -76,7 +76,21 @@ func (journal *txJournal) load(add func(*types.Transaction) error) error {
 	stream := rlp.NewStream(input, 0)
 	total, dropped := 0, 0
 
-	var failure error
+	// Create a method to load a limited batch of transactions and bump the
+	// appropriate progress counters. Then use this method to load all the
+	// journalled transactions in small-ish batches.
+	loadBatch := func(txs types.Transactions) {
+		for _, err := range add(txs) {
+			if err != nil {
+				log.Debug("Failed to add journaled transaction", "err", err)
+				dropped++
+			}
+		}
+	}
+	var (
+		failure error
+		batch   types.Transactions
+	)
 	for {
 		// Parse the next transaction and terminate on error
 		tx := new(types.Transaction)
@@ -84,14 +98,17 @@ func (journal *txJournal) load(add func(*types.Transaction) error) error {
 			if err != io.EOF {
 				failure = err
 			}
+			if batch.Len() > 0 {
+				loadBatch(batch)
+			}
 			break
 		}
-		// Import the transaction and bump the appropriate progress counters
+		// New transaction parsed, queue up for later, import if threnshold is reached
 		total++
-		if err = add(tx); err != nil {
-			log.Debug("Failed to add journaled transaction", "err", err)
-			dropped++
-			continue
+
+		if batch = append(batch, tx); batch.Len() > 1024 {
+			loadBatch(batch)
+			batch = batch[:0]
 		}
 	}
 	log.Info("Loaded local transaction journal", "transactions", total, "dropped", dropped)

core/tx_list.go

@@ -397,13 +397,13 @@ func (h *priceHeap) Pop() interface{} {
 // txPricedList is a price-sorted heap to allow operating on transactions pool
 // contents in a price-incrementing way.
 type txPricedList struct {
-	all    *map[common.Hash]*types.Transaction // Pointer to the map of all transactions
-	items  *priceHeap                          // Heap of prices of all the stored transactions
-	stales int                                 // Number of stale price points to (re-heap trigger)
+	all    *txLookup  // Pointer to the map of all transactions
+	items  *priceHeap // Heap of prices of all the stored transactions
+	stales int        // Number of stale price points to (re-heap trigger)
 }
 
 // newTxPricedList creates a new price-sorted transaction heap.
-func newTxPricedList(all *map[common.Hash]*types.Transaction) *txPricedList {
+func newTxPricedList(all *txLookup) *txPricedList {
 	return &txPricedList{
 		all:   all,
 		items: new(priceHeap),
@@ -425,12 +425,13 @@ func (l *txPricedList) Removed() {
 		return
 	}
 	// Seems we've reached a critical number of stale transactions, reheap
-	reheap := make(priceHeap, 0, len(*l.all))
+	reheap := make(priceHeap, 0, l.all.Count())
 
 	l.stales, l.items = 0, &reheap
-	for _, tx := range *l.all {
+	l.all.Range(func(hash common.Hash, tx *types.Transaction) bool {
 		*l.items = append(*l.items, tx)
-	}
+		return true
+	})
 	heap.Init(l.items)
 }
 
@@ -443,7 +444,7 @@ func (l *txPricedList) Cap(threshold *big.Int, local *accountSet) types.Transact
 	for len(*l.items) > 0 {
 		// Discard stale transactions if found during cleanup
 		tx := heap.Pop(l.items).(*types.Transaction)
-		if _, ok := (*l.all)[tx.Hash()]; !ok {
+		if l.all.Get(tx.Hash()) == nil {
 			l.stales--
 			continue
 		}
@@ -475,7 +476,7 @@ func (l *txPricedList) Underpriced(tx *types.Transaction, local *accountSet) boo
 	// Discard stale price points if found at the heap start
 	for len(*l.items) > 0 {
 		head := []*types.Transaction(*l.items)[0]
-		if _, ok := (*l.all)[head.Hash()]; !ok {
+		if l.all.Get(head.Hash()) == nil {
 			l.stales--
 			heap.Pop(l.items)
 			continue
@@ -500,7 +501,7 @@ func (l *txPricedList) Discard(count int, local *accountSet) types.Transactions
 	for len(*l.items) > 0 && count > 0 {
 		// Discard stale transactions if found during cleanup
 		tx := heap.Pop(l.items).(*types.Transaction)
-		if _, ok := (*l.all)[tx.Hash()]; !ok {
+		if l.all.Get(tx.Hash()) == nil {
 			l.stales--
 			continue
 		}

core/tx_pool.go

@@ -200,11 +200,11 @@ type TxPool struct {
 	locals  *accountSet // Set of local transaction to exempt from eviction rules
 	journal *txJournal  // Journal of local transaction to back up to disk
 
-	pending map[common.Address]*txList         // All currently processable transactions
-	queue   map[common.Address]*txList         // Queued but non-processable transactions
-	beats   map[common.Address]time.Time       // Last heartbeat from each known account
-	all     map[common.Hash]*types.Transaction // All transactions to allow lookups
-	priced  *txPricedList                      // All transactions sorted by price
+	pending map[common.Address]*txList   // All currently processable transactions
+	queue   map[common.Address]*txList   // Queued but non-processable transactions
+	beats   map[common.Address]time.Time // Last heartbeat from each known account
+	all     *txLookup                    // All transactions to allow lookups
+	priced  *txPricedList                // All transactions sorted by price
 
 	wg sync.WaitGroup // for shutdown sync
 
@@ -226,19 +226,19 @@ func NewTxPool(config TxPoolConfig, chainconfig *params.ChainConfig, chain block
 		pending:     make(map[common.Address]*txList),
 		queue:       make(map[common.Address]*txList),
 		beats:       make(map[common.Address]time.Time),
-		all:         make(map[common.Hash]*types.Transaction),
+		all:         newTxLookup(),
 		chainHeadCh: make(chan ChainHeadEvent, chainHeadChanSize),
 		gasPrice:    new(big.Int).SetUint64(config.PriceLimit),
 	}
 	pool.locals = newAccountSet(pool.signer)
-	pool.priced = newTxPricedList(&pool.all)
+	pool.priced = newTxPricedList(pool.all)
 	pool.reset(nil, chain.CurrentBlock().Header())
 
 	// If local transactions and journaling is enabled, load from disk
 	if !config.NoLocals && config.Journal != "" {
 		pool.journal = newTxJournal(config.Journal)
 
-		if err := pool.journal.load(pool.AddLocal); err != nil {
+		if err := pool.journal.load(pool.AddLocals); err != nil {
 			log.Warn("Failed to load transaction journal", "err", err)
 		}
 		if err := pool.journal.rotate(pool.local()); err != nil {
@@ -444,9 +444,9 @@ func (pool *TxPool) Stop() {
 	log.Info("Transaction pool stopped")
 }
 
-// SubscribeTxPreEvent registers a subscription of TxPreEvent and
+// SubscribeNewTxsEvent registers a subscription of NewTxsEvent and
 // starts sending event to the given channel.
-func (pool *TxPool) SubscribeTxPreEvent(ch chan<- TxPreEvent) event.Subscription {
+func (pool *TxPool) SubscribeNewTxsEvent(ch chan<- NewTxsEvent) event.Subscription {
 	return pool.scope.Track(pool.txFeed.Subscribe(ch))
 }
 
@@ -605,7 +605,7 @@ func (pool *TxPool) validateTx(tx *types.Transaction, local bool) error {
 func (pool *TxPool) add(tx *types.Transaction, local bool) (bool, error) {
 	// If the transaction is already known, discard it
 	hash := tx.Hash()
-	if pool.all[hash] != nil {
+	if pool.all.Get(hash) != nil {
 		log.Trace("Discarding already known transaction", "hash", hash)
 		return false, fmt.Errorf("known transaction: %x", hash)
 	}
@@ -616,7 +616,7 @@ func (pool *TxPool) add(tx *types.Transaction, local bool) (bool, error) {
 		return false, err
 	}
 	// If the transaction pool is full, discard underpriced transactions
-	if uint64(len(pool.all)) >= pool.config.GlobalSlots+pool.config.GlobalQueue {
+	if uint64(pool.all.Count()) >= pool.config.GlobalSlots+pool.config.GlobalQueue {
 		// If the new transaction is underpriced, don't accept it
 		if !local && pool.priced.Underpriced(tx, pool.locals) {
 			log.Trace("Discarding underpriced transaction", "hash", hash, "price", tx.GasPrice())
@@ -624,7 +624,7 @@ func (pool *TxPool) add(tx *types.Transaction, local bool) (bool, error) {
 			return false, ErrUnderpriced
 		}
 		// New transaction is better than our worse ones, make room for it
-		drop := pool.priced.Discard(len(pool.all)-int(pool.config.GlobalSlots+pool.config.GlobalQueue-1), pool.locals)
+		drop := pool.priced.Discard(pool.all.Count()-int(pool.config.GlobalSlots+pool.config.GlobalQueue-1), pool.locals)
 		for _, tx := range drop {
 			log.Trace("Discarding freshly underpriced transaction", "hash", tx.Hash(), "price", tx.GasPrice())
 			underpricedTxCounter.Inc(1)
@@ -642,18 +642,18 @@ func (pool *TxPool) add(tx *types.Transaction, local bool) (bool, error) {
 		}
 		// New transaction is better, replace old one
 		if old != nil {
-			delete(pool.all, old.Hash())
+			pool.all.Remove(old.Hash())
 			pool.priced.Removed()
 			pendingReplaceCounter.Inc(1)
 		}
-		pool.all[tx.Hash()] = tx
+		pool.all.Add(tx)
 		pool.priced.Put(tx)
 		pool.journalTx(from, tx)
 
 		log.Trace("Pooled new executable transaction", "hash", hash, "from", from, "to", tx.To())
 
 		// We've directly injected a replacement transaction, notify subsystems
-		go pool.txFeed.Send(TxPreEvent{tx})
+		go pool.txFeed.Send(NewTxsEvent{types.Transactions{tx}})
 
 		return old != nil, nil
 	}
@@ -689,12 +689,12 @@ func (pool *TxPool) enqueueTx(hash common.Hash, tx *types.Transaction) (bool, er
 	}
 	// Discard any previous transaction and mark this
 	if old != nil {
-		delete(pool.all, old.Hash())
+		pool.all.Remove(old.Hash())
 		pool.priced.Removed()
 		queuedReplaceCounter.Inc(1)
 	}
-	if pool.all[hash] == nil {
-		pool.all[hash] = tx
+	if pool.all.Get(hash) == nil {
+		pool.all.Add(tx)
 		pool.priced.Put(tx)
 	}
 	return old != nil, nil
@@ -712,10 +712,11 @@ func (pool *TxPool) journalTx(from common.Address, tx *types.Transaction) {
 	}
 }
 
-// promoteTx adds a transaction to the pending (processable) list of transactions.
+// promoteTx adds a transaction to the pending (processable) list of transactions
+// and returns whether it was inserted or an older was better.
 //
 // Note, this method assumes the pool lock is held!
-func (pool *TxPool) promoteTx(addr common.Address, hash common.Hash, tx *types.Transaction) {
+func (pool *TxPool) promoteTx(addr common.Address, hash common.Hash, tx *types.Transaction) bool {
 	// Try to insert the transaction into the pending queue
 	if pool.pending[addr] == nil {
 		pool.pending[addr] = newTxList(true)
@@ -725,29 +726,29 @@ func (pool *TxPool) promoteTx(addr common.Address, hash common.Hash, tx *types.T
 	inserted, old := list.Add(tx, pool.config.PriceBump)
 	if !inserted {
 		// An older transaction was better, discard this
-		delete(pool.all, hash)
+		pool.all.Remove(hash)
 		pool.priced.Removed()
 
 		pendingDiscardCounter.Inc(1)
-		return
+		return false
 	}
 	// Otherwise discard any previous transaction and mark this
 	if old != nil {
-		delete(pool.all, old.Hash())
+		pool.all.Remove(old.Hash())
 		pool.priced.Removed()
 
 		pendingReplaceCounter.Inc(1)
 	}
 	// Failsafe to work around direct pending inserts (tests)
-	if pool.all[hash] == nil {
-		pool.all[hash] = tx
+	if pool.all.Get(hash) == nil {
+		pool.all.Add(tx)
 		pool.priced.Put(tx)
 	}
 	// Set the potentially new pending nonce and notify any subsystems of the new tx
 	pool.beats[addr] = time.Now()
 	pool.pendingState.SetNonce(addr, tx.Nonce()+1)
 
-	go pool.txFeed.Send(TxPreEvent{tx})
+	return true
 }
 
 // AddLocal enqueues a single transaction into the pool if it is valid, marking
@@ -839,7 +840,7 @@ func (pool *TxPool) Status(hashes []common.Hash) []TxStatus {
 
 	status := make([]TxStatus, len(hashes))
 	for i, hash := range hashes {
-		if tx := pool.all[hash]; tx != nil {
+		if tx := pool.all.Get(hash); tx != nil {
 			from, _ := types.Sender(pool.signer, tx) // already validated
 			if pool.pending[from] != nil && pool.pending[from].txs.items[tx.Nonce()] != nil {
 				status[i] = TxStatusPending
@@ -854,24 +855,21 @@ func (pool *TxPool) Status(hashes []common.Hash) []TxStatus {
 // Get returns a transaction if it is contained in the pool
 // and nil otherwise.
 func (pool *TxPool) Get(hash common.Hash) *types.Transaction {
-	pool.mu.RLock()
-	defer pool.mu.RUnlock()
-
-	return pool.all[hash]
+	return pool.all.Get(hash)
 }
 
 // removeTx removes a single transaction from the queue, moving all subsequent
 // transactions back to the future queue.
 func (pool *TxPool) removeTx(hash common.Hash, outofbound bool) {
 	// Fetch the transaction we wish to delete
-	tx, ok := pool.all[hash]
-	if !ok {
+	tx := pool.all.Get(hash)
+	if tx == nil {
 		return
 	}
 	addr, _ := types.Sender(pool.signer, tx) // already validated during insertion
 
 	// Remove it from the list of known transactions
-	delete(pool.all, hash)
+	pool.all.Remove(hash)
 	if outofbound {
 		pool.priced.Removed()
 	}
@@ -907,6 +905,9 @@ func (pool *TxPool) removeTx(hash common.Hash, outofbound bool) {
 // future queue to the set of pending transactions. During this process, all
 // invalidated transactions (low nonce, low balance) are deleted.
 func (pool *TxPool) promoteExecutables(accounts []common.Address) {
+	// Track the promoted transactions to broadcast them at once
+	var promoted []*types.Transaction
+
 	// Gather all the accounts potentially needing updates
 	if accounts == nil {
 		accounts = make([]common.Address, 0, len(pool.queue))
@@ -924,7 +925,7 @@ func (pool *TxPool) promoteExecutables(accounts []common.Address) {
 		for _, tx := range list.Forward(pool.currentState.GetNonce(addr)) {
 			hash := tx.Hash()
 			log.Trace("Removed old queued transaction", "hash", hash)
-			delete(pool.all, hash)
+			pool.all.Remove(hash)
 			pool.priced.Removed()
 		}
 		// Drop all transactions that are too costly (low balance or out of gas)
@@ -932,21 +933,23 @@ func (pool *TxPool) promoteExecutables(accounts []common.Address) {
 		for _, tx := range drops {
 			hash := tx.Hash()
 			log.Trace("Removed unpayable queued transaction", "hash", hash)
-			delete(pool.all, hash)
+			pool.all.Remove(hash)
 			pool.priced.Removed()
 			queuedNofundsCounter.Inc(1)
 		}
 		// Gather all executable transactions and promote them
 		for _, tx := range list.Ready(pool.pendingState.GetNonce(addr)) {
 			hash := tx.Hash()
-			log.Trace("Promoting queued transaction", "hash", hash)
-			pool.promoteTx(addr, hash, tx)
+			if pool.promoteTx(addr, hash, tx) {
+				log.Trace("Promoting queued transaction", "hash", hash)
+				promoted = append(promoted, tx)
+			}
 		}
 		// Drop all transactions over the allowed limit
 		if !pool.locals.contains(addr) {
 			for _, tx := range list.Cap(int(pool.config.AccountQueue)) {
 				hash := tx.Hash()
-				delete(pool.all, hash)
+				pool.all.Remove(hash)
 				pool.priced.Removed()
 				queuedRateLimitCounter.Inc(1)
 				log.Trace("Removed cap-exceeding queued transaction", "hash", hash)
@@ -957,6 +960,10 @@ func (pool *TxPool) promoteExecutables(accounts []common.Address) {
 			delete(pool.queue, addr)
 		}
 	}
+	// Notify subsystem for new promoted transactions.
+	if len(promoted) > 0 {
+		pool.txFeed.Send(NewTxsEvent{promoted})
+	}
 	// If the pending limit is overflown, start equalizing allowances
 	pending := uint64(0)
 	for _, list := range pool.pending {
@@ -991,7 +998,7 @@ func (pool *TxPool) promoteExecutables(accounts []common.Address) {
 						for _, tx := range list.Cap(list.Len() - 1) {
 							// Drop the transaction from the global pools too
 							hash := tx.Hash()
-							delete(pool.all, hash)
+							pool.all.Remove(hash)
 							pool.priced.Removed()
 
 							// Update the account nonce to the dropped transaction
@@ -1013,7 +1020,7 @@ func (pool *TxPool) promoteExecutables(accounts []common.Address) {
 					for _, tx := range list.Cap(list.Len() - 1) {
 						// Drop the transaction from the global pools too
 						hash := tx.Hash()
-						delete(pool.all, hash)
+						pool.all.Remove(hash)
 						pool.priced.Removed()
 
 						// Update the account nonce to the dropped transaction
@@ -1082,7 +1089,7 @@ func (pool *TxPool) demoteUnexecutables() {
 		for _, tx := range list.Forward(nonce) {
 			hash := tx.Hash()
 			log.Trace("Removed old pending transaction", "hash", hash)
-			delete(pool.all, hash)
+			pool.all.Remove(hash)
 			pool.priced.Removed()
 		}
 		// Drop all transactions that are too costly (low balance or out of gas), and queue any invalids back for later
@@ -1090,7 +1097,7 @@ func (pool *TxPool) demoteUnexecutables() {
 		for _, tx := range drops {
 			hash := tx.Hash()
 			log.Trace("Removed unpayable pending transaction", "hash", hash)
-			delete(pool.all, hash)
+			pool.all.Remove(hash)
 			pool.priced.Removed()
 			pendingNofundsCounter.Inc(1)
 		}
@@ -1162,3 +1169,68 @@ func (as *accountSet) containsTx(tx *types.Transaction) bool {
 func (as *accountSet) add(addr common.Address) {
 	as.accounts[addr] = struct{}{}
 }
+
+// txLookup is used internally by TxPool to track transactions while allowing lookup without
+// mutex contention.
+//
+// Note, although this type is properly protected against concurrent access, it
+// is **not** a type that should ever be mutated or even exposed outside of the
+// transaction pool, since its internal state is tightly coupled with the pools
+// internal mechanisms. The sole purpose of the type is to permit out-of-bound
+// peeking into the pool in TxPool.Get without having to acquire the widely scoped
+// TxPool.mu mutex.
+type txLookup struct {
+	all  map[common.Hash]*types.Transaction
+	lock sync.RWMutex
+}
+
+// newTxLookup returns a new txLookup structure.
+func newTxLookup() *txLookup {
+	return &txLookup{
+		all: make(map[common.Hash]*types.Transaction),
+	}
+}
+
+// Range calls f on each key and value present in the map.
+func (t *txLookup) Range(f func(hash common.Hash, tx *types.Transaction) bool) {
+	t.lock.RLock()
+	defer t.lock.RUnlock()
+
+	for key, value := range t.all {
+		if !f(key, value) {
+			break
+		}
+	}
+}
+
+// Get returns a transaction if it exists in the lookup, or nil if not found.
+func (t *txLookup) Get(hash common.Hash) *types.Transaction {
+	t.lock.RLock()
+	defer t.lock.RUnlock()
+
+	return t.all[hash]
+}
+
+// Count returns the current number of items in the lookup.
+func (t *txLookup) Count() int {
+	t.lock.RLock()
+	defer t.lock.RUnlock()
+
+	return len(t.all)
+}
+
+// Add adds a transaction to the lookup.
+func (t *txLookup) Add(tx *types.Transaction) {
+	t.lock.Lock()
+	defer t.lock.Unlock()
+
+	t.all[tx.Hash()] = tx
+}
+
+// Remove removes a transaction from the lookup.
+func (t *txLookup) Remove(hash common.Hash) {
+	t.lock.Lock()
+	defer t.lock.Unlock()
+
+	delete(t.all, hash)
+}

core/tx_pool_test.go

@@ -94,7 +94,7 @@ func validateTxPoolInternals(pool *TxPool) error {
 
 	// Ensure the total transaction set is consistent with pending + queued
 	pending, queued := pool.stats()
-	if total := len(pool.all); total != pending+queued {
+	if total := pool.all.Count(); total != pending+queued {
 		return fmt.Errorf("total transaction count %d != %d pending + %d queued", total, pending, queued)
 	}
 	if priced := pool.priced.items.Len() - pool.priced.stales; priced != pending+queued {
@@ -118,21 +118,27 @@ func validateTxPoolInternals(pool *TxPool) error {
 
 // validateEvents checks that the correct number of transaction addition events
 // were fired on the pool's event feed.
-func validateEvents(events chan TxPreEvent, count int) error {
-	for i := 0; i < count; i++ {
+func validateEvents(events chan NewTxsEvent, count int) error {
+	var received []*types.Transaction
+
+	for len(received) < count {
 		select {
-		case <-events:
+		case ev := <-events:
+			received = append(received, ev.Txs...)
 		case <-time.After(time.Second):
-			return fmt.Errorf("event #%d not fired", i)
+			return fmt.Errorf("event #%d not fired", received)
 		}
 	}
+	if len(received) > count {
+		return fmt.Errorf("more than %d events fired: %v", count, received[count:])
+	}
 	select {
-	case tx := <-events:
-		return fmt.Errorf("more than %d events fired: %v", count, tx.Tx)
+	case ev := <-events:
+		return fmt.Errorf("more than %d events fired: %v", count, ev.Txs)
 
 	case <-time.After(50 * time.Millisecond):
 		// This branch should be "default", but it's a data race between goroutines,
-		// reading the event channel and pushng into it, so better wait a bit ensuring
+		// reading the event channel and pushing into it, so better wait a bit ensuring
 		// really nothing gets injected.
 	}
 	return nil
@@ -395,8 +401,8 @@ func TestTransactionDoubleNonce(t *testing.T) {
 		t.Errorf("transaction mismatch: have %x, want %x", tx.Hash(), tx2.Hash())
 	}
 	// Ensure the total transaction count is correct
-	if len(pool.all) != 1 {
-		t.Error("expected 1 total transactions, got", len(pool.all))
+	if pool.all.Count() != 1 {
+		t.Error("expected 1 total transactions, got", pool.all.Count())
 	}
 }
 
@@ -418,8 +424,8 @@ func TestTransactionMissingNonce(t *testing.T) {
 	if pool.queue[addr].Len() != 1 {
 		t.Error("expected 1 queued transaction, got", pool.queue[addr].Len())
 	}
-	if len(pool.all) != 1 {
-		t.Error("expected 1 total transactions, got", len(pool.all))
+	if pool.all.Count() != 1 {
+		t.Error("expected 1 total transactions, got", pool.all.Count())
 	}
 }
 
@@ -482,8 +488,8 @@ func TestTransactionDropping(t *testing.T) {
 	if pool.queue[account].Len() != 3 {
 		t.Errorf("queued transaction mismatch: have %d, want %d", pool.queue[account].Len(), 3)
 	}
-	if len(pool.all) != 6 {
-		t.Errorf("total transaction mismatch: have %d, want %d", len(pool.all), 6)
+	if pool.all.Count() != 6 {
+		t.Errorf("total transaction mismatch: have %d, want %d", pool.all.Count(), 6)
 	}
 	pool.lockedReset(nil, nil)
 	if pool.pending[account].Len() != 3 {
@@ -492,8 +498,8 @@ func TestTransactionDropping(t *testing.T) {
 	if pool.queue[account].Len() != 3 {
 		t.Errorf("queued transaction mismatch: have %d, want %d", pool.queue[account].Len(), 3)
 	}
-	if len(pool.all) != 6 {
-		t.Errorf("total transaction mismatch: have %d, want %d", len(pool.all), 6)
+	if pool.all.Count() != 6 {
+		t.Errorf("total transaction mismatch: have %d, want %d", pool.all.Count(), 6)
 	}
 	// Reduce the balance of the account, and check that invalidated transactions are dropped
 	pool.currentState.AddBalance(account, big.NewInt(-650))
@@ -517,8 +523,8 @@ func TestTransactionDropping(t *testing.T) {
 	if _, ok := pool.queue[account].txs.items[tx12.Nonce()]; ok {
 		t.Errorf("out-of-fund queued transaction present: %v", tx11)
 	}
-	if len(pool.all) != 4 {
-		t.Errorf("total transaction mismatch: have %d, want %d", len(pool.all), 4)
+	if pool.all.Count() != 4 {
+		t.Errorf("total transaction mismatch: have %d, want %d", pool.all.Count(), 4)
 	}
 	// Reduce the block gas limit, check that invalidated transactions are dropped
 	pool.chain.(*testBlockChain).gasLimit = 100
@@ -536,8 +542,8 @@ func TestTransactionDropping(t *testing.T) {
 	if _, ok := pool.queue[account].txs.items[tx11.Nonce()]; ok {
 		t.Errorf("over-gased queued transaction present: %v", tx11)
 	}
-	if len(pool.all) != 2 {
-		t.Errorf("total transaction mismatch: have %d, want %d", len(pool.all), 2)
+	if pool.all.Count() != 2 {
+		t.Errorf("total transaction mismatch: have %d, want %d", pool.all.Count(), 2)
 	}
 }
 
@@ -590,8 +596,8 @@ func TestTransactionPostponing(t *testing.T) {
 	if len(pool.queue) != 0 {
 		t.Errorf("queued accounts mismatch: have %d, want %d", len(pool.queue), 0)
 	}
-	if len(pool.all) != len(txs) {
-		t.Errorf("total transaction mismatch: have %d, want %d", len(pool.all), len(txs))
+	if pool.all.Count() != len(txs) {
+		t.Errorf("total transaction mismatch: have %d, want %d", pool.all.Count(), len(txs))
 	}
 	pool.lockedReset(nil, nil)
 	if pending := pool.pending[accs[0]].Len() + pool.pending[accs[1]].Len(); pending != len(txs) {
@@ -600,8 +606,8 @@ func TestTransactionPostponing(t *testing.T) {
 	if len(pool.queue) != 0 {
 		t.Errorf("queued accounts mismatch: have %d, want %d", len(pool.queue), 0)
 	}
-	if len(pool.all) != len(txs) {
-		t.Errorf("total transaction mismatch: have %d, want %d", len(pool.all), len(txs))
+	if pool.all.Count() != len(txs) {
+		t.Errorf("total transaction mismatch: have %d, want %d", pool.all.Count(), len(txs))
 	}
 	// Reduce the balance of the account, and check that transactions are reorganised
 	for _, addr := range accs {
@@ -650,8 +656,8 @@ func TestTransactionPostponing(t *testing.T) {
 			}
 		}
 	}
-	if len(pool.all) != len(txs)/2 {
-		t.Errorf("total transaction mismatch: have %d, want %d", len(pool.all), len(txs)/2)
+	if pool.all.Count() != len(txs)/2 {
+		t.Errorf("total transaction mismatch: have %d, want %d", pool.all.Count(), len(txs)/2)
 	}
 }
 
@@ -669,7 +675,7 @@ func TestTransactionGapFilling(t *testing.T) {
 	pool.currentState.AddBalance(account, big.NewInt(1000000))
 
 	// Keep track of transaction events to ensure all executables get announced
-	events := make(chan TxPreEvent, testTxPoolConfig.AccountQueue+5)
+	events := make(chan NewTxsEvent, testTxPoolConfig.AccountQueue+5)
 	sub := pool.txFeed.Subscribe(events)
 	defer sub.Unsubscribe()
 
@@ -742,8 +748,8 @@ func TestTransactionQueueAccountLimiting(t *testing.T) {
 			}
 		}
 	}
-	if len(pool.all) != int(testTxPoolConfig.AccountQueue) {
-		t.Errorf("total transaction mismatch: have %d, want %d", len(pool.all), testTxPoolConfig.AccountQueue)
+	if pool.all.Count() != int(testTxPoolConfig.AccountQueue) {
+		t.Errorf("total transaction mismatch: have %d, want %d", pool.all.Count(), testTxPoolConfig.AccountQueue)
 	}
 }
 
@@ -920,7 +926,7 @@ func TestTransactionPendingLimiting(t *testing.T) {
 	pool.currentState.AddBalance(account, big.NewInt(1000000))
 
 	// Keep track of transaction events to ensure all executables get announced
-	events := make(chan TxPreEvent, testTxPoolConfig.AccountQueue+5)
+	events := make(chan NewTxsEvent, testTxPoolConfig.AccountQueue+5)
 	sub := pool.txFeed.Subscribe(events)
 	defer sub.Unsubscribe()
 
@@ -936,8 +942,8 @@ func TestTransactionPendingLimiting(t *testing.T) {
 			t.Errorf("tx %d: queue size mismatch: have %d, want %d", i, pool.queue[account].Len(), 0)
 		}
 	}
-	if len(pool.all) != int(testTxPoolConfig.AccountQueue+5) {
-		t.Errorf("total transaction mismatch: have %d, want %d", len(pool.all), testTxPoolConfig.AccountQueue+5)
+	if pool.all.Count() != int(testTxPoolConfig.AccountQueue+5) {
+		t.Errorf("total transaction mismatch: have %d, want %d", pool.all.Count(), testTxPoolConfig.AccountQueue+5)
 	}
 	if err := validateEvents(events, int(testTxPoolConfig.AccountQueue+5)); err != nil {
 		t.Fatalf("event firing failed: %v", err)
@@ -987,8 +993,8 @@ func testTransactionLimitingEquivalency(t *testing.T, origin uint64) {
 	if len(pool1.queue) != len(pool2.queue) {
 		t.Errorf("queued transaction count mismatch: one-by-one algo: %d, batch algo: %d", len(pool1.queue), len(pool2.queue))
 	}
-	if len(pool1.all) != len(pool2.all) {
-		t.Errorf("total transaction count mismatch: one-by-one algo %d, batch algo %d", len(pool1.all), len(pool2.all))
+	if pool1.all.Count() != pool2.all.Count() {
+		t.Errorf("total transaction count mismatch: one-by-one algo %d, batch algo %d", pool1.all.Count(), pool2.all.Count())
 	}
 	if err := validateTxPoolInternals(pool1); err != nil {
 		t.Errorf("pool 1 internal state corrupted: %v", err)
@@ -1140,7 +1146,7 @@ func TestTransactionPoolRepricing(t *testing.T) {
 	defer pool.Stop()
 
 	// Keep track of transaction events to ensure all executables get announced
-	events := make(chan TxPreEvent, 32)
+	events := make(chan NewTxsEvent, 32)
 	sub := pool.txFeed.Subscribe(events)
 	defer sub.Unsubscribe()
 
@@ -1327,7 +1333,7 @@ func TestTransactionPoolUnderpricing(t *testing.T) {
 	defer pool.Stop()
 
 	// Keep track of transaction events to ensure all executables get announced
-	events := make(chan TxPreEvent, 32)
+	events := make(chan NewTxsEvent, 32)
 	sub := pool.txFeed.Subscribe(events)
 	defer sub.Unsubscribe()
 
@@ -1433,7 +1439,7 @@ func TestTransactionPoolStableUnderpricing(t *testing.T) {
 	defer pool.Stop()
 
 	// Keep track of transaction events to ensure all executables get announced
-	events := make(chan TxPreEvent, 32)
+	events := make(chan NewTxsEvent, 32)
 	sub := pool.txFeed.Subscribe(events)
 	defer sub.Unsubscribe()
 
@@ -1495,7 +1501,7 @@ func TestTransactionReplacement(t *testing.T) {
 	defer pool.Stop()
 
 	// Keep track of transaction events to ensure all executables get announced
-	events := make(chan TxPreEvent, 32)
+	events := make(chan NewTxsEvent, 32)
 	sub := pool.txFeed.Subscribe(events)
 	defer sub.Unsubscribe()
 

core/types/gen_receipt_json.go

@@ -12,10 +12,11 @@ import (
 
 var _ = (*receiptMarshaling)(nil)
 
+// MarshalJSON marshals as JSON.
 func (r Receipt) MarshalJSON() ([]byte, error) {
 	type Receipt struct {
 		PostState         hexutil.Bytes  `json:"root"`
-		Status            hexutil.Uint   `json:"status"`
+		Status            hexutil.Uint64 `json:"status"`
 		CumulativeGasUsed hexutil.Uint64 `json:"cumulativeGasUsed" gencodec:"required"`
 		Bloom             Bloom          `json:"logsBloom"         gencodec:"required"`
 		Logs              []*Log         `json:"logs"              gencodec:"required"`
@@ -25,7 +26,7 @@ func (r Receipt) MarshalJSON() ([]byte, error) {
 	}
 	var enc Receipt
 	enc.PostState = r.PostState
-	enc.Status = hexutil.Uint(r.Status)
+	enc.Status = hexutil.Uint64(r.Status)
 	enc.CumulativeGasUsed = hexutil.Uint64(r.CumulativeGasUsed)
 	enc.Bloom = r.Bloom
 	enc.Logs = r.Logs
@@ -35,10 +36,11 @@ func (r Receipt) MarshalJSON() ([]byte, error) {
 	return json.Marshal(&enc)
 }
 
+// UnmarshalJSON unmarshals from JSON.
 func (r *Receipt) UnmarshalJSON(input []byte) error {
 	type Receipt struct {
 		PostState         *hexutil.Bytes  `json:"root"`
-		Status            *hexutil.Uint   `json:"status"`
+		Status            *hexutil.Uint64 `json:"status"`
 		CumulativeGasUsed *hexutil.Uint64 `json:"cumulativeGasUsed" gencodec:"required"`
 		Bloom             *Bloom          `json:"logsBloom"         gencodec:"required"`
 		Logs              []*Log          `json:"logs"              gencodec:"required"`
@@ -54,7 +56,7 @@ func (r *Receipt) UnmarshalJSON(input []byte) error {
 		r.PostState = *dec.PostState
 	}
 	if dec.Status != nil {
-		r.Status = uint(*dec.Status)
+		r.Status = uint64(*dec.Status)
 	}
 	if dec.CumulativeGasUsed == nil {
 		return errors.New("missing required field 'cumulativeGasUsed' for Receipt")

core/types/receipt.go

@@ -36,17 +36,17 @@ var (
 
 const (
 	// ReceiptStatusFailed is the status code of a transaction if execution failed.
-	ReceiptStatusFailed = uint(0)
+	ReceiptStatusFailed = uint64(0)
 
 	// ReceiptStatusSuccessful is the status code of a transaction if execution succeeded.
-	ReceiptStatusSuccessful = uint(1)
+	ReceiptStatusSuccessful = uint64(1)
 )
 
 // Receipt represents the results of a transaction.
 type Receipt struct {
 	// Consensus fields
 	PostState         []byte `json:"root"`
-	Status            uint   `json:"status"`
+	Status            uint64 `json:"status"`
 	CumulativeGasUsed uint64 `json:"cumulativeGasUsed" gencodec:"required"`
 	Bloom             Bloom  `json:"logsBloom"         gencodec:"required"`
 	Logs              []*Log `json:"logs"              gencodec:"required"`
@@ -59,7 +59,7 @@ type Receipt struct {
 
 type receiptMarshaling struct {
 	PostState         hexutil.Bytes
-	Status            hexutil.Uint
+	Status            hexutil.Uint64
 	CumulativeGasUsed hexutil.Uint64
 	GasUsed           hexutil.Uint64
 }

core/vm/instructions.go

@@ -850,7 +850,7 @@ func makePush(size uint64, pushByteSize int) executionFunc {
 	}
 }
 
-// make push instruction function
+// make dup instruction function
 func makeDup(size int64) executionFunc {
 	return func(pc *uint64, evm *EVM, contract *Contract, memory *Memory, stack *Stack) ([]byte, error) {
 		stack.dup(evm.interpreter.intPool, int(size))

eth/api_backend.go

@@ -188,8 +188,8 @@ func (b *EthAPIBackend) TxPoolContent() (map[common.Address]types.Transactions,
 	return b.eth.TxPool().Content()
 }
 
-func (b *EthAPIBackend) SubscribeTxPreEvent(ch chan<- core.TxPreEvent) event.Subscription {
-	return b.eth.TxPool().SubscribeTxPreEvent(ch)
+func (b *EthAPIBackend) SubscribeNewTxsEvent(ch chan<- core.NewTxsEvent) event.Subscription {
+	return b.eth.TxPool().SubscribeNewTxsEvent(ch)
 }
 
 func (b *EthAPIBackend) Downloader() *downloader.Downloader {

eth/backend.go

@@ -215,14 +215,14 @@ func CreateConsensusEngine(ctx *node.ServiceContext, config *ethash.Config, chai
 		return clique.New(chainConfig.Clique, db)
 	}
 	// Otherwise assume proof-of-work
-	switch {
-	case config.PowMode == ethash.ModeFake:
+	switch config.PowMode {
+	case ethash.ModeFake:
 		log.Warn("Ethash used in fake mode")
 		return ethash.NewFaker()
-	case config.PowMode == ethash.ModeTest:
+	case ethash.ModeTest:
 		log.Warn("Ethash used in test mode")
 		return ethash.NewTester()
-	case config.PowMode == ethash.ModeShared:
+	case ethash.ModeShared:
 		log.Warn("Ethash used in shared mode")
 		return ethash.NewShared()
 	default:
@@ -239,7 +239,7 @@ func CreateConsensusEngine(ctx *node.ServiceContext, config *ethash.Config, chai
 	}
 }
 
-// APIs returns the collection of RPC services the ethereum package offers.
+// APIs return the collection of RPC services the ethereum package offers.
 // NOTE, some of these services probably need to be moved to somewhere else.
 func (s *Ethereum) APIs() []rpc.API {
 	apis := ethapi.GetAPIs(s.APIBackend)

eth/filters/api.go

@@ -104,8 +104,8 @@ func (api *PublicFilterAPI) timeoutLoop() {
 // https://github.com/ethereum/wiki/wiki/JSON-RPC#eth_newpendingtransactionfilter
 func (api *PublicFilterAPI) NewPendingTransactionFilter() rpc.ID {
 	var (
-		pendingTxs   = make(chan common.Hash)
-		pendingTxSub = api.events.SubscribePendingTxEvents(pendingTxs)
+		pendingTxs   = make(chan []common.Hash)
+		pendingTxSub = api.events.SubscribePendingTxs(pendingTxs)
 	)
 
 	api.filtersMu.Lock()
@@ -118,7 +118,7 @@ func (api *PublicFilterAPI) NewPendingTransactionFilter() rpc.ID {
 			case ph := <-pendingTxs:
 				api.filtersMu.Lock()
 				if f, found := api.filters[pendingTxSub.ID]; found {
-					f.hashes = append(f.hashes, ph)
+					f.hashes = append(f.hashes, ph...)
 				}
 				api.filtersMu.Unlock()
 			case <-pendingTxSub.Err():
@@ -144,13 +144,17 @@ func (api *PublicFilterAPI) NewPendingTransactions(ctx context.Context) (*rpc.Su
 	rpcSub := notifier.CreateSubscription()
 
 	go func() {
-		txHashes := make(chan common.Hash)
-		pendingTxSub := api.events.SubscribePendingTxEvents(txHashes)
+		txHashes := make(chan []common.Hash, 128)
+		pendingTxSub := api.events.SubscribePendingTxs(txHashes)
 
 		for {
 			select {
-			case h := <-txHashes:
-				notifier.Notify(rpcSub.ID, h)
+			case hashes := <-txHashes:
+				// To keep the original behaviour, send a single tx hash in one notification.
+				// TODO(rjl493456442) Send a batch of tx hashes in one notification
+				for _, h := range hashes {
+					notifier.Notify(rpcSub.ID, h)
+				}
 			case <-rpcSub.Err():
 				pendingTxSub.Unsubscribe()
 				return

eth/filters/filter.go

@@ -36,7 +36,7 @@ type Backend interface {
 	GetReceipts(ctx context.Context, blockHash common.Hash) (types.Receipts, error)
 	GetLogs(ctx context.Context, blockHash common.Hash) ([][]*types.Log, error)
 
-	SubscribeTxPreEvent(chan<- core.TxPreEvent) event.Subscription
+	SubscribeNewTxsEvent(chan<- core.NewTxsEvent) event.Subscription
 	SubscribeChainEvent(ch chan<- core.ChainEvent) event.Subscription
 	SubscribeRemovedLogsEvent(ch chan<- core.RemovedLogsEvent) event.Subscription
 	SubscribeLogsEvent(ch chan<- []*types.Log) event.Subscription

eth/filters/filter_system.go

@@ -59,7 +59,7 @@ const (
 
 const (
 
-	// txChanSize is the size of channel listening to TxPreEvent.
+	// txChanSize is the size of channel listening to NewTxsEvent.
 	// The number is referenced from the size of tx pool.
 	txChanSize = 4096
 	// rmLogsChanSize is the size of channel listening to RemovedLogsEvent.
@@ -80,7 +80,7 @@ type subscription struct {
 	created   time.Time
 	logsCrit  ethereum.FilterQuery
 	logs      chan []*types.Log
-	hashes    chan common.Hash
+	hashes    chan []common.Hash
 	headers   chan *types.Header
 	installed chan struct{} // closed when the filter is installed
 	err       chan error    // closed when the filter is uninstalled
@@ -95,7 +95,7 @@ type EventSystem struct {
 	lastHead  *types.Header
 
 	// Subscriptions
-	txSub         event.Subscription         // Subscription for new transaction event
+	txsSub        event.Subscription         // Subscription for new transaction event
 	logsSub       event.Subscription         // Subscription for new log event
 	rmLogsSub     event.Subscription         // Subscription for removed log event
 	chainSub      event.Subscription         // Subscription for new chain event
@@ -104,7 +104,7 @@ type EventSystem struct {
 	// Channels
 	install   chan *subscription         // install filter for event notification
 	uninstall chan *subscription         // remove filter for event notification
-	txCh      chan core.TxPreEvent       // Channel to receive new transaction event
+	txsCh     chan core.NewTxsEvent      // Channel to receive new transactions event
 	logsCh    chan []*types.Log          // Channel to receive new log event
 	rmLogsCh  chan core.RemovedLogsEvent // Channel to receive removed log event
 	chainCh   chan core.ChainEvent       // Channel to receive new chain event
@@ -123,14 +123,14 @@ func NewEventSystem(mux *event.TypeMux, backend Backend, lightMode bool) *EventS
 		lightMode: lightMode,
 		install:   make(chan *subscription),
 		uninstall: make(chan *subscription),
-		txCh:      make(chan core.TxPreEvent, txChanSize),
+		txsCh:     make(chan core.NewTxsEvent, txChanSize),
 		logsCh:    make(chan []*types.Log, logsChanSize),
 		rmLogsCh:  make(chan core.RemovedLogsEvent, rmLogsChanSize),
 		chainCh:   make(chan core.ChainEvent, chainEvChanSize),
 	}
 
 	// Subscribe events
-	m.txSub = m.backend.SubscribeTxPreEvent(m.txCh)
+	m.txsSub = m.backend.SubscribeNewTxsEvent(m.txsCh)
 	m.logsSub = m.backend.SubscribeLogsEvent(m.logsCh)
 	m.rmLogsSub = m.backend.SubscribeRemovedLogsEvent(m.rmLogsCh)
 	m.chainSub = m.backend.SubscribeChainEvent(m.chainCh)
@@ -138,7 +138,7 @@ func NewEventSystem(mux *event.TypeMux, backend Backend, lightMode bool) *EventS
 	m.pendingLogSub = m.mux.Subscribe(core.PendingLogsEvent{})
 
 	// Make sure none of the subscriptions are empty
-	if m.txSub == nil || m.logsSub == nil || m.rmLogsSub == nil || m.chainSub == nil ||
+	if m.txsSub == nil || m.logsSub == nil || m.rmLogsSub == nil || m.chainSub == nil ||
 		m.pendingLogSub.Closed() {
 		log.Crit("Subscribe for event system failed")
 	}
@@ -240,7 +240,7 @@ func (es *EventSystem) subscribeMinedPendingLogs(crit ethereum.FilterQuery, logs
 		logsCrit:  crit,
 		created:   time.Now(),
 		logs:      logs,
-		hashes:    make(chan common.Hash),
+		hashes:    make(chan []common.Hash),
 		headers:   make(chan *types.Header),
 		installed: make(chan struct{}),
 		err:       make(chan error),
@@ -257,7 +257,7 @@ func (es *EventSystem) subscribeLogs(crit ethereum.FilterQuery, logs chan []*typ
 		logsCrit:  crit,
 		created:   time.Now(),
 		logs:      logs,
-		hashes:    make(chan common.Hash),
+		hashes:    make(chan []common.Hash),
 		headers:   make(chan *types.Header),
 		installed: make(chan struct{}),
 		err:       make(chan error),
@@ -274,7 +274,7 @@ func (es *EventSystem) subscribePendingLogs(crit ethereum.FilterQuery, logs chan
 		logsCrit:  crit,
 		created:   time.Now(),
 		logs:      logs,
-		hashes:    make(chan common.Hash),
+		hashes:    make(chan []common.Hash),
 		headers:   make(chan *types.Header),
 		installed: make(chan struct{}),
 		err:       make(chan error),
@@ -290,7 +290,7 @@ func (es *EventSystem) SubscribeNewHeads(headers chan *types.Header) *Subscripti
 		typ:       BlocksSubscription,
 		created:   time.Now(),
 		logs:      make(chan []*types.Log),
-		hashes:    make(chan common.Hash),
+		hashes:    make(chan []common.Hash),
 		headers:   headers,
 		installed: make(chan struct{}),
 		err:       make(chan error),
@@ -298,9 +298,9 @@ func (es *EventSystem) SubscribeNewHeads(headers chan *types.Header) *Subscripti
 	return es.subscribe(sub)
 }
 
-// SubscribePendingTxEvents creates a subscription that writes transaction hashes for
+// SubscribePendingTxs creates a subscription that writes transaction hashes for
 // transactions that enter the transaction pool.
-func (es *EventSystem) SubscribePendingTxEvents(hashes chan common.Hash) *Subscription {
+func (es *EventSystem) SubscribePendingTxs(hashes chan []common.Hash) *Subscription {
 	sub := &subscription{
 		id:        rpc.NewID(),
 		typ:       PendingTransactionsSubscription,
@@ -348,9 +348,13 @@ func (es *EventSystem) broadcast(filters filterIndex, ev interface{}) {
 				}
 			}
 		}
-	case core.TxPreEvent:
+	case core.NewTxsEvent:
+		hashes := make([]common.Hash, 0, len(e.Txs))
+		for _, tx := range e.Txs {
+			hashes = append(hashes, tx.Hash())
+		}
 		for _, f := range filters[PendingTransactionsSubscription] {
-			f.hashes <- e.Tx.Hash()
+			f.hashes <- hashes
 		}
 	case core.ChainEvent:
 		for _, f := range filters[BlocksSubscription] {
@@ -446,7 +450,7 @@ func (es *EventSystem) eventLoop() {
 	// Ensure all subscriptions get cleaned up
 	defer func() {
 		es.pendingLogSub.Unsubscribe()
-		es.txSub.Unsubscribe()
+		es.txsSub.Unsubscribe()
 		es.logsSub.Unsubscribe()
 		es.rmLogsSub.Unsubscribe()
 		es.chainSub.Unsubscribe()
@@ -460,7 +464,7 @@ func (es *EventSystem) eventLoop() {
 	for {
 		select {
 		// Handle subscribed events
-		case ev := <-es.txCh:
+		case ev := <-es.txsCh:
 			es.broadcast(index, ev)
 		case ev := <-es.logsCh:
 			es.broadcast(index, ev)
@@ -495,7 +499,7 @@ func (es *EventSystem) eventLoop() {
 			close(f.err)
 
 		// System stopped
-		case <-es.txSub.Err():
+		case <-es.txsSub.Err():
 			return
 		case <-es.logsSub.Err():
 			return

eth/filters/filter_system_test.go

@@ -96,7 +96,7 @@ func (b *testBackend) GetLogs(ctx context.Context, hash common.Hash) ([][]*types
 	return logs, nil
 }
 
-func (b *testBackend) SubscribeTxPreEvent(ch chan<- core.TxPreEvent) event.Subscription {
+func (b *testBackend) SubscribeNewTxsEvent(ch chan<- core.NewTxsEvent) event.Subscription {
 	return b.txFeed.Subscribe(ch)
 }
 
@@ -232,10 +232,7 @@ func TestPendingTxFilter(t *testing.T) {
 	fid0 := api.NewPendingTransactionFilter()
 
 	time.Sleep(1 * time.Second)
-	for _, tx := range transactions {
-		ev := core.TxPreEvent{Tx: tx}
-		txFeed.Send(ev)
-	}
+	txFeed.Send(core.NewTxsEvent{Txs: transactions})
 
 	timeout := time.Now().Add(1 * time.Second)
 	for {

eth/handler.go

@@ -46,7 +46,7 @@ const (
 	softResponseLimit = 2 * 1024 * 1024 // Target maximum size of returned blocks, headers or node data.
 	estHeaderRlpSize  = 500             // Approximate size of an RLP encoded block header
 
-	// txChanSize is the size of channel listening to TxPreEvent.
+	// txChanSize is the size of channel listening to NewTxsEvent.
 	// The number is referenced from the size of tx pool.
 	txChanSize = 4096
 )
@@ -81,8 +81,8 @@ type ProtocolManager struct {
 	SubProtocols []p2p.Protocol
 
 	eventMux      *event.TypeMux
-	txCh          chan core.TxPreEvent
-	txSub         event.Subscription
+	txsCh         chan core.NewTxsEvent
+	txsSub        event.Subscription
 	minedBlockSub *event.TypeMuxSubscription
 
 	// channels for fetcher, syncer, txsyncLoop
@@ -204,8 +204,8 @@ func (pm *ProtocolManager) Start(maxPeers int) {
 	pm.maxPeers = maxPeers
 
 	// broadcast transactions
-	pm.txCh = make(chan core.TxPreEvent, txChanSize)
-	pm.txSub = pm.txpool.SubscribeTxPreEvent(pm.txCh)
+	pm.txsCh = make(chan core.NewTxsEvent, txChanSize)
+	pm.txsSub = pm.txpool.SubscribeNewTxsEvent(pm.txsCh)
 	go pm.txBroadcastLoop()
 
 	// broadcast mined blocks
@@ -220,7 +220,7 @@ func (pm *ProtocolManager) Start(maxPeers int) {
 func (pm *ProtocolManager) Stop() {
 	log.Info("Stopping Ethereum protocol")
 
-	pm.txSub.Unsubscribe()         // quits txBroadcastLoop
+	pm.txsSub.Unsubscribe()        // quits txBroadcastLoop
 	pm.minedBlockSub.Unsubscribe() // quits blockBroadcastLoop
 
 	// Quit the sync loop.
@@ -698,7 +698,7 @@ func (pm *ProtocolManager) BroadcastBlock(block *types.Block, propagate bool) {
 		// Send the block to a subset of our peers
 		transfer := peers[:int(math.Sqrt(float64(len(peers))))]
 		for _, peer := range transfer {
-			peer.SendNewBlock(block, td)
+			peer.AsyncSendNewBlock(block, td)
 		}
 		log.Trace("Propagated block", "hash", hash, "recipients", len(transfer), "duration", common.PrettyDuration(time.Since(block.ReceivedAt)))
 		return
@@ -706,22 +706,29 @@ func (pm *ProtocolManager) BroadcastBlock(block *types.Block, propagate bool) {
 	// Otherwise if the block is indeed in out own chain, announce it
 	if pm.blockchain.HasBlock(hash, block.NumberU64()) {
 		for _, peer := range peers {
-			peer.SendNewBlockHashes([]common.Hash{hash}, []uint64{block.NumberU64()})
+			peer.AsyncSendNewBlockHash(block)
 		}
 		log.Trace("Announced block", "hash", hash, "recipients", len(peers), "duration", common.PrettyDuration(time.Since(block.ReceivedAt)))
 	}
 }
 
-// BroadcastTx will propagate a transaction to all peers which are not known to
+// BroadcastTxs will propagate a batch of transactions to all peers which are not known to
 // already have the given transaction.
-func (pm *ProtocolManager) BroadcastTx(hash common.Hash, tx *types.Transaction) {
-	// Broadcast transaction to a batch of peers not knowing about it
-	peers := pm.peers.PeersWithoutTx(hash)
-	//FIXME include this again: peers = peers[:int(math.Sqrt(float64(len(peers))))]
-	for _, peer := range peers {
-		peer.SendTransactions(types.Transactions{tx})
+func (pm *ProtocolManager) BroadcastTxs(txs types.Transactions) {
+	var txset = make(map[*peer]types.Transactions)
+
+	// Broadcast transactions to a batch of peers not knowing about it
+	for _, tx := range txs {
+		peers := pm.peers.PeersWithoutTx(tx.Hash())
+		for _, peer := range peers {
+			txset[peer] = append(txset[peer], tx)
+		}
+		log.Trace("Broadcast transaction", "hash", tx.Hash(), "recipients", len(peers))
+	}
+	// FIXME include this again: peers = peers[:int(math.Sqrt(float64(len(peers))))]
+	for peer, txs := range txset {
+		peer.AsyncSendTransactions(txs)
 	}
-	log.Trace("Broadcast transaction", "hash", hash, "recipients", len(peers))
 }
 
 // Mined broadcast loop
@@ -739,11 +746,11 @@ func (pm *ProtocolManager) minedBroadcastLoop() {
 func (pm *ProtocolManager) txBroadcastLoop() {
 	for {
 		select {
-		case event := <-pm.txCh:
-			pm.BroadcastTx(event.Tx.Hash(), event.Tx)
+		case event := <-pm.txsCh:
+			pm.BroadcastTxs(event.Txs)
 
 		// Err() channel will be closed when unsubscribing.
-		case <-pm.txSub.Err():
+		case <-pm.txsSub.Err():
 			return
 		}
 	}

eth/helper_test.go

@@ -124,7 +124,7 @@ func (p *testTxPool) Pending() (map[common.Address]types.Transactions, error) {
 	return batches, nil
 }
 
-func (p *testTxPool) SubscribeTxPreEvent(ch chan<- core.TxPreEvent) event.Subscription {
+func (p *testTxPool) SubscribeNewTxsEvent(ch chan<- core.NewTxsEvent) event.Subscription {
 	return p.txFeed.Subscribe(ch)
 }
 

eth/peer.go

@@ -37,8 +37,24 @@ var (
 )
 
 const (
-	maxKnownTxs      = 32768 // Maximum transactions hashes to keep in the known list (prevent DOS)
-	maxKnownBlocks   = 1024  // Maximum block hashes to keep in the known list (prevent DOS)
+	maxKnownTxs    = 32768 // Maximum transactions hashes to keep in the known list (prevent DOS)
+	maxKnownBlocks = 1024  // Maximum block hashes to keep in the known list (prevent DOS)
+
+	// maxQueuedTxs is the maximum number of transaction lists to queue up before
+	// dropping broadcasts. This is a sensitive number as a transaction list might
+	// contain a single transaction, or thousands.
+	maxQueuedTxs = 128
+
+	// maxQueuedProps is the maximum number of block propagations to queue up before
+	// dropping broadcasts. There's not much point in queueing stale blocks, so a few
+	// that might cover uncles should be enough.
+	maxQueuedProps = 4
+
+	// maxQueuedAnns is the maximum number of block announcements to queue up before
+	// dropping broadcasts. Similarly to block propagations, there's no point to queue
+	// above some healthy uncle limit, so use that.
+	maxQueuedAnns = 4
+
 	handshakeTimeout = 5 * time.Second
 )
 
@@ -50,6 +66,12 @@ type PeerInfo struct {
 	Head       string   `json:"head"`       // SHA3 hash of the peer's best owned block
 }
 
+// propEvent is a block propagation, waiting for its turn in the broadcast queue.
+type propEvent struct {
+	block *types.Block
+	td    *big.Int
+}
+
 type peer struct {
 	id string
 
@@ -63,23 +85,64 @@ type peer struct {
 	td   *big.Int
 	lock sync.RWMutex
 
-	knownTxs    *set.Set // Set of transaction hashes known to be known by this peer
-	knownBlocks *set.Set // Set of block hashes known to be known by this peer
+	knownTxs    *set.Set                  // Set of transaction hashes known to be known by this peer
+	knownBlocks *set.Set                  // Set of block hashes known to be known by this peer
+	queuedTxs   chan []*types.Transaction // Queue of transactions to broadcast to the peer
+	queuedProps chan *propEvent           // Queue of blocks to broadcast to the peer
+	queuedAnns  chan *types.Block         // Queue of blocks to announce to the peer
+	term        chan struct{}             // Termination channel to stop the broadcaster
 }
 
 func newPeer(version int, p *p2p.Peer, rw p2p.MsgReadWriter) *peer {
-	id := p.ID()
-
 	return &peer{
 		Peer:        p,
 		rw:          rw,
 		version:     version,
-		id:          fmt.Sprintf("%x", id[:8]),
+		id:          fmt.Sprintf("%x", p.ID().Bytes()[:8]),
 		knownTxs:    set.New(),
 		knownBlocks: set.New(),
+		queuedTxs:   make(chan []*types.Transaction, maxQueuedTxs),
+		queuedProps: make(chan *propEvent, maxQueuedProps),
+		queuedAnns:  make(chan *types.Block, maxQueuedAnns),
+		term:        make(chan struct{}),
 	}
 }
 
+// broadcast is a write loop that multiplexes block propagations, announcements
+// and transaction broadcasts into the remote peer. The goal is to have an async
+// writer that does not lock up node internals.
+func (p *peer) broadcast() {
+	for {
+		select {
+		case txs := <-p.queuedTxs:
+			if err := p.SendTransactions(txs); err != nil {
+				return
+			}
+			p.Log().Trace("Broadcast transactions", "count", len(txs))
+
+		case prop := <-p.queuedProps:
+			if err := p.SendNewBlock(prop.block, prop.td); err != nil {
+				return
+			}
+			p.Log().Trace("Propagated block", "number", prop.block.Number(), "hash", prop.block.Hash(), "td", prop.td)
+
+		case block := <-p.queuedAnns:
+			if err := p.SendNewBlockHashes([]common.Hash{block.Hash()}, []uint64{block.NumberU64()}); err != nil {
+				return
+			}
+			p.Log().Trace("Announced block", "number", block.Number(), "hash", block.Hash())
+
+		case <-p.term:
+			return
+		}
+	}
+}
+
+// close signals the broadcast goroutine to terminate.
+func (p *peer) close() {
+	close(p.term)
+}
+
 // Info gathers and returns a collection of metadata known about a peer.
 func (p *peer) Info() *PeerInfo {
 	hash, td := p.Head()
@@ -139,6 +202,19 @@ func (p *peer) SendTransactions(txs types.Transactions) error {
 	return p2p.Send(p.rw, TxMsg, txs)
 }
 
+// AsyncSendTransactions queues list of transactions propagation to a remote
+// peer. If the peer's broadcast queue is full, the event is silently dropped.
+func (p *peer) AsyncSendTransactions(txs []*types.Transaction) {
+	select {
+	case p.queuedTxs <- txs:
+		for _, tx := range txs {
+			p.knownTxs.Add(tx.Hash())
+		}
+	default:
+		p.Log().Debug("Dropping transaction propagation", "count", len(txs))
+	}
+}
+
 // SendNewBlockHashes announces the availability of a number of blocks through
 // a hash notification.
 func (p *peer) SendNewBlockHashes(hashes []common.Hash, numbers []uint64) error {
@@ -153,12 +229,35 @@ func (p *peer) SendNewBlockHashes(hashes []common.Hash, numbers []uint64) error
 	return p2p.Send(p.rw, NewBlockHashesMsg, request)
 }
 
+// AsyncSendNewBlockHash queues the availability of a block for propagation to a
+// remote peer. If the peer's broadcast queue is full, the event is silently
+// dropped.
+func (p *peer) AsyncSendNewBlockHash(block *types.Block) {
+	select {
+	case p.queuedAnns <- block:
+		p.knownBlocks.Add(block.Hash())
+	default:
+		p.Log().Debug("Dropping block announcement", "number", block.NumberU64(), "hash", block.Hash())
+	}
+}
+
 // SendNewBlock propagates an entire block to a remote peer.
 func (p *peer) SendNewBlock(block *types.Block, td *big.Int) error {
 	p.knownBlocks.Add(block.Hash())
 	return p2p.Send(p.rw, NewBlockMsg, []interface{}{block, td})
 }
 
+// AsyncSendNewBlock queues an entire block for propagation to a remote peer. If
+// the peer's broadcast queue is full, the event is silently dropped.
+func (p *peer) AsyncSendNewBlock(block *types.Block, td *big.Int) {
+	select {
+	case p.queuedProps <- &propEvent{block: block, td: td}:
+		p.knownBlocks.Add(block.Hash())
+	default:
+		p.Log().Debug("Dropping block propagation", "number", block.NumberU64(), "hash", block.Hash())
+	}
+}
+
 // SendBlockHeaders sends a batch of block headers to the remote peer.
 func (p *peer) SendBlockHeaders(headers []*types.Header) error {
 	return p2p.Send(p.rw, BlockHeadersMsg, headers)
@@ -313,7 +412,8 @@ func newPeerSet() *peerSet {
 }
 
 // Register injects a new peer into the working set, or returns an error if the
-// peer is already known.
+// peer is already known. If a new peer it registered, its broadcast loop is also
+// started.
 func (ps *peerSet) Register(p *peer) error {
 	ps.lock.Lock()
 	defer ps.lock.Unlock()
@@ -325,6 +425,8 @@ func (ps *peerSet) Register(p *peer) error {
 		return errAlreadyRegistered
 	}
 	ps.peers[p.id] = p
+	go p.broadcast()
+
 	return nil
 }
 
@@ -334,10 +436,13 @@ func (ps *peerSet) Unregister(id string) error {
 	ps.lock.Lock()
 	defer ps.lock.Unlock()
 
-	if _, ok := ps.peers[id]; !ok {
+	p, ok := ps.peers[id]
+	if !ok {
 		return errNotRegistered
 	}
 	delete(ps.peers, id)
+	p.close()
+
 	return nil
 }
 

eth/protocol.go

@@ -103,9 +103,9 @@ type txPool interface {
 	// The slice should be modifiable by the caller.
 	Pending() (map[common.Address]types.Transactions, error)
 
-	// SubscribeTxPreEvent should return an event subscription of
-	// TxPreEvent and send events to the given channel.
-	SubscribeTxPreEvent(chan<- core.TxPreEvent) event.Subscription
+	// SubscribeNewTxsEvent should return an event subscription of
+	// NewTxsEvent and send events to the given channel.
+	SubscribeNewTxsEvent(chan<- core.NewTxsEvent) event.Subscription
 }
 
 // statusData is the network packet for the status message.

eth/protocol_test.go

@@ -116,7 +116,7 @@ func testRecvTransactions(t *testing.T, protocol int) {
 			t.Errorf("added wrong tx hash: got %v, want %v", added[0].Hash(), tx.Hash())
 		}
 	case <-time.After(2 * time.Second):
-		t.Errorf("no TxPreEvent received within 2 seconds")
+		t.Errorf("no NewTxsEvent received within 2 seconds")
 	}
 }
 

ethdb/database.go

@@ -207,6 +207,7 @@ func (db *LDBDatabase) meter(refresh time.Duration) {
 		delaystats      [2]int64
 		lastWriteDelay  time.Time
 		lastWriteDelayN time.Time
+		lastWritePaused time.Time
 	)
 
 	// Iterate ad infinitum and collect the stats
@@ -267,8 +268,9 @@ func (db *LDBDatabase) meter(refresh time.Duration) {
 			delayN        int64
 			delayDuration string
 			duration      time.Duration
+			paused        bool
 		)
-		if n, err := fmt.Sscanf(writedelay, "DelayN:%d Delay:%s", &delayN, &delayDuration); n != 2 || err != nil {
+		if n, err := fmt.Sscanf(writedelay, "DelayN:%d Delay:%s Paused:%t", &delayN, &delayDuration, &paused); n != 3 || err != nil {
 			db.log.Error("Write delay statistic not found")
 			return
 		}
@@ -301,6 +303,14 @@ func (db *LDBDatabase) meter(refresh time.Duration) {
 				lastWriteDelay = time.Now()
 			}
 		}
+		// If a warning that db is performing compaction has been displayed, any subsequent
+		// warnings will be withheld for one minute not to overwhelm the user.
+		if paused && delayN-delaystats[0] == 0 && duration.Nanoseconds()-delaystats[1] == 0 &&
+			time.Now().After(lastWritePaused.Add(writeDelayWarningThrottler)) {
+			db.log.Warn("Database compacting, degraded performance")
+			lastWritePaused = time.Now()
+		}
+
 		delaystats[0], delaystats[1] = delayN, duration.Nanoseconds()
 
 		// Retrieve the database iostats.

ethstats/ethstats.go

@@ -49,7 +49,7 @@ const (
 	// history request.
 	historyUpdateRange = 50
 
-	// txChanSize is the size of channel listening to TxPreEvent.
+	// txChanSize is the size of channel listening to NewTxsEvent.
 	// The number is referenced from the size of tx pool.
 	txChanSize = 4096
 	// chainHeadChanSize is the size of channel listening to ChainHeadEvent.
@@ -57,9 +57,9 @@ const (
 )
 
 type txPool interface {
-	// SubscribeTxPreEvent should return an event subscription of
-	// TxPreEvent and send events to the given channel.
-	SubscribeTxPreEvent(chan<- core.TxPreEvent) event.Subscription
+	// SubscribeNewTxsEvent should return an event subscription of
+	// NewTxsEvent and send events to the given channel.
+	SubscribeNewTxsEvent(chan<- core.NewTxsEvent) event.Subscription
 }
 
 type blockChain interface {
@@ -150,8 +150,8 @@ func (s *Service) loop() {
 	headSub := blockchain.SubscribeChainHeadEvent(chainHeadCh)
 	defer headSub.Unsubscribe()
 
-	txEventCh := make(chan core.TxPreEvent, txChanSize)
-	txSub := txpool.SubscribeTxPreEvent(txEventCh)
+	txEventCh := make(chan core.NewTxsEvent, txChanSize)
+	txSub := txpool.SubscribeNewTxsEvent(txEventCh)
 	defer txSub.Unsubscribe()
 
 	// Start a goroutine that exhausts the subsciptions to avoid events piling up

internal/ethapi/backend.go

@@ -65,7 +65,7 @@ type Backend interface {
 	GetPoolNonce(ctx context.Context, addr common.Address) (uint64, error)
 	Stats() (pending int, queued int)
 	TxPoolContent() (map[common.Address]types.Transactions, map[common.Address]types.Transactions)
-	SubscribeTxPreEvent(chan<- core.TxPreEvent) event.Subscription
+	SubscribeNewTxsEvent(chan<- core.NewTxsEvent) event.Subscription
 
 	ChainConfig() *params.ChainConfig
 	CurrentBlock() *types.Block

les/api_backend.go

@@ -136,8 +136,8 @@ func (b *LesApiBackend) TxPoolContent() (map[common.Address]types.Transactions,
 	return b.eth.txPool.Content()
 }
 
-func (b *LesApiBackend) SubscribeTxPreEvent(ch chan<- core.TxPreEvent) event.Subscription {
-	return b.eth.txPool.SubscribeTxPreEvent(ch)
+func (b *LesApiBackend) SubscribeNewTxsEvent(ch chan<- core.NewTxsEvent) event.Subscription {
+	return b.eth.txPool.SubscribeNewTxsEvent(ch)
 }
 
 func (b *LesApiBackend) SubscribeChainEvent(ch chan<- core.ChainEvent) event.Subscription {

les/odr_requests.go

@@ -230,7 +230,7 @@ func (r *TrieRequest) Validate(db ethdb.Database, msg *Msg) error {
 		}
 		nodeSet := proofs[0].NodeSet()
 		// Verify the proof and store if checks out
-		if _, err, _ := trie.VerifyProof(r.Id.Root, r.Key, nodeSet); err != nil {
+		if _, _, err := trie.VerifyProof(r.Id.Root, r.Key, nodeSet); err != nil {
 			return fmt.Errorf("merkle proof verification failed: %v", err)
 		}
 		r.Proof = nodeSet
@@ -241,7 +241,7 @@ func (r *TrieRequest) Validate(db ethdb.Database, msg *Msg) error {
 		// Verify the proof and store if checks out
 		nodeSet := proofs.NodeSet()
 		reads := &readTraceDB{db: nodeSet}
-		if _, err, _ := trie.VerifyProof(r.Id.Root, r.Key, reads); err != nil {
+		if _, _, err := trie.VerifyProof(r.Id.Root, r.Key, reads); err != nil {
 			return fmt.Errorf("merkle proof verification failed: %v", err)
 		}
 		// check if all nodes have been read by VerifyProof
@@ -400,7 +400,7 @@ func (r *ChtRequest) Validate(db ethdb.Database, msg *Msg) error {
 		var encNumber [8]byte
 		binary.BigEndian.PutUint64(encNumber[:], r.BlockNum)
 
-		value, err, _ := trie.VerifyProof(r.ChtRoot, encNumber[:], light.NodeList(proof.Proof).NodeSet())
+		value, _, err := trie.VerifyProof(r.ChtRoot, encNumber[:], light.NodeList(proof.Proof).NodeSet())
 		if err != nil {
 			return err
 		}
@@ -435,7 +435,7 @@ func (r *ChtRequest) Validate(db ethdb.Database, msg *Msg) error {
 		binary.BigEndian.PutUint64(encNumber[:], r.BlockNum)
 
 		reads := &readTraceDB{db: nodeSet}
-		value, err, _ := trie.VerifyProof(r.ChtRoot, encNumber[:], reads)
+		value, _, err := trie.VerifyProof(r.ChtRoot, encNumber[:], reads)
 		if err != nil {
 			return fmt.Errorf("merkle proof verification failed: %v", err)
 		}
@@ -529,7 +529,7 @@ func (r *BloomRequest) Validate(db ethdb.Database, msg *Msg) error {
 
 	for i, idx := range r.SectionIdxList {
 		binary.BigEndian.PutUint64(encNumber[2:], idx)
-		value, err, _ := trie.VerifyProof(r.BloomTrieRoot, encNumber[:], reads)
+		value, _, err := trie.VerifyProof(r.BloomTrieRoot, encNumber[:], reads)
 		if err != nil {
 			return err
 		}

light/txpool.go

@@ -321,9 +321,9 @@ func (pool *TxPool) Stop() {
 	log.Info("Transaction pool stopped")
 }
 
-// SubscribeTxPreEvent registers a subscription of core.TxPreEvent and
+// SubscribeNewTxsEvent registers a subscription of core.NewTxsEvent and
 // starts sending event to the given channel.
-func (pool *TxPool) SubscribeTxPreEvent(ch chan<- core.TxPreEvent) event.Subscription {
+func (pool *TxPool) SubscribeNewTxsEvent(ch chan<- core.NewTxsEvent) event.Subscription {
 	return pool.scope.Track(pool.txFeed.Subscribe(ch))
 }
 
@@ -412,7 +412,7 @@ func (self *TxPool) add(ctx context.Context, tx *types.Transaction) error {
 		// Notify the subscribers. This event is posted in a goroutine
 		// because it's possible that somewhere during the post "Remove transaction"
 		// gets called which will then wait for the global tx pool lock and deadlock.
-		go self.txFeed.Send(core.TxPreEvent{Tx: tx})
+		go self.txFeed.Send(core.NewTxsEvent{Txs: types.Transactions{tx}})
 	}
 
 	// Print a log message if low enough level is set

log/README.md

@@ -45,7 +45,7 @@ srvlog.SetHandler(log.MultiHandler(
     log.StreamHandler(os.Stderr, log.LogfmtFormat()),
     log.LvlFilterHandler(
         log.LvlError,
-        log.Must.FileHandler("errors.json", log.JsonFormat()))))
+        log.Must.FileHandler("errors.json", log.JSONFormat()))))
 ```
 
 Will result in output that looks like this:

log/doc.go

@@ -86,7 +86,7 @@ from the rpc package in logfmt to standard out. The other prints records at Erro
 or above in JSON formatted output to the file /var/log/service.json
 
     handler := log.MultiHandler(
-        log.LvlFilterHandler(log.LvlError, log.Must.FileHandler("/var/log/service.json", log.JsonFormat())),
+        log.LvlFilterHandler(log.LvlError, log.Must.FileHandler("/var/log/service.json", log.JSONFormat())),
         log.MatchFilterHandler("pkg", "app/rpc" log.StdoutHandler())
     )
 
@@ -304,8 +304,8 @@ For all Handler functions which can return an error, there is a version of that
 function which will return no error but panics on failure. They are all available
 on the Must object. For example:
 
-    log.Must.FileHandler("/path", log.JsonFormat)
-    log.Must.NetHandler("tcp", ":1234", log.JsonFormat)
+    log.Must.FileHandler("/path", log.JSONFormat)
+    log.Must.NetHandler("tcp", ":1234", log.JSONFormat)
 
 Inspiration and Credit
 

log/format.go

@@ -196,16 +196,16 @@ func logfmt(buf *bytes.Buffer, ctx []interface{}, color int, term bool) {
 	buf.WriteByte('\n')
 }
 
-// JsonFormat formats log records as JSON objects separated by newlines.
-// It is the equivalent of JsonFormatEx(false, true).
-func JsonFormat() Format {
-	return JsonFormatEx(false, true)
+// JSONFormat formats log records as JSON objects separated by newlines.
+// It is the equivalent of JSONFormatEx(false, true).
+func JSONFormat() Format {
+	return JSONFormatEx(false, true)
 }
 
-// JsonFormatEx formats log records as JSON objects. If pretty is true,
+// JSONFormatEx formats log records as JSON objects. If pretty is true,
 // records will be pretty-printed. If lineSeparated is true, records
 // will be logged with a new line between each record.
-func JsonFormatEx(pretty, lineSeparated bool) Format {
+func JSONFormatEx(pretty, lineSeparated bool) Format {
 	jsonMarshal := json.Marshal
 	if pretty {
 		jsonMarshal = func(v interface{}) ([]byte, error) {
@@ -225,7 +225,7 @@ func JsonFormatEx(pretty, lineSeparated bool) Format {
 			if !ok {
 				props[errorKey] = fmt.Sprintf("%+v is not a string key", r.Ctx[i])
 			}
-			props[k] = formatJsonValue(r.Ctx[i+1])
+			props[k] = formatJSONValue(r.Ctx[i+1])
 		}
 
 		b, err := jsonMarshal(props)
@@ -270,7 +270,7 @@ func formatShared(value interface{}) (result interface{}) {
 	}
 }
 
-func formatJsonValue(value interface{}) interface{} {
+func formatJSONValue(value interface{}) interface{} {
 	value = formatShared(value)
 	switch value.(type) {
 	case int, int8, int16, int32, int64, float32, float64, uint, uint8, uint16, uint32, uint64, string:

log/handler.go

@@ -11,8 +11,8 @@ import (
 	"github.com/go-stack/stack"
 )
 
+// Handler defines where and how log records are written.
 // A Logger prints its log records by writing to a Handler.
-// The Handler interface defines where and how log records are written.
 // Handlers are composable, providing you great flexibility in combining
 // them to achieve the logging structure that suits your applications.
 type Handler interface {
@@ -193,7 +193,7 @@ func LvlFilterHandler(maxLvl Lvl, h Handler) Handler {
 	}, h)
 }
 
-// A MultiHandler dispatches any write to each of its handlers.
+// MultiHandler dispatches any write to each of its handlers.
 // This is useful for writing different types of log information
 // to different locations. For example, to log to a file and
 // standard error:
@@ -212,7 +212,7 @@ func MultiHandler(hs ...Handler) Handler {
 	})
 }
 
-// A FailoverHandler writes all log records to the first handler
+// FailoverHandler writes all log records to the first handler
 // specified, but will failover and write to the second handler if
 // the first handler has failed, and so on for all handlers specified.
 // For example you might want to log to a network socket, but failover
@@ -220,7 +220,7 @@ func MultiHandler(hs ...Handler) Handler {
 // standard out if the file write fails:
 //
 //     log.FailoverHandler(
-//         log.Must.NetHandler("tcp", ":9090", log.JsonFormat()),
+//         log.Must.NetHandler("tcp", ":9090", log.JSONFormat()),
 //         log.Must.FileHandler("/var/log/app.log", log.LogfmtFormat()),
 //         log.StdoutHandler)
 //
@@ -336,7 +336,7 @@ func DiscardHandler() Handler {
 	})
 }
 
-// The Must object provides the following Handler creation functions
+// Must provides the following Handler creation functions
 // which instead of returning an error parameter only return a Handler
 // and panic on failure: FileHandler, NetHandler, SyslogHandler, SyslogNetHandler
 var Must muster

log/logger.go

@@ -24,7 +24,7 @@ const (
 	LvlTrace
 )
 
-// Aligned returns a 5-character string containing the name of a Lvl.
+// AlignedString returns a 5-character string containing the name of a Lvl.
 func (l Lvl) AlignedString() string {
 	switch l {
 	case LvlTrace:
@@ -64,7 +64,7 @@ func (l Lvl) String() string {
 	}
 }
 
-// Returns the appropriate Lvl from a string name.
+// LvlFromString returns the appropriate Lvl from a string name.
 // Useful for parsing command line args and configuration files.
 func LvlFromString(lvlString string) (Lvl, error) {
 	switch lvlString {
@@ -95,6 +95,7 @@ type Record struct {
 	KeyNames RecordKeyNames
 }
 
+// RecordKeyNames gets stored in a Record when the write function is executed.
 type RecordKeyNames struct {
 	Time string
 	Msg  string

miner/worker.go

@@ -42,7 +42,7 @@ const (
 	resultQueueSize  = 10
 	miningLogAtDepth = 5
 
-	// txChanSize is the size of channel listening to TxPreEvent.
+	// txChanSize is the size of channel listening to NewTxsEvent.
 	// The number is referenced from the size of tx pool.
 	txChanSize = 4096
 	// chainHeadChanSize is the size of channel listening to ChainHeadEvent.
@@ -71,6 +71,7 @@ type Work struct {
 	family    *set.Set       // family set (used for checking uncle invalidity)
 	uncles    *set.Set       // uncle set
 	tcount    int            // tx count in cycle
+	gasPool   *core.GasPool  // available gas used to pack transactions
 
 	Block *types.Block // the new block
 
@@ -95,8 +96,8 @@ type worker struct {
 
 	// update loop
 	mux          *event.TypeMux
-	txCh         chan core.TxPreEvent
-	txSub        event.Subscription
+	txsCh        chan core.NewTxsEvent
+	txsSub       event.Subscription
 	chainHeadCh  chan core.ChainHeadEvent
 	chainHeadSub event.Subscription
 	chainSideCh  chan core.ChainSideEvent
@@ -137,7 +138,7 @@ func newWorker(config *params.ChainConfig, engine consensus.Engine, coinbase com
 		engine:         engine,
 		eth:            eth,
 		mux:            mux,
-		txCh:           make(chan core.TxPreEvent, txChanSize),
+		txsCh:          make(chan core.NewTxsEvent, txChanSize),
 		chainHeadCh:    make(chan core.ChainHeadEvent, chainHeadChanSize),
 		chainSideCh:    make(chan core.ChainSideEvent, chainSideChanSize),
 		chainDb:        eth.ChainDb(),
@@ -149,8 +150,8 @@ func newWorker(config *params.ChainConfig, engine consensus.Engine, coinbase com
 		agents:         make(map[Agent]struct{}),
 		unconfirmed:    newUnconfirmedBlocks(eth.BlockChain(), miningLogAtDepth),
 	}
-	// Subscribe TxPreEvent for tx pool
-	worker.txSub = eth.TxPool().SubscribeTxPreEvent(worker.txCh)
+	// Subscribe NewTxsEvent for tx pool
+	worker.txsSub = eth.TxPool().SubscribeNewTxsEvent(worker.txsCh)
 	// Subscribe events for blockchain
 	worker.chainHeadSub = eth.BlockChain().SubscribeChainHeadEvent(worker.chainHeadCh)
 	worker.chainSideSub = eth.BlockChain().SubscribeChainSideEvent(worker.chainSideCh)
@@ -241,7 +242,7 @@ func (self *worker) unregister(agent Agent) {
 }
 
 func (self *worker) update() {
-	defer self.txSub.Unsubscribe()
+	defer self.txsSub.Unsubscribe()
 	defer self.chainHeadSub.Unsubscribe()
 	defer self.chainSideSub.Unsubscribe()
 
@@ -258,15 +259,21 @@ func (self *worker) update() {
 			self.possibleUncles[ev.Block.Hash()] = ev.Block
 			self.uncleMu.Unlock()
 
-		// Handle TxPreEvent
-		case ev := <-self.txCh:
-			// Apply transaction to the pending state if we're not mining
+		// Handle NewTxsEvent
+		case ev := <-self.txsCh:
+			// Apply transactions to the pending state if we're not mining.
+			//
+			// Note all transactions received may not be continuous with transactions
+			// already included in the current mining block. These transactions will
+			// be automatically eliminated.
 			if atomic.LoadInt32(&self.mining) == 0 {
 				self.currentMu.Lock()
-				acc, _ := types.Sender(self.current.signer, ev.Tx)
-				txs := map[common.Address]types.Transactions{acc: {ev.Tx}}
+				txs := make(map[common.Address]types.Transactions)
+				for _, tx := range ev.Txs {
+					acc, _ := types.Sender(self.current.signer, tx)
+					txs[acc] = append(txs[acc], tx)
+				}
 				txset := types.NewTransactionsByPriceAndNonce(self.current.signer, txs)
-
 				self.current.commitTransactions(self.mux, txset, self.chain, self.coinbase)
 				self.updateSnapshot()
 				self.currentMu.Unlock()
@@ -278,7 +285,7 @@ func (self *worker) update() {
 			}
 
 		// System stopped
-		case <-self.txSub.Err():
+		case <-self.txsSub.Err():
 			return
 		case <-self.chainHeadSub.Err():
 			return
@@ -522,14 +529,16 @@ func (self *worker) updateSnapshot() {
 }
 
 func (env *Work) commitTransactions(mux *event.TypeMux, txs *types.TransactionsByPriceAndNonce, bc *core.BlockChain, coinbase common.Address) {
-	gp := new(core.GasPool).AddGas(env.header.GasLimit)
+	if env.gasPool == nil {
+		env.gasPool = new(core.GasPool).AddGas(env.header.GasLimit)
+	}
 
 	var coalescedLogs []*types.Log
 
 	for {
 		// If we don't have enough gas for any further transactions then we're done
-		if gp.Gas() < params.TxGas {
-			log.Trace("Not enough gas for further transactions", "gp", gp)
+		if env.gasPool.Gas() < params.TxGas {
+			log.Trace("Not enough gas for further transactions", "have", env.gasPool, "want", params.TxGas)
 			break
 		}
 		// Retrieve the next transaction and abort if all done
@@ -553,7 +562,7 @@ func (env *Work) commitTransactions(mux *event.TypeMux, txs *types.TransactionsB
 		// Start executing the transaction
 		env.state.Prepare(tx.Hash(), common.Hash{}, env.tcount)
 
-		err, logs := env.commitTransaction(tx, bc, coinbase, gp)
+		err, logs := env.commitTransaction(tx, bc, coinbase, env.gasPool)
 		switch err {
 		case core.ErrGasLimitReached:
 			// Pop the current out-of-gas transaction without shifting in the next from the account

node/api.go

@@ -217,7 +217,7 @@ func (api *PrivateAdminAPI) StartWS(host *string, port *int, allowedOrigins *str
 	return true, nil
 }
 
-// StopRPC terminates an already running websocket RPC API endpoint.
+// StopWS terminates an already running websocket RPC API endpoint.
 func (api *PrivateAdminAPI) StopWS() (bool, error) {
 	api.node.lock.Lock()
 	defer api.node.lock.Unlock()

node/doc.go

@@ -59,7 +59,7 @@ using the same data directory will store this information in different subdirect
 the data directory.
 
 LevelDB databases are also stored within the instance subdirectory. If multiple node
-instances use the same data directory, openening the databases with identical names will
+instances use the same data directory, opening the databases with identical names will
 create one database for each instance.
 
 The account key store is shared among all node instances using the same data directory
@@ -84,7 +84,7 @@ directory. Mode instance A opens the database "db", node instance B opens the da
             static-nodes.json  -- devp2p static node list of instance B
             db/                -- LevelDB content for "db"
             db-2/              -- LevelDB content for "db-2"
-        B.ipc                  -- JSON-RPC UNIX domain socket endpoint of instance A
+        B.ipc                  -- JSON-RPC UNIX domain socket endpoint of instance B
         keystore/              -- account key store, used by both instances
 */
 package node

node/node_test.go

@@ -507,8 +507,8 @@ func TestAPIGather(t *testing.T) {
 	}
 	// Register a batch of services with some configured APIs
 	calls := make(chan string, 1)
-	makeAPI := func(result string) *OneMethodApi {
-		return &OneMethodApi{fun: func() { calls <- result }}
+	makeAPI := func(result string) *OneMethodAPI {
+		return &OneMethodAPI{fun: func() { calls <- result }}
 	}
 	services := map[string]struct {
 		APIs  []rpc.API

node/utils_test.go

@@ -121,12 +121,12 @@ func InstrumentedServiceMakerC(base ServiceConstructor) ServiceConstructor {
 	return InstrumentingWrapperMaker(base, reflect.TypeOf(InstrumentedServiceC{}))
 }
 
-// OneMethodApi is a single-method API handler to be returned by test services.
-type OneMethodApi struct {
+// OneMethodAPI is a single-method API handler to be returned by test services.
+type OneMethodAPI struct {
 	fun func()
 }
 
-func (api *OneMethodApi) TheOneMethod() {
+func (api *OneMethodAPI) TheOneMethod() {
 	if api.fun != nil {
 		api.fun()
 	}

p2p/enr/enr.go

@@ -29,21 +29,16 @@ package enr
 
 import (
 	"bytes"
-	"crypto/ecdsa"
 	"errors"
 	"fmt"
 	"io"
 	"sort"
 
-	"github.com/ethereum/go-ethereum/crypto"
-	"github.com/ethereum/go-ethereum/crypto/sha3"
 	"github.com/ethereum/go-ethereum/rlp"
 )
 
 const SizeLimit = 300 // maximum encoded size of a node record in bytes
 
-const ID_SECP256k1_KECCAK = ID("secp256k1-keccak") // the default identity scheme
-
 var (
 	errNoID           = errors.New("unknown or unspecified identity scheme")
 	errInvalidSig     = errors.New("invalid signature")
@@ -80,8 +75,8 @@ func (r *Record) Seq() uint64 {
 }
 
 // SetSeq updates the record sequence number. This invalidates any signature on the record.
-// Calling SetSeq is usually not required because signing the redord increments the
-// sequence number.
+// Calling SetSeq is usually not required because setting any key in a signed record
+// increments the sequence number.
 func (r *Record) SetSeq(s uint64) {
 	r.signature = nil
 	r.raw = nil
@@ -104,33 +99,42 @@ func (r *Record) Load(e Entry) error {
 	return &KeyError{Key: e.ENRKey(), Err: errNotFound}
 }
 
-// Set adds or updates the given entry in the record.
-// It panics if the value can't be encoded.
+// Set adds or updates the given entry in the record. It panics if the value can't be
+// encoded. If the record is signed, Set increments the sequence number and invalidates
+// the sequence number.
 func (r *Record) Set(e Entry) {
-	r.signature = nil
-	r.raw = nil
 	blob, err := rlp.EncodeToBytes(e)
 	if err != nil {
 		panic(fmt.Errorf("enr: can't encode %s: %v", e.ENRKey(), err))
 	}
+	r.invalidate()
 
-	i := sort.Search(len(r.pairs), func(i int) bool { return r.pairs[i].k >= e.ENRKey() })
-
-	if i < len(r.pairs) && r.pairs[i].k == e.ENRKey() {
+	pairs := make([]pair, len(r.pairs))
+	copy(pairs, r.pairs)
+	i := sort.Search(len(pairs), func(i int) bool { return pairs[i].k >= e.ENRKey() })
+	switch {
+	case i < len(pairs) && pairs[i].k == e.ENRKey():
 		// element is present at r.pairs[i]
-		r.pairs[i].v = blob
-		return
-	} else if i < len(r.pairs) {
+		pairs[i].v = blob
+	case i < len(r.pairs):
 		// insert pair before i-th elem
 		el := pair{e.ENRKey(), blob}
-		r.pairs = append(r.pairs, pair{})
-		copy(r.pairs[i+1:], r.pairs[i:])
-		r.pairs[i] = el
-		return
+		pairs = append(pairs, pair{})
+		copy(pairs[i+1:], pairs[i:])
+		pairs[i] = el
+	default:
+		// element should be placed at the end of r.pairs
+		pairs = append(pairs, pair{e.ENRKey(), blob})
 	}
+	r.pairs = pairs
+}
 
-	// element should be placed at the end of r.pairs
-	r.pairs = append(r.pairs, pair{e.ENRKey(), blob})
+func (r *Record) invalidate() {
+	if r.signature == nil {
+		r.seq++
+	}
+	r.signature = nil
+	r.raw = nil
 }
 
 // EncodeRLP implements rlp.Encoder. Encoding fails if
@@ -196,39 +200,55 @@ func (r *Record) DecodeRLP(s *rlp.Stream) error {
 		return err
 	}
 
-	// Verify signature.
-	if err = dec.verifySignature(); err != nil {
+	_, scheme := dec.idScheme()
+	if scheme == nil {
+		return errNoID
+	}
+	if err := scheme.Verify(&dec, dec.signature); err != nil {
 		return err
 	}
 	*r = dec
 	return nil
 }
 
-type s256raw []byte
-
-func (s256raw) ENRKey() string { return "secp256k1" }
-
 // NodeAddr returns the node address. The return value will be nil if the record is
-// unsigned.
+// unsigned or uses an unknown identity scheme.
 func (r *Record) NodeAddr() []byte {
-	var entry s256raw
-	if r.Load(&entry) != nil {
+	_, scheme := r.idScheme()
+	if scheme == nil {
 		return nil
 	}
-	return crypto.Keccak256(entry)
+	return scheme.NodeAddr(r)
 }
 
-// Sign signs the record with the given private key. It updates the record's identity
-// scheme, public key and increments the sequence number. Sign returns an error if the
-// encoded record is larger than the size limit.
-func (r *Record) Sign(privkey *ecdsa.PrivateKey) error {
-	r.seq = r.seq + 1
-	r.Set(ID_SECP256k1_KECCAK)
-	r.Set(Secp256k1(privkey.PublicKey))
-	return r.signAndEncode(privkey)
+// SetSig sets the record signature. It returns an error if the encoded record is larger
+// than the size limit or if the signature is invalid according to the passed scheme.
+func (r *Record) SetSig(idscheme string, sig []byte) error {
+	// Check that "id" is set and matches the given scheme. This panics because
+	// inconsitencies here are always implementation bugs in the signing function calling
+	// this method.
+	id, s := r.idScheme()
+	if s == nil {
+		panic(errNoID)
+	}
+	if id != idscheme {
+		panic(fmt.Errorf("identity scheme mismatch in Sign: record has %s, want %s", id, idscheme))
+	}
+
+	// Verify against the scheme.
+	if err := s.Verify(r, sig); err != nil {
+		return err
+	}
+	raw, err := r.encode(sig)
+	if err != nil {
+		return err
+	}
+	r.signature, r.raw = sig, raw
+	return nil
 }
 
-func (r *Record) appendPairs(list []interface{}) []interface{} {
+// AppendElements appends the sequence number and entries to the given slice.
+func (r *Record) AppendElements(list []interface{}) []interface{} {
 	list = append(list, r.seq)
 	for _, p := range r.pairs {
 		list = append(list, p.k, p.v)
@@ -236,54 +256,23 @@ func (r *Record) appendPairs(list []interface{}) []interface{} {
 	return list
 }
 
-func (r *Record) signAndEncode(privkey *ecdsa.PrivateKey) error {
-	// Put record elements into a flat list. Leave room for the signature.
-	list := make([]interface{}, 1, len(r.pairs)*2+2)
-	list = r.appendPairs(list)
-
-	// Sign the tail of the list.
-	h := sha3.NewKeccak256()
-	rlp.Encode(h, list[1:])
-	sig, err := crypto.Sign(h.Sum(nil), privkey)
-	if err != nil {
-		return err
+func (r *Record) encode(sig []byte) (raw []byte, err error) {
+	list := make([]interface{}, 1, 2*len(r.pairs)+1)
+	list[0] = sig
+	list = r.AppendElements(list)
+	if raw, err = rlp.EncodeToBytes(list); err != nil {
+		return nil, err
 	}
-	sig = sig[:len(sig)-1] // remove v
-
-	// Put signature in front.
-	r.signature, list[0] = sig, sig
-	r.raw, err = rlp.EncodeToBytes(list)
-	if err != nil {
-		return err
+	if len(raw) > SizeLimit {
+		return nil, errTooBig
 	}
-	if len(r.raw) > SizeLimit {
-		return errTooBig
-	}
-	return nil
+	return raw, nil
 }
 
-func (r *Record) verifySignature() error {
-	// Get identity scheme, public key, signature.
+func (r *Record) idScheme() (string, IdentityScheme) {
 	var id ID
-	var entry s256raw
 	if err := r.Load(&id); err != nil {
-		return err
-	} else if id != ID_SECP256k1_KECCAK {
-		return errNoID
+		return "", nil
 	}
-	if err := r.Load(&entry); err != nil {
-		return err
-	} else if len(entry) != 33 {
-		return fmt.Errorf("invalid public key")
-	}
-
-	// Verify the signature.
-	list := make([]interface{}, 0, len(r.pairs)*2+1)
-	list = r.appendPairs(list)
-	h := sha3.NewKeccak256()
-	rlp.Encode(h, list)
-	if !crypto.VerifySignature(entry, h.Sum(nil), r.signature) {
-		return errInvalidSig
-	}
-	return nil
+	return string(id), FindIdentityScheme(string(id))
 }

p2p/enr/enr_test.go

@@ -54,35 +54,35 @@ func TestGetSetID(t *testing.T) {
 	assert.Equal(t, id, id2)
 }
 
-// TestGetSetIP4 tests encoding/decoding and setting/getting of the IP4 key.
+// TestGetSetIP4 tests encoding/decoding and setting/getting of the IP key.
 func TestGetSetIP4(t *testing.T) {
-	ip := IP4{192, 168, 0, 3}
+	ip := IP{192, 168, 0, 3}
 	var r Record
 	r.Set(ip)
 
-	var ip2 IP4
+	var ip2 IP
 	require.NoError(t, r.Load(&ip2))
 	assert.Equal(t, ip, ip2)
 }
 
-// TestGetSetIP6 tests encoding/decoding and setting/getting of the IP6 key.
+// TestGetSetIP6 tests encoding/decoding and setting/getting of the IP key.
 func TestGetSetIP6(t *testing.T) {
-	ip := IP6{0x20, 0x01, 0x48, 0x60, 0, 0, 0x20, 0x01, 0, 0, 0, 0, 0, 0, 0x00, 0x68}
+	ip := IP{0x20, 0x01, 0x48, 0x60, 0, 0, 0x20, 0x01, 0, 0, 0, 0, 0, 0, 0x00, 0x68}
 	var r Record
 	r.Set(ip)
 
-	var ip2 IP6
+	var ip2 IP
 	require.NoError(t, r.Load(&ip2))
 	assert.Equal(t, ip, ip2)
 }
 
 // TestGetSetDiscPort tests encoding/decoding and setting/getting of the DiscPort key.
-func TestGetSetDiscPort(t *testing.T) {
-	port := DiscPort(30309)
+func TestGetSetUDP(t *testing.T) {
+	port := UDP(30309)
 	var r Record
 	r.Set(port)
 
-	var port2 DiscPort
+	var port2 UDP
 	require.NoError(t, r.Load(&port2))
 	assert.Equal(t, port, port2)
 }
@@ -90,7 +90,7 @@ func TestGetSetDiscPort(t *testing.T) {
 // TestGetSetSecp256k1 tests encoding/decoding and setting/getting of the Secp256k1 key.
 func TestGetSetSecp256k1(t *testing.T) {
 	var r Record
-	if err := r.Sign(privkey); err != nil {
+	if err := SignV4(&r, privkey); err != nil {
 		t.Fatal(err)
 	}
 
@@ -101,16 +101,16 @@ func TestGetSetSecp256k1(t *testing.T) {
 
 func TestLoadErrors(t *testing.T) {
 	var r Record
-	ip4 := IP4{127, 0, 0, 1}
+	ip4 := IP{127, 0, 0, 1}
 	r.Set(ip4)
 
 	// Check error for missing keys.
-	var ip6 IP6
-	err := r.Load(&ip6)
+	var udp UDP
+	err := r.Load(&udp)
 	if !IsNotFound(err) {
 		t.Error("IsNotFound should return true for missing key")
 	}
-	assert.Equal(t, &KeyError{Key: ip6.ENRKey(), Err: errNotFound}, err)
+	assert.Equal(t, &KeyError{Key: udp.ENRKey(), Err: errNotFound}, err)
 
 	// Check error for invalid keys.
 	var list []uint
@@ -174,7 +174,7 @@ func TestDirty(t *testing.T) {
 		t.Errorf("expected errEncodeUnsigned, got %#v", err)
 	}
 
-	require.NoError(t, r.Sign(privkey))
+	require.NoError(t, SignV4(&r, privkey))
 	if !r.Signed() {
 		t.Error("Signed return false for signed record")
 	}
@@ -194,13 +194,13 @@ func TestDirty(t *testing.T) {
 func TestGetSetOverwrite(t *testing.T) {
 	var r Record
 
-	ip := IP4{192, 168, 0, 3}
+	ip := IP{192, 168, 0, 3}
 	r.Set(ip)
 
-	ip2 := IP4{192, 168, 0, 4}
+	ip2 := IP{192, 168, 0, 4}
 	r.Set(ip2)
 
-	var ip3 IP4
+	var ip3 IP
 	require.NoError(t, r.Load(&ip3))
 	assert.Equal(t, ip2, ip3)
 }
@@ -208,9 +208,9 @@ func TestGetSetOverwrite(t *testing.T) {
 // TestSignEncodeAndDecode tests signing, RLP encoding and RLP decoding of a record.
 func TestSignEncodeAndDecode(t *testing.T) {
 	var r Record
-	r.Set(DiscPort(30303))
-	r.Set(IP4{127, 0, 0, 1})
-	require.NoError(t, r.Sign(privkey))
+	r.Set(UDP(30303))
+	r.Set(IP{127, 0, 0, 1})
+	require.NoError(t, SignV4(&r, privkey))
 
 	blob, err := rlp.EncodeToBytes(r)
 	require.NoError(t, err)
@@ -230,12 +230,12 @@ func TestNodeAddr(t *testing.T) {
 		t.Errorf("wrong address on empty record: got %v, want %v", addr, nil)
 	}
 
-	require.NoError(t, r.Sign(privkey))
-	expected := "caaa1485d83b18b32ed9ad666026151bf0cae8a0a88c857ae2d4c5be2daa6726"
+	require.NoError(t, SignV4(&r, privkey))
+	expected := "a448f24c6d18e575453db13171562b71999873db5b286df957af199ec94617f7"
 	assert.Equal(t, expected, hex.EncodeToString(r.NodeAddr()))
 }
 
-var pyRecord, _ = hex.DecodeString("f896b840954dc36583c1f4b69ab59b1375f362f06ee99f3723cd77e64b6de6d211c27d7870642a79d4516997f94091325d2a7ca6215376971455fb221d34f35b277149a1018664697363763582765f82696490736563703235366b312d6b656363616b83697034847f00000189736563703235366b31a103ca634cae0d49acb401d8a4c6b6fe8c55b70d115bf400769cc1400f3258cd3138")
+var pyRecord, _ = hex.DecodeString("f884b8407098ad865b00a582051940cb9cf36836572411a47278783077011599ed5cd16b76f2635f4e234738f30813a89eb9137e3e3df5266e3a1f11df72ecf1145ccb9c01826964827634826970847f00000189736563703235366b31a103ca634cae0d49acb401d8a4c6b6fe8c55b70d115bf400769cc1400f3258cd31388375647082765f")
 
 // TestPythonInterop checks that we can decode and verify a record produced by the Python
 // implementation.
@@ -246,10 +246,10 @@ func TestPythonInterop(t *testing.T) {
 	}
 
 	var (
-		wantAddr, _  = hex.DecodeString("caaa1485d83b18b32ed9ad666026151bf0cae8a0a88c857ae2d4c5be2daa6726")
-		wantSeq      = uint64(1)
-		wantIP       = IP4{127, 0, 0, 1}
-		wantDiscport = DiscPort(30303)
+		wantAddr, _ = hex.DecodeString("a448f24c6d18e575453db13171562b71999873db5b286df957af199ec94617f7")
+		wantSeq     = uint64(1)
+		wantIP      = IP{127, 0, 0, 1}
+		wantUDP     = UDP(30303)
 	)
 	if r.Seq() != wantSeq {
 		t.Errorf("wrong seq: got %d, want %d", r.Seq(), wantSeq)
@@ -257,7 +257,7 @@ func TestPythonInterop(t *testing.T) {
 	if addr := r.NodeAddr(); !bytes.Equal(addr, wantAddr) {
 		t.Errorf("wrong addr: got %x, want %x", addr, wantAddr)
 	}
-	want := map[Entry]interface{}{new(IP4): &wantIP, new(DiscPort): &wantDiscport}
+	want := map[Entry]interface{}{new(IP): &wantIP, new(UDP): &wantUDP}
 	for k, v := range want {
 		desc := fmt.Sprintf("loading key %q", k.ENRKey())
 		if assert.NoError(t, r.Load(k), desc) {
@@ -272,14 +272,14 @@ func TestRecordTooBig(t *testing.T) {
 	key := randomString(10)
 
 	// set a big value for random key, expect error
-	r.Set(WithEntry(key, randomString(300)))
-	if err := r.Sign(privkey); err != errTooBig {
+	r.Set(WithEntry(key, randomString(SizeLimit)))
+	if err := SignV4(&r, privkey); err != errTooBig {
 		t.Fatalf("expected to get errTooBig, got %#v", err)
 	}
 
 	// set an acceptable value for random key, expect no error
 	r.Set(WithEntry(key, randomString(100)))
-	require.NoError(t, r.Sign(privkey))
+	require.NoError(t, SignV4(&r, privkey))
 }
 
 // TestSignEncodeAndDecodeRandom tests encoding/decoding of records containing random key/value pairs.
@@ -295,7 +295,7 @@ func TestSignEncodeAndDecodeRandom(t *testing.T) {
 		r.Set(WithEntry(key, &value))
 	}
 
-	require.NoError(t, r.Sign(privkey))
+	require.NoError(t, SignV4(&r, privkey))
 	_, err := rlp.EncodeToBytes(r)
 	require.NoError(t, err)
 

p2p/enr/entries.go

@@ -57,59 +57,43 @@ func WithEntry(k string, v interface{}) Entry {
 	return &generic{key: k, value: v}
 }
 
-// DiscPort is the "discv5" key, which holds the UDP port for discovery v5.
-type DiscPort uint16
+// TCP is the "tcp" key, which holds the TCP port of the node.
+type TCP uint16
 
-func (v DiscPort) ENRKey() string { return "discv5" }
+func (v TCP) ENRKey() string { return "tcp" }
+
+// UDP is the "udp" key, which holds the UDP port of the node.
+type UDP uint16
+
+func (v UDP) ENRKey() string { return "udp" }
 
 // ID is the "id" key, which holds the name of the identity scheme.
 type ID string
 
+const IDv4 = ID("v4") // the default identity scheme
+
 func (v ID) ENRKey() string { return "id" }
 
-// IP4 is the "ip4" key, which holds a 4-byte IPv4 address.
-type IP4 net.IP
+// IP is the "ip" key, which holds the IP address of the node.
+type IP net.IP
 
-func (v IP4) ENRKey() string { return "ip4" }
+func (v IP) ENRKey() string { return "ip" }
 
 // EncodeRLP implements rlp.Encoder.
-func (v IP4) EncodeRLP(w io.Writer) error {
-	ip4 := net.IP(v).To4()
-	if ip4 == nil {
-		return fmt.Errorf("invalid IPv4 address: %v", v)
+func (v IP) EncodeRLP(w io.Writer) error {
+	if ip4 := net.IP(v).To4(); ip4 != nil {
+		return rlp.Encode(w, ip4)
 	}
-	return rlp.Encode(w, ip4)
+	return rlp.Encode(w, net.IP(v))
 }
 
 // DecodeRLP implements rlp.Decoder.
-func (v *IP4) DecodeRLP(s *rlp.Stream) error {
+func (v *IP) DecodeRLP(s *rlp.Stream) error {
 	if err := s.Decode((*net.IP)(v)); err != nil {
 		return err
 	}
-	if len(*v) != 4 {
-		return fmt.Errorf("invalid IPv4 address, want 4 bytes: %v", *v)
-	}
-	return nil
-}
-
-// IP6 is the "ip6" key, which holds a 16-byte IPv6 address.
-type IP6 net.IP
-
-func (v IP6) ENRKey() string { return "ip6" }
-
-// EncodeRLP implements rlp.Encoder.
-func (v IP6) EncodeRLP(w io.Writer) error {
-	ip6 := net.IP(v)
-	return rlp.Encode(w, ip6)
-}
-
-// DecodeRLP implements rlp.Decoder.
-func (v *IP6) DecodeRLP(s *rlp.Stream) error {
-	if err := s.Decode((*net.IP)(v)); err != nil {
-		return err
-	}
-	if len(*v) != 16 {
-		return fmt.Errorf("invalid IPv6 address, want 16 bytes: %v", *v)
+	if len(*v) != 4 && len(*v) != 16 {
+		return fmt.Errorf("invalid IP address, want 4 or 16 bytes: %v", *v)
 	}
 	return nil
 }

p2p/enr/idscheme.go

@@ -0,0 +1,114 @@
+// Copyright 2018 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package enr
+
+import (
+	"crypto/ecdsa"
+	"fmt"
+	"sync"
+
+	"github.com/ethereum/go-ethereum/common/math"
+	"github.com/ethereum/go-ethereum/crypto"
+	"github.com/ethereum/go-ethereum/crypto/sha3"
+	"github.com/ethereum/go-ethereum/rlp"
+)
+
+// Registry of known identity schemes.
+var schemes sync.Map
+
+// An IdentityScheme is capable of verifying record signatures and
+// deriving node addresses.
+type IdentityScheme interface {
+	Verify(r *Record, sig []byte) error
+	NodeAddr(r *Record) []byte
+}
+
+// RegisterIdentityScheme adds an identity scheme to the global registry.
+func RegisterIdentityScheme(name string, scheme IdentityScheme) {
+	if _, loaded := schemes.LoadOrStore(name, scheme); loaded {
+		panic("identity scheme " + name + " already registered")
+	}
+}
+
+// FindIdentityScheme resolves name to an identity scheme in the global registry.
+func FindIdentityScheme(name string) IdentityScheme {
+	s, ok := schemes.Load(name)
+	if !ok {
+		return nil
+	}
+	return s.(IdentityScheme)
+}
+
+// v4ID is the "v4" identity scheme.
+type v4ID struct{}
+
+func init() {
+	RegisterIdentityScheme("v4", v4ID{})
+}
+
+// SignV4 signs a record using the v4 scheme.
+func SignV4(r *Record, privkey *ecdsa.PrivateKey) error {
+	// Copy r to avoid modifying it if signing fails.
+	cpy := *r
+	cpy.Set(ID("v4"))
+	cpy.Set(Secp256k1(privkey.PublicKey))
+
+	h := sha3.NewKeccak256()
+	rlp.Encode(h, cpy.AppendElements(nil))
+	sig, err := crypto.Sign(h.Sum(nil), privkey)
+	if err != nil {
+		return err
+	}
+	sig = sig[:len(sig)-1] // remove v
+	if err = cpy.SetSig("v4", sig); err == nil {
+		*r = cpy
+	}
+	return err
+}
+
+// s256raw is an unparsed secp256k1 public key entry.
+type s256raw []byte
+
+func (s256raw) ENRKey() string { return "secp256k1" }
+
+func (v4ID) Verify(r *Record, sig []byte) error {
+	var entry s256raw
+	if err := r.Load(&entry); err != nil {
+		return err
+	} else if len(entry) != 33 {
+		return fmt.Errorf("invalid public key")
+	}
+
+	h := sha3.NewKeccak256()
+	rlp.Encode(h, r.AppendElements(nil))
+	if !crypto.VerifySignature(entry, h.Sum(nil), sig) {
+		return errInvalidSig
+	}
+	return nil
+}
+
+func (v4ID) NodeAddr(r *Record) []byte {
+	var pubkey Secp256k1
+	err := r.Load(&pubkey)
+	if err != nil {
+		return nil
+	}
+	buf := make([]byte, 64)
+	math.ReadBits(pubkey.X, buf[:32])
+	math.ReadBits(pubkey.Y, buf[32:])
+	return crypto.Keccak256(buf)
+}

p2p/enr/idscheme_test.go

@@ -0,0 +1,36 @@
+// Copyright 2018 The go-ethereum Authors
+// This file is part of the go-ethereum library.
+//
+// The go-ethereum library is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// The go-ethereum library is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.
+
+package enr
+
+import (
+	"crypto/ecdsa"
+	"math/big"
+	"testing"
+)
+
+// Checks that failure to sign leaves the record unmodified.
+func TestSignError(t *testing.T) {
+	invalidKey := &ecdsa.PrivateKey{D: new(big.Int), PublicKey: *pubkey}
+
+	var r Record
+	if err := SignV4(&r, invalidKey); err == nil {
+		t.Fatal("expected error from SignV4")
+	}
+	if len(r.pairs) > 0 {
+		t.Fatal("expected empty record, have", r.pairs)
+	}
+}

params/version.go

@@ -21,10 +21,10 @@ import (
 )
 
 const (
-	VersionMajor = 1        // Major version component of the current release
-	VersionMinor = 8        // Minor version component of the current release
-	VersionPatch = 8        // Patch version component of the current release
-	VersionMeta  = "stable" // Version metadata to append to the version string
+	VersionMajor = 1          // Major version component of the current release
+	VersionMinor = 8          // Minor version component of the current release
+	VersionPatch = 9          // Patch version component of the current release
+	VersionMeta  = "unstable" // Version metadata to append to the version string
 )
 
 // Version holds the textual version string.

trie/hasher.go

@@ -196,12 +196,12 @@ func (h *hasher) store(n node, db *Database, force bool) (node, error) {
 		if h.onleaf != nil {
 			switch n := n.(type) {
 			case *shortNode:
-				if child, ok := n.Val.(valueNode); ok {
+				if child, ok := n.Val.(valueNode); ok && child != nil {
 					h.onleaf(child, hash)
 				}
 			case *fullNode:
 				for i := 0; i < 16; i++ {
-					if child, ok := n.Children[i].(valueNode); ok {
+					if child, ok := n.Children[i].(valueNode); ok && child != nil {
 						h.onleaf(child, hash)
 					}
 				}

trie/iterator.go

@@ -22,6 +22,7 @@ import (
 	"errors"
 
 	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/rlp"
 )
 
 // Iterator is a key-value trie iterator that traverses a Trie.
@@ -55,31 +56,50 @@ func (it *Iterator) Next() bool {
 	return false
 }
 
+// Prove generates the Merkle proof for the leaf node the iterator is currently
+// positioned on.
+func (it *Iterator) Prove() [][]byte {
+	return it.nodeIt.LeafProof()
+}
+
 // NodeIterator is an iterator to traverse the trie pre-order.
 type NodeIterator interface {
 	// Next moves the iterator to the next node. If the parameter is false, any child
 	// nodes will be skipped.
 	Next(bool) bool
+
 	// Error returns the error status of the iterator.
 	Error() error
 
 	// Hash returns the hash of the current node.
 	Hash() common.Hash
+
 	// Parent returns the hash of the parent of the current node. The hash may be the one
 	// grandparent if the immediate parent is an internal node with no hash.
 	Parent() common.Hash
+
 	// Path returns the hex-encoded path to the current node.
 	// Callers must not retain references to the return value after calling Next.
 	// For leaf nodes, the last element of the path is the 'terminator symbol' 0x10.
 	Path() []byte
 
 	// Leaf returns true iff the current node is a leaf node.
-	// LeafBlob, LeafKey return the contents and key of the leaf node. These
-	// method panic if the iterator is not positioned at a leaf.
-	// Callers must not retain references to their return value after calling Next
 	Leaf() bool
-	LeafBlob() []byte
+
+	// LeafKey returns the key of the leaf. The method panics if the iterator is not
+	// positioned at a leaf. Callers must not retain references to the value after
+	// calling Next.
 	LeafKey() []byte
+
+	// LeafBlob returns the content of the leaf. The method panics if the iterator
+	// is not positioned at a leaf. Callers must not retain references to the value
+	// after calling Next.
+	LeafBlob() []byte
+
+	// LeafProof returns the Merkle proof of the leaf. The method panics if the
+	// iterator is not positioned at a leaf. Callers must not retain references
+	// to the value after calling Next.
+	LeafProof() [][]byte
 }
 
 // nodeIteratorState represents the iteration state at one particular node of the
@@ -99,8 +119,8 @@ type nodeIterator struct {
 	err   error                // Failure set in case of an internal error in the iterator
 }
 
-// iteratorEnd is stored in nodeIterator.err when iteration is done.
-var iteratorEnd = errors.New("end of iteration")
+// errIteratorEnd is stored in nodeIterator.err when iteration is done.
+var errIteratorEnd = errors.New("end of iteration")
 
 // seekError is stored in nodeIterator.err if the initial seek has failed.
 type seekError struct {
@@ -139,6 +159,15 @@ func (it *nodeIterator) Leaf() bool {
 	return hasTerm(it.path)
 }
 
+func (it *nodeIterator) LeafKey() []byte {
+	if len(it.stack) > 0 {
+		if _, ok := it.stack[len(it.stack)-1].node.(valueNode); ok {
+			return hexToKeybytes(it.path)
+		}
+	}
+	panic("not at leaf")
+}
+
 func (it *nodeIterator) LeafBlob() []byte {
 	if len(it.stack) > 0 {
 		if node, ok := it.stack[len(it.stack)-1].node.(valueNode); ok {
@@ -148,10 +177,22 @@ func (it *nodeIterator) LeafBlob() []byte {
 	panic("not at leaf")
 }
 
-func (it *nodeIterator) LeafKey() []byte {
+func (it *nodeIterator) LeafProof() [][]byte {
 	if len(it.stack) > 0 {
 		if _, ok := it.stack[len(it.stack)-1].node.(valueNode); ok {
-			return hexToKeybytes(it.path)
+			hasher := newHasher(0, 0, nil)
+			proofs := make([][]byte, 0, len(it.stack))
+
+			for i, item := range it.stack[:len(it.stack)-1] {
+				// Gather nodes that end up as hash nodes (or the root)
+				node, _, _ := hasher.hashChildren(item.node, nil)
+				hashed, _ := hasher.store(node, nil, false)
+				if _, ok := hashed.(hashNode); ok || i == 0 {
+					enc, _ := rlp.EncodeToBytes(node)
+					proofs = append(proofs, enc)
+				}
+			}
+			return proofs
 		}
 	}
 	panic("not at leaf")
@@ -162,7 +203,7 @@ func (it *nodeIterator) Path() []byte {
 }
 
 func (it *nodeIterator) Error() error {
-	if it.err == iteratorEnd {
+	if it.err == errIteratorEnd {
 		return nil
 	}
 	if seek, ok := it.err.(seekError); ok {
@@ -176,7 +217,7 @@ func (it *nodeIterator) Error() error {
 // sets the Error field to the encountered failure. If `descend` is false,
 // skips iterating over any subnodes of the current node.
 func (it *nodeIterator) Next(descend bool) bool {
-	if it.err == iteratorEnd {
+	if it.err == errIteratorEnd {
 		return false
 	}
 	if seek, ok := it.err.(seekError); ok {
@@ -201,8 +242,8 @@ func (it *nodeIterator) seek(prefix []byte) error {
 	// Move forward until we're just before the closest match to key.
 	for {
 		state, parentIndex, path, err := it.peek(bytes.HasPrefix(key, it.path))
-		if err == iteratorEnd {
-			return iteratorEnd
+		if err == errIteratorEnd {
+			return errIteratorEnd
 		} else if err != nil {
 			return seekError{prefix, err}
 		} else if bytes.Compare(path, key) >= 0 {
@@ -246,7 +287,7 @@ func (it *nodeIterator) peek(descend bool) (*nodeIteratorState, *int, []byte, er
 		// No more child nodes, move back up.
 		it.pop()
 	}
-	return nil, nil, nil, iteratorEnd
+	return nil, nil, nil, errIteratorEnd
 }
 
 func (st *nodeIteratorState) resolve(tr *Trie, path []byte) error {
@@ -361,12 +402,16 @@ func (it *differenceIterator) Leaf() bool {
 	return it.b.Leaf()
 }
 
+func (it *differenceIterator) LeafKey() []byte {
+	return it.b.LeafKey()
+}
+
 func (it *differenceIterator) LeafBlob() []byte {
 	return it.b.LeafBlob()
 }
 
-func (it *differenceIterator) LeafKey() []byte {
-	return it.b.LeafKey()
+func (it *differenceIterator) LeafProof() [][]byte {
+	return it.b.LeafProof()
 }
 
 func (it *differenceIterator) Path() []byte {
@@ -464,12 +509,16 @@ func (it *unionIterator) Leaf() bool {
 	return (*it.items)[0].Leaf()
 }
 
+func (it *unionIterator) LeafKey() []byte {
+	return (*it.items)[0].LeafKey()
+}
+
 func (it *unionIterator) LeafBlob() []byte {
 	return (*it.items)[0].LeafBlob()
 }
 
-func (it *unionIterator) LeafKey() []byte {
-	return (*it.items)[0].LeafKey()
+func (it *unionIterator) LeafProof() [][]byte {
+	return (*it.items)[0].LeafProof()
 }
 
 func (it *unionIterator) Path() []byte {
@@ -509,12 +558,10 @@ func (it *unionIterator) Next(descend bool) bool {
 			heap.Push(it.items, skipped)
 		}
 	}
-
 	if least.Next(descend) {
 		it.count++
 		heap.Push(it.items, least)
 	}
-
 	return len(*it.items) > 0
 }
 

trie/proof.go

@@ -102,28 +102,28 @@ func (t *SecureTrie) Prove(key []byte, fromLevel uint, proofDb ethdb.Putter) err
 // VerifyProof checks merkle proofs. The given proof must contain the value for
 // key in a trie with the given root hash. VerifyProof returns an error if the
 // proof contains invalid trie nodes or the wrong value.
-func VerifyProof(rootHash common.Hash, key []byte, proofDb DatabaseReader) (value []byte, err error, nodes int) {
+func VerifyProof(rootHash common.Hash, key []byte, proofDb DatabaseReader) (value []byte, nodes int, err error) {
 	key = keybytesToHex(key)
 	wantHash := rootHash
 	for i := 0; ; i++ {
 		buf, _ := proofDb.Get(wantHash[:])
 		if buf == nil {
-			return nil, fmt.Errorf("proof node %d (hash %064x) missing", i, wantHash), i
+			return nil, i, fmt.Errorf("proof node %d (hash %064x) missing", i, wantHash)
 		}
 		n, err := decodeNode(wantHash[:], buf, 0)
 		if err != nil {
-			return nil, fmt.Errorf("bad proof node %d: %v", i, err), i
+			return nil, i, fmt.Errorf("bad proof node %d: %v", i, err)
 		}
 		keyrest, cld := get(n, key)
 		switch cld := cld.(type) {
 		case nil:
 			// The trie doesn't contain the key.
-			return nil, nil, i
+			return nil, i, nil
 		case hashNode:
 			key = keyrest
 			copy(wantHash[:], cld)
 		case valueNode:
-			return cld, nil, i + 1
+			return cld, i + 1, nil
 		}
 	}
 }

trie/proof_test.go

@@ -32,20 +32,46 @@ func init() {
 	mrand.Seed(time.Now().Unix())
 }
 
+// makeProvers creates Merkle trie provers based on different implementations to
+// test all variations.
+func makeProvers(trie *Trie) []func(key []byte) *ethdb.MemDatabase {
+	var provers []func(key []byte) *ethdb.MemDatabase
+
+	// Create a direct trie based Merkle prover
+	provers = append(provers, func(key []byte) *ethdb.MemDatabase {
+		proof := ethdb.NewMemDatabase()
+		trie.Prove(key, 0, proof)
+		return proof
+	})
+	// Create a leaf iterator based Merkle prover
+	provers = append(provers, func(key []byte) *ethdb.MemDatabase {
+		proof := ethdb.NewMemDatabase()
+		if it := NewIterator(trie.NodeIterator(key)); it.Next() && bytes.Equal(key, it.Key) {
+			for _, p := range it.Prove() {
+				proof.Put(crypto.Keccak256(p), p)
+			}
+		}
+		return proof
+	})
+	return provers
+}
+
 func TestProof(t *testing.T) {
 	trie, vals := randomTrie(500)
 	root := trie.Hash()
-	for _, kv := range vals {
-		proofs := ethdb.NewMemDatabase()
-		if trie.Prove(kv.k, 0, proofs) != nil {
-			t.Fatalf("missing key %x while constructing proof", kv.k)
-		}
-		val, err, _ := VerifyProof(root, kv.k, proofs)
-		if err != nil {
-			t.Fatalf("VerifyProof error for key %x: %v\nraw proof: %v", kv.k, err, proofs)
-		}
-		if !bytes.Equal(val, kv.v) {
-			t.Fatalf("VerifyProof returned wrong value for key %x: got %x, want %x", kv.k, val, kv.v)
+	for i, prover := range makeProvers(trie) {
+		for _, kv := range vals {
+			proof := prover(kv.k)
+			if proof == nil {
+				t.Fatalf("prover %d: missing key %x while constructing proof", i, kv.k)
+			}
+			val, _, err := VerifyProof(root, kv.k, proof)
+			if err != nil {
+				t.Fatalf("prover %d: failed to verify proof for key %x: %v\nraw proof: %x", i, kv.k, err, proof)
+			}
+			if !bytes.Equal(val, kv.v) {
+				t.Fatalf("prover %d: verified value mismatch for key %x: have %x, want %x", i, kv.k, val, kv.v)
+			}
 		}
 	}
 }
@@ -53,37 +79,66 @@ func TestProof(t *testing.T) {
 func TestOneElementProof(t *testing.T) {
 	trie := new(Trie)
 	updateString(trie, "k", "v")
-	proofs := ethdb.NewMemDatabase()
-	trie.Prove([]byte("k"), 0, proofs)
-	if len(proofs.Keys()) != 1 {
-		t.Error("proof should have one element")
-	}
-	val, err, _ := VerifyProof(trie.Hash(), []byte("k"), proofs)
-	if err != nil {
-		t.Fatalf("VerifyProof error: %v\nproof hashes: %v", err, proofs.Keys())
-	}
-	if !bytes.Equal(val, []byte("v")) {
-		t.Fatalf("VerifyProof returned wrong value: got %x, want 'k'", val)
+	for i, prover := range makeProvers(trie) {
+		proof := prover([]byte("k"))
+		if proof == nil {
+			t.Fatalf("prover %d: nil proof", i)
+		}
+		if proof.Len() != 1 {
+			t.Errorf("prover %d: proof should have one element", i)
+		}
+		val, _, err := VerifyProof(trie.Hash(), []byte("k"), proof)
+		if err != nil {
+			t.Fatalf("prover %d: failed to verify proof: %v\nraw proof: %x", i, err, proof)
+		}
+		if !bytes.Equal(val, []byte("v")) {
+			t.Fatalf("prover %d: verified value mismatch: have %x, want 'k'", i, val)
+		}
 	}
 }
 
-func TestVerifyBadProof(t *testing.T) {
+func TestBadProof(t *testing.T) {
 	trie, vals := randomTrie(800)
 	root := trie.Hash()
-	for _, kv := range vals {
-		proofs := ethdb.NewMemDatabase()
-		trie.Prove(kv.k, 0, proofs)
-		if len(proofs.Keys()) == 0 {
-			t.Fatal("zero length proof")
+	for i, prover := range makeProvers(trie) {
+		for _, kv := range vals {
+			proof := prover(kv.k)
+			if proof == nil {
+				t.Fatalf("prover %d: nil proof", i)
+			}
+			key := proof.Keys()[mrand.Intn(proof.Len())]
+			val, _ := proof.Get(key)
+			proof.Delete(key)
+
+			mutateByte(val)
+			proof.Put(crypto.Keccak256(val), val)
+
+			if _, _, err := VerifyProof(root, kv.k, proof); err == nil {
+				t.Fatalf("prover %d: expected proof to fail for key %x", i, kv.k)
+			}
 		}
-		keys := proofs.Keys()
-		key := keys[mrand.Intn(len(keys))]
-		node, _ := proofs.Get(key)
-		proofs.Delete(key)
-		mutateByte(node)
-		proofs.Put(crypto.Keccak256(node), node)
-		if _, err, _ := VerifyProof(root, kv.k, proofs); err == nil {
-			t.Fatalf("expected proof to fail for key %x", kv.k)
+	}
+}
+
+// Tests that missing keys can also be proven. The test explicitly uses a single
+// entry trie and checks for missing keys both before and after the single entry.
+func TestMissingKeyProof(t *testing.T) {
+	trie := new(Trie)
+	updateString(trie, "k", "v")
+
+	for i, key := range []string{"a", "j", "l", "z"} {
+		proof := ethdb.NewMemDatabase()
+		trie.Prove([]byte(key), 0, proof)
+
+		if proof.Len() != 1 {
+			t.Errorf("test %d: proof should have one element", i)
+		}
+		val, _, err := VerifyProof(trie.Hash(), []byte(key), proof)
+		if err != nil {
+			t.Fatalf("test %d: failed to verify proof: %v\nraw proof: %x", i, err, proof)
+		}
+		if val != nil {
+			t.Fatalf("test %d: verified value mismatch: have %x, want nil", i, val)
 		}
 	}
 }
@@ -131,7 +186,7 @@ func BenchmarkVerifyProof(b *testing.B) {
 	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
 		im := i % len(keys)
-		if _, err, _ := VerifyProof(root, []byte(keys[im]), proofs[im]); err != nil {
+		if _, _, err := VerifyProof(root, []byte(keys[im]), proofs[im]); err != nil {
 			b.Fatalf("key %x: %v", keys[im], err)
 		}
 	}

trie/secure_trie.go

@@ -155,14 +155,19 @@ func (t *SecureTrie) Commit(onleaf LeafCallback) (root common.Hash, err error) {
 	return t.trie.Commit(onleaf)
 }
 
+// Hash returns the root hash of SecureTrie. It does not write to the
+// database and can be used even if the trie doesn't have one.
 func (t *SecureTrie) Hash() common.Hash {
 	return t.trie.Hash()
 }
 
+// Root returns the root hash of SecureTrie.
+// Deprecated: use Hash instead.
 func (t *SecureTrie) Root() []byte {
 	return t.trie.Root()
 }
 
+// Copy returns a copy of SecureTrie.
 func (t *SecureTrie) Copy() *SecureTrie {
 	cpy := *t
 	return &cpy

trie/trie.go

@@ -101,7 +101,7 @@ func New(root common.Hash, db *Database) (*Trie, error) {
 		db:           db,
 		originalRoot: root,
 	}
-	if (root != common.Hash{}) && root != emptyRoot {
+	if root != (common.Hash{}) && root != emptyRoot {
 		rootnode, err := trie.resolveHash(root[:], nil)
 		if err != nil {
 			return nil, err

vendor/github.com/syndtr/goleveldb/leveldb/db.go

@@ -35,6 +35,7 @@ type DB struct {
 	// Stats. Need 64-bit alignment.
 	cWriteDelay            int64 // The cumulative duration of write delays
 	cWriteDelayN           int32 // The cumulative number of write delays
+	inWritePaused          int32 // The indicator whether write operation is paused by compaction
 	aliveSnaps, aliveIters int32
 
 	// Session.
@@ -967,7 +968,8 @@ func (db *DB) GetProperty(name string) (value string, err error) {
 			float64(db.s.stor.writes())/1048576.0)
 	case p == "writedelay":
 		writeDelayN, writeDelay := atomic.LoadInt32(&db.cWriteDelayN), time.Duration(atomic.LoadInt64(&db.cWriteDelay))
-		value = fmt.Sprintf("DelayN:%d Delay:%s", writeDelayN, writeDelay)
+		paused := atomic.LoadInt32(&db.inWritePaused) == 1
+		value = fmt.Sprintf("DelayN:%d Delay:%s Paused:%t", writeDelayN, writeDelay, paused)
 	case p == "sstables":
 		for level, tables := range v.levels {
 			value += fmt.Sprintf("--- level %d ---\n", level)
@@ -996,6 +998,75 @@ func (db *DB) GetProperty(name string) (value string, err error) {
 	return
 }
 
+// DBStats is database statistics.
+type DBStats struct {
+	WriteDelayCount    int32
+	WriteDelayDuration time.Duration
+	WritePaused        bool
+
+	AliveSnapshots int32
+	AliveIterators int32
+
+	IOWrite uint64
+	IORead  uint64
+
+	BlockCacheSize    int
+	OpenedTablesCount int
+
+	LevelSizes        []int64
+	LevelTablesCounts []int
+	LevelRead         []int64
+	LevelWrite        []int64
+	LevelDurations    []time.Duration
+}
+
+// Stats populates s with database statistics.
+func (db *DB) Stats(s *DBStats) error {
+	err := db.ok()
+	if err != nil {
+		return err
+	}
+
+	s.IORead = db.s.stor.reads()
+	s.IOWrite = db.s.stor.writes()
+	s.WriteDelayCount = atomic.LoadInt32(&db.cWriteDelayN)
+	s.WriteDelayDuration = time.Duration(atomic.LoadInt64(&db.cWriteDelay))
+	s.WritePaused = atomic.LoadInt32(&db.inWritePaused) == 1
+
+	s.OpenedTablesCount = db.s.tops.cache.Size()
+	if db.s.tops.bcache != nil {
+		s.BlockCacheSize = db.s.tops.bcache.Size()
+	} else {
+		s.BlockCacheSize = 0
+	}
+
+	s.AliveIterators = atomic.LoadInt32(&db.aliveIters)
+	s.AliveSnapshots = atomic.LoadInt32(&db.aliveSnaps)
+
+	s.LevelDurations = s.LevelDurations[:0]
+	s.LevelRead = s.LevelRead[:0]
+	s.LevelWrite = s.LevelWrite[:0]
+	s.LevelSizes = s.LevelSizes[:0]
+	s.LevelTablesCounts = s.LevelTablesCounts[:0]
+
+	v := db.s.version()
+	defer v.release()
+
+	for level, tables := range v.levels {
+		duration, read, write := db.compStats.getStat(level)
+		if len(tables) == 0 && duration == 0 {
+			continue
+		}
+		s.LevelDurations = append(s.LevelDurations, duration)
+		s.LevelRead = append(s.LevelRead, read)
+		s.LevelWrite = append(s.LevelWrite, write)
+		s.LevelSizes = append(s.LevelSizes, tables.size())
+		s.LevelTablesCounts = append(s.LevelTablesCounts, len(tables))
+	}
+
+	return nil
+}
+
 // SizeOf calculates approximate sizes of the given key ranges.
 // The length of the returned sizes are equal with the length of the given
 // ranges. The returned sizes measure storage space usage, so if the user

vendor/github.com/syndtr/goleveldb/leveldb/db_write.go

@@ -89,7 +89,11 @@ func (db *DB) flush(n int) (mdb *memDB, mdbFree int, err error) {
 			return false
 		case tLen >= pauseTrigger:
 			delayed = true
+			// Set the write paused flag explicitly.
+			atomic.StoreInt32(&db.inWritePaused, 1)
 			err = db.compTriggerWait(db.tcompCmdC)
+			// Unset the write paused flag.
+			atomic.StoreInt32(&db.inWritePaused, 0)
 			if err != nil {
 				return false
 			}

vendor/vendor.json

@@ -418,10 +418,10 @@
 			"revisionTime": "2017-07-05T02:17:15Z"
 		},
 		{
-			"checksumSHA1": "k13cCuMJO7+KhR8ZXx5oUqDKGQA=",
+			"checksumSHA1": "TJV50D0q8E3vtc90ibC+qOYdjrw=",
 			"path": "github.com/syndtr/goleveldb/leveldb",
-			"revision": "ae970a0732be3a1f5311da86118d37b9f4bd2a5a",
-			"revisionTime": "2018-05-02T07:23:49Z"
+			"revision": "59047f74db0d042c8d8dd8e30bb030bc774a7d7a",
+			"revisionTime": "2018-05-21T04:45:49Z"
 		},
 		{
 			"checksumSHA1": "EKIow7XkgNdWvR/982ffIZxKG8Y=",