swarm/storage: fix process leak in dpa

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
 	}

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()
 				}

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

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) {
+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, nil, wg)
+	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() {
+func (self *DPA) retrieveWorker() {
-	RETRIEVE:
+		for chunk := range self.retrieveC {
-		for ch := range self.retrieveC {
-
-			go func(chunk *Chunk) {
 				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
+func (self *DPA) storeWorker() {
-// store channel to its ChunkStore (NetStore or LocalStore)
+
-func (self *DPA) storeLoop() {
+		for chunk := range self.storeC {
-	self.storeC = make(chan *Chunk)
-	go func() {
-	STORE:
-		for ch := range self.storeC {
-			go func(chunk *Chunk) {
 				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,

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)
 	}