forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-17 09:23:48 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

Fix storageWG in pyramid chunker

Browse source 
storageWG did not work because it was possible to start to wait on earlier than the first Add happened
Also go routine for prepareChunks in Split and Append was unnecessary, because the processors are started in goroutines anyway
This commit is contained in:
Balint Gabor 2018-01-04 12:49:03 +01:00
parent 755cf50d6f
commit 1ad50b1563
1 changed files with 7 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
swarm/storage/pyramid.go
View file

@ -168,13 +168,14 @@ func (self *PyramidChunker) Split(data io.Reader, size int64, chunkC chan *Chunk
jobC := make(chan *chunkJob, 2*ChunkProcessors) jobC := make(chan *chunkJob, 2*ChunkProcessors)
wg := &sync.WaitGroup{} wg := &sync.WaitGroup{}
storageWG := &sync.WaitGroup{} storageWG := &sync.WaitGroup{}
storageWG.Add(1)
errC := make(chan error) errC := make(chan error)
quitC := make(chan bool) quitC := make(chan bool)
rootKey := make([]byte, self.hashSize) rootKey := make([]byte, self.hashSize)
chunkLevel := make([][]*TreeEntry, self.branches) chunkLevel := make([][]*TreeEntry, self.branches)
wg.Add(1) wg.Add(1)
go self.prepareChunks(false, chunkLevel, data, rootKey, quitC, wg, jobC, chunkC, errC, storageWG) self.prepareChunks(false, chunkLevel, data, rootKey, quitC, wg, jobC, chunkC, errC, storageWG)
// closes internal error channel if all subprocesses in the workgroup finished // closes internal error channel if all subprocesses in the workgroup finished
go func() { go func() {
@ -214,9 +215,10 @@ func (self *PyramidChunker) Append(key Key, data io.Reader, chunkC chan *Chunk)
errC := make(chan error) errC := make(chan error)
storageWG := &sync.WaitGroup{} storageWG := &sync.WaitGroup{}
storageWG.Add(1)
wg.Add(1) wg.Add(1)
go self.prepareChunks(true, chunkLevel, data, rootKey, quitC, wg, jobC, chunkC, errC, storageWG) self.prepareChunks(true, chunkLevel, data, rootKey, quitC, wg, jobC, chunkC, errC, storageWG)
// closes internal error channel if all subprocesses in the workgroup finished // closes internal error channel if all subprocesses in the workgroup finished
go func() { go func() {
@ -242,7 +244,7 @@ func (self *PyramidChunker) Append(key Key, data io.Reader, chunkC chan *Chunk)
func (self *PyramidChunker) processor(id int64, jobC chan *chunkJob, chunkC chan *Chunk, errC chan error, quitC chan bool, storageWG *sync.WaitGroup) { func (self *PyramidChunker) processor(id int64, jobC chan *chunkJob, chunkC chan *Chunk, errC chan error, quitC chan bool, storageWG *sync.WaitGroup) {
defer self.decrementWorkerCount() defer self.decrementWorkerCount()
defer storageWG.Done()
hasher := self.hashFunc() hasher := self.hashFunc()
for { for {
select { select {
@ -370,6 +372,7 @@ func (self *PyramidChunker) prepareChunks(isAppend bool, chunkLevel [][]*TreeEnt
totalDataSize := 0 totalDataSize := 0
self.incrementWorkerCount() self.incrementWorkerCount()
go self.processor(self.workerCount, jobC, chunkC, errC, quitC, storageWG) go self.processor(self.workerCount, jobC, chunkC, errC, quitC, storageWG)
parent := NewTreeEntry(self) parent := NewTreeEntry(self)
@ -471,6 +474,7 @@ func (self *PyramidChunker) prepareChunks(isAppend bool, chunkLevel [][]*TreeEnt
workers := self.getWorkerCount() workers := self.getWorkerCount()
if int64(len(jobC)) > workers && workers < ChunkProcessors { if int64(len(jobC)) > workers && workers < ChunkProcessors {
self.incrementWorkerCount() self.incrementWorkerCount()
storageWG.Add(1)
go self.processor(self.workerCount, jobC, chunkC, errC, quitC, storageWG) go self.processor(self.workerCount, jobC, chunkC, errC, quitC, storageWG)
} }