Initial commit for branch 1.8.12

.travis.yml

@@ -146,7 +146,7 @@ matrix:
       git:
         submodules: false # avoid cloning ethereum/tests
       before_install:
-        - curl https://storage.googleapis.com/golang/go1.10.2.linux-amd64.tar.gz | tar -xz
+        - curl https://storage.googleapis.com/golang/go1.10.3.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.11
+1.8.12

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.2.windows-%GETH_ARCH%.zip
+  - appveyor DownloadFile https://storage.googleapis.com/golang/go1.10.3.windows-%GETH_ARCH%.zip
-  - 7z x go1.10.2.windows-%GETH_ARCH%.zip -y -oC:\ > NUL
+  - 7z x go1.10.3.windows-%GETH_ARCH%.zip -y -oC:\ > NUL
   - go version
   - gcc --version
 

build/ci.go

@@ -330,6 +330,7 @@ func doLint(cmdline []string) {
 	configs := []string{
 		"--vendor",
 		"--tests",
+		"--deadline=2m",
 		"--disable-all",
 		"--enable=goimports",
 		"--enable=varcheck",

cmd/utils/flags.go

@@ -1084,6 +1084,9 @@ func SetEthConfig(ctx *cli.Context, stack *node.Node, cfg *eth.Config) {
 		}
 		cfg.Genesis = core.DefaultRinkebyGenesisBlock()
 	case ctx.GlobalBool(DeveloperFlag.Name):
+		if !ctx.GlobalIsSet(NetworkIdFlag.Name) {
+			cfg.NetworkId = 1337
+		}
 		// Create new developer account or reuse existing one
 		var (
 			developer accounts.Account

common/hexutil/hexutil.go

@@ -39,6 +39,7 @@ import (
 
 const uintBits = 32 << (uint64(^uint(0)) >> 63)
 
+// Errors
 var (
 	ErrEmptyString   = &decError{"empty hex string"}
 	ErrSyntax        = &decError{"invalid hex string"}

common/math/big.go

@@ -22,12 +22,13 @@ import (
 	"math/big"
 )
 
+// Various big integer limit values.
 var (
 	tt255     = BigPow(2, 255)
 	tt256     = BigPow(2, 256)
 	tt256m1   = new(big.Int).Sub(tt256, big.NewInt(1))
-	MaxBig256 = new(big.Int).Set(tt256m1)
 	tt63      = BigPow(2, 63)
+	MaxBig256 = new(big.Int).Set(tt256m1)
 	MaxBig63  = new(big.Int).Sub(tt63, big.NewInt(1))
 )
 

common/math/integer.go

@@ -21,8 +21,8 @@ import (
 	"strconv"
 )
 
-const (
 // Integer limit values.
+const (
 	MaxInt8   = 1<<7 - 1
 	MinInt8   = -1 << 7
 	MaxInt16  = 1<<15 - 1

common/mclock/mclock.go

@@ -14,7 +14,7 @@
 // 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 mclock is a wrapper for a monotonic clock source
+// Package mclock is a wrapper for a monotonic clock source
 package mclock
 
 import (
@@ -23,8 +23,10 @@ import (
 	"github.com/aristanetworks/goarista/monotime"
 )
 
-type AbsTime time.Duration // absolute monotonic time
+// AbsTime represents absolute monotonic time.
+type AbsTime time.Duration
 
+// Now returns the current absolute monotonic time.
 func Now() AbsTime {
 	return AbsTime(monotime.Now())
 }

common/number/int.go

@@ -22,9 +22,11 @@ import (
 	"github.com/EthereumCommonwealth/go-callisto/common"
 )
 
-var tt256 = new(big.Int).Lsh(big.NewInt(1), 256)
+var (
-var tt256m1 = new(big.Int).Sub(new(big.Int).Lsh(big.NewInt(1), 256), big.NewInt(1))
+	tt256   = new(big.Int).Lsh(big.NewInt(1), 256)
-var tt255 = new(big.Int).Lsh(big.NewInt(1), 255)
+	tt256m1 = new(big.Int).Sub(new(big.Int).Lsh(big.NewInt(1), 256), big.NewInt(1))
+	tt255   = new(big.Int).Lsh(big.NewInt(1), 255)
+)
 
 func limitUnsigned256(x *Number) *Number {
 	x.num.And(x.num, tt256m1)
@@ -181,7 +183,6 @@ func (i *Number) FirstBitSet() int {
 }
 
 // Variables
-
 var (
 	Zero       = Uint(0)
 	One        = Uint(1)

common/types.go

@@ -29,6 +29,7 @@ import (
 	"github.com/EthereumCommonwealth/go-callisto/crypto/sha3"
 )
 
+// Lengths of hashes and addresses in bytes.
 const (
 	HashLength    = 32
 	AddressLength = 20

console/console.go

@@ -60,7 +60,7 @@ type Config struct {
 	Preload  []string     // Absolute paths to JavaScript files to preload
 }
 
-// Console is a JavaScript interpreted runtime environment. It is a fully fleged
+// Console is a JavaScript interpreted runtime environment. It is a fully fledged
 // JavaScript console attached to a running node via an external or in-process RPC
 // client.
 type Console struct {

contracts/ens/ens.go

@@ -95,7 +95,7 @@ func ensParentNode(name string) (common.Hash, common.Hash) {
 	}
 }
 
-func ensNode(name string) common.Hash {
+func EnsNode(name string) common.Hash {
 	parentNode, parentLabel := ensParentNode(name)
 	return crypto.Keccak256Hash(parentNode[:], parentLabel[:])
 }
@@ -136,7 +136,7 @@ func (self *ENS) getRegistrar(node [32]byte) (*contract.FIFSRegistrarSession, er
 
 // Resolve is a non-transactional call that returns the content hash associated with a name.
 func (self *ENS) Resolve(name string) (common.Hash, error) {
-	node := ensNode(name)
+	node := EnsNode(name)
 
 	resolver, err := self.getResolver(node)
 	if err != nil {
@@ -165,7 +165,7 @@ func (self *ENS) Register(name string) (*types.Transaction, error) {
 // SetContentHash sets the content hash associated with a name. Only works if the caller
 // owns the name, and the associated resolver implements a `setContent` function.
 func (self *ENS) SetContentHash(name string, hash common.Hash) (*types.Transaction, error) {
-	node := ensNode(name)
+	node := EnsNode(name)
 
 	resolver, err := self.getResolver(node)
 	if err != nil {

contracts/ens/ens_test.go

@@ -55,7 +55,7 @@ func TestENS(t *testing.T) {
 	if err != nil {
 		t.Fatalf("can't deploy resolver: %v", err)
 	}
-	if _, err := ens.SetResolver(ensNode(name), resolverAddr); err != nil {
+	if _, err := ens.SetResolver(EnsNode(name), resolverAddr); err != nil {
 		t.Fatalf("can't set resolver: %v", err)
 	}
 	contractBackend.Commit()

core/asm/lexer.go

@@ -242,7 +242,7 @@ func lexLabel(l *lexer) stateFn {
 }
 
 // lexInsideString lexes the inside of a string until
-// until the state function finds the closing quote.
+// the state function finds the closing quote.
 // It returns the lex text state function.
 func lexInsideString(l *lexer) stateFn {
 	if l.acceptRunUntil('"') {

core/tx_pool.go

@@ -1107,7 +1107,7 @@ func (pool *TxPool) demoteUnexecutables() {
 			log.Trace("Demoting pending transaction", "hash", hash)
 			pool.enqueueTx(hash, tx)
 		}
-		// If there's a gap in front, warn (should never happen) and postpone all transactions
+		// If there's a gap in front, alert (should never happen) and postpone all transactions
 		if list.Len() > 0 && list.txs.Get(nonce) == nil {
 			for _, tx := range list.Cap(0) {
 				hash := tx.Hash()

core/vm/instructions.go

@@ -556,7 +556,7 @@ func opMload(pc *uint64, evm *EVM, contract *Contract, memory *Memory, stack *St
 func opMstore(pc *uint64, evm *EVM, contract *Contract, memory *Memory, stack *Stack) ([]byte, error) {
 	// pop value of the stack
 	mStart, val := stack.pop(), stack.pop()
-	memory.Set(mStart.Uint64(), 32, math.PaddedBigBytes(val, 32))
+	memory.Set32(mStart.Uint64(), val)
 
 	evm.interpreter.intPool.put(mStart, val)
 	return nil, nil
@@ -570,9 +570,9 @@ func opMstore8(pc *uint64, evm *EVM, contract *Contract, memory *Memory, stack *
 }
 
 func opSload(pc *uint64, evm *EVM, contract *Contract, memory *Memory, stack *Stack) ([]byte, error) {
-	loc := common.BigToHash(stack.pop())
+	loc := stack.peek()
-	val := evm.StateDB.GetState(contract.Address(), loc).Big()
+	val := evm.StateDB.GetState(contract.Address(), common.BigToHash(loc))
-	stack.push(val)
+	loc.SetBytes(val.Bytes())
 	return nil, nil
 }
 

core/vm/instructions_test.go

@@ -425,3 +425,42 @@ func BenchmarkOpIsZero(b *testing.B) {
 	x := "FBCDEF090807060504030201ffffffffFBCDEF090807060504030201ffffffff"
 	opBenchmark(b, opIszero, x)
 }
+
+func TestOpMstore(t *testing.T) {
+	var (
+		env   = NewEVM(Context{}, nil, params.TestChainConfig, Config{})
+		stack = newstack()
+		mem   = NewMemory()
+	)
+	mem.Resize(64)
+	pc := uint64(0)
+	v := "abcdef00000000000000abba000000000deaf000000c0de00100000000133700"
+	stack.pushN(new(big.Int).SetBytes(common.Hex2Bytes(v)), big.NewInt(0))
+	opMstore(&pc, env, nil, mem, stack)
+	if got := common.Bytes2Hex(mem.Get(0, 32)); got != v {
+		t.Fatalf("Mstore fail, got %v, expected %v", got, v)
+	}
+	stack.pushN(big.NewInt(0x1), big.NewInt(0))
+	opMstore(&pc, env, nil, mem, stack)
+	if common.Bytes2Hex(mem.Get(0, 32)) != "0000000000000000000000000000000000000000000000000000000000000001" {
+		t.Fatalf("Mstore failed to overwrite previous value")
+	}
+}
+
+func BenchmarkOpMstore(bench *testing.B) {
+	var (
+		env   = NewEVM(Context{}, nil, params.TestChainConfig, Config{})
+		stack = newstack()
+		mem   = NewMemory()
+	)
+	mem.Resize(64)
+	pc := uint64(0)
+	memStart := big.NewInt(0)
+	value := big.NewInt(0x1337)
+
+	bench.ResetTimer()
+	for i := 0; i < bench.N; i++ {
+		stack.pushN(value, memStart)
+		opMstore(&pc, env, nil, mem, stack)
+	}
+}

core/vm/memory.go

@@ -16,7 +16,12 @@
 
 package vm
 
-import "fmt"
+import (
+	"fmt"
+	"math/big"
+
+	"github.com/EthereumCommonwealth/go-callisto/common/math"
+)
 
 // Memory implements a simple memory model for the ethereum virtual machine.
 type Memory struct {
@@ -30,19 +35,32 @@ func NewMemory() *Memory {
 
 // Set sets offset + size to value
 func (m *Memory) Set(offset, size uint64, value []byte) {
-	// length of store may never be less than offset + size.
-	// The store should be resized PRIOR to setting the memory
-	if size > uint64(len(m.store)) {
-		panic("INVALID memory: store empty")
-	}
-
 	// It's possible the offset is greater than 0 and size equals 0. This is because
 	// the calcMemSize (common.go) could potentially return 0 when size is zero (NO-OP)
 	if size > 0 {
+		// length of store may never be less than offset + size.
+		// The store should be resized PRIOR to setting the memory
+		if offset+size > uint64(len(m.store)) {
+			panic("invalid memory: store empty")
+		}
 		copy(m.store[offset:offset+size], value)
 	}
 }
 
+// Set32 sets the 32 bytes starting at offset to the value of val, left-padded with zeroes to
+// 32 bytes.
+func (m *Memory) Set32(offset uint64, val *big.Int) {
+	// length of store may never be less than offset + size.
+	// The store should be resized PRIOR to setting the memory
+	if offset+32 > uint64(len(m.store)) {
+		panic("invalid memory: store empty")
+	}
+	// Zero the memory area
+	copy(m.store[offset:offset+32], []byte{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0})
+	// Fill in relevant bits
+	math.ReadBits(val, m.store[offset:offset+32])
+}
+
 // Resize resizes the memory to size
 func (m *Memory) Resize(size uint64) {
 	if uint64(m.Len()) < size {

eth/api.go

@@ -359,7 +359,7 @@ type BadBlockArgs struct {
 	RLP   string                 `json:"rlp"`
 }
 
-// GetBadBLocks returns a list of the last 'bad blocks' that the client has seen on the network
+// GetBadBlocks returns a list of the last 'bad blocks' that the client has seen on the network
 // and returns them as a JSON list of block-hashes
 func (api *PrivateDebugAPI) GetBadBlocks(ctx context.Context) ([]*BadBlockArgs, error) {
 	blocks := api.eth.BlockChain().BadBlocks()
@@ -431,7 +431,7 @@ func storageRangeAt(st state.Trie, start []byte, maxResult int) (StorageRangeRes
 	return result, nil
 }
 
-// GetModifiedAccountsByumber returns all accounts that have changed between the
+// GetModifiedAccountsByNumber returns all accounts that have changed between the
 // two blocks specified. A change is defined as a difference in nonce, balance,
 // code hash, or storage hash.
 //

eth/api_backend.go

@@ -43,6 +43,7 @@ type EthAPIBackend struct {
 	gpo *gasprice.Oracle
 }
 
+// ChainConfig returns the active chain configuration.
 func (b *EthAPIBackend) ChainConfig() *params.ChainConfig {
 	return b.eth.chainConfig
 }

eth/backend.go

@@ -88,7 +88,7 @@ type Ethereum struct {
 	gasPrice  *big.Int
 	etherbase common.Address
 
-	networkId     uint64
+	networkID     uint64
 	netRPCService *ethapi.PublicNetAPI
 
 	lock sync.RWMutex // Protects the variadic fields (e.g. gas price and etherbase)
@@ -126,7 +126,7 @@ func New(ctx *node.ServiceContext, config *Config) (*Ethereum, error) {
 		accountManager: ctx.AccountManager,
 		engine:         CreateConsensusEngine(ctx, &config.Ethash, chainConfig, chainDb),
 		shutdownChan:   make(chan bool),
-		networkId:      config.NetworkId,
+		networkID:      config.NetworkId,
 		gasPrice:       config.GasPrice,
 		etherbase:      config.Etherbase,
 		bloomRequests:  make(chan chan *bloombits.Retrieval),
@@ -369,7 +369,7 @@ func (s *Ethereum) Engine() consensus.Engine           { return s.engine }
 func (s *Ethereum) ChainDb() ethdb.Database            { return s.chainDb }
 func (s *Ethereum) IsListening() bool                  { return true } // Always listening
 func (s *Ethereum) EthVersion() int                    { return int(s.protocolManager.SubProtocols[0].Version) }
-func (s *Ethereum) NetVersion() uint64                 { return s.networkId }
+func (s *Ethereum) NetVersion() uint64                 { return s.networkID }
 func (s *Ethereum) Downloader() *downloader.Downloader { return s.protocolManager.downloader }
 
 // Protocols implements node.Service, returning all the currently configured

eth/downloader/downloader.go

@@ -900,7 +900,7 @@ func (d *Downloader) fillHeaderSkeleton(from uint64, skeleton []*types.Header) (
 	var (
 		deliver = func(packet dataPack) (int, error) {
 			pack := packet.(*headerPack)
-			return d.queue.DeliverHeaders(pack.peerId, pack.headers, d.headerProcCh)
+			return d.queue.DeliverHeaders(pack.peerID, pack.headers, d.headerProcCh)
 		}
 		expire   = func() map[string]int { return d.queue.ExpireHeaders(d.requestTTL()) }
 		throttle = func() bool { return false }
@@ -930,7 +930,7 @@ func (d *Downloader) fetchBodies(from uint64) error {
 	var (
 		deliver = func(packet dataPack) (int, error) {
 			pack := packet.(*bodyPack)
-			return d.queue.DeliverBodies(pack.peerId, pack.transactions, pack.uncles)
+			return d.queue.DeliverBodies(pack.peerID, pack.transactions, pack.uncles)
 		}
 		expire   = func() map[string]int { return d.queue.ExpireBodies(d.requestTTL()) }
 		fetch    = func(p *peerConnection, req *fetchRequest) error { return p.FetchBodies(req) }
@@ -954,7 +954,7 @@ func (d *Downloader) fetchReceipts(from uint64) error {
 	var (
 		deliver = func(packet dataPack) (int, error) {
 			pack := packet.(*receiptPack)
-			return d.queue.DeliverReceipts(pack.peerId, pack.receipts)
+			return d.queue.DeliverReceipts(pack.peerID, pack.receipts)
 		}
 		expire   = func() map[string]int { return d.queue.ExpireReceipts(d.requestTTL()) }
 		fetch    = func(p *peerConnection, req *fetchRequest) error { return p.FetchReceipts(req) }

eth/downloader/queue.go

@@ -596,21 +596,21 @@ func (q *queue) cancel(request *fetchRequest, taskQueue *prque.Prque, pendPool m
 // Revoke cancels all pending requests belonging to a given peer. This method is
 // meant to be called during a peer drop to quickly reassign owned data fetches
 // to remaining nodes.
-func (q *queue) Revoke(peerId string) {
+func (q *queue) Revoke(peerID string) {
 	q.lock.Lock()
 	defer q.lock.Unlock()
 
-	if request, ok := q.blockPendPool[peerId]; ok {
+	if request, ok := q.blockPendPool[peerID]; ok {
 		for _, header := range request.Headers {
 			q.blockTaskQueue.Push(header, -float32(header.Number.Uint64()))
 		}
-		delete(q.blockPendPool, peerId)
+		delete(q.blockPendPool, peerID)
 	}
-	if request, ok := q.receiptPendPool[peerId]; ok {
+	if request, ok := q.receiptPendPool[peerID]; ok {
 		for _, header := range request.Headers {
 			q.receiptTaskQueue.Push(header, -float32(header.Number.Uint64()))
 		}
-		delete(q.receiptPendPool, peerId)
+		delete(q.receiptPendPool, peerID)
 	}
 }
 

eth/downloader/types.go

@@ -34,22 +34,22 @@ type dataPack interface {
 
 // headerPack is a batch of block headers returned by a peer.
 type headerPack struct {
-	peerId  string
+	peerID  string
 	headers []*types.Header
 }
 
-func (p *headerPack) PeerId() string { return p.peerId }
+func (p *headerPack) PeerId() string { return p.peerID }
 func (p *headerPack) Items() int     { return len(p.headers) }
 func (p *headerPack) Stats() string  { return fmt.Sprintf("%d", len(p.headers)) }
 
 // bodyPack is a batch of block bodies returned by a peer.
 type bodyPack struct {
-	peerId       string
+	peerID       string
 	transactions [][]*types.Transaction
 	uncles       [][]*types.Header
 }
 
-func (p *bodyPack) PeerId() string { return p.peerId }
+func (p *bodyPack) PeerId() string { return p.peerID }
 func (p *bodyPack) Items() int {
 	if len(p.transactions) <= len(p.uncles) {
 		return len(p.transactions)
@@ -60,20 +60,20 @@ func (p *bodyPack) Stats() string { return fmt.Sprintf("%d:%d", len(p.transactio
 
 // receiptPack is a batch of receipts returned by a peer.
 type receiptPack struct {
-	peerId   string
+	peerID   string
 	receipts [][]*types.Receipt
 }
 
-func (p *receiptPack) PeerId() string { return p.peerId }
+func (p *receiptPack) PeerId() string { return p.peerID }
 func (p *receiptPack) Items() int     { return len(p.receipts) }
 func (p *receiptPack) Stats() string  { return fmt.Sprintf("%d", len(p.receipts)) }
 
 // statePack is a batch of states returned by a peer.
 type statePack struct {
-	peerId string
+	peerID string
 	states [][]byte
 }
 
-func (p *statePack) PeerId() string { return p.peerId }
+func (p *statePack) PeerId() string { return p.peerID }
 func (p *statePack) Items() int     { return len(p.states) }
 func (p *statePack) Stats() string  { return fmt.Sprintf("%d", len(p.states)) }

eth/fetcher/fetcher.go

@@ -88,7 +88,7 @@ type headerFilterTask struct {
 	time    time.Time       // Arrival time of the headers
 }
 
-// headerFilterTask represents a batch of block bodies (transactions and uncles)
+// bodyFilterTask represents a batch of block bodies (transactions and uncles)
 // needing fetcher filtering.
 type bodyFilterTask struct {
 	peer         string                 // The source peer of block bodies

eth/filters/filter.go

@@ -258,9 +258,9 @@ Logs:
 		if len(topics) > len(log.Topics) {
 			continue Logs
 		}
-		for i, topics := range topics {
+		for i, sub := range topics {
-			match := len(topics) == 0 // empty rule set == wildcard
+			match := len(sub) == 0 // empty rule set == wildcard
-			for _, topic := range topics {
+			for _, topic := range sub {
 				if log.Topics[i] == topic {
 					match = true
 					break

eth/handler.go

@@ -64,7 +64,7 @@ func errResp(code errCode, format string, v ...interface{}) error {
 }
 
 type ProtocolManager struct {
-	networkId uint64
+	networkID uint64
 
 	fastSync  uint32 // Flag whether fast sync is enabled (gets disabled if we already have blocks)
 	acceptTxs uint32 // Flag whether we're considered synchronised (enables transaction processing)
@@ -98,10 +98,10 @@ type ProtocolManager struct {
 
 // NewProtocolManager returns a new Ethereum sub protocol manager. The Ethereum sub protocol manages peers capable
 // with the Ethereum network.
-func NewProtocolManager(config *params.ChainConfig, mode downloader.SyncMode, networkId uint64, mux *event.TypeMux, txpool txPool, engine consensus.Engine, blockchain *core.BlockChain, chaindb ethdb.Database) (*ProtocolManager, error) {
+func NewProtocolManager(config *params.ChainConfig, mode downloader.SyncMode, networkID uint64, mux *event.TypeMux, txpool txPool, engine consensus.Engine, blockchain *core.BlockChain, chaindb ethdb.Database) (*ProtocolManager, error) {
 	// Create the protocol manager with the base fields
 	manager := &ProtocolManager{
-		networkId:   networkId,
+		networkID:   networkID,
 		eventMux:    mux,
 		txpool:      txpool,
 		blockchain:  blockchain,
@@ -263,7 +263,7 @@ func (pm *ProtocolManager) handle(p *peer) error {
 		number  = head.Number.Uint64()
 		td      = pm.blockchain.GetTd(hash, number)
 	)
-	if err := p.Handshake(pm.networkId, td, hash, genesis.Hash()); err != nil {
+	if err := p.Handshake(pm.networkID, td, hash, genesis.Hash()); err != nil {
 		p.Log().Debug("Ethereum handshake failed", "err", err)
 		return err
 	}
@@ -779,7 +779,7 @@ type NodeInfo struct {
 func (pm *ProtocolManager) NodeInfo() *NodeInfo {
 	currentBlock := pm.blockchain.CurrentBlock()
 	return &NodeInfo{
-		Network:    pm.networkId,
+		Network:    pm.networkID,
 		Difficulty: pm.blockchain.GetTd(currentBlock.Hash(), currentBlock.NumberU64()),
 		Genesis:    pm.blockchain.Genesis().Hash(),
 		Config:     pm.blockchain.Config(),

internal/web3ext/web3ext.go

@@ -313,8 +313,8 @@ web3._extend({
 			params: 2
 		}),
 		new web3._extend.Method({
-			name: 'setMutexProfileRate',
+			name: 'setMutexProfileFraction',
-			call: 'debug_setMutexProfileRate',
+			call: 'debug_setMutexProfileFraction',
 			params: 1
 		}),
 		new web3._extend.Method({

log/format.go

@@ -15,7 +15,7 @@ import (
 
 const (
 	timeFormat     = "2006-01-02T15:04:05-0700"
-	termTimeFormat = "01-02|15:04:05"
+	termTimeFormat = "01-02|15:04:05.999999"
 	floatFormat    = 'f'
 	termMsgJust    = 40
 )

log/logger.go

@@ -12,6 +12,7 @@ const timeKey = "t"
 const lvlKey = "lvl"
 const msgKey = "msg"
 const errorKey = "LOG15_ERROR"
+const skipLevel = 2
 
 type Lvl int
 
@@ -127,13 +128,13 @@ type logger struct {
 	h   *swapHandler
 }
 
-func (l *logger) write(msg string, lvl Lvl, ctx []interface{}) {
+func (l *logger) write(msg string, lvl Lvl, ctx []interface{}, skip int) {
 	l.h.Log(&Record{
 		Time: time.Now(),
 		Lvl:  lvl,
 		Msg:  msg,
 		Ctx:  newContext(l.ctx, ctx),
-		Call: stack.Caller(2),
+		Call: stack.Caller(skip),
 		KeyNames: RecordKeyNames{
 			Time: timeKey,
 			Msg:  msgKey,
@@ -157,27 +158,27 @@ func newContext(prefix []interface{}, suffix []interface{}) []interface{} {
 }
 
 func (l *logger) Trace(msg string, ctx ...interface{}) {
-	l.write(msg, LvlTrace, ctx)
+	l.write(msg, LvlTrace, ctx, skipLevel)
 }
 
 func (l *logger) Debug(msg string, ctx ...interface{}) {
-	l.write(msg, LvlDebug, ctx)
+	l.write(msg, LvlDebug, ctx, skipLevel)
 }
 
 func (l *logger) Info(msg string, ctx ...interface{}) {
-	l.write(msg, LvlInfo, ctx)
+	l.write(msg, LvlInfo, ctx, skipLevel)
 }
 
 func (l *logger) Warn(msg string, ctx ...interface{}) {
-	l.write(msg, LvlWarn, ctx)
+	l.write(msg, LvlWarn, ctx, skipLevel)
 }
 
 func (l *logger) Error(msg string, ctx ...interface{}) {
-	l.write(msg, LvlError, ctx)
+	l.write(msg, LvlError, ctx, skipLevel)
 }
 
 func (l *logger) Crit(msg string, ctx ...interface{}) {
-	l.write(msg, LvlCrit, ctx)
+	l.write(msg, LvlCrit, ctx, skipLevel)
 	os.Exit(1)
 }
 

log/root.go

@@ -31,31 +31,40 @@ func Root() Logger {
 
 // Trace is a convenient alias for Root().Trace
 func Trace(msg string, ctx ...interface{}) {
-	root.write(msg, LvlTrace, ctx)
+	root.write(msg, LvlTrace, ctx, skipLevel)
 }
 
 // Debug is a convenient alias for Root().Debug
 func Debug(msg string, ctx ...interface{}) {
-	root.write(msg, LvlDebug, ctx)
+	root.write(msg, LvlDebug, ctx, skipLevel)
 }
 
 // Info is a convenient alias for Root().Info
 func Info(msg string, ctx ...interface{}) {
-	root.write(msg, LvlInfo, ctx)
+	root.write(msg, LvlInfo, ctx, skipLevel)
 }
 
 // Warn is a convenient alias for Root().Warn
 func Warn(msg string, ctx ...interface{}) {
-	root.write(msg, LvlWarn, ctx)
+	root.write(msg, LvlWarn, ctx, skipLevel)
 }
 
 // Error is a convenient alias for Root().Error
 func Error(msg string, ctx ...interface{}) {
-	root.write(msg, LvlError, ctx)
+	root.write(msg, LvlError, ctx, skipLevel)
 }
 
 // Crit is a convenient alias for Root().Crit
 func Crit(msg string, ctx ...interface{}) {
-	root.write(msg, LvlCrit, ctx)
+	root.write(msg, LvlCrit, ctx, skipLevel)
 	os.Exit(1)
 }
+
+// Output is a convenient alias for write, allowing for the modification of
+// the calldepth (number of stack frames to skip).
+// calldepth influences the reported line number of the log message.
+// A calldepth of zero reports the immediate caller of Output.
+// Non-zero calldepth skips as many stack frames.
+func Output(msg string, lvl Lvl, calldepth int, ctx ...interface{}) {
+	root.write(msg, lvl, ctx, calldepth+skipLevel)
+}

metrics/timer_test.go

@@ -47,8 +47,8 @@ func TestTimerStop(t *testing.T) {
 func TestTimerFunc(t *testing.T) {
 	tm := NewTimer()
 	tm.Time(func() { time.Sleep(50e6) })
-	if max := tm.Max(); 35e6 > max || max > 95e6 {
+	if max := tm.Max(); 35e6 > max || max > 145e6 {
-		t.Errorf("tm.Max(): 35e6 > %v || %v > 95e6\n", max, max)
+		t.Errorf("tm.Max(): 35e6 > %v || %v > 145e6\n", max, max)
 	}
 }
 

p2p/discover/table.go

@@ -480,16 +480,16 @@ func (tab *Table) doRevalidate(done chan<- struct{}) {
 	b := tab.buckets[bi]
 	if err == nil {
 		// The node responded, move it to the front.
-		log.Debug("Revalidated node", "b", bi, "id", last.ID)
+		log.Trace("Revalidated node", "b", bi, "id", last.ID)
 		b.bump(last)
 		return
 	}
 	// No reply received, pick a replacement or delete the node if there aren't
 	// any replacements.
 	if r := tab.replace(b, last); r != nil {
-		log.Debug("Replaced dead node", "b", bi, "id", last.ID, "ip", last.IP, "r", r.ID, "rip", r.IP)
+		log.Trace("Replaced dead node", "b", bi, "id", last.ID, "ip", last.IP, "r", r.ID, "rip", r.IP)
 	} else {
-		log.Debug("Removed dead node", "b", bi, "id", last.ID, "ip", last.IP)
+		log.Trace("Removed dead node", "b", bi, "id", last.ID, "ip", last.IP)
 	}
 }
 

p2p/protocols/protocol.go

@@ -33,7 +33,9 @@ import (
 	"fmt"
 	"reflect"
 	"sync"
+	"time"
 
+	"github.com/EthereumCommonwealth/go-callisto/metrics"
 	"github.com/EthereumCommonwealth/go-callisto/p2p"
 )
 
@@ -217,6 +219,8 @@ func (p *Peer) Drop(err error) {
 // this low level call will be wrapped by libraries providing routed or broadcast sends
 // but often just used to forward and push messages to directly connected peers
 func (p *Peer) Send(msg interface{}) error {
+	defer metrics.GetOrRegisterResettingTimer("peer.send_t", nil).UpdateSince(time.Now())
+	metrics.GetOrRegisterCounter("peer.send", nil).Inc(1)
 	code, found := p.spec.GetCode(msg)
 	if !found {
 		return errorf(ErrInvalidMsgType, "%v", code)

p2p/protocols/protocol_test.go

@@ -373,15 +373,14 @@ WAIT:
 	}
 
 }
-
+func XTestMultiplePeersDropSelf(t *testing.T) {
-func TestMultiplePeersDropSelf(t *testing.T) {
 	runMultiplePeers(t, 0,
 		fmt.Errorf("subprotocol error"),
 		fmt.Errorf("Message handler error: (msg code 3): dropped"),
 	)
 }
 
-func TestMultiplePeersDropOther(t *testing.T) {
+func XTestMultiplePeersDropOther(t *testing.T) {
 	runMultiplePeers(t, 1,
 		fmt.Errorf("Message handler error: (msg code 3): dropped"),
 		fmt.Errorf("subprotocol error"),

p2p/rlpx_test.go

@@ -30,12 +30,13 @@ import (
 	"testing"
 	"time"
 
-	"github.com/davecgh/go-spew/spew"
 	"github.com/EthereumCommonwealth/go-callisto/crypto"
 	"github.com/EthereumCommonwealth/go-callisto/crypto/ecies"
 	"github.com/EthereumCommonwealth/go-callisto/crypto/sha3"
 	"github.com/EthereumCommonwealth/go-callisto/p2p/discover"
+	"github.com/EthereumCommonwealth/go-callisto/p2p/simulations/pipes"
 	"github.com/EthereumCommonwealth/go-callisto/rlp"
+	"github.com/davecgh/go-spew/spew"
 )
 
 func TestSharedSecret(t *testing.T) {
@@ -159,7 +160,7 @@ func TestProtocolHandshake(t *testing.T) {
 		wg sync.WaitGroup
 	)
 
-	fd0, fd1, err := tcpPipe()
+	fd0, fd1, err := pipes.TCPPipe()
 	if err != nil {
 		t.Fatal(err)
 	}
@@ -601,31 +602,3 @@ func TestHandshakeForwardCompatibility(t *testing.T) {
 		t.Errorf("ingress-mac('foo') mismatch:\ngot %x\nwant %x", fooIngressHash, wantFooIngressHash)
 	}
 }
-
-// tcpPipe creates an in process full duplex pipe based on a localhost TCP socket
-func tcpPipe() (net.Conn, net.Conn, error) {
-	l, err := net.Listen("tcp", "127.0.0.1:0")
-	if err != nil {
-		return nil, nil, err
-	}
-	defer l.Close()
-
-	var aconn net.Conn
-	aerr := make(chan error, 1)
-	go func() {
-		var err error
-		aconn, err = l.Accept()
-		aerr <- err
-	}()
-
-	dconn, err := net.Dial("tcp", l.Addr().String())
-	if err != nil {
-		<-aerr
-		return nil, nil, err
-	}
-	if err := <-aerr; err != nil {
-		dconn.Close()
-		return nil, nil, err
-	}
-	return aconn, dconn, nil
-}

p2p/server.go

@@ -594,13 +594,13 @@ running:
 			// This channel is used by AddPeer to add to the
 			// ephemeral static peer list. Add it to the dialer,
 			// it will keep the node connected.
-			srv.log.Debug("Adding static node", "node", n)
+			srv.log.Trace("Adding static node", "node", n)
 			dialstate.addStatic(n)
 		case n := <-srv.removestatic:
 			// This channel is used by RemovePeer to send a
 			// disconnect request to a peer and begin the
 			// stop keeping the node connected
-			srv.log.Debug("Removing static node", "node", n)
+			srv.log.Trace("Removing static node", "node", n)
 			dialstate.removeStatic(n)
 			if p, ok := peers[n.ID]; ok {
 				p.Disconnect(DiscRequested)

p2p/simulations/adapters/docker.go

@@ -27,12 +27,15 @@ import (
 	"runtime"
 	"strings"
 
-	"github.com/EthereumCommonwealth/go-callisto/log"
 	"github.com/EthereumCommonwealth/go-callisto/node"
 	"github.com/EthereumCommonwealth/go-callisto/p2p/discover"
 	"github.com/docker/docker/pkg/reexec"
 )
 
+var (
+	ErrLinuxOnly = errors.New("DockerAdapter can only be used on Linux as it uses the current binary (which must be a Linux binary)")
+)
+
 // DockerAdapter is a NodeAdapter which runs simulation nodes inside Docker
 // containers.
 //
@@ -52,7 +55,7 @@ func NewDockerAdapter() (*DockerAdapter, error) {
 	// It is reasonable to require this because the caller can just
 	// compile the current binary in a Docker container.
 	if runtime.GOOS != "linux" {
-		return nil, errors.New("DockerAdapter can only be used on Linux as it uses the current binary (which must be a Linux binary)")
+		return nil, ErrLinuxOnly
 	}
 
 	if err := buildDockerImage(); err != nil {
@@ -95,7 +98,10 @@ func (d *DockerAdapter) NewNode(config *NodeConfig) (Node, error) {
 	conf.Stack.P2P.NoDiscovery = true
 	conf.Stack.P2P.NAT = nil
 	conf.Stack.NoUSB = true
-	conf.Stack.Logger = log.New("node.id", config.ID.String())
+
+	// listen on all interfaces on a given port, which we set when we
+	// initialise NodeConfig (usually a random port)
+	conf.Stack.P2P.ListenAddr = fmt.Sprintf(":%d", config.Port)
 
 	node := &DockerNode{
 		ExecNode: ExecNode{

p2p/simulations/adapters/exec.go

@@ -17,6 +17,7 @@
 package adapters
 
 import (
+	"bufio"
 	"context"
 	"crypto/ecdsa"
 	"encoding/json"
@@ -103,9 +104,9 @@ func (e *ExecAdapter) NewNode(config *NodeConfig) (Node, error) {
 	conf.Stack.P2P.NAT = nil
 	conf.Stack.NoUSB = true
 
-	// listen on a random localhost port (we'll get the actual port after
+	// listen on a localhost port, which we set when we
-	// starting the node through the RPC admin.nodeInfo method)
+	// initialise NodeConfig (usually a random port)
-	conf.Stack.P2P.ListenAddr = "127.0.0.1:0"
+	conf.Stack.P2P.ListenAddr = fmt.Sprintf(":%d", config.Port)
 
 	node := &ExecNode{
 		ID:      config.ID,
@@ -190,9 +191,23 @@ func (n *ExecNode) Start(snapshots map[string][]byte) (err error) {
 	n.Cmd = cmd
 
 	// read the WebSocket address from the stderr logs
-	wsAddr, err := findWSAddr(stderrR, 10*time.Second)
+	var wsAddr string
-	if err != nil {
+	wsAddrC := make(chan string)
-		return fmt.Errorf("error getting WebSocket address: %s", err)
+	go func() {
+		s := bufio.NewScanner(stderrR)
+		for s.Scan() {
+			if strings.Contains(s.Text(), "WebSocket endpoint opened") {
+				wsAddrC <- wsAddrPattern.FindString(s.Text())
+			}
+		}
+	}()
+	select {
+	case wsAddr = <-wsAddrC:
+		if wsAddr == "" {
+			return errors.New("failed to read WebSocket address from stderr")
+		}
+	case <-time.After(10 * time.Second):
+		return errors.New("timed out waiting for WebSocket address on stderr")
 	}
 
 	// create the RPC client and load the node info
@@ -318,6 +333,21 @@ type execNodeConfig struct {
 	PeerAddrs map[string]string `json:"peer_addrs,omitempty"`
 }
 
+// ExternalIP gets an external IP address so that Enode URL is usable
+func ExternalIP() net.IP {
+	addrs, err := net.InterfaceAddrs()
+	if err != nil {
+		log.Crit("error getting IP address", "err", err)
+	}
+	for _, addr := range addrs {
+		if ip, ok := addr.(*net.IPNet); ok && !ip.IP.IsLoopback() && !ip.IP.IsLinkLocalUnicast() {
+			return ip.IP
+		}
+	}
+	log.Warn("unable to determine explicit IP address, falling back to loopback")
+	return net.IP{127, 0, 0, 1}
+}
+
 // execP2PNode starts a devp2p node when the current binary is executed with
 // argv[0] being "p2p-node", reading the service / ID from argv[1] / argv[2]
 // and the node config from the _P2P_NODE_CONFIG environment variable
@@ -341,25 +371,11 @@ func execP2PNode() {
 	conf.Stack.P2P.PrivateKey = conf.Node.PrivateKey
 	conf.Stack.Logger = log.New("node.id", conf.Node.ID.String())
 
-	// use explicit IP address in ListenAddr so that Enode URL is usable
-	externalIP := func() string {
-		addrs, err := net.InterfaceAddrs()
-		if err != nil {
-			log.Crit("error getting IP address", "err", err)
-		}
-		for _, addr := range addrs {
-			if ip, ok := addr.(*net.IPNet); ok && !ip.IP.IsLoopback() {
-				return ip.IP.String()
-			}
-		}
-		log.Crit("unable to determine explicit IP address")
-		return ""
-	}
 	if strings.HasPrefix(conf.Stack.P2P.ListenAddr, ":") {
-		conf.Stack.P2P.ListenAddr = externalIP() + conf.Stack.P2P.ListenAddr
+		conf.Stack.P2P.ListenAddr = ExternalIP().String() + conf.Stack.P2P.ListenAddr
 	}
 	if conf.Stack.WSHost == "0.0.0.0" {
-		conf.Stack.WSHost = externalIP()
+		conf.Stack.WSHost = ExternalIP().String()
 	}
 
 	// initialize the devp2p stack

p2p/simulations/adapters/inproc.go

@@ -28,12 +28,14 @@ import (
 	"github.com/EthereumCommonwealth/go-callisto/node"
 	"github.com/EthereumCommonwealth/go-callisto/p2p"
 	"github.com/EthereumCommonwealth/go-callisto/p2p/discover"
+	"github.com/EthereumCommonwealth/go-callisto/p2p/simulations/pipes"
 	"github.com/EthereumCommonwealth/go-callisto/rpc"
 )
 
 // SimAdapter is a NodeAdapter which creates in-memory simulation nodes and
-// connects them using in-memory net.Pipe connections
+// connects them using net.Pipe
 type SimAdapter struct {
+	pipe     func() (net.Conn, net.Conn, error)
 	mtx      sync.RWMutex
 	nodes    map[discover.NodeID]*SimNode
 	services map[string]ServiceFunc
@@ -42,8 +44,18 @@ type SimAdapter struct {
 // NewSimAdapter creates a SimAdapter which is capable of running in-memory
 // simulation nodes running any of the given services (the services to run on a
 // particular node are passed to the NewNode function in the NodeConfig)
+// the adapter uses a net.Pipe for in-memory simulated network connections
 func NewSimAdapter(services map[string]ServiceFunc) *SimAdapter {
 	return &SimAdapter{
+		pipe:     pipes.NetPipe,
+		nodes:    make(map[discover.NodeID]*SimNode),
+		services: services,
+	}
+}
+
+func NewTCPAdapter(services map[string]ServiceFunc) *SimAdapter {
+	return &SimAdapter{
+		pipe:     pipes.TCPPipe,
 		nodes:    make(map[discover.NodeID]*SimNode),
 		services: services,
 	}
@@ -81,7 +93,7 @@ func (s *SimAdapter) NewNode(config *NodeConfig) (Node, error) {
 			MaxPeers:        math.MaxInt32,
 			NoDiscovery:     true,
 			Dialer:          s,
-			EnableMsgEvents: true,
+			EnableMsgEvents: config.EnableMsgEvents,
 		},
 		NoUSB:  true,
 		Logger: log.New("node.id", id.String()),
@@ -102,7 +114,7 @@ func (s *SimAdapter) NewNode(config *NodeConfig) (Node, error) {
 }
 
 // Dial implements the p2p.NodeDialer interface by connecting to the node using
-// an in-memory net.Pipe connection
+// an in-memory net.Pipe
 func (s *SimAdapter) Dial(dest *discover.Node) (conn net.Conn, err error) {
 	node, ok := s.GetNode(dest.ID)
 	if !ok {
@@ -112,7 +124,14 @@ func (s *SimAdapter) Dial(dest *discover.Node) (conn net.Conn, err error) {
 	if srv == nil {
 		return nil, fmt.Errorf("node not running: %s", dest.ID)
 	}
-	pipe1, pipe2 := net.Pipe()
+	// SimAdapter.pipe is net.Pipe (NewSimAdapter)
+	pipe1, pipe2, err := s.pipe()
+	if err != nil {
+		return nil, err
+	}
+	// this is simulated 'listening'
+	// asynchronously call the dialed destintion node's p2p server
+	// to set up connection on the 'listening' side
 	go srv.SetupConn(pipe1, 0, nil)
 	return pipe2, nil
 }
@@ -140,8 +159,8 @@ func (s *SimAdapter) GetNode(id discover.NodeID) (*SimNode, bool) {
 }
 
 // SimNode is an in-memory simulation node which connects to other nodes using
-// an in-memory net.Pipe connection (see SimAdapter.Dial), running devp2p
+// net.Pipe (see SimAdapter.Dial), running devp2p protocols directly over that
-// protocols directly over that pipe
+// pipe
 type SimNode struct {
 	lock         sync.RWMutex
 	ID           discover.NodeID
@@ -241,7 +260,7 @@ func (sn *SimNode) Start(snapshots map[string][]byte) error {
 		for _, name := range sn.config.Services {
 			if err := sn.node.Register(newService(name)); err != nil {
 				regErr = err
-				return
+				break
 			}
 		}
 	})
@@ -314,3 +333,18 @@ func (sn *SimNode) NodeInfo() *p2p.NodeInfo {
 	}
 	return server.NodeInfo()
 }
+
+func setSocketBuffer(conn net.Conn, socketReadBuffer int, socketWriteBuffer int) error {
+	switch v := conn.(type) {
+	case *net.UnixConn:
+		err := v.SetReadBuffer(socketReadBuffer)
+		if err != nil {
+			return err
+		}
+		err = v.SetWriteBuffer(socketWriteBuffer)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}

p2p/simulations/adapters/inproc_test.go

@@ -0,0 +1,259 @@
+// 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 adapters
+
+import (
+	"bytes"
+	"encoding/binary"
+	"fmt"
+	"testing"
+	"time"
+
+	"github.com/EthereumCommonwealth/go-callisto/p2p/simulations/pipes"
+)
+
+func TestTCPPipe(t *testing.T) {
+	c1, c2, err := pipes.TCPPipe()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	done := make(chan struct{})
+
+	go func() {
+		msgs := 50
+		size := 1024
+		for i := 0; i < msgs; i++ {
+			msg := make([]byte, size)
+			_ = binary.PutUvarint(msg, uint64(i))
+
+			_, err := c1.Write(msg)
+			if err != nil {
+				t.Fatal(err)
+			}
+		}
+
+		for i := 0; i < msgs; i++ {
+			msg := make([]byte, size)
+			_ = binary.PutUvarint(msg, uint64(i))
+
+			out := make([]byte, size)
+			_, err := c2.Read(out)
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			if !bytes.Equal(msg, out) {
+				t.Fatalf("expected %#v, got %#v", msg, out)
+			}
+		}
+		done <- struct{}{}
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(5 * time.Second):
+		t.Fatal("test timeout")
+	}
+}
+
+func TestTCPPipeBidirections(t *testing.T) {
+	c1, c2, err := pipes.TCPPipe()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	done := make(chan struct{})
+
+	go func() {
+		msgs := 50
+		size := 7
+		for i := 0; i < msgs; i++ {
+			msg := []byte(fmt.Sprintf("ping %02d", i))
+
+			_, err := c1.Write(msg)
+			if err != nil {
+				t.Fatal(err)
+			}
+		}
+
+		for i := 0; i < msgs; i++ {
+			expected := []byte(fmt.Sprintf("ping %02d", i))
+
+			out := make([]byte, size)
+			_, err := c2.Read(out)
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			if !bytes.Equal(expected, out) {
+				t.Fatalf("expected %#v, got %#v", out, expected)
+			} else {
+				msg := []byte(fmt.Sprintf("pong %02d", i))
+				_, err := c2.Write(msg)
+				if err != nil {
+					t.Fatal(err)
+				}
+			}
+		}
+
+		for i := 0; i < msgs; i++ {
+			expected := []byte(fmt.Sprintf("pong %02d", i))
+
+			out := make([]byte, size)
+			_, err := c1.Read(out)
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			if !bytes.Equal(expected, out) {
+				t.Fatalf("expected %#v, got %#v", out, expected)
+			}
+		}
+		done <- struct{}{}
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(5 * time.Second):
+		t.Fatal("test timeout")
+	}
+}
+
+func TestNetPipe(t *testing.T) {
+	c1, c2, err := pipes.NetPipe()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	done := make(chan struct{})
+
+	go func() {
+		msgs := 50
+		size := 1024
+		// netPipe is blocking, so writes are emitted asynchronously
+		go func() {
+			for i := 0; i < msgs; i++ {
+				msg := make([]byte, size)
+				_ = binary.PutUvarint(msg, uint64(i))
+
+				_, err := c1.Write(msg)
+				if err != nil {
+					t.Fatal(err)
+				}
+			}
+		}()
+
+		for i := 0; i < msgs; i++ {
+			msg := make([]byte, size)
+			_ = binary.PutUvarint(msg, uint64(i))
+
+			out := make([]byte, size)
+			_, err := c2.Read(out)
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			if !bytes.Equal(msg, out) {
+				t.Fatalf("expected %#v, got %#v", msg, out)
+			}
+		}
+
+		done <- struct{}{}
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(5 * time.Second):
+		t.Fatal("test timeout")
+	}
+}
+
+func TestNetPipeBidirections(t *testing.T) {
+	c1, c2, err := pipes.NetPipe()
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	done := make(chan struct{})
+
+	go func() {
+		msgs := 1000
+		size := 8
+		pingTemplate := "ping %03d"
+		pongTemplate := "pong %03d"
+
+		// netPipe is blocking, so writes are emitted asynchronously
+		go func() {
+			for i := 0; i < msgs; i++ {
+				msg := []byte(fmt.Sprintf(pingTemplate, i))
+
+				_, err := c1.Write(msg)
+				if err != nil {
+					t.Fatal(err)
+				}
+			}
+		}()
+
+		// netPipe is blocking, so reads for pong are emitted asynchronously
+		go func() {
+			for i := 0; i < msgs; i++ {
+				expected := []byte(fmt.Sprintf(pongTemplate, i))
+
+				out := make([]byte, size)
+				_, err := c1.Read(out)
+				if err != nil {
+					t.Fatal(err)
+				}
+
+				if !bytes.Equal(expected, out) {
+					t.Fatalf("expected %#v, got %#v", expected, out)
+				}
+			}
+
+			done <- struct{}{}
+		}()
+
+		// expect to read pings, and respond with pongs to the alternate connection
+		for i := 0; i < msgs; i++ {
+			expected := []byte(fmt.Sprintf(pingTemplate, i))
+
+			out := make([]byte, size)
+			_, err := c2.Read(out)
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			if !bytes.Equal(expected, out) {
+				t.Fatalf("expected %#v, got %#v", expected, out)
+			} else {
+				msg := []byte(fmt.Sprintf(pongTemplate, i))
+
+				_, err := c2.Write(msg)
+				if err != nil {
+					t.Fatal(err)
+				}
+			}
+		}
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(5 * time.Second):
+		t.Fatal("test timeout")
+	}
+}

p2p/simulations/adapters/types.go

@@ -23,6 +23,7 @@ import (
 	"fmt"
 	"net"
 	"os"
+	"strconv"
 
 	"github.com/docker/docker/pkg/reexec"
 	"github.com/EthereumCommonwealth/go-callisto/crypto"
@@ -97,6 +98,8 @@ type NodeConfig struct {
 
 	// function to sanction or prevent suggesting a peer
 	Reachable func(id discover.NodeID) bool
+
+	Port uint16
 }
 
 // nodeConfigJSON is used to encode and decode NodeConfig as JSON by encoding
@@ -106,6 +109,8 @@ type nodeConfigJSON struct {
 	PrivateKey      string   `json:"private_key"`
 	Name            string   `json:"name"`
 	Services        []string `json:"services"`
+	EnableMsgEvents bool     `json:"enable_msg_events"`
+	Port            uint16   `json:"port"`
 }
 
 // MarshalJSON implements the json.Marshaler interface by encoding the config
@@ -115,6 +120,8 @@ func (n *NodeConfig) MarshalJSON() ([]byte, error) {
 		ID:              n.ID.String(),
 		Name:            n.Name,
 		Services:        n.Services,
+		Port:            n.Port,
+		EnableMsgEvents: n.EnableMsgEvents,
 	}
 	if n.PrivateKey != nil {
 		confJSON.PrivateKey = hex.EncodeToString(crypto.FromECDSA(n.PrivateKey))
@@ -152,6 +159,8 @@ func (n *NodeConfig) UnmarshalJSON(data []byte) error {
 
 	n.Name = confJSON.Name
 	n.Services = confJSON.Services
+	n.Port = confJSON.Port
+	n.EnableMsgEvents = confJSON.EnableMsgEvents
 
 	return nil
 }
@@ -163,15 +172,38 @@ func RandomNodeConfig() *NodeConfig {
 	if err != nil {
 		panic("unable to generate key")
 	}
-	var id discover.NodeID
+
-	pubkey := crypto.FromECDSAPub(&key.PublicKey)
+	id := discover.PubkeyID(&key.PublicKey)
-	copy(id[:], pubkey[1:])
+	port, err := assignTCPPort()
+	if err != nil {
+		panic("unable to assign tcp port")
+	}
 	return &NodeConfig{
 		ID:              id,
+		Name:            fmt.Sprintf("node_%s", id.String()),
 		PrivateKey:      key,
+		Port:            port,
+		EnableMsgEvents: true,
 	}
 }
 
+func assignTCPPort() (uint16, error) {
+	l, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		return 0, err
+	}
+	l.Close()
+	_, port, err := net.SplitHostPort(l.Addr().String())
+	if err != nil {
+		return 0, err
+	}
+	p, err := strconv.ParseInt(port, 10, 32)
+	if err != nil {
+		return 0, err
+	}
+	return uint16(p), nil
+}
+
 // ServiceContext is a collection of options and methods which can be utilised
 // when starting services
 type ServiceContext struct {

p2p/simulations/http.go

@@ -561,7 +561,8 @@ func (s *Server) LoadSnapshot(w http.ResponseWriter, req *http.Request) {
 
 // CreateNode creates a node in the network using the given configuration
 func (s *Server) CreateNode(w http.ResponseWriter, req *http.Request) {
-	config := adapters.RandomNodeConfig()
+	config := &adapters.NodeConfig{}
+
 	err := json.NewDecoder(req.Body).Decode(config)
 	if err != nil && err != io.EOF {
 		http.Error(w, err.Error(), http.StatusBadRequest)

p2p/simulations/http_test.go

@@ -348,7 +348,8 @@ func startTestNetwork(t *testing.T, client *Client) []string {
 	nodeCount := 2
 	nodeIDs := make([]string, nodeCount)
 	for i := 0; i < nodeCount; i++ {
-		node, err := client.CreateNode(nil)
+		config := adapters.RandomNodeConfig()
+		node, err := client.CreateNode(config)
 		if err != nil {
 			t.Fatalf("error creating node: %s", err)
 		}
@@ -527,7 +528,9 @@ func TestHTTPNodeRPC(t *testing.T) {
 
 	// start a node in the network
 	client := NewClient(s.URL)
-	node, err := client.CreateNode(nil)
+
+	config := adapters.RandomNodeConfig()
+	node, err := client.CreateNode(config)
 	if err != nil {
 		t.Fatalf("error creating node: %s", err)
 	}
@@ -589,7 +592,8 @@ func TestHTTPSnapshot(t *testing.T) {
 	nodeCount := 2
 	nodes := make([]*p2p.NodeInfo, nodeCount)
 	for i := 0; i < nodeCount; i++ {
-		node, err := client.CreateNode(nil)
+		config := adapters.RandomNodeConfig()
+		node, err := client.CreateNode(config)
 		if err != nil {
 			t.Fatalf("error creating node: %s", err)
 		}

p2p/simulations/mocker.go

@@ -26,6 +26,7 @@ import (
 
 	"github.com/EthereumCommonwealth/go-callisto/log"
 	"github.com/EthereumCommonwealth/go-callisto/p2p/discover"
+	"github.com/EthereumCommonwealth/go-callisto/p2p/simulations/adapters"
 )
 
 //a map of mocker names to its function
@@ -102,8 +103,14 @@ func startStop(net *Network, quit chan struct{}, nodeCount int) {
 func probabilistic(net *Network, quit chan struct{}, nodeCount int) {
 	nodes, err := connectNodesInRing(net, nodeCount)
 	if err != nil {
+		select {
+		case <-quit:
+			//error may be due to abortion of mocking; so the quit channel is closed
+			return
+		default:
 			panic("Could not startup node network for mocker")
 		}
+	}
 	for {
 		select {
 		case <-quit:
@@ -143,7 +150,7 @@ func probabilistic(net *Network, quit chan struct{}, nodeCount int) {
 			log.Debug(fmt.Sprintf("node %v shutting down", nodes[i]))
 			err := net.Stop(nodes[i])
 			if err != nil {
-				log.Error(fmt.Sprintf("Error stopping node %s", nodes[i]))
+				log.Error("Error stopping node", "node", nodes[i])
 				wg.Done()
 				continue
 			}
@@ -151,7 +158,7 @@ func probabilistic(net *Network, quit chan struct{}, nodeCount int) {
 				time.Sleep(randWait)
 				err := net.Start(id)
 				if err != nil {
-					log.Error(fmt.Sprintf("Error starting node %s", id))
+					log.Error("Error starting node", "node", id)
 				}
 				wg.Done()
 			}(nodes[i])
@@ -165,9 +172,10 @@ func probabilistic(net *Network, quit chan struct{}, nodeCount int) {
 func connectNodesInRing(net *Network, nodeCount int) ([]discover.NodeID, error) {
 	ids := make([]discover.NodeID, nodeCount)
 	for i := 0; i < nodeCount; i++ {
-		node, err := net.NewNode()
+		conf := adapters.RandomNodeConfig()
+		node, err := net.NewNodeWithConfig(conf)
 		if err != nil {
-			log.Error("Error creating a node! %s", err)
+			log.Error("Error creating a node!", "err", err)
 			return nil, err
 		}
 		ids[i] = node.ID()
@@ -175,7 +183,7 @@ func connectNodesInRing(net *Network, nodeCount int) ([]discover.NodeID, error)
 
 	for _, id := range ids {
 		if err := net.Start(id); err != nil {
-			log.Error("Error starting a node! %s", err)
+			log.Error("Error starting a node!", "err", err)
 			return nil, err
 		}
 		log.Debug(fmt.Sprintf("node %v starting up", id))
@@ -183,7 +191,7 @@ func connectNodesInRing(net *Network, nodeCount int) ([]discover.NodeID, error)
 	for i, id := range ids {
 		peerID := ids[(i+1)%len(ids)]
 		if err := net.Connect(id, peerID); err != nil {
-			log.Error("Error connecting a node to a peer! %s", err)
+			log.Error("Error connecting a node to a peer!", "err", err)
 			return nil, err
 		}
 	}

p2p/simulations/network.go

@@ -382,6 +382,15 @@ func (net *Network) GetNodeByName(name string) *Node {
 	return net.getNodeByName(name)
 }
 
+// GetNodes returns the existing nodes
+func (net *Network) GetNodes() (nodes []*Node) {
+	net.lock.Lock()
+	defer net.lock.Unlock()
+
+	nodes = append(nodes, net.Nodes...)
+	return nodes
+}
+
 func (net *Network) getNode(id discover.NodeID) *Node {
 	i, found := net.nodeMap[id]
 	if !found {
@@ -399,15 +408,6 @@ func (net *Network) getNodeByName(name string) *Node {
 	return nil
 }
 
-// GetNodes returns the existing nodes
-func (net *Network) GetNodes() (nodes []*Node) {
-	net.lock.Lock()
-	defer net.lock.Unlock()
-
-	nodes = append(nodes, net.Nodes...)
-	return nodes
-}
-
 // GetConn returns the connection which exists between "one" and "other"
 // regardless of which node initiated the connection
 func (net *Network) GetConn(oneID, otherID discover.NodeID) *Conn {

p2p/simulations/network_test.go

@@ -41,7 +41,8 @@ func TestNetworkSimulation(t *testing.T) {
 	nodeCount := 20
 	ids := make([]discover.NodeID, nodeCount)
 	for i := 0; i < nodeCount; i++ {
-		node, err := network.NewNode()
+		conf := adapters.RandomNodeConfig()
+		node, err := network.NewNodeWithConfig(conf)
 		if err != nil {
 			t.Fatalf("error creating node: %s", err)
 		}

p2p/simulations/pipes/pipes.go

@@ -0,0 +1,55 @@
+// 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 pipes
+
+import (
+	"net"
+)
+
+// NetPipe wraps net.Pipe in a signature returning an error
+func NetPipe() (net.Conn, net.Conn, error) {
+	p1, p2 := net.Pipe()
+	return p1, p2, nil
+}
+
+// TCPPipe creates an in process full duplex pipe based on a localhost TCP socket
+func TCPPipe() (net.Conn, net.Conn, error) {
+	l, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		return nil, nil, err
+	}
+	defer l.Close()
+
+	var aconn net.Conn
+	aerr := make(chan error, 1)
+	go func() {
+		var err error
+		aconn, err = l.Accept()
+		aerr <- err
+	}()
+
+	dconn, err := net.Dial("tcp", l.Addr().String())
+	if err != nil {
+		<-aerr
+		return nil, nil, err
+	}
+	if err := <-aerr; err != nil {
+		dconn.Close()
+		return nil, nil, err
+	}
+	return aconn, dconn, nil
+}

params/version.go

@@ -23,8 +23,8 @@ import (
 const (
 	VersionMajor = 1              // Major version component of the current release
 	VersionMinor = 8              // Minor version component of the current release
-	VersionPatch = 11           // Patch version component of the current release
+	VersionPatch = 12             // Patch version component of the current release
-	VersionMeta  = "CLO stable" // Version metadata to append to the version string
+	VersionMeta  = "CLO Unstable" // Version metadata to append to the version string
 )
 
 // Version holds the textual version string.

rpc/client.go

@@ -61,7 +61,7 @@ const (
 	// The approach taken here is to maintain a per-subscription linked list buffer
 	// shrinks on demand. If the buffer reaches the size below, the subscription is
 	// dropped.
-	maxClientSubscriptionBuffer = 8000
+	maxClientSubscriptionBuffer = 20000
 )
 
 // BatchElem is an element in a batch request.

whisper/whisperv5/peer_test.go

@@ -19,7 +19,6 @@ package whisperv5
 import (
 	"bytes"
 	"crypto/ecdsa"
-	"fmt"
 	"net"
 	"sync"
 	"testing"
@@ -108,8 +107,6 @@ func TestSimulation(t *testing.T) {
 
 func initialize(t *testing.T) {
 	var err error
-	ip := net.IPv4(127, 0, 0, 1)
-	port0 := 30303
 
 	for i := 0; i < NumNodes; i++ {
 		var node TestNode
@@ -128,29 +125,15 @@ func initialize(t *testing.T) {
 		if err != nil {
 			t.Fatalf("failed convert the key: %s.", keys[i])
 		}
-		port := port0 + i
-		addr := fmt.Sprintf(":%d", port) // e.g. ":30303"
 		name := common.MakeName("whisper-go", "2.0")
-		var peers []*discover.Node
-		if i > 0 {
-			peerNodeId := nodes[i-1].id
-			peerPort := uint16(port - 1)
-			peerNode := discover.PubkeyID(&peerNodeId.PublicKey)
-			peer := discover.NewNode(peerNode, ip, peerPort, peerPort)
-			peers = append(peers, peer)
-		}
-
 		node.server = &p2p.Server{
 			Config: p2p.Config{
 				PrivateKey: node.id,
 				MaxPeers:   NumNodes/2 + 1,
 				Name:       name,
 				Protocols:  node.shh.Protocols(),
-				ListenAddr:     addr,
+				ListenAddr: "127.0.0.1:0",
 				NAT:        nat.Any(),
-				BootstrapNodes: peers,
-				StaticNodes:    peers,
-				TrustedNodes:   peers,
 			},
 		}
 
@@ -159,6 +142,15 @@ func initialize(t *testing.T) {
 			t.Fatalf("failed to start server %d.", i)
 		}
 
+		for j := 0; j < i; j++ {
+			peerNodeId := nodes[j].id
+			address, _ := net.ResolveTCPAddr("tcp", nodes[j].server.ListenAddr)
+			peerPort := uint16(address.Port)
+			peerNode := discover.PubkeyID(&peerNodeId.PublicKey)
+			peer := discover.NewNode(peerNode, address.IP, peerPort, peerPort)
+			node.server.AddPeer(peer)
+		}
+
 		nodes[i] = &node
 	}
 }

whisper/whisperv6/peer_test.go

@@ -21,12 +21,13 @@ import (
 	"crypto/ecdsa"
 	"fmt"
 	mrand "math/rand"
-	"net"
 	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
 
+	"net"
+
 	"github.com/EthereumCommonwealth/go-callisto/common"
 	"github.com/EthereumCommonwealth/go-callisto/common/hexutil"
 	"github.com/EthereumCommonwealth/go-callisto/crypto"
@@ -173,8 +174,6 @@ func initialize(t *testing.T) {
 	initBloom(t)
 
 	var err error
-	ip := net.IPv4(127, 0, 0, 1)
-	port0 := 30303
 
 	for i := 0; i < NumNodes; i++ {
 		var node TestNode
@@ -199,17 +198,7 @@ func initialize(t *testing.T) {
 		if err != nil {
 			t.Fatalf("failed convert the key: %s.", keys[i])
 		}
-		port := port0 + i
-		addr := fmt.Sprintf(":%d", port) // e.g. ":30303"
 		name := common.MakeName("whisper-go", "2.0")
-		var peers []*discover.Node
-		if i > 0 {
-			peerNodeID := nodes[i-1].id
-			peerPort := uint16(port - 1)
-			peerNode := discover.PubkeyID(&peerNodeID.PublicKey)
-			peer := discover.NewNode(peerNode, ip, peerPort, peerPort)
-			peers = append(peers, peer)
-		}
 
 		node.server = &p2p.Server{
 			Config: p2p.Config{
@@ -217,22 +206,28 @@ func initialize(t *testing.T) {
 				MaxPeers:   NumNodes/2 + 1,
 				Name:       name,
 				Protocols:  node.shh.Protocols(),
-				ListenAddr:     addr,
+				ListenAddr: "127.0.0.1:0",
 				NAT:        nat.Any(),
-				BootstrapNodes: peers,
-				StaticNodes:    peers,
-				TrustedNodes:   peers,
 			},
 		}
 
+		go startServer(t, node.server)
+
 		nodes[i] = &node
 	}
 
-	for i := 0; i < NumNodes; i++ {
-		go startServer(t, nodes[i].server)
-	}
-
 	waitForServersToStart(t)
+
+	for i := 0; i < NumNodes; i++ {
+		for j := 0; j < i; j++ {
+			peerNodeId := nodes[j].id
+			address, _ := net.ResolveTCPAddr("tcp", nodes[j].server.ListenAddr)
+			peerPort := uint16(address.Port)
+			peerNode := discover.PubkeyID(&peerNodeId.PublicKey)
+			peer := discover.NewNode(peerNode, address.IP, peerPort, peerPort)
+			nodes[i].server.AddPeer(peer)
+		}
+	}
 }
 
 func startServer(t *testing.T, s *p2p.Server) {