Locks, refactor, tests

* Added additional chain tests
* Added proper mutex' on chain
* Removed ethereum dependencies

core/block_manager.go

@@ -256,12 +256,12 @@ func (sm *BlockManager) CalculateTD(block *types.Block) (*big.Int, bool) {
 
 	// TD(genesis_block) = 0 and TD(B) = TD(B.parent) + sum(u.difficulty for u in B.uncles) + B.difficulty
 	td := new(big.Int)
-	td = td.Add(sm.bc.TD, uncleDiff)
+	td = td.Add(sm.bc.Td(), uncleDiff)
 	td = td.Add(td, block.Difficulty)
 
 	// The new TD will only be accepted if the new difficulty is
 	// is greater than the previous.
-	if td.Cmp(sm.bc.TD) > 0 {
+	if td.Cmp(sm.bc.Td()) > 0 {
 		return td, true
 	}
 
@@ -279,7 +279,7 @@ func (sm *BlockManager) ValidateBlock(block, parent *types.Block) error {
 
 	diff := block.Time - parent.Time
 	if diff < 0 {
-		return ValidationError("Block timestamp less then prev block %v (%v - %v)", diff, block.Time, sm.bc.CurrentBlock.Time)
+		return ValidationError("Block timestamp less then prev block %v (%v - %v)", diff, block.Time, sm.bc.CurrentBlock().Time)
 	}
 
 	/* XXX

core/chain_manager.go

@@ -3,6 +3,7 @@ package core
 import (
 	"fmt"
 	"math/big"
+	"sync"
 
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/ethutil"
@@ -50,14 +51,34 @@ type ChainManager struct {
 	eventMux     *event.TypeMux
 	genesisBlock *types.Block
 	// Last known total difficulty
-	TD *big.Int
+	mu              sync.RWMutex
+	td              *big.Int
+	lastBlockNumber uint64
+	currentBlock    *types.Block
+	lastBlockHash   []byte
 
-	LastBlockNumber uint64
+	transState *state.StateDB
+}
 
-	CurrentBlock  *types.Block
+func (self *ChainManager) Td() *big.Int {
-	LastBlockHash []byte
+	self.mu.RLock()
+	defer self.mu.RUnlock()
 
-	transState *state.StateDB
+	return self.td
+}
+
+func (self *ChainManager) LastBlockNumber() uint64 {
+	self.mu.RLock()
+	defer self.mu.RUnlock()
+
+	return self.lastBlockNumber
+}
+
+func (self *ChainManager) CurrentBlock() *types.Block {
+	self.mu.RLock()
+	defer self.mu.RUnlock()
+
+	return self.currentBlock
 }
 
 func NewChainManager(mux *event.TypeMux) *ChainManager {
@@ -77,7 +98,7 @@ func (self *ChainManager) SetProcessor(proc types.BlockProcessor) {
 }
 
 func (self *ChainManager) State() *state.StateDB {
-	return self.CurrentBlock.State()
+	return self.CurrentBlock().State()
 }
 
 func (self *ChainManager) TransState() *state.StateDB {
@@ -91,27 +112,30 @@ func (bc *ChainManager) setLastBlock() {
 		AddTestNetFunds(bc.genesisBlock)
 
 		block := types.NewBlockFromBytes(data)
-		bc.CurrentBlock = block
+		bc.currentBlock = block
-		bc.LastBlockHash = block.Hash()
+		bc.lastBlockHash = block.Hash()
-		bc.LastBlockNumber = block.Number.Uint64()
+		bc.lastBlockNumber = block.Number.Uint64()
 
 		// Set the last know difficulty (might be 0x0 as initial value, Genesis)
-		bc.TD = ethutil.BigD(ethutil.Config.Db.LastKnownTD())
+		bc.td = ethutil.BigD(ethutil.Config.Db.LastKnownTD())
 	} else {
 		bc.Reset()
 	}
 
-	chainlogger.Infof("Last block (#%d) %x\n", bc.LastBlockNumber, bc.CurrentBlock.Hash())
+	chainlogger.Infof("Last block (#%d) %x\n", bc.lastBlockNumber, bc.currentBlock.Hash())
 }
 
 // Block creation & chain handling
 func (bc *ChainManager) NewBlock(coinbase []byte) *types.Block {
+	bc.mu.RLock()
+	defer bc.mu.RUnlock()
+
 	var root interface{}
 	hash := ZeroHash256
 
 	if bc.CurrentBlock != nil {
-		root = bc.CurrentBlock.Root()
+		root = bc.currentBlock.Root()
-		hash = bc.LastBlockHash
+		hash = bc.lastBlockHash
 	}
 
 	block := types.CreateBlock(
@@ -122,11 +146,11 @@ func (bc *ChainManager) NewBlock(coinbase []byte) *types.Block {
 		nil,
 		"")
 
-	parent := bc.CurrentBlock
+	parent := bc.currentBlock
 	if parent != nil {
 		block.Difficulty = CalcDifficulty(block, parent)
-		block.Number = new(big.Int).Add(bc.CurrentBlock.Number, ethutil.Big1)
+		block.Number = new(big.Int).Add(bc.currentBlock.Number, ethutil.Big1)
-		block.GasLimit = block.CalcGasLimit(bc.CurrentBlock)
+		block.GasLimit = block.CalcGasLimit(bc.currentBlock)
 
 	}
 
@@ -134,35 +158,42 @@ func (bc *ChainManager) NewBlock(coinbase []byte) *types.Block {
 }
 
 func (bc *ChainManager) Reset() {
+	bc.mu.Lock()
+	defer bc.mu.Unlock()
+
 	AddTestNetFunds(bc.genesisBlock)
 
 	bc.genesisBlock.Trie().Sync()
 	// Prepare the genesis block
 	bc.write(bc.genesisBlock)
 	bc.insert(bc.genesisBlock)
-	bc.CurrentBlock = bc.genesisBlock
+	bc.currentBlock = bc.genesisBlock
 
-	bc.SetTotalDifficulty(ethutil.Big("0"))
+	bc.setTotalDifficulty(ethutil.Big("0"))
 
 	// Set the last know difficulty (might be 0x0 as initial value, Genesis)
-	bc.TD = ethutil.BigD(ethutil.Config.Db.LastKnownTD())
+	bc.td = ethutil.BigD(ethutil.Config.Db.LastKnownTD())
 }
 
 func (self *ChainManager) Export() []byte {
-	chainlogger.Infoln("exporting", self.CurrentBlock.Number, "blocks")
+	self.mu.RLock()
+	defer self.mu.RUnlock()
+
+	chainlogger.Infof("exporting %v blocks...\n", self.currentBlock.Number)
 
-	blocks := make(types.Blocks, int(self.CurrentBlock.Number.Int64())+1)
+	blocks := make([]*types.Block, int(self.currentBlock.Number.Int64())+1)
-	for block := self.CurrentBlock; block != nil; block = self.GetBlock(block.PrevHash) {
+	for block := self.currentBlock; block != nil; block = self.GetBlock(block.PrevHash) {
 		blocks[block.Number.Int64()] = block
 	}
+
 	return ethutil.Encode(blocks)
 }
 
 func (bc *ChainManager) insert(block *types.Block) {
 	encodedBlock := block.RlpEncode()
 	ethutil.Config.Db.Put([]byte("LastBlock"), encodedBlock)
-	bc.CurrentBlock = block
+	bc.currentBlock = block
-	bc.LastBlockHash = block.Hash()
+	bc.lastBlockHash = block.Hash()
 }
 
 func (bc *ChainManager) write(block *types.Block) {
@@ -213,7 +244,10 @@ func (self *ChainManager) GetBlock(hash []byte) *types.Block {
 }
 
 func (self *ChainManager) GetBlockByNumber(num uint64) *types.Block {
-	block := self.CurrentBlock
+	self.mu.RLock()
+	defer self.mu.RUnlock()
+
+	block := self.currentBlock
 	for ; block != nil; block = self.GetBlock(block.PrevHash) {
 		if block.Number.Uint64() == num {
 			break
@@ -227,9 +261,9 @@ func (self *ChainManager) GetBlockByNumber(num uint64) *types.Block {
 	return block
 }
 
-func (bc *ChainManager) SetTotalDifficulty(td *big.Int) {
+func (bc *ChainManager) setTotalDifficulty(td *big.Int) {
 	ethutil.Config.Db.Put([]byte("LTD"), td.Bytes())
-	bc.TD = td
+	bc.td = td
 }
 
 func (self *ChainManager) CalcTotalDiff(block *types.Block) (*big.Int, error) {
@@ -262,8 +296,8 @@ func (bc *ChainManager) BlockInfo(block *types.Block) types.BlockInfo {
 
 // Unexported method for writing extra non-essential block info to the db
 func (bc *ChainManager) writeBlockInfo(block *types.Block) {
-	bc.LastBlockNumber++
+	bc.lastBlockNumber++
-	bi := types.BlockInfo{Number: bc.LastBlockNumber, Hash: block.Hash(), Parent: block.PrevHash, TD: bc.TD}
+	bi := types.BlockInfo{Number: bc.lastBlockNumber, Hash: block.Hash(), Parent: block.PrevHash, TD: bc.td}
 
 	// For now we use the block hash with the words "info" appended as key
 	ethutil.Config.Db.Put(append(block.Hash(), []byte("Info")...), bi.RlpEncode())
@@ -289,17 +323,22 @@ func (self *ChainManager) InsertChain(chain types.Blocks) error {
 			return err
 		}
 
-		self.write(block)
+		self.mu.Lock()
-		if td.Cmp(self.TD) > 0 {
+		{
-			if block.Number.Cmp(new(big.Int).Add(self.CurrentBlock.Number, ethutil.Big1)) < 0 {
+
-				chainlogger.Infof("Split detected. New head #%v (%x), was #%v (%x)\n", block.Number, block.Hash()[:4], self.CurrentBlock.Number, self.CurrentBlock.Hash()[:4])
+			self.write(block)
+			if td.Cmp(self.td) > 0 {
+				if block.Number.Cmp(new(big.Int).Add(self.currentBlock.Number, ethutil.Big1)) < 0 {
+					chainlogger.Infof("Split detected. New head #%v (%x), was #%v (%x)\n", block.Number, block.Hash()[:4], self.currentBlock.Number, self.currentBlock.Hash()[:4])
+				}
+
+				self.setTotalDifficulty(td)
+				self.insert(block)
+				self.transState = self.currentBlock.State().Copy()
 			}
 
-			self.SetTotalDifficulty(td)
-			self.insert(block)
-			self.transState = self.State().Copy()
-			//sm.eth.TxPool().RemoveSet(block.Transactions())
 		}
+		self.mu.Unlock()
 
 		self.eventMux.Post(NewBlockEvent{block})
 		self.eventMux.Post(messages)

core/chain_manager_test.go

@@ -3,17 +3,75 @@ package core
 import (
 	"fmt"
 	"path"
+	"runtime"
 	"testing"
 
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/ethdb"
 	"github.com/ethereum/go-ethereum/ethutil"
+	"github.com/ethereum/go-ethereum/event"
+	//logpkg "github.com/ethereum/go-ethereum/logger"
 )
 
-func TestChainInsertions(t *testing.T) {
+//var Logger logpkg.LogSystem
-	c1, err := ethutil.ReadAllFile(path.Join("..", "_data", "chain1"))
+//var Log = logpkg.NewLogger("TEST")
+
+func init() {
+	runtime.GOMAXPROCS(runtime.NumCPU())
+	//Logger = logpkg.NewStdLogSystem(os.Stdout, log.LstdFlags, logpkg.InfoLevel)
+	//logpkg.AddLogSystem(Logger)
+
+	ethutil.ReadConfig("/tmp/ethtest", "/tmp/ethtest", "ETH")
+
+	db, err := ethdb.NewMemDatabase()
+	if err != nil {
+		panic("Could not create mem-db, failing")
+	}
+	ethutil.Config.Db = db
+}
+
+func loadChain(fn string, t *testing.T) types.Blocks {
+	c1, err := ethutil.ReadAllFile(path.Join("..", "_data", fn))
 	if err != nil {
 		fmt.Println(err)
 		t.FailNow()
 	}
-	data1, _ := ethutil.Decode([]byte(c1), 0)
+	value := ethutil.NewValueFromBytes([]byte(c1))
-	fmt.Println(data1)
+	blocks := make(types.Blocks, value.Len())
+	it := value.NewIterator()
+	for it.Next() {
+		blocks[it.Idx()] = types.NewBlockFromRlpValue(it.Value())
+	}
+
+	return blocks
+}
+
+func insertChain(done chan bool, chainMan *ChainManager, chain types.Blocks, t *testing.T) {
+	err := chainMan.InsertChain(chain)
+	if err != nil {
+		fmt.Println(err)
+		t.FailNow()
+	}
+	done <- true
+}
+
+func TestChainInsertions(t *testing.T) {
+	chain1 := loadChain("chain1", t)
+	chain2 := loadChain("chain2", t)
+	var eventMux event.TypeMux
+	chainMan := NewChainManager(&eventMux)
+	txPool := NewTxPool(chainMan, nil, &eventMux)
+	blockMan := NewBlockManager(txPool, chainMan, &eventMux)
+	chainMan.SetProcessor(blockMan)
+
+	const max = 2
+	done := make(chan bool, max)
+
+	go insertChain(done, chainMan, chain1, t)
+	go insertChain(done, chainMan, chain2, t)
+
+	for i := 0; i < max; i++ {
+		<-done
+	}
+	fmt.Println(chainMan.CurrentBlock())
 }

core/filter.go

@@ -78,11 +78,11 @@ func (self *Filter) SetSkip(skip int) {
 func (self *Filter) Find() []*state.Message {
 	var earliestBlockNo uint64 = uint64(self.earliest)
 	if self.earliest == -1 {
-		earliestBlockNo = self.eth.ChainManager().CurrentBlock.Number.Uint64()
+		earliestBlockNo = self.eth.ChainManager().CurrentBlock().Number.Uint64()
 	}
 	var latestBlockNo uint64 = uint64(self.latest)
 	if self.latest == -1 {
-		latestBlockNo = self.eth.ChainManager().CurrentBlock.Number.Uint64()
+		latestBlockNo = self.eth.ChainManager().CurrentBlock().Number.Uint64()
 	}
 
 	var (

core/filter_test.go

@@ -1,7 +1 @@
 package core
-
-// import "testing"
-
-// func TestFilter(t *testing.T) {
-// 	NewFilter(NewTestManager())
-// }

core/transaction_pool.go

@@ -8,6 +8,7 @@ import (
 	"sync"
 
 	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/event"
 	"github.com/ethereum/go-ethereum/logger"
 	"github.com/ethereum/go-ethereum/state"
 	"github.com/ethereum/go-ethereum/wire"
@@ -61,7 +62,6 @@ type TxProcessor interface {
 // pool is being drained or synced for whatever reason the transactions
 // will simple queue up and handled when the mutex is freed.
 type TxPool struct {
-	Ethereum EthManager
 	// The mutex for accessing the Tx pool.
 	mutex sync.Mutex
 	// Queueing channel for reading and writing incoming
@@ -75,14 +75,20 @@ type TxPool struct {
 	SecondaryProcessor TxProcessor
 
 	subscribers []chan TxMsg
+
+	broadcaster  types.Broadcaster
+	chainManager *ChainManager
+	eventMux     *event.TypeMux
 }
 
-func NewTxPool(ethereum EthManager) *TxPool {
+func NewTxPool(chainManager *ChainManager, broadcaster types.Broadcaster, eventMux *event.TypeMux) *TxPool {
 	return &TxPool{
-		pool:      list.New(),
+		pool:         list.New(),
-		queueChan: make(chan *types.Transaction, txPoolQueueSize),
+		queueChan:    make(chan *types.Transaction, txPoolQueueSize),
-		quit:      make(chan bool),
+		quit:         make(chan bool),
-		Ethereum:  ethereum,
+		chainManager: chainManager,
+		eventMux:     eventMux,
+		broadcaster:  broadcaster,
 	}
 }
 
@@ -94,13 +100,13 @@ func (pool *TxPool) addTransaction(tx *types.Transaction) {
 	pool.pool.PushBack(tx)
 
 	// Broadcast the transaction to the rest of the peers
-	pool.Ethereum.Broadcast(wire.MsgTxTy, []interface{}{tx.RlpData()})
+	pool.broadcaster.Broadcast(wire.MsgTxTy, []interface{}{tx.RlpData()})
 }
 
 func (pool *TxPool) ValidateTransaction(tx *types.Transaction) error {
 	// Get the last block so we can retrieve the sender and receiver from
 	// the merkle trie
-	block := pool.Ethereum.ChainManager().CurrentBlock
+	block := pool.chainManager.CurrentBlock
 	// Something has gone horribly wrong if this happens
 	if block == nil {
 		return fmt.Errorf("No last block on the block chain")
@@ -116,7 +122,7 @@ func (pool *TxPool) ValidateTransaction(tx *types.Transaction) error {
 	}
 
 	// Get the sender
-	sender := pool.Ethereum.ChainManager().State().GetAccount(tx.Sender())
+	sender := pool.chainManager.State().GetAccount(tx.Sender())
 
 	totAmount := new(big.Int).Set(tx.Value)
 	// Make sure there's enough in the sender's account. Having insufficient
@@ -160,7 +166,7 @@ func (self *TxPool) Add(tx *types.Transaction) error {
 	txplogger.Debugf("(t) %x => %x (%v) %x\n", tx.Sender()[:4], tmp, tx.Value, tx.Hash())
 
 	// Notify the subscribers
-	go self.Ethereum.EventMux().Post(TxPreEvent{tx})
+	go self.eventMux.Post(TxPreEvent{tx})
 
 	return nil
 }

core/types/block.go

@@ -409,7 +409,7 @@ func (self *Block) Size() ethutil.StorageSize {
 
 // Implement RlpEncodable
 func (self *Block) RlpData() interface{} {
-	return self.Value().Val
+	return []interface{}{self.header(), self.transactions, self.rlpUncles()}
 }
 
 // Implement pow.Block

core/types/common.go

@@ -4,8 +4,13 @@ import (
 	"math/big"
 
 	"github.com/ethereum/go-ethereum/state"
+	"github.com/ethereum/go-ethereum/wire"
 )
 
 type BlockProcessor interface {
 	Process(*Block) (*big.Int, state.Messages, error)
 }
+
+type Broadcaster interface {
+	Broadcast(wire.MsgType, []interface{})
+}

ethereum.go

@@ -128,9 +128,9 @@ func New(db ethutil.Database, clientIdentity wire.ClientIdentity, keyManager *cr
 	}
 
 	ethereum.blockPool = NewBlockPool(ethereum)
-	ethereum.txPool = core.NewTxPool(ethereum)
 	ethereum.blockChain = core.NewChainManager(ethereum.EventMux())
-	ethereum.blockManager = core.NewBlockManager(ethereum)
+	ethereum.txPool = core.NewTxPool(ethereum.blockChain, ethereum, ethereum.EventMux())
+	ethereum.blockManager = core.NewBlockManager(ethereum.txPool, ethereum.blockChain, ethereum.EventMux())
 	ethereum.blockChain.SetProcessor(ethereum.blockManager)
 
 	// Start the tx pool

pow/ezp/pow.go

@@ -21,7 +21,7 @@ type EasyPow struct {
 }
 
 func New() *EasyPow {
-	return &EasyPow{}
+	return &EasyPow{turbo: true}
 }
 
 func (pow *EasyPow) GetHashrate() int64 {