diff --git a/bzz/dpa.go b/bzz/dpa.go index 1a6380a975..85d77ff047 100644 --- a/bzz/dpa.go +++ b/bzz/dpa.go @@ -3,8 +3,6 @@ package bzz import ( "errors" "sync" - // "time" - // "fmt" ethlogger "github.com/ethereum/go-ethereum/logger" ) diff --git a/bzz/hive.go b/bzz/hive.go index 52cf8efdf7..d1408e6675 100644 --- a/bzz/hive.go +++ b/bzz/hive.go @@ -3,7 +3,6 @@ package bzz import ( // "fmt" - "github.com/ethereum/go-ethereum/common" "github.com/ethereum/go-ethereum/common/kademlia" ) @@ -29,15 +28,15 @@ type hive struct { path string } -func newHive(address common.Hash, hivepath string) *hive { +func newHive(hivepath string) *hive { return &hive{ path: hivepath, - kad: kademlia.New(kademlia.Address(address)), + kad: kademlia.New(), } } -func (self *hive) start() (err error) { - self.kad.Start() +func (self *hive) start(address kademlia.Address) (err error) { + self.kad.Start(address) err = self.kad.Load(self.path) if err != nil { dpaLogger.Warnf("Warning: error reading kademlia node db (skipping): %v", err) diff --git a/bzz/netstore.go b/bzz/netstore.go index 7f803adb8f..cf7bc26ec1 100644 --- a/bzz/netstore.go +++ b/bzz/netstore.go @@ -6,13 +6,15 @@ import ( "sync" "time" - "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/common/kademlia" + "github.com/ethereum/go-ethereum/p2p/discover" ) type NetStore struct { localStore *localStore lock sync.Mutex hive *hive + self *discover.Node } /* @@ -45,19 +47,24 @@ type requestStatus struct { C chan bool } -func NewNetStore(addr common.Hash, path, hivepath string) (netstore *NetStore, err error) { +func NewNetStore(path, hivepath string) (netstore *NetStore, err error) { dbStore, err := newDbStore(path) if err != nil { return } - hive := newHive(addr, hivepath) + hive := newHive(hivepath) netstore = &NetStore{ localStore: &localStore{ memStore: newMemStore(dbStore), dbStore: dbStore, }, hive: hive, } - err = hive.start() + return +} + +func (self *NetStore) Start(node *discover.Node) (err error) { + self.self = node + err = self.hive.start(kademlia.Address(node.Sha())) if err != nil { return } diff --git a/bzz/protocol.go b/bzz/protocol.go index ee6340a64a..0b13973eed 100644 --- a/bzz/protocol.go +++ b/bzz/protocol.go @@ -58,7 +58,6 @@ var errorToString = map[int]string{ // bzzProtocol represents the swarm wire protocol // instance is running on each peer type bzzProtocol struct { - self func() *discover.Node node *discover.Node netStore *NetStore peer *p2p.Peer @@ -188,22 +187,21 @@ main entrypoint, wrappers starting a server running the bzz protocol use this constructor to attach the protocol ("class") to server caps the Dev p2p layer then runs the protocol instance on each peer */ -func BzzProtocol(netStore *NetStore, self func() *discover.Node) p2p.Protocol { +func BzzProtocol(netStore *NetStore) p2p.Protocol { return p2p.Protocol{ Name: "bzz", Version: Version, Length: ProtocolLength, Run: func(p *p2p.Peer, rw p2p.MsgReadWriter) error { - return runBzzProtocol(netStore, self, p, rw) + return runBzzProtocol(netStore, p, rw) }, } } // the main loop that handles incoming messages // note RemovePeer in the post-disconnect hook -func runBzzProtocol(netStore *NetStore, selfF func() *discover.Node, p *p2p.Peer, rw p2p.MsgReadWriter) (err error) { +func runBzzProtocol(netStore *NetStore, p *p2p.Peer, rw p2p.MsgReadWriter) (err error) { self := &bzzProtocol{ - self: selfF, netStore: netStore, rw: rw, peer: p, @@ -284,12 +282,12 @@ func (self *bzzProtocol) handle() error { func (self *bzzProtocol) handleStatus() (err error) { // send precanned status message - sliceNodeID := self.self().ID + sliceNodeID := self.netStore.self.ID handshake := &statusMsgData{ Version: uint64(Version), ID: "honey", NodeID: sliceNodeID[:], - Addr: newPeerAddrFromNode(self.self()), + Addr: newPeerAddrFromNode(self.netStore.self), NetworkId: uint64(NetworkId), Caps: []p2p.Cap{}, } diff --git a/common/kademlia/kademlia.go b/common/kademlia/kademlia.go index 553d35a350..5a699a9bb6 100644 --- a/common/kademlia/kademlia.go +++ b/common/kademlia/kademlia.go @@ -77,12 +77,10 @@ type kadDB struct { Nodes [][]*NodeRecord `json:nodes` } -// public constructor with compulsory arguments +// public constructor // hash is a byte slice of length equal to self.HashBytes -func New(a Address) *Kademlia { - return &Kademlia{ - addr: a, // compulsory fields without default - } +func New() *Kademlia { + return &Kademlia{} } // accessor for KAD self address @@ -97,12 +95,13 @@ func (self *Kademlia) Count() int { // Start brings up a pool of entries potentially from an offline persisted source // and sets default values for optional parameters -func (self *Kademlia) Start() error { +func (self *Kademlia) Start(addr Address) error { self.lock.Lock() defer self.lock.Unlock() if self.quitC != nil { return nil } + self.addr = addr if self.MaxProx == 0 { self.MaxProx = maxProx } @@ -454,6 +453,7 @@ bit first). proximity(x, y) counts the common zeros in the front of this distance measure. */ func proximity(one, other Address) (ret int) { + fmt.Printf("%08b ^ %08b = %08b\n", one[0], other[0], one[0]^other[0]) for i := 0; i < len(one); i++ { oxo := one[i] ^ other[i] for j := 0; j < 8; j++ { diff --git a/common/kademlia/kademlia_test.go b/common/kademlia/kademlia_test.go index 5a7a66fff2..c0e78f351d 100644 --- a/common/kademlia/kademlia_test.go +++ b/common/kademlia/kademlia_test.go @@ -61,8 +61,8 @@ func TestAddNode(t *testing.T) { if !ok { t.Errorf("oops") } - kad := New(addr) - kad.Start() + kad := New() + kad.Start(addr) err := kad.AddNode(&testNode{addr: other}) _ = err } @@ -73,9 +73,9 @@ func TestGetNodes(t *testing.T) { test := func(test *getNodesTest) bool { // for any node kad.le, Target and N - kad := New(test.Self) + kad := New() kad.MaxProx = 10 - kad.Start() + kad.Start(test.Self) var err error t.Logf("getNodesTest %v: %v\n", len(test.All), test) for _, node := range test.All { @@ -152,9 +152,9 @@ func TestProxAdjust(t *testing.T) { r := rand.New(rand.NewSource(time.Now().UnixNano())) self := gen(Address{}, r).(Address) - kad := New(self) + kad := New() kad.MaxProx = 10 - kad.Start() + kad.Start(self) var err error for i := 0; i < 100; i++ { a := gen(Address{}, r).(Address) @@ -186,7 +186,7 @@ func TestProxAdjust(t *testing.T) { func TestCallback(t *testing.T) { r := rand.New(rand.NewSource(time.Now().UnixNano())) self := gen(Address{}, r).(Address) - kad := New(self) + kad := New() var bucket int var called chan bool kad.MaxProx = 5 @@ -195,7 +195,7 @@ func TestCallback(t *testing.T) { bucket = b close(called) } - kad.Start() + kad.Start(self) var err error for i := 0; i < 100; i++ { @@ -232,9 +232,9 @@ func TestSaveLoad(t *testing.T) { r := rand.New(rand.NewSource(time.Now().UnixNano())) addresses := gen([]Address{}, r).([]Address) self := addresses[0] - kad := New(self) + kad := New() kad.MaxProx = 10 - kad.Start() + kad.Start(self) var err error for _, a := range addresses[1:] { err = kad.AddNode(&testNode{addr: a}) @@ -246,8 +246,8 @@ func TestSaveLoad(t *testing.T) { nodes := kad.GetNodes(self, 100) path := "/tmp/bzz.peers" kad.Stop(path) - kad = New(self) - kad.Start() + kad = New() + kad.Start(self) kad.Load(path) for _, b := range kad.DB() { for _, node := range b { diff --git a/eth/backend.go b/eth/backend.go index 6250838f92..8e8bdb1314 100644 --- a/eth/backend.go +++ b/eth/backend.go @@ -184,6 +184,8 @@ type Ethereum struct { pow *ethash.Ethash protocolManager *ProtocolManager downloader *downloader.Downloader + DPA *bzz.DPA + netStore *bzz.NetStore net *p2p.Server eventMux *event.TypeMux @@ -285,42 +287,40 @@ func New(config *Config) (*Ethereum, error) { protocols := []p2p.Protocol{eth.protocolManager.SubProtocol} - eth.net = &p2p.Server{ - PrivateKey: netprv, - Name: config.Name, - MaxPeers: config.MaxPeers, - MaxPendingPeers: config.MaxPendingPeers, - // Protocols: protocols, - NAT: config.NAT, - NoDial: !config.Dial, - BootstrapNodes: config.parseBootNodes(), - StaticNodes: config.parseNodes(staticNodes), - TrustedNodes: config.parseNodes(trustedNodes), - NodeDatabase: nodeDb, - } - if len(config.Port) > 0 { - eth.net.ListenAddr = ":" + config.Port - } if config.Bzz { - netStore, err := bzz.NewNetStore(eth.net.Self().Sha(), path.Join(config.DataDir, "bzz"), path.Join(config.DataDir, "bzzpeers.json")) + eth.netStore, err = bzz.NewNetStore(path.Join(config.DataDir, "bzz"), path.Join(config.DataDir, "bzzpeers.json")) if err != nil { glog.V(logger.Warn).Infof("BZZ: error creating net store: %v. Protocol skipped", err) } else { chunker := &bzz.TreeChunker{} chunker.Init() - dpa := &bzz.DPA{ + eth.DPA = &bzz.DPA{ Chunker: chunker, - ChunkStore: netStore, + ChunkStore: eth.netStore, } - dpa.Start() - protocols = append(protocols, bzz.BzzProtocol(netStore, eth.net.Self)) - go bzz.StartHttpServer(dpa) + protocols = append(protocols, bzz.BzzProtocol(eth.netStore)) } } if config.Shh { protocols = append(protocols, eth.whisper.Protocol()) } + eth.net = &p2p.Server{ + PrivateKey: netprv, + Name: config.Name, + MaxPeers: config.MaxPeers, + MaxPendingPeers: config.MaxPendingPeers, + Protocols: protocols, + NAT: config.NAT, + NoDial: !config.Dial, + BootstrapNodes: config.parseBootNodes(), + StaticNodes: config.parseNodes(staticNodes), + TrustedNodes: config.parseNodes(trustedNodes), + NodeDatabase: nodeDb, + } + if len(config.Port) > 0 { + eth.net.ListenAddr = ":" + config.Port + } eth.net.Protocols = protocols @@ -466,6 +466,12 @@ func (s *Ethereum) Start() error { s.whisper.Start() } + if s.DPA != nil { + s.DPA.Start() + s.netStore.Start(s.net.Self()) + go bzz.StartHttpServer(s.DPA) + } + // broadcast transactions s.txSub = s.eventMux.Subscribe(core.TxPreEvent{}) go s.txBroadcastLoop()