diff --git a/swarm/api/api.go b/swarm/api/api.go index 321837bda8..13c81b783b 100644 --- a/swarm/api/api.go +++ b/swarm/api/api.go @@ -48,7 +48,7 @@ func (self *Api) Retrieve(key storage.Key) storage.LazySectionReader { } func (self *Api) Store(data io.Reader, size int64, wg *sync.WaitGroup) (key storage.Key, err error) { - return self.dpa.Store(data, size, wg) + return self.dpa.Store(data, size, wg, nil) } type ErrResolve error @@ -107,13 +107,13 @@ func (self *Api) parseAndResolve(uri string, nameresolver bool) (contentHash sto func (self *Api) Put(content, contentType string) (string, error) { r := strings.NewReader(content) wg := &sync.WaitGroup{} - key, err := self.dpa.Store(r, int64(len(content)), wg) + key, err := self.dpa.Store(r, int64(len(content)), wg, nil) if err != nil { return "", err } manifest := fmt.Sprintf(`{"entries":[{"hash":"%v","contentType":"%s"}]}`, key, contentType) r = strings.NewReader(manifest) - key, err = self.dpa.Store(r, int64(len(manifest)), wg) + key, err = self.dpa.Store(r, int64(len(manifest)), wg, nil) if err != nil { return "", err } diff --git a/swarm/api/filesystem.go b/swarm/api/filesystem.go index ac09602bc6..86e26bccf4 100644 --- a/swarm/api/filesystem.go +++ b/swarm/api/filesystem.go @@ -100,7 +100,7 @@ func (self *FileSystem) Upload(lpath, index string) (string, error) { stat, _ := f.Stat() var hash storage.Key wg := &sync.WaitGroup{} - hash, err = self.api.dpa.Store(f, stat.Size(), wg) + hash, err = self.api.dpa.Store(f, stat.Size(), wg, nil) if hash != nil { list[i].Hash = hash.String() } diff --git a/swarm/api/manifest.go b/swarm/api/manifest.go index a78d86f485..7f87d24591 100644 --- a/swarm/api/manifest.go +++ b/swarm/api/manifest.go @@ -199,7 +199,7 @@ func (self *manifestTrie) recalcAndStore() error { sr := bytes.NewReader(manifest) wg := &sync.WaitGroup{} - key, err2 := self.dpa.Store(sr, int64(len(manifest)), wg) + key, err2 := self.dpa.Store(sr, int64(len(manifest)), wg, nil) wg.Wait() self.hash = key return err2 diff --git a/swarm/storage/dpa.go b/swarm/storage/dpa.go index 34a4639a6d..2a23a06ae0 100644 --- a/swarm/storage/dpa.go +++ b/swarm/storage/dpa.go @@ -27,6 +27,8 @@ const ( retrieveChanCapacity = 100 singletonSwarmDbCapacity = 50000 singletonSwarmCacheCapacity = 500 + maxStoreProcesses = 100 + maxRetrieveProcesses = 100 ) var ( @@ -42,7 +44,7 @@ type DPA struct { lock sync.Mutex running bool wg *sync.WaitGroup - quitC chan bool + quitC chan bool } // for testing locally @@ -79,8 +81,8 @@ func (self *DPA) Retrieve(key Key) LazySectionReader { // Public API. Main entry point for document storage directly. Used by the // FS-aware API and httpaccess -func (self *DPA) Store(data io.Reader, size int64, wg *sync.WaitGroup) (key Key, err error) { - return self.Chunker.Split(data, size, self.storeC, nil, wg) +func (self *DPA) Store(data io.Reader, size int64, swg *sync.WaitGroup, wwg *sync.WaitGroup) (key Key, err error) { + return self.Chunker.Split(data, size, self.storeC, swg, wwg) } func (self *DPA) Start() { @@ -90,6 +92,8 @@ func (self *DPA) Start() { return } self.running = true + self.retrieveC = make(chan *Chunk, retrieveChanCapacity) + self.storeC = make(chan *Chunk, storeChanCapacity) self.quitC = make(chan bool) self.storeLoop() self.retrieveLoop() @@ -108,13 +112,14 @@ func (self *DPA) Stop() { // retrieveLoop dispatches the parallel chunk retrieval requests received on the // retrieve channel to its ChunkStore (NetStore or LocalStore) func (self *DPA) retrieveLoop() { - self.retrieveC = make(chan *Chunk, retrieveChanCapacity) + for i:=0; i< maxRetrieveProcesses; i++ { + go self.retrieveWorker() + } + glog.V(logger.Detail).Infof("[BZZ] dpa: retrieve loop spawning %v workers", maxRetrieveProcesses) +} - go func() { - RETRIEVE: - for ch := range self.retrieveC { - - go func(chunk *Chunk) { +func (self *DPA) retrieveWorker() { + for chunk := range self.retrieveC { glog.V(logger.Detail).Infof("[BZZ] dpa: retrieve loop : chunk %v", chunk.Key.Log()) storedChunk, err := self.Get(chunk.Key) if err == notFound { @@ -126,37 +131,39 @@ func (self *DPA) retrieveLoop() { chunk.Size = storedChunk.Size } close(chunk.C) - }(ch) - select { + + select { case <-self.quitC: - break RETRIEVE + return default: - } +} } - }() + } + +// storeLoop dispatches the parallel chunk store request processors +// received on the store channel to its ChunkStore (NetStore or LocalStore) +func (self *DPA) storeLoop() { + for i:=0; i< maxStoreProcesses; i++ { + go self.storeWorker() + } + glog.V(logger.Detail).Infof("[BZZ] dpa: store spawning %v workers", maxStoreProcesses) } -// storeLoop dispatches the parallel chunk store requests received on the -// store channel to its ChunkStore (NetStore or LocalStore) -func (self *DPA) storeLoop() { - self.storeC = make(chan *Chunk) - go func() { - STORE: - for ch := range self.storeC { - go func(chunk *Chunk) { +func (self *DPA) storeWorker() { + + for chunk := range self.storeC { self.Put(chunk) if chunk.wg != nil { - glog.V(logger.Detail).Infof("[BZZ] dpa: store loop %v", chunk.Key.Log()) + glog.V(logger.Detail).Infof("[BZZ] dpa: store processor %v", chunk.Key.Log()) chunk.wg.Done() + } - }(ch) - select { + select { case <-self.quitC: - break STORE + return default: - } - } - }() +} +} } // DpaChunkStore implements the ChunkStore interface, diff --git a/swarm/storage/dpa_test.go b/swarm/storage/dpa_test.go index 4c50a7214f..f820d8e693 100644 --- a/swarm/storage/dpa_test.go +++ b/swarm/storage/dpa_test.go @@ -14,10 +14,10 @@ const testDataSize = 0x1000000 func TestDPArandom(t *testing.T) { os.RemoveAll("/tmp/bzz") dbStore, err := NewDbStore("/tmp/bzz", MakeHashFunc(defaultHash), defaultDbCapacity, defaultRadius) - dbStore.setCapacity(50000) if err != nil { t.Errorf("DB error: %v", err) } + dbStore.setCapacity(50000) memStore := NewMemStore(dbStore, defaultCacheCapacity) localStore := &LocalStore{ memStore, @@ -29,9 +29,12 @@ func TestDPArandom(t *testing.T) { ChunkStore: localStore, } dpa.Start() + defer dpa.Stop() + defer os.RemoveAll("/tmp/bzz") + reader, slice := testDataReaderAndSlice(testDataSize) wg := &sync.WaitGroup{} - key, err := dpa.Store(reader, testDataSize, wg) + key, err := dpa.Store(reader, testDataSize, wg, nil) if err != nil { t.Errorf("Store error: %v", err) } @@ -87,7 +90,7 @@ func TestDPA_capacity(t *testing.T) { dpa.Start() reader, slice := testDataReaderAndSlice(testDataSize) wg := &sync.WaitGroup{} - key, err := dpa.Store(reader, testDataSize, wg) + key, err := dpa.Store(reader, testDataSize, wg, nil) if err != nil { t.Errorf("Store error: %v", err) }