OLD | NEW |
---|---|
(Empty) | |
1 // Copyright 2016 The LUCI Authors. All rights reserved. | |
2 // Use of this source code is governed under the Apache License, Version 2.0 | |
3 // that can be found in the LICENSE file. | |
4 | |
5 /** | |
6 | |
7 WARNING: THIS FUNCTION IS SLOWER THAN THE NON-PARALLEL VERSION! | |
M-A Ruel
2016/09/15 14:31:03
// Testing shows that this function is slower than
mithro
2016/09/20 12:41:45
Done.
The problem is frequently around locking of
M-A Ruel
2016/09/20 16:37:27
That is actually a more helpful comment that shoul
| |
8 | |
9 **/ | |
10 package dirwalk | |
11 | |
12 import ( | |
13 "io/ioutil" | |
14 "os" | |
15 "path/filepath" | |
16 "sort" | |
17 "sync/atomic" | |
18 | |
19 "github.com/eapache/channels" | |
20 ) | |
21 | |
22 type fileQueue struct { | |
23 queued uint64 | |
24 finished uint64 | |
25 items channels.Channel | |
26 waiton chan bool | |
27 } | |
28 | |
29 func (q *fileQueue) add(s string) { | |
30 atomic.AddUint64(&q.queued, 1) | |
31 q.items.In() <- s | |
32 } | |
33 | |
34 func (q *fileQueue) done() { | |
35 atomic.AddUint64(&q.finished, 1) | |
36 | |
37 if q.queued == q.finished { | |
38 q.items.Close() | |
39 q.waiton <- true | |
40 } | |
41 } | |
42 | |
43 func (q *fileQueue) wait() { | |
44 <-q.waiton | |
45 } | |
46 | |
47 func examinePath(queue *fileQueue, smallfile_limit int64, obs WalkObserver) { | |
48 for ipath := range queue.items.Out() { | |
49 path := ipath.(string) | |
50 | |
51 fi, err := os.Stat(path) | |
52 if err != nil { | |
53 obs.Error(path, err) | |
54 return | |
55 } | |
56 | |
57 if fi.IsDir() { | |
58 f, err := os.Open(path) | |
59 if err != nil { | |
60 obs.Error(path, err) | |
61 } | |
62 | |
63 dircontents, err := f.Readdirnames(-1) | |
64 if err != nil { | |
65 obs.Error(path, err) | |
66 } | |
67 sort.Strings(dircontents) | |
68 for _, name := range dircontents { | |
69 fname := filepath.Join(path, name) | |
70 queue.add(fname) | |
71 } | |
72 } else { | |
73 if fi.Size() < smallfile_limit { | |
74 data, err := ioutil.ReadFile(path) | |
75 if err != nil { | |
76 obs.Error(path, err) | |
77 return | |
78 } | |
79 if int64(len(data)) != fi.Size() { | |
80 panic("file size was wrong!") | |
81 } | |
82 obs.SmallFile(path, data) | |
83 } else { | |
84 obs.LargeFile(path) | |
85 } | |
86 } | |
87 queue.done() | |
88 } | |
89 } | |
90 | |
91 func WalkParallel(root string, smallfile_limit int64, obs WalkObserver) { | |
92 queue := fileQueue{queued: 0, finished: 0, items: channels.NewInfiniteCh annel(), waiton: make(chan bool)} | |
93 | |
94 for w := 0; w <= 10; w++ { | |
95 go examinePath(&queue, smallfile_limit, obs) | |
96 } | |
97 | |
98 queue.add(root) | |
99 queue.wait() | |
100 obs.Finished() | |
101 } | |
OLD | NEW |