forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-23 05:06:43 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

fixed too many open files bug, limited parallel processing of files

Browse source
This commit is contained in:
zsfelfoldi 2015-05-29 03:33:18 +02:00
parent c3bb14e352
commit ab75309ffa
1 changed files with 19 additions and 7 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

26
bzz/api.go
View file

@ -108,12 +108,14 @@ func (self *Api) Download(bzzpath, localpath string) (string, error) {
return "", nil return "", nil
} }
const maxParallelFiles = 5
// Upload replicates a local directory as a manifest file and uploads it // Upload replicates a local directory as a manifest file and uploads it
// using dpa store // using dpa store
// TODO: localpath should point to a manifest // TODO: localpath should point to a manifest
func (self *Api) Upload(lpath string) (string, error) { func (self *Api) Upload(lpath string) (string, error) {
var files []string var files []string
localpath, err1 := filepath.Abs(lpath) localpath, err1 := filepath.Abs(filepath.Clean(lpath))
if err1 != nil { if err1 != nil {
return "", err1 return "", err1
} }
@ -143,16 +145,22 @@ func (self *Api) Upload(lpath string) (string, error) {
hashes := make([]Key, cnt) hashes := make([]Key, cnt)
errors := make([]error, cnt) errors := make([]error, cnt)
ctypes := make([]string, cnt) ctypes := make([]string, cnt)
wg := &sync.WaitGroup{} done := make(chan bool, maxParallelFiles)
dcnt := 0
for i, path := range files { for i, path := range files {
wg.Add(1) if i >= dcnt+maxParallelFiles {
go func(i int, path string) { <-done
dcnt++
}
go func(i int, path string, done chan bool) {
f, err := os.Open(path) f, err := os.Open(path)
if err == nil { if err == nil {
stat, _ := f.Stat() stat, _ := f.Stat()
sr := io.NewSectionReader(f, 0, stat.Size()) sr := io.NewSectionReader(f, 0, stat.Size())
wg := &sync.WaitGroup{}
hashes[i], err = self.dpa.Store(sr, wg) hashes[i], err = self.dpa.Store(sr, wg)
wg.Wait()
} }
if err == nil { if err == nil {
cmd := exec.Command("file", "--mime-type", "-b", path) cmd := exec.Command("file", "--mime-type", "-b", path)
@ -164,10 +172,13 @@ func (self *Api) Upload(lpath string) (string, error) {
} }
} }
errors[i] = err errors[i] = err
wg.Done() done <- true
}(i, path) }(i, path, done)
}
for dcnt < cnt {
<-done
dcnt++
} }
wg.Wait()
var buffer bytes.Buffer var buffer bytes.Buffer
buffer.WriteString(`{"entries":[`) buffer.WriteString(`{"entries":[`)
@ -188,6 +199,7 @@ func (self *Api) Upload(lpath string) (string, error) {
manifest := buffer.Bytes() manifest := buffer.Bytes()
sr := io.NewSectionReader(bytes.NewReader(manifest), 0, int64(len(manifest))) sr := io.NewSectionReader(bytes.NewReader(manifest), 0, int64(len(manifest)))
wg := &sync.WaitGroup{}
key, err2 := self.dpa.Store(sr, wg) key, err2 := self.dpa.Store(sr, wg)
wg.Wait() wg.Wait()
return fmt.Sprintf("%064x", key), err2 return fmt.Sprintf("%064x", key), err2