forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-25 14:16:44 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

swarm/storage: chunker join fix process leak and keep parallelisation limited to depth 1

Browse source
This commit is contained in:
zelig 2016-06-14 11:47:39 +01:00
parent f84c16f203
commit fb83c1fcec
1 changed files with 17 additions and 15 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

22
swarm/storage/chunker.go
View file

@ -336,17 +336,14 @@ func (self *LazyChunkReader) ReadAt(b []byte, off int64) (read int, err error) {
wg.Wait() wg.Wait()
close(self.errC) close(self.errC)
}() }()
select {
case err = <-self.errC: err = <-self.errC
// glog.V(logger.Detail).Infof("[BZZ] ReadAt received %v", err) // glog.V(logger.Detail).Infof("[BZZ] ReadAt received %v", err)
read = len(b) read = len(b)
if off+int64(read) == self.size { if off+int64(read) == self.size {
err = io.EOF err = io.EOF
} }
// glog.V(logger.Detail).Infof("[BZZ] ReadAt returning at %d: %v", read, err) // glog.V(logger.Detail).Infof("[BZZ] ReadAt returning at %d: %v", read, err)
case <-self.quitC:
// glog.V(logger.Detail).Infof("[BZZ] ReadAt aborted at %d: %v", read, err)
}
return return
} }
@ -374,10 +371,10 @@ func (self *LazyChunkReader) join(b []byte, off int64, eoff int64, depth int, tr
return // simply give back the chunks reader for content chunks return // simply give back the chunks reader for content chunks
} }
// subtree index // subtree
start := off / treeSize start := off / treeSize
end := (eoff + treeSize - 1) / treeSize end := (eoff + treeSize - 1) / treeSize
wg := sync.WaitGroup{} wg := &sync.WaitGroup{}
for i := start; i < end; i++ { for i := start; i < end; i++ {
@ -391,7 +388,9 @@ func (self *LazyChunkReader) join(b []byte, off int64, eoff int64, depth int, tr
if seoff > eoff { if seoff > eoff {
seoff = eoff seoff = eoff
} }
if depth > 1 {
wg.Wait()
}
wg.Add(1) wg.Add(1)
go func(j int64) { go func(j int64) {
childKey := chunk.SData[8+j*self.chunker.hashSize : 8+(j+1)*self.chunker.hashSize] childKey := chunk.SData[8+j*self.chunker.hashSize : 8+(j+1)*self.chunker.hashSize]
@ -416,10 +415,13 @@ func (self *LazyChunkReader) join(b []byte, off int64, eoff int64, depth int, tr
soff = off soff = off
} }
if len(ch.SData) == 0 { if len(ch.SData) == 0 {
self.errC <- fmt.Errorf("chunk %v-%v not found", off, off+treeSize) select {
case self.errC <- fmt.Errorf("chunk %v-%v not found", off, off+treeSize):
case <-self.quitC:
}
return return
} }
self.join(b[soff-off:seoff-off], soff-roff, seoff-roff, depth-1, treeSize/self.chunker.branches, ch, &wg) self.join(b[soff-off:seoff-off], soff-roff, seoff-roff, depth-1, treeSize/self.chunker.branches, ch, wg)
}(i) }(i)
} //for } //for
wg.Wait() wg.Wait()