diff --git a/bzz/hive.go b/bzz/hive.go index 79ba71dde5..4ee17cb190 100644 --- a/bzz/hive.go +++ b/bzz/hive.go @@ -10,6 +10,12 @@ type hive struct { pool map[string]peer } +func newHive() *hive { + return &hive{ + pool: make(map[string]peer), + } +} + func (self *hive) addPeer(p peer) { self.pool[string(p.pubkey)] = p } diff --git a/bzz/netstore.go b/bzz/netstore.go index f7b1ef5edd..69d28abf2a 100644 --- a/bzz/netstore.go +++ b/bzz/netstore.go @@ -6,7 +6,7 @@ import ( "time" ) -type netStore struct { +type NetStore struct { localStore *localStore lock sync.Mutex hive *hive @@ -42,7 +42,18 @@ type requestStatus struct { C chan bool } -func (self *netStore) Put(entry *Chunk) { +func NewNetStore(path string) *NetStore { + dbStore, _ := newDbStore(path) + return &NetStore{ + localStore: &localStore{ + memStore: newMemStore(dbStore), + dbStore: dbStore, + }, + hive: newHive(), + } +} + +func (self *NetStore) Put(entry *Chunk) { chunk, err := self.localStore.Get(entry.Key) if err != nil { chunk = entry @@ -55,7 +66,7 @@ func (self *netStore) Put(entry *Chunk) { self.put(chunk) } -func (self *netStore) put(entry *Chunk) { +func (self *NetStore) put(entry *Chunk) { self.localStore.Put(entry) self.store(entry) // only send responses once @@ -66,7 +77,7 @@ func (self *netStore) put(entry *Chunk) { } } -func (self *netStore) addStoreRequest(req *storeRequestMsgData) { +func (self *NetStore) addStoreRequest(req *storeRequestMsgData) { self.lock.Lock() defer self.lock.Unlock() chunk, err := self.localStore.Get(req.Key) @@ -87,7 +98,7 @@ func (self *netStore) addStoreRequest(req *storeRequestMsgData) { } // waits for response or times out -func (self *netStore) Get(key Key) (chunk *Chunk, err error) { +func (self *NetStore) Get(key Key) (chunk *Chunk, err error) { chunk = self.get(key) id := generateId() timeout := time.Now().Add(searchTimeout) @@ -103,7 +114,7 @@ func (self *netStore) Get(key Key) (chunk *Chunk, err error) { return } -func (self *netStore) get(key Key) (chunk *Chunk) { +func (self *NetStore) get(key Key) (chunk *Chunk) { var err error chunk, err = self.localStore.Get(key) // we assume that a returned chunk is the one stored in the memory cache @@ -121,7 +132,7 @@ func (self *netStore) get(key Key) (chunk *Chunk) { return } -func (self *netStore) addRetrieveRequest(req *retrieveRequestMsgData) { +func (self *NetStore) addRetrieveRequest(req *retrieveRequestMsgData) { self.lock.Lock() defer self.lock.Unlock() @@ -143,7 +154,7 @@ func (self *netStore) addRetrieveRequest(req *retrieveRequestMsgData) { } // it's assumed that caller holds the lock -func (self *netStore) startSearch(chunk *Chunk, id int64, timeout time.Time) { +func (self *NetStore) startSearch(chunk *Chunk, id int64, timeout time.Time) { chunk.req.status = reqSearching peers := self.hive.getPeers(chunk.Key) req := &retrieveRequestMsgData{ @@ -166,7 +177,7 @@ adds a new peer to an existing open request only add if less than requesterCount peers forwarded the same request id so far note this is done irrespective of status (searching or found/timedOut) */ -func (self *netStore) addRequester(rs *requestStatus, req *retrieveRequestMsgData) { +func (self *NetStore) addRequester(rs *requestStatus, req *retrieveRequestMsgData) { list := rs.requesters[req.Id] rs.requesters[req.Id] = append(list, req) } @@ -183,7 +194,7 @@ this is the most simplistic implementation: - respond with peers and timeout if still searching ! in the last case as well, we should respond with reject if already got requesterCount peers with that exact id */ -func (self *netStore) strategyUpdateRequest(rs *requestStatus, req *retrieveRequestMsgData) (msgTyp int, timeout *time.Time) { +func (self *NetStore) strategyUpdateRequest(rs *requestStatus, req *retrieveRequestMsgData) (msgTyp int, timeout *time.Time) { switch rs.status { case reqSearching: @@ -198,7 +209,7 @@ func (self *netStore) strategyUpdateRequest(rs *requestStatus, req *retrieveRequ } -func (self *netStore) propagateResponse(chunk *Chunk) { +func (self *NetStore) propagateResponse(chunk *Chunk) { for id, requesters := range chunk.req.requesters { counter := requesterCount msg := &storeRequestMsgData{ @@ -219,7 +230,7 @@ func (self *netStore) propagateResponse(chunk *Chunk) { } } -func (self *netStore) deliver(req *retrieveRequestMsgData, chunk *Chunk) { +func (self *NetStore) deliver(req *retrieveRequestMsgData, chunk *Chunk) { storeReq := &storeRequestMsgData{ Key: req.Key, Id: req.Id, @@ -232,7 +243,7 @@ func (self *netStore) deliver(req *retrieveRequestMsgData, chunk *Chunk) { req.peer.store(storeReq) } -func (self *netStore) store(chunk *Chunk) { +func (self *NetStore) store(chunk *Chunk) { id := generateId() req := &storeRequestMsgData{ Key: chunk.Key, @@ -245,7 +256,7 @@ func (self *netStore) store(chunk *Chunk) { } } -func (self *netStore) peers(req *retrieveRequestMsgData, chunk *Chunk, timeout time.Time) { +func (self *NetStore) peers(req *retrieveRequestMsgData, chunk *Chunk, timeout time.Time) { peersData := &peersMsgData{ Peers: []*peerAddr{}, // get proximity bin from cademlia routing table Key: req.Key, @@ -255,7 +266,7 @@ func (self *netStore) peers(req *retrieveRequestMsgData, chunk *Chunk, timeout t req.peer.peers(peersData) } -func (self *netStore) searchTimeout(rs *requestStatus, req *retrieveRequestMsgData) (timeout *time.Time) { +func (self *NetStore) searchTimeout(rs *requestStatus, req *retrieveRequestMsgData) (timeout *time.Time) { t := time.Now().Add(searchTimeout) if req.Timeout.Before(t) { return &req.Timeout diff --git a/bzz/protocol.go b/bzz/protocol.go index da79179dbd..0765668666 100644 --- a/bzz/protocol.go +++ b/bzz/protocol.go @@ -32,8 +32,7 @@ const ( // bzzProtocol represents the swarm wire protocol // instance is running on each peer type bzzProtocol struct { - netStore *netStore - hive *hive + netStore *NetStore peer *p2p.Peer rw p2p.MsgReadWriter } @@ -141,23 +140,22 @@ 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, hive *hive) 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, hive, 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, hive *hive, p *p2p.Peer, rw p2p.MsgReadWriter) (err error) { +func runBzzProtocol(netStore *NetStore, p *p2p.Peer, rw p2p.MsgReadWriter) (err error) { self := &bzzProtocol{ netStore: netStore, - hive: hive, rw: rw, peer: p, } @@ -166,7 +164,7 @@ func runBzzProtocol(netStore *netStore, hive *hive, p *p2p.Peer, rw p2p.MsgReadW for { err = self.handle() if err != nil { - self.hive.removePeer(peer{bzzProtocol: self}) + self.netStore.hive.removePeer(peer{bzzProtocol: self}) break } } @@ -216,7 +214,7 @@ func (self *bzzProtocol) handle() error { return self.protoError(ErrDecode, "->msg %v: %v", msg, err) } req.peer = peer{bzzProtocol: self} - self.hive.addPeers(&req) + self.netStore.hive.addPeers(&req) default: return self.protoError(ErrInvalidMsgCode, "%v", msg.Code) @@ -269,7 +267,7 @@ func (self *bzzProtocol) handleStatus() (err error) { self.peer.Infof("Peer is [bzz] capable (%d/%d)\n", status.Version, status.NetworkId) - self.hive.addPeer(peer{bzzProtocol: self, pubkey: status.NodeID}) + self.netStore.hive.addPeer(peer{bzzProtocol: self, pubkey: status.NodeID}) return nil } diff --git a/eth/backend.go b/eth/backend.go index 50c7f171ae..8029c25ca2 100644 --- a/eth/backend.go +++ b/eth/backend.go @@ -138,7 +138,7 @@ func New(config *Config) (*Ethereum, error) { protocols := []p2p.Protocol{ ethProto, eth.whisper.Protocol(), - bzz.BzzProtocol(nil, nil), + bzz.BzzProtocol(bzz.NewNetStore(config.DataDir + "/bzz")), } nat, err := p2p.ParseNAT(config.NATType, config.PMPGateway)