From 279219eb407455598118ebf4887491327d3cfd51 Mon Sep 17 00:00:00 2001 From: zelig Date: Fri, 6 Feb 2015 16:21:31 +0100 Subject: [PATCH] netstore Get/addRetrieveRequest logic --- bzz/netstore.go | 183 +++++++++++++++++++++++++++--------------------- 1 file changed, 104 insertions(+), 79 deletions(-) diff --git a/bzz/netstore.go b/bzz/netstore.go index de2e738e75..c0a94000be 100644 --- a/bzz/netstore.go +++ b/bzz/netstore.go @@ -58,7 +58,13 @@ const ( reqFound ) -const requesterCount = 3 +const ( + requesterCount = 3 +) + +var ( + searchTimeout = 3 * time.Second +) type peer struct { *bzzProtocol @@ -69,6 +75,101 @@ type requestStatus struct { key Key status int requesters map[int64][]*retrieveRequestMsgData + C chan bool +} + +func (self *netStore) Put(entry *Chunk) { + chunk, err := self.localStore.Get(entry.Key) + if err != nil { + chunk = entry + } else if chunk.Data == nil { + chunk.Data = entry.Data + chunk.Size = entry.Size + } else { + return + } + self.put(chunk) +} + +func (self *netStore) put(entry *Chunk) { + self.localStore.Put(entry) + self.store(entry) + // only send responses once + if entry.req != nil && entry.req.status == reqSearching { + entry.req.status = reqFound + close(entry.req.C) + self.propagateResponse(entry) + } +} + +func (self *netStore) addStoreRequest(req *storeRequestMsgData) { + self.lock.Lock() + defer self.lock.Unlock() + chunk, err := self.localStore.Get(req.Key) + // we assume that a returned chunk is the one stored in the memory cache + if err != nil { + chunk = &Chunk{ + Key: req.Key, + Data: req.Data, + Size: req.Size, + } + } else if chunk.Data == nil { + chunk.Data = req.Data + chunk.Size = req.Size + } else { + return + } + self.put(chunk) +} + +func (self *netStore) Get(key Key) (chunk *Chunk, err error) { + chunk = self.get(key) + timeout := time.After(searchTimeout) + select { + case <-timeout: + err = notFound + case <-chunk.req.C: + } + return +} + +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 + if err != nil { + // no data and no request status + chunk = &Chunk{ + Key: key, + } + self.localStore.memStore.Put(chunk) + } + + if chunk.req == nil { + chunk.req = new(requestStatus) + if chunk.Data == nil { + self.startSearch(chunk) + } + } + return +} + +func (self *netStore) addRetrieveRequest(req *retrieveRequestMsgData) { + + self.lock.Lock() + defer self.lock.Unlock() + + chunk := self.get(req.Key) + + send, timeout := self.strategyUpdateRequest(chunk.req, req) // may change req status + + if send == storeRequestMsg { + self.deliver(req, chunk) + } else { + // we might need chunk.req to cache relevant peers response, or would it expire? + self.peers(req, chunk, timeout) + } + } // it's assumed that caller holds the lock @@ -114,52 +215,9 @@ func (self *netStore) strategyUpdateRequest(rs *requestStatus, req *retrieveRequ } -func (self *netStore) put(entry *Chunk) { - self.localStore.Put(entry) - self.store(entry) - // only send responses once - if entry.req != nil && entry.req.status == reqSearching { - entry.req.status = reqFound - self.propagateResponse(entry) - } -} - -func (self *netStore) Put(entry *Chunk) { - chunk, err := self.localStore.Get(entry.Key) - if err != nil { - chunk = entry - } else if chunk.Data == nil { - chunk.Data = entry.Data - chunk.Size = entry.Size - } else { - return - } - self.put(chunk) -} - -func (self *netStore) addStoreRequest(req *storeRequestMsgData) { - self.lock.Lock() - defer self.lock.Unlock() - chunk, err := self.localStore.Get(req.Key) - // we assume that a returned chunk is the one stored in the memory cache - if err != nil { - chunk = &Chunk{ - Key: req.Key, - Data: req.Data, - Size: req.Size, - } - } else if chunk.Data == nil { - chunk.Data = req.Data - chunk.Size = req.Size - } else { - return - } - self.put(chunk) -} - func (self *netStore) propagateResponse(chunk *Chunk) { for id, requesters := range chunk.req.requesters { - counter = requesterCount + counter := requesterCount msg := &storeRequestMsgData{ Key: chunk.Key, Data: chunk.Data, @@ -178,39 +236,6 @@ func (self *netStore) propagateResponse(chunk *Chunk) { } } -func (self *netStore) addRetrieveRequest(req *retrieveRequestMsgData) { - - self.lock.Lock() - defer self.lock.Unlock() - - chunk, err := self.localStore.Get(req.Key) - // we assume that a returned chunk is the one stored in the memory cache - if err != nil { - // no data and no request status - chunk = &Chunk{ - Key: req.Key, - } - self.localStore.memStore.Put(chunk) - } - - if chunk.req == nil { - chunk.req = new(requestStatus) - if chunk.Data == nil { - self.startSearch(chunk) - } - } - - send, timeout := self.strategyUpdateRequest(chunk.req, req) // may change req status - - if send == storeRequestMsg { - self.deliver(req, chunk) - } else { - // we might need chunk.req to cache relevant peers response, or would it expire? - self.peers(req, chunk, timeout) - } - -} - func (self *netStore) deliver(req *retrieveRequestMsgData, chunk *Chunk) { storeReq := &storeRequestMsgData{ Key: req.Key, @@ -248,7 +273,7 @@ func (self *netStore) peers(req *retrieveRequestMsgData, chunk *Chunk, timeout t } func (self *netStore) searchTimeout(rs *requestStatus, req *retrieveRequestMsgData) (timeout time.Time) { - t := time.Now().Add(3 * time.Second) + t := time.Now().Add(searchTimeout) if req.Timeout.Before(t) { return req.Timeout } else {