forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-27 15:16:43 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

Merge branch 'master' into master

Browse source
This commit is contained in:
Jeffrey Wilcke 2017-05-22 09:00:47 +02:00 committed by GitHub
commit 7b6f14158f
40 changed files with 1311 additions and 294 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
README.md
View file

@ -84,7 +84,7 @@ $ geth --testnet --fast --cache=512 console
```
The `--fast`, `--cache` flags and `console` subcommand have the exact same meaning as above and they
are equially useful on the testnet too. Please see above for their explanations if you've skipped to
are equally useful on the testnet too. Please see above for their explanations if you've skipped to
here.
Specifying the `--testnet` flag however will reconfigure your Geth instance a bit:

2
cmd/geth/main.go
View file

@ -55,6 +55,8 @@ var (
utils.UnlockedAccountFlag,
utils.PasswordFileFlag,
utils.BootnodesFlag,
utils.BootnodesV4Flag,
utils.BootnodesV5Flag,
utils.DataDirFlag,
utils.KeyStoreDirFlag,
utils.NoUSBFlag,

2
cmd/geth/usage.go
View file

@ -129,6 +129,8 @@ var AppHelpFlagGroups = []flagGroup{
Name: "NETWORKING",
Flags: []cli.Flag{
utils.BootnodesFlag,
utils.BootnodesV4Flag,
utils.BootnodesV5Flag,
utils.ListenPortFlag,
utils.MaxPeersFlag,
utils.MaxPendingPeersFlag,

25
cmd/puppeth/module_node.go
View file

@ -40,7 +40,7 @@ ADD genesis.json /genesis.json
RUN \
echo '/geth init /genesis.json' > geth.sh && \{{if .Unlock}}
echo 'mkdir -p /root/.ethereum/keystore/ && cp /signer.json /root/.ethereum/keystore/' >> geth.sh && \{{end}}
echo $'/geth --networkid {{.NetworkID}} --cache 512 --port {{.Port}} --maxpeers {{.Peers}} {{.LightFlag}} --ethstats \'{{.Ethstats}}\' {{if .Bootnodes}}--bootnodes {{.Bootnodes}}{{end}} {{if .Etherbase}}--etherbase {{.Etherbase}} --mine{{end}}{{if .Unlock}}--unlock 0 --password /signer.pass --mine{{end}}' >> geth.sh
echo $'/geth --networkid {{.NetworkID}} --cache 512 --port {{.Port}} --maxpeers {{.Peers}} {{.LightFlag}} --ethstats \'{{.Ethstats}}\' {{if .BootV4}}--bootnodesv4 {{.BootV4}}{{end}} {{if .BootV5}}--bootnodesv5 {{.BootV5}}{{end}} {{if .Etherbase}}--etherbase {{.Etherbase}} --mine{{end}}{{if .Unlock}}--unlock 0 --password /signer.pass --mine{{end}} --targetgaslimit {{.GasTarget}} --gasprice {{.GasPrice}}' >> geth.sh
ENTRYPOINT ["/bin/sh", "geth.sh"]
`
@ -66,17 +66,20 @@ services:
- LIGHT_PEERS={{.LightPeers}}
- STATS_NAME={{.Ethstats}}
- MINER_NAME={{.Etherbase}}
- GAS_TARGET={{.GasTarget}}
- GAS_PRICE={{.GasPrice}}
restart: always
`
// deployNode deploys a new Ethereum node container to a remote machine via SSH,
// docker and docker-compose. If an instance with the specified network name
// already exists there, it will be overwritten!
func deployNode(client *sshClient, network string, bootnodes []string, config *nodeInfos) ([]byte, error) {
func deployNode(client *sshClient, network string, bootv4, bootv5 []string, config *nodeInfos) ([]byte, error) {
kind := "sealnode"
if config.keyJSON == "" && config.etherbase == "" {
kind = "bootnode"
bootnodes = make([]string, 0)
bootv4 = make([]string, 0)
bootv5 = make([]string, 0)
}
// Generate the content to upload to the server
workdir := fmt.Sprintf("%d", rand.Int63())
@ -92,9 +95,12 @@ func deployNode(client *sshClient, network string, bootnodes []string, config *n
"Port": config.portFull,
"Peers": config.peersTotal,
"LightFlag": lightFlag,
"Bootnodes": strings.Join(bootnodes, ","),
"BootV4": strings.Join(bootv4, ","),
"BootV5": strings.Join(bootv5, ","),
"Ethstats": config.ethstats,
"Etherbase": config.etherbase,
"GasTarget": uint64(1000000 * config.gasTarget),
"GasPrice": uint64(1000000000 * config.gasPrice),
"Unlock": config.keyJSON != "",
})
files[filepath.Join(workdir, "Dockerfile")] = dockerfile.Bytes()
@ -111,6 +117,8 @@ func deployNode(client *sshClient, network string, bootnodes []string, config *n
"LightPeers": config.peersLight,
"Ethstats": config.ethstats[:strings.Index(config.ethstats, ":")],
"Etherbase": config.etherbase,
"GasTarget": config.gasTarget,
"GasPrice": config.gasPrice,
})
files[filepath.Join(workdir, "docker-compose.yaml")] = composefile.Bytes()
@ -147,6 +155,8 @@ type nodeInfos struct {
etherbase string
keyJSON string
keyPass string
gasTarget float64
gasPrice float64
}
// String implements the stringer interface.
@ -155,7 +165,8 @@ func (info *nodeInfos) String() string {
if info.peersLight > 0 {
discv5 = fmt.Sprintf(", portv5=%d", info.portLight)
}
return fmt.Sprintf("port=%d%s, datadir=%s, peers=%d, lights=%d, ethstats=%s", info.portFull, discv5, info.datadir, info.peersTotal, info.peersLight, info.ethstats)
return fmt.Sprintf("port=%d%s, datadir=%s, peers=%d, lights=%d, ethstats=%s, gastarget=%0.3f MGas, gasprice=%0.3f GWei",
info.portFull, discv5, info.datadir, info.peersTotal, info.peersLight, info.ethstats, info.gasTarget, info.gasPrice)
}
// checkNode does a health-check against an boot or seal node server to verify
@ -176,6 +187,8 @@ func checkNode(client *sshClient, network string, boot bool) (*nodeInfos, error)
// Resolve a few types from the environmental variables
totalPeers, _ := strconv.Atoi(infos.envvars["TOTAL_PEERS"])
lightPeers, _ := strconv.Atoi(infos.envvars["LIGHT_PEERS"])
gasTarget, _ := strconv.ParseFloat(infos.envvars["GAS_TARGET"], 64)
gasPrice, _ := strconv.ParseFloat(infos.envvars["GAS_PRICE"], 64)
// Container available, retrieve its node ID and its genesis json
var out []byte
@ -213,6 +226,8 @@ func checkNode(client *sshClient, network string, boot bool) (*nodeInfos, error)
etherbase: infos.envvars["MINER_NAME"],
keyJSON: keyJSON,
keyPass: keyPass,
gasTarget: gasTarget,
gasPrice: gasPrice,
}
stats.enodeFull = fmt.Sprintf("enode://%s@%s:%d", id, client.address, stats.portFull)
if stats.portLight != 0 {

42
cmd/puppeth/wizard.go
View file

@ -162,6 +162,48 @@ func (w *wizard) readDefaultInt(def int) int {
}
}
// readFloat reads a single line from stdin, trimming if from spaces, enforcing it
// to parse into a float.
func (w *wizard) readFloat() float64 {
for {
fmt.Printf("> ")
text, err := w.in.ReadString('\n')
if err != nil {
log.Crit("Failed to read user input", "err", err)
}
if text = strings.TrimSpace(text); text == "" {
continue
}
val, err := strconv.ParseFloat(strings.TrimSpace(text), 64)
if err != nil {
log.Error("Invalid input, expected float", "err", err)
continue
}
return val
}
}
// readDefaultFloat reads a single line from stdin, trimming if from spaces, enforcing
// it to parse into a float. If an empty line is entered, the default value is returned.
func (w *wizard) readDefaultFloat(def float64) float64 {
for {
fmt.Printf("> ")
text, err := w.in.ReadString('\n')
if err != nil {
log.Crit("Failed to read user input", "err", err)
}
if text = strings.TrimSpace(text); text == "" {
return def
}
val, err := strconv.ParseFloat(strings.TrimSpace(text), 64)
if err != nil {
log.Error("Invalid input, expected float", "err", err)
continue
}
return val
}
}
// readPassword reads a single line from stdin, trimming it from the trailing new
// line and returns it. The input will not be echoed.
func (w *wizard) readPassword() string {

2
cmd/puppeth/wizard_netstats.go
View file

@ -39,7 +39,7 @@ func (w *wizard) networkStats(tips bool) {
// Iterate over all the specified hosts and check their status
stats := tablewriter.NewWriter(os.Stdout)
stats.SetHeader([]string{"Server", "IP", "Status", "Service", "Details"})
stats.SetColWidth(128)
stats.SetColWidth(100)
for server, pubkey := range w.conf.Servers {
client := w.servers[server]

12
cmd/puppeth/wizard_node.go
View file

@ -50,7 +50,7 @@ func (w *wizard) deployNode(boot bool) {
if boot {
infos = &nodeInfos{portFull: 30303, peersTotal: 512, peersLight: 256}
} else {
infos = &nodeInfos{portFull: 30303, peersTotal: 50, peersLight: 0}
infos = &nodeInfos{portFull: 30303, peersTotal: 50, peersLight: 0, gasTarget: 4.7, gasPrice: 18}
}
}
infos.genesis, _ = json.MarshalIndent(w.conf.genesis, "", " ")
@ -136,9 +136,17 @@ func (w *wizard) deployNode(boot bool) {
}
}
}
// Establish the gas dynamics to be enforced by the signer
fmt.Println()
fmt.Printf("What gas limit should empty blocks target (MGas)? (default = %0.3f)\n", infos.gasTarget)
infos.gasTarget = w.readDefaultFloat(infos.gasTarget)
fmt.Println()
fmt.Printf("What gas price should the signer require (GWei)? (default = %0.3f)\n", infos.gasPrice)
infos.gasPrice = w.readDefaultFloat(infos.gasPrice)
}
// Try to deploy the full node on the host
if out, err := deployNode(client, w.network, w.conf.bootFull, infos); err != nil {
if out, err := deployNode(client, w.network, w.conf.bootFull, w.conf.bootLight, infos); err != nil {
log.Error("Failed to deploy Ethereum node container", "err", err)
if len(out) > 0 {
fmt.Printf("%s\n", out)

8
cmd/swarm/main.go
View file

@ -67,6 +67,10 @@ var (
Name: "bzzaccount",
Usage: "Swarm account key file",
}
SwarmListenAddrFlag = cli.StringFlag{
Name: "httpaddr",
Usage: "Swarm HTTP API listening interface",
}
SwarmPortFlag = cli.StringFlag{
Name: "bzzport",
Usage: "Swarm local http api port",
@ -249,6 +253,7 @@ Cleans database of corrupted entries.
SwarmConfigPathFlag,
SwarmSwapEnabledFlag,
SwarmSyncEnabledFlag,
SwarmListenAddrFlag,
SwarmPortFlag,
SwarmAccountFlag,
SwarmNetworkIdFlag,
@ -345,6 +350,9 @@ func registerBzzService(ctx *cli.Context, stack *node.Node) {
if len(bzzport) > 0 {
bzzconfig.Port = bzzport
}
if bzzaddr := ctx.GlobalString(SwarmListenAddrFlag.Name); bzzaddr != "" {
bzzconfig.ListenAddr = bzzaddr
}
swapEnabled := ctx.GlobalBool(SwarmSwapEnabledFlag.Name)
syncEnabled := ctx.GlobalBoolT(SwarmSyncEnabledFlag.Name)

27
cmd/utils/flags.go
View file

@ -237,7 +237,7 @@ var (
GasPriceFlag = BigFlag{
Name: "gasprice",
Usage: "Minimal gas price to accept for mining a transactions",
Value: big.NewInt(20 * params.Shannon),
Value: eth.DefaultConfig.GasPrice,
}
ExtraDataFlag = cli.StringFlag{
Name: "extradata",
@ -360,7 +360,17 @@ var (
}
BootnodesFlag = cli.StringFlag{
Name: "bootnodes",
Usage: "Comma separated enode URLs for P2P discovery bootstrap",
Usage: "Comma separated enode URLs for P2P discovery bootstrap (set v4+v5 instead for light servers)",
Value: "",
}
BootnodesV4Flag = cli.StringFlag{
Name: "bootnodesv4",
Usage: "Comma separated enode URLs for P2P v4 discovery bootstrap (light server, full nodes)",
Value: "",
}
BootnodesV5Flag = cli.StringFlag{
Name: "bootnodesv5",
Usage: "Comma separated enode URLs for P2P v5 discovery bootstrap (light server, light nodes)",
Value: "",
}
NodeKeyFileFlag = cli.StringFlag{
@ -469,8 +479,12 @@ func setNodeUserIdent(ctx *cli.Context, cfg *node.Config) {
func setBootstrapNodes(ctx *cli.Context, cfg *p2p.Config) {
urls := params.MainnetBootnodes
switch {
case ctx.GlobalIsSet(BootnodesFlag.Name):
case ctx.GlobalIsSet(BootnodesFlag.Name) || ctx.GlobalIsSet(BootnodesV4Flag.Name):
if ctx.GlobalIsSet(BootnodesV4Flag.Name) {
urls = strings.Split(ctx.GlobalString(BootnodesV4Flag.Name), ",")
} else {
urls = strings.Split(ctx.GlobalString(BootnodesFlag.Name), ",")
}
case ctx.GlobalBool(TestnetFlag.Name):
urls = params.TestnetBootnodes
case ctx.GlobalBool(RinkebyFlag.Name):
@ -493,8 +507,12 @@ func setBootstrapNodes(ctx *cli.Context, cfg *p2p.Config) {
func setBootstrapNodesV5(ctx *cli.Context, cfg *p2p.Config) {
urls := params.DiscoveryV5Bootnodes
switch {
case ctx.GlobalIsSet(BootnodesFlag.Name):
case ctx.GlobalIsSet(BootnodesFlag.Name) || ctx.GlobalIsSet(BootnodesV5Flag.Name):
if ctx.GlobalIsSet(BootnodesV5Flag.Name) {
urls = strings.Split(ctx.GlobalString(BootnodesV5Flag.Name), ",")
} else {
urls = strings.Split(ctx.GlobalString(BootnodesFlag.Name), ",")
}
case ctx.GlobalBool(RinkebyFlag.Name):
urls = params.RinkebyV5Bootnodes
case cfg.BootstrapNodesV5 != nil:
@ -717,6 +735,7 @@ func SetP2PConfig(ctx *cli.Context, cfg *p2p.Config) {
// --dev mode can't use p2p networking.
cfg.MaxPeers = 0
cfg.ListenAddr = ":0"
cfg.DiscoveryV5Addr = ":0"
cfg.NoDiscovery = true
cfg.DiscoveryV5 = false
}

1
containers/vagrant/.gitignore vendored Normal file
View file

@ -0,0 +1 @@
.vagrant

51
containers/vagrant/Vagrantfile vendored
View file

@ -1,29 +1,38 @@
# -*- mode: ruby -*-
# vi: set ft=ruby :
Vagrant.configure(2) do |config|
config.vm.box = "ubuntu/trusty64"
require 'yaml'
config.vm.provider "virtualbox" do |vb|
vb.memory = "2048"
VAGRANTFILE_API_VERSION = 2
VM_RAM = 2048
Vagrant.configure(VAGRANTFILE_API_VERSION) do |config|
config.vm.define "ubuntu", :primary => true do |ubuntu|
ubuntu.vm.box = "ubuntu/trusty64"
ubuntu.vm.provision "shell", :path => "provisioners/shell/ubuntu.sh"
end
config.vm.define "debian", :primary => true do |debian|
debian.vm.box = "debian/jessie64"
debian.vm.provision "shell", :path => "provisioners/shell/debian.sh"
end
config.vm.define "centos", :autostart => false do |centos|
centos.vm.box = "centos/7"
centos.vm.provision "shell", :path => "provisioners/shell/centos.sh"
end
config.vm.provider "virtualbox" do |vb|
vb.memory = VM_RAM
end
config.vm.provider "libvirt" do |lv|
lv.memory = VM_RAM
config.vm.synced_folder ".", "/home/vagrant/sync", :disabled => true
end
config.vm.synced_folder ".", "/vagrant", :disabled => true
config.vm.synced_folder "../../", "/home/vagrant/go/src/github.com/ethereum/go-ethereum"
config.vm.synced_folder ".", "/vagrant", disabled: true
config.vm.provision "shell", inline: <<-SHELL
sudo apt-get install software-properties-common
sudo add-apt-repository -y ppa:ethereum/ethereum
sudo add-apt-repository -y ppa:ethereum/ethereum-dev
sudo apt-get update
sudo apt-get install -y build-essential golang git-all
GOPATH=/home/vagrant/go go get github.com/tools/godep
sudo chown -R vagrant:vagrant ~vagrant/go
echo "export GOPATH=/home/vagrant/go" >> ~vagrant/.bashrc
echo "export PATH=\\\$PATH:\\\$GOPATH/bin:/usr/local/go/bin" >> ~vagrant/.bashrc
SHELL
end

11
containers/vagrant/provisioners/shell/centos.sh Executable file
View file

@ -0,0 +1,11 @@
#!/bin/bash
sudo yum install -y git wget
sudo yum update -y
wget --continue https://storage.googleapis.com/golang/go1.8.1.linux-amd64.tar.gz
sudo tar -C /usr/local -xzf go1.8.1.linux-amd64.tar.gz
GETH_PATH="~vagrant/go/src/github.com/ethereum/go-ethereum/build/bin/"
echo "export PATH=$PATH:/usr/local/go/bin:$GETH_PATH" >> ~vagrant/.bashrc

11
containers/vagrant/provisioners/shell/debian.sh Executable file
View file

@ -0,0 +1,11 @@
#!/bin/bash
sudo apt-get install -y build-essential git-all wget
sudo apt-get update
wget --continue https://storage.googleapis.com/golang/go1.8.1.linux-amd64.tar.gz
sudo tar -C /usr/local -xzf go1.8.1.linux-amd64.tar.gz
GETH_PATH="~vagrant/go/src/github.com/ethereum/go-ethereum/build/bin/"
echo "export PATH=$PATH:/usr/local/go/bin:$GETH_PATH" >> ~vagrant/.bashrc

11
containers/vagrant/provisioners/shell/ubuntu.sh Executable file
View file

@ -0,0 +1,11 @@
#!/bin/bash
sudo apt-get install -y build-essential git-all wget
sudo apt-get update
wget --continue https://storage.googleapis.com/golang/go1.8.1.linux-amd64.tar.gz
sudo tar -C /usr/local -xzf go1.8.1.linux-amd64.tar.gz
GETH_PATH="~vagrant/go/src/github.com/ethereum/go-ethereum/build/bin/"
echo "export PATH=$PATH:/usr/local/go/bin:$GETH_PATH" >> ~vagrant/.bashrc

4
core/events.go
View file

@ -17,8 +17,6 @@
package core
import (
"math/big"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/types"
)
@ -67,8 +65,6 @@ type ChainUncleEvent struct {
type ChainHeadEvent struct{ Block *types.Block }
type GasPriceChanged struct{ Price *big.Int }
// Mining operation events
type StartMining struct{}
type TopMining struct{}

164
core/tx_list.go
View file

@ -22,7 +22,9 @@ import (
"math/big"
"sort"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/log"
)
// nonceHeap is a heap.Interface implementation over 64bit unsigned integers for
@ -53,11 +55,11 @@ type txSortedMap struct {
cache types.Transactions // Cache of the transactions already sorted
}
// newTxSortedMap creates a new sorted transaction map.
// newTxSortedMap creates a new nonce-sorted transaction map.
func newTxSortedMap() *txSortedMap {
return &txSortedMap{
items: make(map[uint64]*types.Transaction),
index: &nonceHeap{},
index: new(nonceHeap),
}
}
@ -233,6 +235,12 @@ func newTxList(strict bool) *txList {
}
}
// Overlaps returns whether the transaction specified has the same nonce as one
// already contained within the list.
func (l *txList) Overlaps(tx *types.Transaction) bool {
return l.txs.Get(tx.Nonce()) != nil
}
// Add tries to insert a new transaction into the list, returning whether the
// transaction was accepted, and if yes, any previous transaction it replaced.
//
@ -241,9 +249,12 @@ func newTxList(strict bool) *txList {
func (l *txList) Add(tx *types.Transaction) (bool, *types.Transaction) {
// If there's an older better transaction, abort
old := l.txs.Get(tx.Nonce())
if old != nil && old.GasPrice().Cmp(tx.GasPrice()) >= 0 {
if old != nil {
threshold := new(big.Int).Div(new(big.Int).Mul(old.GasPrice(), big.NewInt(100+minPriceBumpPercent)), big.NewInt(100))
if threshold.Cmp(tx.GasPrice()) >= 0 {
return false, nil
}
}
// Otherwise overwrite the old transaction with the current one
l.txs.Put(tx)
if cost := tx.Cost(); l.costcap.Cmp(cost) < 0 {
@ -340,3 +351,150 @@ func (l *txList) Empty() bool {
func (l *txList) Flatten() types.Transactions {
return l.txs.Flatten()
}
// priceHeap is a heap.Interface implementation over transactions for retrieving
// price-sorted transactions to discard when the pool fills up.
type priceHeap []*types.Transaction
func (h priceHeap) Len() int { return len(h) }
func (h priceHeap) Less(i, j int) bool { return h[i].GasPrice().Cmp(h[j].GasPrice()) < 0 }
func (h priceHeap) Swap(i, j int) { h[i], h[j] = h[j], h[i] }
func (h *priceHeap) Push(x interface{}) {
*h = append(*h, x.(*types.Transaction))
}
func (h *priceHeap) Pop() interface{} {
old := *h
n := len(old)
x := old[n-1]
*h = old[0 : n-1]
return x
}
// 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)
}
// newTxPricedList creates a new price-sorted transaction heap.
func newTxPricedList(all *map[common.Hash]*types.Transaction) *txPricedList {
return &txPricedList{
all: all,
items: new(priceHeap),
}
}
// Put inserts a new transaction into the heap.
func (l *txPricedList) Put(tx *types.Transaction) {
heap.Push(l.items, tx)
}
// Removed notifies the prices transaction list that an old transaction dropped
// from the pool. The list will just keep a counter of stale objects and update
// the heap if a large enough ratio of transactions go stale.
func (l *txPricedList) Removed() {
// Bump the stale counter, but exit if still too low (< 25%)
l.stales++
if l.stales <= len(*l.items)/4 {
return
}
// Seems we've reached a critical number of stale transactions, reheap
reheap := make(priceHeap, 0, len(*l.all))
l.stales, l.items = 0, &reheap
for _, tx := range *l.all {
*l.items = append(*l.items, tx)
}
heap.Init(l.items)
}
// Discard finds all the transactions below the given price threshold, drops them
// from the priced list and returs them for further removal from the entire pool.
func (l *txPricedList) Cap(threshold *big.Int, local *txSet) types.Transactions {
drop := make(types.Transactions, 0, 128) // Remote underpriced transactions to drop
save := make(types.Transactions, 0, 64) // Local underpriced transactions to keep
for len(*l.items) > 0 {
// Discard stale transactions if found during cleanup
tx := heap.Pop(l.items).(*types.Transaction)
hash := tx.Hash()
if _, ok := (*l.all)[hash]; !ok {
l.stales--
continue
}
// Stop the discards if we've reached the threshold
if tx.GasPrice().Cmp(threshold) >= 0 {
break
}
// Non stale transaction found, discard unless local
if local.contains(hash) {
save = append(save, tx)
} else {
drop = append(drop, tx)
}
}
for _, tx := range save {
heap.Push(l.items, tx)
}
return drop
}
// Underpriced checks whether a transaction is cheaper than (or as cheap as) the
// lowest priced transaction currently being tracked.
func (l *txPricedList) Underpriced(tx *types.Transaction, local *txSet) bool {
// Local transactions cannot be underpriced
if local.contains(tx.Hash()) {
return false
}
// 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 {
l.stales--
heap.Pop(l.items)
continue
}
break
}
// Check if the transaction is underpriced or not
if len(*l.items) == 0 {
log.Error("Pricing query for empty pool") // This cannot happen, print to catch programming errors
return false
}
cheapest := []*types.Transaction(*l.items)[0]
return cheapest.GasPrice().Cmp(tx.GasPrice()) >= 0
}
// Discard finds a number of most underpriced transactions, removes them from the
// priced list and returs them for further removal from the entire pool.
func (l *txPricedList) Discard(count int, local *txSet) types.Transactions {
drop := make(types.Transactions, 0, count) // Remote underpriced transactions to drop
save := make(types.Transactions, 0, 64) // Local underpriced transactions to keep
for len(*l.items) > 0 && count > 0 {
// Discard stale transactions if found during cleanup
tx := heap.Pop(l.items).(*types.Transaction)
hash := tx.Hash()
if _, ok := (*l.all)[hash]; !ok {
l.stales--
continue
}
// Non stale transaction found, discard unless local
if local.contains(hash) {
save = append(save, tx)
} else {
drop = append(drop, tx)
count--
}
}
for _, tx := range save {
heap.Push(l.items, tx)
}
return drop
}

235
core/tx_pool.go
View file

@ -36,23 +36,26 @@ import (
var (
// Transaction Pool Errors
ErrInvalidSender = errors.New("Invalid sender")
ErrNonce = errors.New("Nonce too low")
ErrCheap = errors.New("Gas price too low for acceptance")
ErrBalance = errors.New("Insufficient balance")
ErrInsufficientFunds = errors.New("Insufficient funds for gas * price + value")
ErrIntrinsicGas = errors.New("Intrinsic gas too low")
ErrGasLimit = errors.New("Exceeds block gas limit")
ErrNegativeValue = errors.New("Negative value")
ErrInvalidSender = errors.New("invalid sender")
ErrNonce = errors.New("nonce too low")
ErrUnderpriced = errors.New("transaction underpriced")
ErrReplaceUnderpriced = errors.New("replacement transaction underpriced")
ErrBalance = errors.New("insufficient balance")
ErrInsufficientFunds = errors.New("insufficient funds for gas * price + value")
ErrIntrinsicGas = errors.New("intrinsic gas too low")
ErrGasLimit = errors.New("exceeds block gas limit")
ErrNegativeValue = errors.New("negative value")
)
var (
minPendingPerAccount = uint64(16) // Min number of guaranteed transaction slots per address
maxPendingTotal = uint64(4096) // Max limit of pending transactions from all accounts (soft)
maxQueuedPerAccount = uint64(64) // Max limit of queued transactions per address
maxQueuedInTotal = uint64(1024) // Max limit of queued transactions from all accounts
maxQueuedTotal = uint64(1024) // Max limit of queued transactions from all accounts
maxQueuedLifetime = 3 * time.Hour // Max amount of time transactions from idle accounts are queued
minPriceBumpPercent = int64(10) // Minimum price bump needed to replace an old transaction
evictionInterval = time.Minute // Time interval to check for evictable transactions
statsReportInterval = 8 * time.Second // Time interval to report transaction pool stats
)
var (
@ -70,6 +73,7 @@ var (
// General tx metrics
invalidTxCounter = metrics.NewCounter("txpool/invalid")
underpricedTxCounter = metrics.NewCounter("txpool/underpriced")
)
type stateFn func() (*state.StateDB, error)
@ -86,17 +90,18 @@ type TxPool struct {
currentState stateFn // The state function which will allow us to do some pre checks
pendingState *state.ManagedState
gasLimit func() *big.Int // The current gas limit function callback
minGasPrice *big.Int
gasPrice *big.Int
eventMux *event.TypeMux
events *event.TypeMuxSubscription
localTx *txSet
locals *txSet
signer types.Signer
mu sync.RWMutex
pending map[common.Address]*txList // All currently processable transactions
queue map[common.Address]*txList // Queued but non-processable transactions
all map[common.Hash]*types.Transaction // All transactions to allow lookups
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
wg sync.WaitGroup // for shutdown sync
quit chan struct{}
@ -110,18 +115,18 @@ func NewTxPool(config *params.ChainConfig, eventMux *event.TypeMux, currentState
signer: types.NewEIP155Signer(config.ChainId),
pending: make(map[common.Address]*txList),
queue: make(map[common.Address]*txList),
all: make(map[common.Hash]*types.Transaction),
beats: make(map[common.Address]time.Time),
all: make(map[common.Hash]*types.Transaction),
eventMux: eventMux,
currentState: currentStateFn,
gasLimit: gasLimitFn,
minGasPrice: new(big.Int),
gasPrice: big.NewInt(1),
pendingState: nil,
localTx: newTxSet(),
events: eventMux.Subscribe(ChainHeadEvent{}, GasPriceChanged{}, RemovedTransactionEvent{}),
locals: newTxSet(),
events: eventMux.Subscribe(ChainHeadEvent{}, RemovedTransactionEvent{}),
quit: make(chan struct{}),
}
pool.priced = newTxPricedList(&pool.all)
pool.resetState()
pool.wg.Add(2)
@ -134,10 +139,22 @@ func NewTxPool(config *params.ChainConfig, eventMux *event.TypeMux, currentState
func (pool *TxPool) eventLoop() {
defer pool.wg.Done()
// Start a ticker and keep track of interesting pool stats to report
var prevPending, prevQueued, prevStales int
report := time.NewTicker(statsReportInterval)
defer report.Stop()
// Track chain events. When a chain events occurs (new chain canon block)
// we need to know the new state. The new state will help us determine
// the nonces in the managed state
for ev := range pool.events.Chan() {
for {
select {
// Handle any events fired by the system
case ev, ok := <-pool.events.Chan():
if !ok {
return
}
switch ev := ev.Data.(type) {
case ChainHeadEvent:
pool.mu.Lock()
@ -146,16 +163,25 @@ func (pool *TxPool) eventLoop() {
pool.homestead = true
}
}
pool.resetState()
pool.mu.Unlock()
case GasPriceChanged:
pool.mu.Lock()
pool.minGasPrice = ev.Price
pool.mu.Unlock()
case RemovedTransactionEvent:
pool.AddBatch(ev.Txs)
}
// Handle stats reporting ticks
case <-report.C:
pool.mu.RLock()
pending, queued := pool.stats()
stales := pool.priced.stales
pool.mu.RUnlock()
if pending != prevPending || queued != prevQueued || stales != prevStales {
log.Debug("Transaction pool status report", "executable", pending, "queued", queued, "stales", stales)
prevPending, prevQueued, prevStales = pending, queued, stales
}
}
}
}
@ -192,6 +218,27 @@ func (pool *TxPool) Stop() {
log.Info("Transaction pool stopped")
}
// GasPrice returns the current gas price enforced by the transaction pool.
func (pool *TxPool) GasPrice() *big.Int {
pool.mu.RLock()
defer pool.mu.RUnlock()
return new(big.Int).Set(pool.gasPrice)
}
// SetGasPrice updates the minimum price required by the transaction pool for a
// new transaction, and drops all transactions below this threshold.
func (pool *TxPool) SetGasPrice(price *big.Int) {
pool.mu.Lock()
defer pool.mu.Unlock()
pool.gasPrice = price
for _, tx := range pool.priced.Cap(price, pool.locals) {
pool.removeTx(tx.Hash())
}
log.Info("Transaction pool price threshold updated", "price", price)
}
// State returns the state of TxPool
func (pool *TxPool) State() *state.ManagedState {
pool.mu.RLock()
@ -202,17 +249,25 @@ func (pool *TxPool) State() *state.ManagedState {
// Stats retrieves the current pool stats, namely the number of pending and the
// number of queued (non-executable) transactions.
func (pool *TxPool) Stats() (pending int, queued int) {
func (pool *TxPool) Stats() (int, int) {
pool.mu.RLock()
defer pool.mu.RUnlock()
return pool.stats()
}
// stats retrieves the current pool stats, namely the number of pending and the
// number of queued (non-executable) transactions.
func (pool *TxPool) stats() (int, int) {
pending := 0
for _, list := range pool.pending {
pending += list.Len()
}
queued := 0
for _, list := range pool.queue {
queued += list.Len()
}
return
return pending, queued
}
// Content retrieves the data content of the transaction pool, returning all the
@ -262,16 +317,16 @@ func (pool *TxPool) Pending() (map[common.Address]types.Transactions, error) {
func (pool *TxPool) SetLocal(tx *types.Transaction) {
pool.mu.Lock()
defer pool.mu.Unlock()
pool.localTx.add(tx.Hash())
pool.locals.add(tx.Hash())
}
// validateTx checks whether a transaction is valid according
// to the consensus rules.
func (pool *TxPool) validateTx(tx *types.Transaction) error {
local := pool.localTx.contains(tx.Hash())
local := pool.locals.contains(tx.Hash())
// Drop transactions under our own minimal accepted gas price
if !local && pool.minGasPrice.Cmp(tx.GasPrice()) > 0 {
return ErrCheap
if !local && pool.gasPrice.Cmp(tx.GasPrice()) > 0 {
return ErrUnderpriced
}
currentState, err := pool.currentState()
@ -316,31 +371,72 @@ func (pool *TxPool) validateTx(tx *types.Transaction) error {
}
// add validates a transaction and inserts it into the non-executable queue for
// later pending promotion and execution.
func (pool *TxPool) add(tx *types.Transaction) error {
// later pending promotion and execution. If the transaction is a replacement for
// an already pending or queued one, it overwrites the previous and returns this
// so outer code doesn't uselessly call promote.
func (pool *TxPool) add(tx *types.Transaction) (bool, error) {
// If the transaction is already known, discard it
hash := tx.Hash()
if pool.all[hash] != nil {
log.Trace("Discarding already known transaction", "hash", hash)
return fmt.Errorf("known transaction: %x", hash)
return false, fmt.Errorf("known transaction: %x", hash)
}
// Otherwise ensure basic validation passes and queue it up
// If the transaction fails basic validation, discard it
if err := pool.validateTx(tx); err != nil {
log.Trace("Discarding invalid transaction", "hash", hash, "err", err)
invalidTxCounter.Inc(1)
return err
return false, err
}
pool.enqueueTx(hash, tx)
// If the transaction pool is full, discard underpriced transactions
if uint64(len(pool.all)) >= maxPendingTotal+maxQueuedTotal {
// If the new transaction is underpriced, don't accept it
if pool.priced.Underpriced(tx, pool.locals) {
log.Trace("Discarding underpriced transaction", "hash", hash, "price", tx.GasPrice())
underpricedTxCounter.Inc(1)
return false, ErrUnderpriced
}
// New transaction is better than our worse ones, make room for it
drop := pool.priced.Discard(len(pool.all)-int(maxPendingTotal+maxQueuedTotal-1), pool.locals)
for _, tx := range drop {
log.Trace("Discarding freshly underpriced transaction", "hash", tx.Hash(), "price", tx.GasPrice())
underpricedTxCounter.Inc(1)
pool.removeTx(tx.Hash())
}
}
// If the transaction is replacing an already pending one, do directly
from, _ := types.Sender(pool.signer, tx) // already validated
if list := pool.pending[from]; list != nil && list.Overlaps(tx) {
// Nonce already pending, check if required price bump is met
inserted, old := list.Add(tx)
if !inserted {
pendingDiscardCounter.Inc(1)
return false, ErrReplaceUnderpriced
}
// New transaction is better, replace old one
if old != nil {
delete(pool.all, old.Hash())
pool.priced.Removed()
pendingReplaceCounter.Inc(1)
}
pool.all[tx.Hash()] = tx
pool.priced.Put(tx)
// Print a log message if low enough level is set
log.Debug("Pooled new transaction", "hash", hash, "from", log.Lazy{Fn: func() common.Address { from, _ := types.Sender(pool.signer, tx); return from }}, "to", tx.To())
return nil
log.Trace("Pooled new executable transaction", "hash", hash, "from", from, "to", tx.To())
return old != nil, nil
}
// New transaction isn't replacing a pending one, push into queue
replace, err := pool.enqueueTx(hash, tx)
if err != nil {
return false, err
}
log.Trace("Pooled new future transaction", "hash", hash, "from", from, "to", tx.To())
return replace, nil
}
// enqueueTx inserts a new transaction into the non-executable transaction queue.
//
// Note, this method assumes the pool lock is held!
func (pool *TxPool) enqueueTx(hash common.Hash, tx *types.Transaction) {
func (pool *TxPool) enqueueTx(hash common.Hash, tx *types.Transaction) (bool, error) {
// Try to insert the transaction into the future queue
from, _ := types.Sender(pool.signer, tx) // already validated
if pool.queue[from] == nil {
@ -348,15 +444,19 @@ func (pool *TxPool) enqueueTx(hash common.Hash, tx *types.Transaction) {
}
inserted, old := pool.queue[from].Add(tx)
if !inserted {
// An older transaction was better, discard this
queuedDiscardCounter.Inc(1)
return // An older transaction was better, discard this
return false, ErrReplaceUnderpriced
}
// Discard any previous transaction and mark this
if old != nil {
delete(pool.all, old.Hash())
pool.priced.Removed()
queuedReplaceCounter.Inc(1)
}
pool.all[hash] = tx
pool.priced.Put(tx)
return old != nil, nil
}
// promoteTx adds a transaction to the pending (processable) list of transactions.
@ -373,16 +473,23 @@ func (pool *TxPool) promoteTx(addr common.Address, hash common.Hash, tx *types.T
if !inserted {
// An older transaction was better, discard this
delete(pool.all, hash)
pool.priced.Removed()
pendingDiscardCounter.Inc(1)
return
}
// Otherwise discard any previous transaction and mark this
if old != nil {
delete(pool.all, old.Hash())
pool.priced.Removed()
pendingReplaceCounter.Inc(1)
}
pool.all[hash] = tx // Failsafe to work around direct pending inserts (tests)
// Failsafe to work around direct pending inserts (tests)
if pool.all[hash] == nil {
pool.all[hash] = 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)
@ -394,16 +501,19 @@ func (pool *TxPool) Add(tx *types.Transaction) error {
pool.mu.Lock()
defer pool.mu.Unlock()
if err := pool.add(tx); err != nil {
// Try to inject the transaction and update any state
replace, err := pool.add(tx)
if err != nil {
return err
}
state, err := pool.currentState()
if err != nil {
return err
}
// If we added a new transaction, run promotion checks and return
if !replace {
pool.promoteExecutables(state)
}
return nil
}
@ -413,10 +523,13 @@ func (pool *TxPool) AddBatch(txs []*types.Transaction) error {
defer pool.mu.Unlock()
// Add the batch of transaction, tracking the accepted ones
added := 0
replaced, added := true, 0
for _, tx := range txs {
if err := pool.add(tx); err == nil {
if replace, err := pool.add(tx); err == nil {
added++
if !replace {
replaced = false
}
}
}
// Only reprocess the internal state if something was actually added
@ -425,8 +538,10 @@ func (pool *TxPool) AddBatch(txs []*types.Transaction) error {
if err != nil {
return err
}
if !replaced {
pool.promoteExecutables(state)
}
}
return nil
}
@ -469,6 +584,7 @@ func (pool *TxPool) removeTx(hash common.Hash) {
// Remove it from the list of known transactions
delete(pool.all, hash)
pool.priced.Removed()
// Remove the transaction from the pending lists and reset the account nonce
if pending := pool.pending[addr]; pending != nil {
@ -508,28 +624,31 @@ func (pool *TxPool) promoteExecutables(state *state.StateDB) {
// Drop all transactions that are deemed too old (low nonce)
for _, tx := range list.Forward(state.GetNonce(addr)) {
hash := tx.Hash()
log.Debug("Removed old queued transaction", "hash", hash)
log.Trace("Removed old queued transaction", "hash", hash)
delete(pool.all, hash)
pool.priced.Removed()
}
// Drop all transactions that are too costly (low balance)
drops, _ := list.Filter(state.GetBalance(addr))
for _, tx := range drops {
hash := tx.Hash()
log.Debug("Removed unpayable queued transaction", "hash", hash)
log.Trace("Removed unpayable queued transaction", "hash", hash)
delete(pool.all, 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.Debug("Promoting queued transaction", "hash", hash)
log.Trace("Promoting queued transaction", "hash", hash)
pool.promoteTx(addr, hash, tx)
}
// Drop all transactions over the allowed limit
for _, tx := range list.Cap(int(maxQueuedPerAccount)) {
hash := tx.Hash()
log.Debug("Removed cap-exceeding queued transaction", "hash", hash)
log.Trace("Removed cap-exceeding queued transaction", "hash", hash)
delete(pool.all, hash)
pool.priced.Removed()
queuedRLCounter.Inc(1)
}
queued += uint64(list.Len())
@ -553,7 +672,7 @@ func (pool *TxPool) promoteExecutables(state *state.StateDB) {
if uint64(list.Len()) > minPendingPerAccount {
// Skip local accounts as pools should maintain backlogs for themselves
for _, tx := range list.txs.items {
if !pool.localTx.contains(tx.Hash()) {
if !pool.locals.contains(tx.Hash()) {
spammers.Push(addr, float32(list.Len()))
}
break // Checking on transaction for locality is enough
@ -595,7 +714,7 @@ func (pool *TxPool) promoteExecutables(state *state.StateDB) {
pendingRLCounter.Inc(int64(pendingBeforeCap - pending))
}
// If we've queued more transactions than the hard limit, drop oldest ones
if queued > maxQueuedInTotal {
if queued > maxQueuedTotal {
// Sort all accounts with queued transactions by heartbeat
addresses := make(addresssByHeartbeat, 0, len(pool.queue))
for addr := range pool.queue {
@ -604,7 +723,7 @@ func (pool *TxPool) promoteExecutables(state *state.StateDB) {
sort.Sort(addresses)
// Drop transactions until the total is below the limit
for drop := queued - maxQueuedInTotal; drop > 0; {
for drop := queued - maxQueuedTotal; drop > 0; {
addr := addresses[len(addresses)-1]
list := pool.queue[addr.address]
@ -641,20 +760,22 @@ func (pool *TxPool) demoteUnexecutables(state *state.StateDB) {
// Drop all transactions that are deemed too old (low nonce)
for _, tx := range list.Forward(nonce) {
hash := tx.Hash()
log.Debug("Removed old pending transaction", "hash", hash)
log.Trace("Removed old pending transaction", "hash", hash)
delete(pool.all, hash)
pool.priced.Removed()
}
// Drop all transactions that are too costly (low balance), and queue any invalids back for later
drops, invalids := list.Filter(state.GetBalance(addr))
for _, tx := range drops {
hash := tx.Hash()
log.Debug("Removed unpayable pending transaction", "hash", hash)
log.Trace("Removed unpayable pending transaction", "hash", hash)
delete(pool.all, hash)
pool.priced.Removed()
pendingNofundsCounter.Inc(1)
}
for _, tx := range invalids {
hash := tx.Hash()
log.Debug("Demoting pending transaction", "hash", hash)
log.Trace("Demoting pending transaction", "hash", hash)
pool.enqueueTx(hash, tx)
}
// Delete the entire queue entry if it became empty.

261
core/tx_pool_test.go
View file

@ -33,7 +33,11 @@ import (
)
func transaction(nonce uint64, gaslimit *big.Int, key *ecdsa.PrivateKey) *types.Transaction {
tx, _ := types.SignTx(types.NewTransaction(nonce, common.Address{}, big.NewInt(100), gaslimit, big.NewInt(1), nil), types.HomesteadSigner{}, key)
return pricedTransaction(nonce, gaslimit, big.NewInt(1), key)
}
func pricedTransaction(nonce uint64, gaslimit, gasprice *big.Int, key *ecdsa.PrivateKey) *types.Transaction {
tx, _ := types.SignTx(types.NewTransaction(nonce, common.Address{}, big.NewInt(100), gaslimit, gasprice, nil), types.HomesteadSigner{}, key)
return tx
}
@ -151,9 +155,9 @@ func TestInvalidTransactions(t *testing.T) {
}
tx = transaction(1, big.NewInt(100000), key)
pool.minGasPrice = big.NewInt(1000)
if err := pool.Add(tx); err != ErrCheap {
t.Error("expected", ErrCheap, "got", err)
pool.gasPrice = big.NewInt(1000)
if err := pool.Add(tx); err != ErrUnderpriced {
t.Error("expected", ErrUnderpriced, "got", err)
}
pool.SetLocal(tx)
@ -262,14 +266,14 @@ func TestTransactionChainFork(t *testing.T) {
resetState()
tx := transaction(0, big.NewInt(100000), key)
if err := pool.add(tx); err != nil {
if _, err := pool.add(tx); err != nil {
t.Error("didn't expect error", err)
}
pool.RemoveBatch([]*types.Transaction{tx})
// reset the pool's internal state
resetState()
if err := pool.add(tx); err != nil {
if _, err := pool.add(tx); err != nil {
t.Error("didn't expect error", err)
}
}
@ -293,11 +297,11 @@ func TestTransactionDoubleNonce(t *testing.T) {
tx3, _ := types.SignTx(types.NewTransaction(0, common.Address{}, big.NewInt(100), big.NewInt(1000000), big.NewInt(1), nil), signer, key)
// Add the first two transaction, ensure higher priced stays only
if err := pool.add(tx1); err != nil {
t.Error("didn't expect error", err)
if replace, err := pool.add(tx1); err != nil || replace {
t.Errorf("first transaction insert failed (%v) or reported replacement (%v)", err, replace)
}
if err := pool.add(tx2); err != nil {
t.Error("didn't expect error", err)
if replace, err := pool.add(tx2); err != nil || !replace {
t.Errorf("second transaction insert failed (%v) or not reported replacement (%v)", err, replace)
}
state, _ := pool.currentState()
pool.promoteExecutables(state)
@ -308,9 +312,7 @@ func TestTransactionDoubleNonce(t *testing.T) {
t.Errorf("transaction mismatch: have %x, want %x", tx.Hash(), tx2.Hash())
}
// Add the thid transaction and ensure it's not saved (smaller price)
if err := pool.add(tx3); err != nil {
t.Error("didn't expect error", err)
}
pool.add(tx3)
pool.promoteExecutables(state)
if pool.pending[addr].Len() != 1 {
t.Error("expected 1 pending transactions, got", pool.pending[addr].Len())
@ -330,7 +332,7 @@ func TestMissingNonce(t *testing.T) {
currentState, _ := pool.currentState()
currentState.AddBalance(addr, big.NewInt(100000000000000))
tx := transaction(1, big.NewInt(100000), key)
if err := pool.add(tx); err != nil {
if _, err := pool.add(tx); err != nil {
t.Error("didn't expect error", err)
}
if len(pool.pending) != 0 {
@ -557,8 +559,8 @@ func TestTransactionQueueAccountLimiting(t *testing.T) {
// some threshold, the higher transactions are dropped to prevent DOS attacks.
func TestTransactionQueueGlobalLimiting(t *testing.T) {
// Reduce the queue limits to shorten test time
defer func(old uint64) { maxQueuedInTotal = old }(maxQueuedInTotal)
maxQueuedInTotal = maxQueuedPerAccount * 3
defer func(old uint64) { maxQueuedTotal = old }(maxQueuedTotal)
maxQueuedTotal = maxQueuedPerAccount * 3
// Create the pool to test the limit enforcement with
db, _ := ethdb.NewMemDatabase()
@ -578,7 +580,7 @@ func TestTransactionQueueGlobalLimiting(t *testing.T) {
// Generate and queue a batch of transactions
nonces := make(map[common.Address]uint64)
txs := make(types.Transactions, 0, 3*maxQueuedInTotal)
txs := make(types.Transactions, 0, 3*maxQueuedTotal)
for len(txs) < cap(txs) {
key := keys[rand.Intn(len(keys))]
addr := crypto.PubkeyToAddress(key.PublicKey)
@ -596,8 +598,8 @@ func TestTransactionQueueGlobalLimiting(t *testing.T) {
}
queued += list.Len()
}
if queued > int(maxQueuedInTotal) {
t.Fatalf("total transactions overflow allowance: %d > %d", queued, maxQueuedInTotal)
if queued > int(maxQueuedTotal) {
t.Fatalf("total transactions overflow allowance: %d > %d", queued, maxQueuedTotal)
}
}
@ -791,6 +793,227 @@ func TestTransactionPendingMinimumAllowance(t *testing.T) {
}
}
// Tests that setting the transaction pool gas price to a higher value correctly
// discards everything cheaper than that and moves any gapped transactions back
// from the pending pool to the queue.
//
// Note, local transactions are never allowed to be dropped.
func TestTransactionPoolRepricing(t *testing.T) {
// Create the pool to test the pricing enforcement with
db, _ := ethdb.NewMemDatabase()
statedb, _ := state.New(common.Hash{}, db)
pool := NewTxPool(params.TestChainConfig, new(event.TypeMux), func() (*state.StateDB, error) { return statedb, nil }, func() *big.Int { return big.NewInt(1000000) })
pool.resetState()
// Create a number of test accounts and fund them
state, _ := pool.currentState()
keys := make([]*ecdsa.PrivateKey, 3)
for i := 0; i < len(keys); i++ {
keys[i], _ = crypto.GenerateKey()
state.AddBalance(crypto.PubkeyToAddress(keys[i].PublicKey), big.NewInt(1000000))
}
// Generate and queue a batch of transactions, both pending and queued
txs := types.Transactions{}
txs = append(txs, pricedTransaction(0, big.NewInt(100000), big.NewInt(2), keys[0]))
txs = append(txs, pricedTransaction(1, big.NewInt(100000), big.NewInt(1), keys[0]))
txs = append(txs, pricedTransaction(2, big.NewInt(100000), big.NewInt(2), keys[0]))
txs = append(txs, pricedTransaction(1, big.NewInt(100000), big.NewInt(2), keys[1]))
txs = append(txs, pricedTransaction(2, big.NewInt(100000), big.NewInt(1), keys[1]))
txs = append(txs, pricedTransaction(3, big.NewInt(100000), big.NewInt(2), keys[1]))
txs = append(txs, pricedTransaction(0, big.NewInt(100000), big.NewInt(1), keys[2]))
pool.SetLocal(txs[len(txs)-1]) // prevent this one from ever being dropped
// Import the batch and that both pending and queued transactions match up
pool.AddBatch(txs)
pending, queued := pool.stats()
if pending != 4 {
t.Fatalf("pending transactions mismatched: have %d, want %d", pending, 4)
}
if queued != 3 {
t.Fatalf("queued transactions mismatched: have %d, want %d", queued, 3)
}
// Reprice the pool and check that underpriced transactions get dropped
pool.SetGasPrice(big.NewInt(2))
pending, queued = pool.stats()
if pending != 2 {
t.Fatalf("pending transactions mismatched: have %d, want %d", pending, 2)
}
if queued != 3 {
t.Fatalf("queued transactions mismatched: have %d, want %d", queued, 3)
}
// Check that we can't add the old transactions back
if err := pool.Add(pricedTransaction(1, big.NewInt(100000), big.NewInt(1), keys[0])); err != ErrUnderpriced {
t.Fatalf("adding underpriced pending transaction error mismatch: have %v, want %v", err, ErrUnderpriced)
}
if err := pool.Add(pricedTransaction(2, big.NewInt(100000), big.NewInt(1), keys[1])); err != ErrUnderpriced {
t.Fatalf("adding underpriced queued transaction error mismatch: have %v, want %v", err, ErrUnderpriced)
}
// However we can add local underpriced transactions
tx := pricedTransaction(1, big.NewInt(100000), big.NewInt(1), keys[2])
pool.SetLocal(tx) // prevent this one from ever being dropped
if err := pool.Add(tx); err != nil {
t.Fatalf("failed to add underpriced local transaction: %v", err)
}
if pending, _ = pool.stats(); pending != 3 {
t.Fatalf("pending transactions mismatched: have %d, want %d", pending, 3)
}
}
// Tests that when the pool reaches its global transaction limit, underpriced
// transactions are gradually shifted out for more expensive ones and any gapped
// pending transactions are moved into te queue.
//
// Note, local transactions are never allowed to be dropped.
func TestTransactionPoolUnderpricing(t *testing.T) {
// Reduce the queue limits to shorten test time
defer func(old uint64) { maxPendingTotal = old }(maxPendingTotal)
maxPendingTotal = 2
defer func(old uint64) { maxQueuedTotal = old }(maxQueuedTotal)
maxQueuedTotal = 2
// Create the pool to test the pricing enforcement with
db, _ := ethdb.NewMemDatabase()
statedb, _ := state.New(common.Hash{}, db)
pool := NewTxPool(params.TestChainConfig, new(event.TypeMux), func() (*state.StateDB, error) { return statedb, nil }, func() *big.Int { return big.NewInt(1000000) })
pool.resetState()
// Create a number of test accounts and fund them
state, _ := pool.currentState()
keys := make([]*ecdsa.PrivateKey, 3)
for i := 0; i < len(keys); i++ {
keys[i], _ = crypto.GenerateKey()
state.AddBalance(crypto.PubkeyToAddress(keys[i].PublicKey), big.NewInt(1000000))
}
// Generate and queue a batch of transactions, both pending and queued
txs := types.Transactions{}
txs = append(txs, pricedTransaction(0, big.NewInt(100000), big.NewInt(1), keys[0]))
txs = append(txs, pricedTransaction(1, big.NewInt(100000), big.NewInt(2), keys[0]))
txs = append(txs, pricedTransaction(1, big.NewInt(100000), big.NewInt(1), keys[1]))
txs = append(txs, pricedTransaction(0, big.NewInt(100000), big.NewInt(1), keys[2]))
pool.SetLocal(txs[len(txs)-1]) // prevent this one from ever being dropped
// Import the batch and that both pending and queued transactions match up
pool.AddBatch(txs)
pending, queued := pool.stats()
if pending != 3 {
t.Fatalf("pending transactions mismatched: have %d, want %d", pending, 3)
}
if queued != 1 {
t.Fatalf("queued transactions mismatched: have %d, want %d", queued, 1)
}
// Ensure that adding an underpriced transaction on block limit fails
if err := pool.Add(pricedTransaction(0, big.NewInt(100000), big.NewInt(1), keys[1])); err != ErrUnderpriced {
t.Fatalf("adding underpriced pending transaction error mismatch: have %v, want %v", err, ErrUnderpriced)
}
// Ensure that adding high priced transactions drops cheap ones, but not own
if err := pool.Add(pricedTransaction(0, big.NewInt(100000), big.NewInt(3), keys[1])); err != nil {
t.Fatalf("failed to add well priced transaction: %v", err)
}
if err := pool.Add(pricedTransaction(2, big.NewInt(100000), big.NewInt(4), keys[1])); err != nil {
t.Fatalf("failed to add well priced transaction: %v", err)
}
if err := pool.Add(pricedTransaction(3, big.NewInt(100000), big.NewInt(5), keys[1])); err != nil {
t.Fatalf("failed to add well priced transaction: %v", err)
}
pending, queued = pool.stats()
if pending != 2 {
t.Fatalf("pending transactions mismatched: have %d, want %d", pending, 2)
}
if queued != 2 {
t.Fatalf("queued transactions mismatched: have %d, want %d", queued, 2)
}
// Ensure that adding local transactions can push out even higher priced ones
tx := pricedTransaction(1, big.NewInt(100000), big.NewInt(0), keys[2])
pool.SetLocal(tx) // prevent this one from ever being dropped
if err := pool.Add(tx); err != nil {
t.Fatalf("failed to add underpriced local transaction: %v", err)
}
pending, queued = pool.stats()
if pending != 2 {
t.Fatalf("pending transactions mismatched: have %d, want %d", pending, 2)
}
if queued != 2 {
t.Fatalf("queued transactions mismatched: have %d, want %d", queued, 2)
}
}
// Tests that the pool rejects replacement transactions that don't meet the minimum
// price bump required.
func TestTransactionReplacement(t *testing.T) {
// Create the pool to test the pricing enforcement with
db, _ := ethdb.NewMemDatabase()
statedb, _ := state.New(common.Hash{}, db)
pool := NewTxPool(params.TestChainConfig, new(event.TypeMux), func() (*state.StateDB, error) { return statedb, nil }, func() *big.Int { return big.NewInt(1000000) })
pool.resetState()
// Create a a test account to add transactions with
key, _ := crypto.GenerateKey()
state, _ := pool.currentState()
state.AddBalance(crypto.PubkeyToAddress(key.PublicKey), big.NewInt(1000000000))
// Add pending transactions, ensuring the minimum price bump is enforced for replacement (for ultra low prices too)
price := int64(100)
threshold := (price * (100 + minPriceBumpPercent)) / 100
if err := pool.Add(pricedTransaction(0, big.NewInt(100000), big.NewInt(1), key)); err != nil {
t.Fatalf("failed to add original cheap pending transaction: %v", err)
}
if err := pool.Add(pricedTransaction(0, big.NewInt(100001), big.NewInt(1), key)); err != ErrReplaceUnderpriced {
t.Fatalf("original cheap pending transaction replacement error mismatch: have %v, want %v", err, ErrReplaceUnderpriced)
}
if err := pool.Add(pricedTransaction(0, big.NewInt(100000), big.NewInt(2), key)); err != nil {
t.Fatalf("failed to replace original cheap pending transaction: %v", err)
}
if err := pool.Add(pricedTransaction(0, big.NewInt(100000), big.NewInt(price), key)); err != nil {
t.Fatalf("failed to add original proper pending transaction: %v", err)
}
if err := pool.Add(pricedTransaction(0, big.NewInt(100000), big.NewInt(threshold), key)); err != ErrReplaceUnderpriced {
t.Fatalf("original proper pending transaction replacement error mismatch: have %v, want %v", err, ErrReplaceUnderpriced)
}
if err := pool.Add(pricedTransaction(0, big.NewInt(100000), big.NewInt(threshold+1), key)); err != nil {
t.Fatalf("failed to replace original proper pending transaction: %v", err)
}
// Add queued transactions, ensuring the minimum price bump is enforced for replacement (for ultra low prices too)
if err := pool.Add(pricedTransaction(2, big.NewInt(100000), big.NewInt(1), key)); err != nil {
t.Fatalf("failed to add original queued transaction: %v", err)
}
if err := pool.Add(pricedTransaction(2, big.NewInt(100001), big.NewInt(1), key)); err != ErrReplaceUnderpriced {
t.Fatalf("original queued transaction replacement error mismatch: have %v, want %v", err, ErrReplaceUnderpriced)
}
if err := pool.Add(pricedTransaction(2, big.NewInt(100000), big.NewInt(2), key)); err != nil {
t.Fatalf("failed to replace original queued transaction: %v", err)
}
if err := pool.Add(pricedTransaction(2, big.NewInt(100000), big.NewInt(price), key)); err != nil {
t.Fatalf("failed to add original queued transaction: %v", err)
}
if err := pool.Add(pricedTransaction(2, big.NewInt(100001), big.NewInt(threshold), key)); err != ErrReplaceUnderpriced {
t.Fatalf("original queued transaction replacement error mismatch: have %v, want %v", err, ErrReplaceUnderpriced)
}
if err := pool.Add(pricedTransaction(2, big.NewInt(100000), big.NewInt(threshold+1), key)); err != nil {
t.Fatalf("failed to replace original queued transaction: %v", err)
}
}
// Benchmarks the speed of validating the contents of the pending queue of the
// transaction pool.
func BenchmarkPendingDemotion100(b *testing.B) { benchmarkPendingDemotion(b, 100) }

4
eth/api.go
View file

@ -153,6 +153,8 @@ func (api *PrivateMinerAPI) Start(threads *int) error {
}
// Start the miner and return
if !api.e.IsMining() {
// Propagate the initial price point to the transaction pool
api.e.txPool.SetGasPrice(api.e.gasPrice)
return api.e.StartMining(true)
}
return nil
@ -180,7 +182,7 @@ func (api *PrivateMinerAPI) SetExtra(extra string) (bool, error) {
// SetGasPrice sets the minimum accepted gas price for the miner.
func (api *PrivateMinerAPI) SetGasPrice(gasPrice hexutil.Big) bool {
api.e.Miner().SetGasPrice((*big.Int)(&gasPrice))
api.e.txPool.SetGasPrice((*big.Int)(&gasPrice))
return true
}

4
eth/backend.go
View file

@ -20,6 +20,7 @@ package eth
import (
"errors"
"fmt"
"math/big"
"runtime"
"sync"
"sync/atomic"
@ -76,6 +77,7 @@ type Ethereum struct {
ApiBackend *EthApiBackend
miner *miner.Miner
gasPrice *big.Int
Mining bool
MinerThreads int
etherbase common.Address
@ -167,7 +169,7 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) {
}
eth.miner = miner.New(eth, eth.chainConfig, eth.EventMux(), eth.engine)
eth.miner.SetGasPrice(config.GasPrice)
eth.gasPrice = config.GasPrice
eth.miner.SetExtra(makeExtraData(config.ExtraData))
eth.ApiBackend = &EthApiBackend{eth, nil}

2
eth/config.go
View file

@ -42,7 +42,7 @@ var DefaultConfig = Config{
NetworkId: 1,
LightPeers: 20,
DatabaseCache: 128,
GasPrice: big.NewInt(20 * params.Shannon),
GasPrice: big.NewInt(18 * params.Shannon),
GPO: gasprice.Config{
Blocks: 10,

4
ethstats/ethstats.go
View file

@ -18,6 +18,7 @@
package ethstats
import (
"context"
"encoding/json"
"errors"
"fmt"
@ -639,7 +640,8 @@ func (s *Service) reportStats(conn *websocket.Conn) error {
sync := s.eth.Downloader().Progress()
syncing = s.eth.BlockChain().CurrentHeader().Number.Uint64() >= sync.HighestBlock
gasprice = int(s.eth.Miner().GasPrice().Uint64())
price, _ := s.eth.ApiBackend.SuggestPrice(context.Background())
gasprice = int(price.Uint64())
} else {
sync := s.les.Downloader().Progress()
syncing = s.les.BlockChain().CurrentHeader().Number.Uint64() >= sync.HighestBlock

23
internal/ethapi/api.go
View file

@ -24,6 +24,7 @@ import (
"fmt"
"math/big"
"strings"
"sync"
"time"
"github.com/ethereum/go-ethereum/accounts"
@ -890,6 +891,12 @@ type PublicTransactionPoolAPI struct {
b Backend
}
// nonceMutex is a global mutex for locking the nonce while a transaction
// is being submitted. This should be used when a nonce has not been provided by the user,
// and we get a nonce from the pools. The mutex prevents the (an identical nonce) from being
// read again during the time that the first transaction is being signed.
var nonceMutex sync.RWMutex
// NewPublicTransactionPoolAPI creates a new RPC service with methods specific for the transaction pool.
func NewPublicTransactionPoolAPI(b Backend) *PublicTransactionPoolAPI {
return &PublicTransactionPoolAPI{b}
@ -1170,6 +1177,14 @@ func submitTransaction(ctx context.Context, b Backend, tx *types.Transaction) (c
// SendTransaction creates a transaction for the given argument, sign it and submit it to the
// transaction pool.
func (s *PublicTransactionPoolAPI) SendTransaction(ctx context.Context, args SendTxArgs) (common.Hash, error) {
if args.Nonce == nil {
// We'll need to set nonce from pool, and thus we need to lock here
nonceMutex.Lock()
defer nonceMutex.Unlock()
}
// Set some sanity defaults and terminate on failure
if err := args.setDefaults(ctx, s.b); err != nil {
return common.Hash{}, err
@ -1257,6 +1272,14 @@ type SignTransactionResult struct {
// The node needs to have the private key of the account corresponding with
// the given from address and it needs to be unlocked.
func (s *PublicTransactionPoolAPI) SignTransaction(ctx context.Context, args SendTxArgs) (*SignTransactionResult, error) {
if args.Nonce == nil {
// We'll need to set nonce from pool, and thus we need to lock here
nonceMutex.Lock()
defer nonceMutex.Unlock()
}
if err := args.setDefaults(ctx, s.b); err != nil {
return nil, err
}

86
les/execqueue.go
View file

@ -16,56 +16,82 @@
package les
import (
"sync/atomic"
)
import "sync"
// ExecQueue implements a queue that executes function calls in a single thread,
// execQueue implements a queue that executes function calls in a single thread,
// in the same order as they have been queued.
type execQueue struct {
chn chan func()
cnt, stop, capacity int32
mu sync.Mutex
cond *sync.Cond
funcs []func()
closeWait chan struct{}
}
// NewExecQueue creates a new execution queue.
func newExecQueue(capacity int32) *execQueue {
q := &execQueue{
chn: make(chan func(), capacity),
capacity: capacity,
}
// newExecQueue creates a new execution queue.
func newExecQueue(capacity int) *execQueue {
q := &execQueue{funcs: make([]func(), 0, capacity)}
q.cond = sync.NewCond(&q.mu)
go q.loop()
return q
}
func (q *execQueue) loop() {
for f := range q.chn {
atomic.AddInt32(&q.cnt, -1)
if atomic.LoadInt32(&q.stop) != 0 {
return
}
for f := q.waitNext(false); f != nil; f = q.waitNext(true) {
f()
}
close(q.closeWait)
}
// CanQueue returns true if more function calls can be added to the execution queue.
func (q *execQueue) waitNext(drop bool) (f func()) {
q.mu.Lock()
if drop {
// Remove the function that just executed. We do this here instead of when
// dequeuing so len(q.funcs) includes the function that is running.
q.funcs = append(q.funcs[:0], q.funcs[1:]...)
}
for !q.isClosed() {
if len(q.funcs) > 0 {
f = q.funcs[0]
break
}
q.cond.Wait()
}
q.mu.Unlock()
return f
}
func (q *execQueue) isClosed() bool {
return q.closeWait != nil
}
// canQueue returns true if more function calls can be added to the execution queue.
func (q *execQueue) canQueue() bool {
return atomic.LoadInt32(&q.stop) == 0 && atomic.LoadInt32(&q.cnt) < q.capacity
q.mu.Lock()
ok := !q.isClosed() && len(q.funcs) < cap(q.funcs)
q.mu.Unlock()
return ok
}
// Queue adds a function call to the execution queue. Returns true if successful.
// queue adds a function call to the execution queue. Returns true if successful.
func (q *execQueue) queue(f func()) bool {
if atomic.LoadInt32(&q.stop) != 0 {
return false
q.mu.Lock()
ok := !q.isClosed() && len(q.funcs) < cap(q.funcs)
if ok {
q.funcs = append(q.funcs, f)
q.cond.Signal()
}
if atomic.AddInt32(&q.cnt, 1) > q.capacity {
atomic.AddInt32(&q.cnt, -1)
return false
}
q.chn <- f
return true
q.mu.Unlock()
return ok
}
// Stop stops the exec queue.
// quit stops the exec queue.
// quit waits for the current execution to finish before returning.
func (q *execQueue) quit() {
atomic.StoreInt32(&q.stop, 1)
q.mu.Lock()
if !q.isClosed() {
q.closeWait = make(chan struct{})
q.cond.Signal()
}
q.mu.Unlock()
<-q.closeWait
}

62
les/execqueue_test.go Normal file
View file

@ -0,0 +1,62 @@
// Copyright 2017 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 les
import (
"testing"
)
func TestExecQueue(t *testing.T) {
var (
N = 10000
q = newExecQueue(N)
counter int
execd = make(chan int)
testexit = make(chan struct{})
)
defer q.quit()
defer close(testexit)
check := func(state string, wantOK bool) {
c := counter
counter++
qf := func() {
select {
case execd <- c:
case <-testexit:
}
}
if q.canQueue() != wantOK {
t.Fatalf("canQueue() == %t for %s", !wantOK, state)
}
if q.queue(qf) != wantOK {
t.Fatalf("canQueue() == %t for %s", !wantOK, state)
}
}
for i := 0; i < N; i++ {
check("queue below cap", true)
}
check("full queue", false)
for i := 0; i < N; i++ {
if c := <-execd; c != i {
t.Fatal("execution out of order")
}
}
q.quit()
check("closed queue", false)
}

13
miner/miner.go
View file

@ -19,7 +19,6 @@ package miner
import (
"fmt"
"math/big"
"sync/atomic"
"github.com/ethereum/go-ethereum/accounts"
@ -104,18 +103,6 @@ out:
}
}
func (m *Miner) GasPrice() *big.Int {
return new(big.Int).Set(m.worker.gasPrice)
}
func (m *Miner) SetGasPrice(price *big.Int) {
// FIXME block tests set a nil gas price. Quick dirty fix
if price == nil {
return
}
m.worker.setGasPrice(price)
}
func (self *Miner) Start(coinbase common.Address) {
atomic.StoreInt32(&self.shouldStart, 1)
self.worker.setEtherbase(coinbase)

56
miner/worker.go
View file

@ -64,8 +64,6 @@ type Work struct {
family *set.Set // family set (used for checking uncle invalidity)
uncles *set.Set // uncle set
tcount int // tx count in cycle
ownedAccounts *set.Set
lowGasTxs types.Transactions
failedTxs types.Transactions
Block *types.Block // the new block
@ -103,7 +101,6 @@ type worker struct {
chainDb ethdb.Database
coinbase common.Address
gasPrice *big.Int
extra []byte
currentMu sync.Mutex
@ -132,7 +129,6 @@ func newWorker(config *params.ChainConfig, engine consensus.Engine, coinbase com
mux: mux,
chainDb: eth.ChainDb(),
recv: make(chan *Result, resultQueueSize),
gasPrice: new(big.Int),
chain: eth.BlockChain(),
proc: eth.BlockChain().Validator(),
possibleUncles: make(map[common.Hash]*types.Block),
@ -252,7 +248,7 @@ func (self *worker) update() {
txs := map[common.Address]types.Transactions{acc: {ev.Tx}}
txset := types.NewTransactionsByPriceAndNonce(txs)
self.current.commitTransactions(self.mux, txset, self.gasPrice, self.chain, self.coinbase)
self.current.commitTransactions(self.mux, txset, self.chain, self.coinbase)
self.currentMu.Unlock()
}
}
@ -375,22 +371,10 @@ func (self *worker) makeCurrent(parent *types.Block, header *types.Header) error
}
// Keep track of transactions which return errors so they can be removed
work.tcount = 0
work.ownedAccounts = accountAddressesSet(accounts)
self.current = work
return nil
}
func (w *worker) setGasPrice(p *big.Int) {
w.mu.Lock()
defer w.mu.Unlock()
// calculate the minimal gas price the miner accepts when sorting out transactions.
const pct = int64(90)
w.gasPrice = gasprice(p, pct)
w.mux.Post(core.GasPriceChanged{Price: w.gasPrice})
}
func (self *worker) commitNewWork() {
self.mu.Lock()
defer self.mu.Unlock()
@ -460,9 +444,8 @@ func (self *worker) commitNewWork() {
return
}
txs := types.NewTransactionsByPriceAndNonce(pending)
work.commitTransactions(self.mux, txs, self.gasPrice, self.chain, self.coinbase)
work.commitTransactions(self.mux, txs, self.chain, self.coinbase)
self.eth.TxPool().RemoveBatch(work.lowGasTxs)
self.eth.TxPool().RemoveBatch(work.failedTxs)
// compute uncles for the new block.
@ -515,7 +498,7 @@ func (self *worker) commitUncle(work *Work, uncle *types.Header) error {
return nil
}
func (env *Work) commitTransactions(mux *event.TypeMux, txs *types.TransactionsByPriceAndNonce, gasPrice *big.Int, bc *core.BlockChain, coinbase common.Address) {
func (env *Work) commitTransactions(mux *event.TypeMux, txs *types.TransactionsByPriceAndNonce, bc *core.BlockChain, coinbase common.Address) {
gp := new(core.GasPool).AddGas(env.header.GasLimit)
var coalescedLogs []*types.Log
@ -539,17 +522,6 @@ func (env *Work) commitTransactions(mux *event.TypeMux, txs *types.TransactionsB
txs.Pop()
continue
}
// Ignore any transactions (and accounts subsequently) with low gas limits
if tx.GasPrice().Cmp(gasPrice) < 0 && !env.ownedAccounts.Has(from) {
// Pop the current low-priced transaction without shifting in the next from the account
log.Warn("Transaction below gas price", "sender", from, "hash", tx.Hash(), "have", tx.GasPrice(), "want", gasPrice)
env.lowGasTxs = append(env.lowGasTxs, tx)
txs.Pop()
continue
}
// Start executing the transaction
env.state.StartRecord(tx.Hash(), common.Hash{}, env.tcount)
@ -607,25 +579,3 @@ func (env *Work) commitTransaction(tx *types.Transaction, bc *core.BlockChain, c
return nil, receipt.Logs
}
// TODO: remove or use
func (self *worker) HashRate() int64 {
return 0
}
// gasprice calculates a reduced gas price based on the pct
// XXX Use big.Rat?
func gasprice(price *big.Int, pct int64) *big.Int {
p := new(big.Int).Set(price)
p.Div(p, big.NewInt(100))
p.Mul(p, big.NewInt(pct))
return p
}
func accountAddressesSet(accounts []accounts.Account) *set.Set {
accountSet := set.New()
for _, account := range accounts {
accountSet.Add(account.Address)
}
return accountSet
}

6
mobile/accounts.go
View file

@ -115,6 +115,9 @@ func (ks *KeyStore) SignHash(address *Address, hash []byte) (signature []byte, _
// SignTx signs the given transaction with the requested account.
func (ks *KeyStore) SignTx(account *Account, tx *Transaction, chainID *BigInt) (*Transaction, error) {
if chainID == nil { // Null passed from mobile app
chainID = new(BigInt)
}
signed, err := ks.keystore.SignTx(account.account, tx.tx, chainID.bigint)
if err != nil {
return nil, err
@ -132,6 +135,9 @@ func (ks *KeyStore) SignHashPassphrase(account *Account, passphrase string, hash
// SignTxPassphrase signs the transaction if the private key matching the
// given address can be decrypted with the given passphrase.
func (ks *KeyStore) SignTxPassphrase(account *Account, passphrase string, tx *Transaction, chainID *BigInt) (*Transaction, error) {
if chainID == nil { // Null passed from mobile app
chainID = new(BigInt)
}
signed, err := ks.keystore.SignTxWithPassphrase(account.account, passphrase, tx.tx, chainID.bigint)
if err != nil {
return nil, err

160
mobile/types.go
View file

@ -19,10 +19,12 @@
package geth
import (
"encoding/json"
"errors"
"fmt"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/rlp"
)
// A Nonce is a 64-bit hash which proves (combined with the mix-hash) that
@ -61,6 +63,45 @@ type Header struct {
header *types.Header
}
// NewHeaderFromRLP parses a header from an RLP data dump.
func NewHeaderFromRLP(data []byte) (*Header, error) {
h := &Header{
header: new(types.Header),
}
if err := rlp.DecodeBytes(data, h.header); err != nil {
return nil, err
}
return h, nil
}
// EncodeRLP encodes a header into an RLP data dump.
func (h *Header) EncodeRLP() ([]byte, error) {
return rlp.EncodeToBytes(h.header)
}
// NewHeaderFromJSON parses a header from an JSON data dump.
func NewHeaderFromJSON(data string) (*Header, error) {
h := &Header{
header: new(types.Header),
}
if err := json.Unmarshal([]byte(data), h.header); err != nil {
return nil, err
}
return h, nil
}
// EncodeJSON encodes a header into an JSON data dump.
func (h *Header) EncodeJSON() (string, error) {
data, err := json.Marshal(h.header)
return string(data), err
}
// String implements the fmt.Stringer interface to print some semi-meaningful
// data dump of the header for debugging purposes.
func (h *Header) String() string {
return h.header.String()
}
func (h *Header) GetParentHash() *Hash { return &Hash{h.header.ParentHash} }
func (h *Header) GetUncleHash() *Hash { return &Hash{h.header.UncleHash} }
func (h *Header) GetCoinbase() *Address { return &Address{h.header.Coinbase} }
@ -76,9 +117,7 @@ func (h *Header) GetTime() int64 { return h.header.Time.Int64() }
func (h *Header) GetExtra() []byte { return h.header.Extra }
func (h *Header) GetMixDigest() *Hash { return &Hash{h.header.MixDigest} }
func (h *Header) GetNonce() *Nonce { return &Nonce{h.header.Nonce} }
func (h *Header) GetHash() *Hash { return &Hash{h.header.Hash()} }
func (h *Header) GetHashNoNonce() *Hash { return &Hash{h.header.HashNoNonce()} }
// Headers represents a slice of headers.
type Headers struct{ headers []*types.Header }
@ -101,6 +140,45 @@ type Block struct {
block *types.Block
}
// NewBlockFromRLP parses a block from an RLP data dump.
func NewBlockFromRLP(data []byte) (*Block, error) {
b := &Block{
block: new(types.Block),
}
if err := rlp.DecodeBytes(data, b.block); err != nil {
return nil, err
}
return b, nil
}
// EncodeRLP encodes a block into an RLP data dump.
func (b *Block) EncodeRLP() ([]byte, error) {
return rlp.EncodeToBytes(b.block)
}
// NewBlockFromJSON parses a block from an JSON data dump.
func NewBlockFromJSON(data string) (*Block, error) {
b := &Block{
block: new(types.Block),
}
if err := json.Unmarshal([]byte(data), b.block); err != nil {
return nil, err
}
return b, nil
}
// EncodeJSON encodes a block into an JSON data dump.
func (b *Block) EncodeJSON() (string, error) {
data, err := json.Marshal(b.block)
return string(data), err
}
// String implements the fmt.Stringer interface to print some semi-meaningful
// data dump of the block for debugging purposes.
func (b *Block) String() string {
return b.block.String()
}
func (b *Block) GetParentHash() *Hash { return &Hash{b.block.ParentHash()} }
func (b *Block) GetUncleHash() *Hash { return &Hash{b.block.UncleHash()} }
func (b *Block) GetCoinbase() *Address { return &Address{b.block.Coinbase()} }
@ -137,6 +215,45 @@ func NewTransaction(nonce int64, to *Address, amount, gasLimit, gasPrice *BigInt
return &Transaction{types.NewTransaction(uint64(nonce), to.address, amount.bigint, gasLimit.bigint, gasPrice.bigint, data)}
}
// NewTransactionFromRLP parses a transaction from an RLP data dump.
func NewTransactionFromRLP(data []byte) (*Transaction, error) {
tx := &Transaction{
tx: new(types.Transaction),
}
if err := rlp.DecodeBytes(data, tx.tx); err != nil {
return nil, err
}
return tx, nil
}
// EncodeRLP encodes a transaction into an RLP data dump.
func (tx *Transaction) EncodeRLP() ([]byte, error) {
return rlp.EncodeToBytes(tx.tx)
}
// NewTransactionFromJSON parses a transaction from an JSON data dump.
func NewTransactionFromJSON(data string) (*Transaction, error) {
tx := &Transaction{
tx: new(types.Transaction),
}
if err := json.Unmarshal([]byte(data), tx.tx); err != nil {
return nil, err
}
return tx, nil
}
// EncodeJSON encodes a transaction into an JSON data dump.
func (tx *Transaction) EncodeJSON() (string, error) {
data, err := json.Marshal(tx.tx)
return string(data), err
}
// String implements the fmt.Stringer interface to print some semi-meaningful
// data dump of the transaction for debugging purposes.
func (tx *Transaction) String() string {
return tx.tx.String()
}
func (tx *Transaction) GetData() []byte { return tx.tx.Data() }
func (tx *Transaction) GetGas() int64 { return tx.tx.Gas().Int64() }
func (tx *Transaction) GetGasPrice() *BigInt { return &BigInt{tx.tx.GasPrice()} }
@ -185,6 +302,45 @@ type Receipt struct {
receipt *types.Receipt
}
// NewReceiptFromRLP parses a transaction receipt from an RLP data dump.
func NewReceiptFromRLP(data []byte) (*Receipt, error) {
r := &Receipt{
receipt: new(types.Receipt),
}
if err := rlp.DecodeBytes(data, r.receipt); err != nil {
return nil, err
}
return r, nil
}
// EncodeRLP encodes a transaction receipt into an RLP data dump.
func (r *Receipt) EncodeRLP() ([]byte, error) {
return rlp.EncodeToBytes(r.receipt)
}
// NewReceiptFromJSON parses a transaction receipt from an JSON data dump.
func NewReceiptFromJSON(data string) (*Receipt, error) {
r := &Receipt{
receipt: new(types.Receipt),
}
if err := json.Unmarshal([]byte(data), r.receipt); err != nil {
return nil, err
}
return r, nil
}
// EncodeJSON encodes a transaction receipt into an JSON data dump.
func (r *Receipt) EncodeJSON() (string, error) {
data, err := rlp.EncodeToBytes(r.receipt)
return string(data), err
}
// String implements the fmt.Stringer interface to print some semi-meaningful
// data dump of the transaction receipt for debugging purposes.
func (r *Receipt) String() string {
return r.receipt.String()
}
func (r *Receipt) GetPostState() []byte { return r.receipt.PostState }
func (r *Receipt) GetCumulativeGasUsed() *BigInt { return &BigInt{r.receipt.CumulativeGasUsed} }
func (r *Receipt) GetBloom() *Bloom { return &Bloom{r.receipt.Bloom} }

1
node/defaults.go
View file

@ -42,6 +42,7 @@ var DefaultConfig = Config{
WSModules: []string{"net", "web3"},
P2P: p2p.Config{
ListenAddr: ":30303",
DiscoveryV5Addr: ":30304",
MaxPeers: 25,
NAT: nat.Any(),
},

2
p2p/discover/udp.go
View file

@ -224,7 +224,7 @@ func ListenUDP(priv *ecdsa.PrivateKey, laddr string, natm nat.Interface, nodeDBP
if err != nil {
return nil, err
}
log.Debug("UDP listener up", "self", tab.self)
log.Info("UDP listener up", "self", tab.self)
return tab, nil
}

43
swarm/api/api.go
View file

@ -25,12 +25,13 @@ import (
"sync"
"bytes"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/swarm/storage"
"mime"
"path/filepath"
"time"
"github.com/ethereum/go-ethereum/common"
"github.com/ethereum/go-ethereum/log"
"github.com/ethereum/go-ethereum/swarm/storage"
)
var (
@ -84,27 +85,33 @@ type ErrResolve error
func (self *Api) Resolve(uri *URI) (storage.Key, error) {
log.Trace(fmt.Sprintf("Resolving : %v", uri.Addr))
var err error
if !uri.Immutable() {
if self.dns != nil {
// if the URI is immutable, check if the address is a hash
isHash := hashMatcher.MatchString(uri.Addr)
if uri.Immutable() {
if !isHash {
return nil, fmt.Errorf("immutable address not a content hash: %q", uri.Addr)
}
return common.Hex2Bytes(uri.Addr), nil
}
// if DNS is not configured, check if the address is a hash
if self.dns == nil {
if !isHash {
return nil, fmt.Errorf("no DNS to resolve name: %q", uri.Addr)
}
return common.Hex2Bytes(uri.Addr), nil
}
// try and resolve the address
resolved, err := self.dns.Resolve(uri.Addr)
if err == nil {
return resolved[:], nil
} else if !isHash {
return nil, err
}
} else {
err = fmt.Errorf("no DNS to resolve name")
}
}
if hashMatcher.MatchString(uri.Addr) {
return storage.Key(common.Hex2Bytes(uri.Addr)), nil
}
if err != nil {
return nil, fmt.Errorf("'%s' does not resolve: %v but is not a content hash", uri.Addr, err)
}
return nil, fmt.Errorf("'%s' is not a content hash", uri.Addr)
return common.Hex2Bytes(uri.Addr), nil
}
// Put provides singleton manifest creation on top of dpa store
func (self *Api) Put(content, contentType string) (storage.Key, error) {
r := strings.NewReader(content)

120
swarm/api/api_test.go
View file

@ -17,6 +17,7 @@
package api
import (
"errors"
"fmt"
"io"
"io/ioutil"
@ -117,3 +118,122 @@ func TestApiPut(t *testing.T) {
checkResponse(t, resp, exp)
})
}
// testResolver implements the Resolver interface and either returns the given
// hash if it is set, or returns a "name not found" error
type testResolver struct {
hash *common.Hash
}
func newTestResolver(addr string) *testResolver {
r := &testResolver{}
if addr != "" {
hash := common.HexToHash(addr)
r.hash = &hash
}
return r
}
func (t *testResolver) Resolve(addr string) (common.Hash, error) {
if t.hash == nil {
return common.Hash{}, fmt.Errorf("DNS name not found: %q", addr)
}
return *t.hash, nil
}
// TestAPIResolve tests resolving URIs which can either contain content hashes
// or ENS names
func TestAPIResolve(t *testing.T) {
ensAddr := "swarm.eth"
hashAddr := "1111111111111111111111111111111111111111111111111111111111111111"
resolvedAddr := "2222222222222222222222222222222222222222222222222222222222222222"
doesResolve := newTestResolver(resolvedAddr)
doesntResolve := newTestResolver("")
type test struct {
desc string
dns Resolver
addr string
immutable bool
result string
expectErr error
}
tests := []*test{
{
desc: "DNS not configured, hash address, returns hash address",
dns: nil,
addr: hashAddr,
result: hashAddr,
},
{
desc: "DNS not configured, ENS address, returns error",
dns: nil,
addr: ensAddr,
expectErr: errors.New(`no DNS to resolve name: "swarm.eth"`),
},
{
desc: "DNS configured, hash address, hash resolves, returns resolved address",
dns: doesResolve,
addr: hashAddr,
result: resolvedAddr,
},
{
desc: "DNS configured, immutable hash address, hash resolves, returns hash address",
dns: doesResolve,
addr: hashAddr,
immutable: true,
result: hashAddr,
},
{
desc: "DNS configured, hash address, hash doesn't resolve, returns hash address",
dns: doesntResolve,
addr: hashAddr,
result: hashAddr,
},
{
desc: "DNS configured, ENS address, name resolves, returns resolved address",
dns: doesResolve,
addr: ensAddr,
result: resolvedAddr,
},
{
desc: "DNS configured, immutable ENS address, name resolves, returns error",
dns: doesResolve,
addr: ensAddr,
immutable: true,
expectErr: errors.New(`immutable address not a content hash: "swarm.eth"`),
},
{
desc: "DNS configured, ENS address, name doesn't resolve, returns error",
dns: doesntResolve,
addr: ensAddr,
expectErr: errors.New(`DNS name not found: "swarm.eth"`),
},
}
for _, x := range tests {
t.Run(x.desc, func(t *testing.T) {
api := &Api{dns: x.dns}
uri := &URI{Addr: x.addr, Scheme: "bzz"}
if x.immutable {
uri.Scheme = "bzzi"
}
res, err := api.Resolve(uri)
if err == nil {
if x.expectErr != nil {
t.Fatalf("expected error %q, got result %q", x.expectErr, res)
}
if res.String() != x.result {
t.Fatalf("expected result %q, got %q", x.result, res)
}
} else {
if x.expectErr == nil {
t.Fatalf("expected no error, got %q", err)
}
if err.Error() != x.expectErr.Error() {
t.Fatalf("expected error %q, got %q", x.expectErr, err)
}
}
})
}
}

7
swarm/api/config.go
View file

@ -32,7 +32,8 @@ import (
)
const (
port = "8500"
DefaultHTTPListenAddr = "127.0.0.1"
DefaultHTTPPort = "8500"
)
var (
@ -49,6 +50,7 @@ type Config struct {
Swap *swap.SwapParams
*network.SyncParams
Path string
ListenAddr string
Port string
PublicKey string
BzzKey string
@ -76,7 +78,8 @@ func NewConfig(path string, contract common.Address, prvKey *ecdsa.PrivateKey, n
HiveParams: network.NewHiveParams(dirpath),
ChunkerParams: storage.NewChunkerParams(),
StoreParams: storage.NewStoreParams(dirpath),
Port: port,
ListenAddr: DefaultHTTPListenAddr,
Port: DefaultHTTPPort,
Path: dirpath,
Swap: swap.DefaultSwapParams(contract, prvKey),
PublicKey: pubkeyhex,

1
swarm/api/config_test.go
View file

@ -80,6 +80,7 @@ var (
false
],
"Path": "TMPDIR",
"ListenAddr": "127.0.0.1",
"Port": "8500",
"PublicKey": "0x045f5cfd26692e48d0017d380349bcf50982488bc11b5145f3ddf88b24924299048450542d43527fbe29a5cb32f38d62755393ac002e6bfdd71b8d7ba725ecd7a3",
"BzzKey": "0xe861964402c0b78e2d44098329b8545726f215afa737d803714a4338552fcb81",

1
swarm/api/http/server.go
View file

@ -69,7 +69,6 @@ func StartHttpServer(api *api.Api, config *ServerConfig) {
hdlr := c.Handler(NewServer(api))
go http.ListenAndServe(config.Addr, hdlr)
log.Info(fmt.Sprintf("Swarm HTTP proxy started on localhost:%s", config.Addr))
}
func NewServer(api *api.Api) *Server {

6
swarm/api/http/server_test.go
View file

@ -106,9 +106,9 @@ func TestBzzrGetPath(t *testing.T) {
}
nonhashresponses := []string{
"error resolving name: 'name' does not resolve: no DNS to resolve name but is not a content hash\n",
"error resolving nonhash: 'nonhash' is not a content hash\n",
"error resolving nonhash: 'nonhash' does not resolve: no DNS to resolve name but is not a content hash\n",
"error resolving name: no DNS to resolve name: \"name\"\n",
"error resolving nonhash: immutable address not a content hash: \"nonhash\"\n",
"error resolving nonhash: no DNS to resolve name: \"nonhash\"\n",
}
for i, url := range nonhashtests {

4
swarm/api/manifest.go
View file

@ -237,12 +237,12 @@ func (self *manifestTrie) addEntry(entry *manifestTrieEntry, quitC chan bool) {
}
b := byte(entry.Path[0])
if (self.entries[b] == nil) || (self.entries[b].Path == entry.Path) {
oldentry := self.entries[b]
if (oldentry == nil) || (oldentry.Path == entry.Path && oldentry.ContentType != ManifestType) {
self.entries[b] = entry
return
}
oldentry := self.entries[b]
cpl := 0
for (len(entry.Path) > cpl) && (len(oldentry.Path) > cpl) && (entry.Path[cpl] == oldentry.Path[cpl]) {
cpl++

33
swarm/api/manifest_test.go
View file

@ -18,6 +18,8 @@ package api
import (
// "encoding/json"
"bytes"
"encoding/json"
"fmt"
"io"
"strings"
@ -78,3 +80,34 @@ func TestGetEntry(t *testing.T) {
func TestDeleteEntry(t *testing.T) {
}
// TestAddFileWithManifestPath tests that adding an entry at a path which
// already exists as a manifest just adds the entry to the manifest rather
// than replacing the manifest with the entry
func TestAddFileWithManifestPath(t *testing.T) {
// create a manifest containing "ab" and "ac"
manifest, _ := json.Marshal(&Manifest{
Entries: []ManifestEntry{
{Path: "ab", Hash: "ab"},
{Path: "ac", Hash: "ac"},
},
})
reader := &storage.LazyTestSectionReader{
SectionReader: io.NewSectionReader(bytes.NewReader(manifest), 0, int64(len(manifest))),
}
trie, err := readManifest(reader, nil, nil, nil)
if err != nil {
t.Fatal(err)
}
checkEntry(t, "ab", "ab", trie)
checkEntry(t, "ac", "ac", trie)
// now add path "a" and check we can still get "ab" and "ac"
entry := &manifestTrieEntry{}
entry.Path = "a"
entry.Hash = "a"
trie.addEntry(entry, nil)
checkEntry(t, "ab", "ab", trie)
checkEntry(t, "ac", "ac", trie)
checkEntry(t, "a", "a", trie)
}

16
swarm/swarm.go
View file

@ -21,6 +21,7 @@ import (
"context"
"crypto/ecdsa"
"fmt"
"net"
"github.com/ethereum/go-ethereum/accounts/abi/bind"
"github.com/ethereum/go-ethereum/common"
@ -166,13 +167,13 @@ Start is called when the stack is started
* TODO: start subservices like sword, swear, swarmdns
*/
// implements the node.Service interface
func (self *Swarm) Start(net *p2p.Server) error {
func (self *Swarm) Start(srv *p2p.Server) error {
connectPeer := func(url string) error {
node, err := discover.ParseNode(url)
if err != nil {
return fmt.Errorf("invalid node URL: %v", err)
}
net.AddPeer(node)
srv.AddPeer(node)
return nil
}
// set chequebook
@ -189,8 +190,8 @@ func (self *Swarm) Start(net *p2p.Server) error {
log.Warn(fmt.Sprintf("Starting Swarm service"))
self.hive.Start(
discover.PubkeyID(&net.PrivateKey.PublicKey),
func() string { return net.ListenAddr },
discover.PubkeyID(&srv.PrivateKey.PublicKey),
func() string { return srv.ListenAddr },
connectPeer,
)
log.Info(fmt.Sprintf("Swarm network started on bzz address: %v", self.hive.Addr()))
@ -200,18 +201,17 @@ func (self *Swarm) Start(net *p2p.Server) error {
// start swarm http proxy server
if self.config.Port != "" {
addr := ":" + self.config.Port
addr := net.JoinHostPort(self.config.ListenAddr, self.config.Port)
go httpapi.StartHttpServer(self.api, &httpapi.ServerConfig{
Addr: addr,
CorsString: self.corsString,
})
}
log.Debug(fmt.Sprintf("Swarm http proxy started on port: %v", self.config.Port))
log.Info(fmt.Sprintf("Swarm http proxy started on %v", addr))
if self.corsString != "" {
log.Debug(fmt.Sprintf("Swarm http proxy started with corsdomain: %v", self.corsString))
}
}
return nil
}