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 // On the majority of systems testing shows that this function is either | |
6 // slower than (or at best comparable) the non-parallel version while consuming | |
7 // many times the resources. | |
8 // | |
9 // Linux Kernel versions newer than >4.8 which disable locks in stat path can | |
10 // make this version faster. | |
11 // | |
12 // Use the performance tests to determine the correct walker for your platform | |
13 // and system! | |
14 | |
15 package dirwalk | |
16 | |
17 import ( | |
18 "io/ioutil" | |
19 "os" | |
20 "path/filepath" | |
21 "sort" | |
22 "sync/atomic" | |
23 | |
24 "github.com/eapache/channels" | |
25 ) | |
26 | |
27 type fileQueue struct { | |
28 queued uint64 | |
29 finished uint64 | |
30 items channels.Channel | |
31 waiton chan bool | |
32 } | |
33 | |
34 func (q *fileQueue) add(s string) { | |
35 atomic.AddUint64(&q.queued, 1) | |
36 q.items.In() <- s | |
37 } | |
38 | |
39 func (q *fileQueue) done() { | |
40 atomic.AddUint64(&q.finished, 1) | |
41 | |
42 if q.queued == q.finished { | |
43 q.items.Close() | |
44 q.waiton <- true | |
45 } | |
46 } | |
47 | |
48 func (q *fileQueue) wait() { | |
49 <-q.waiton | |
50 } | |
51 | |
52 func examinePath(queue *fileQueue, smallfile_limit int64, obs WalkObserver) { | |
53 for ipath := range queue.items.Out() { | |
54 path := ipath.(string) | |
55 | |
56 fi, err := os.Stat(path) | |
57 if err != nil { | |
58 obs.Error(path, err) | |
59 return | |
60 } | |
61 | |
62 if fi.IsDir() { | |
63 f, err := os.Open(path) | |
64 if err != nil { | |
65 obs.Error(path, err) | |
66 } | |
67 | |
68 dircontents, err := f.Readdirnames(-1) | |
69 if err != nil { | |
70 obs.Error(path, err) | |
71 } | |
72 sort.Strings(dircontents) | |
M-A Ruel
2016/09/23 01:48:17
same
| |
73 for _, name := range dircontents { | |
74 fname := filepath.Join(path, name) | |
75 queue.add(fname) | |
76 } | |
77 } else { | |
78 if fi.Size() < smallfile_limit { | |
79 data, err := ioutil.ReadFile(path) | |
80 if err != nil { | |
81 obs.Error(path, err) | |
82 return | |
83 } | |
84 if int64(len(data)) != fi.Size() { | |
85 panic("file size was wrong!") | |
86 } | |
87 obs.SmallFile(path, data) | |
88 } else { | |
89 obs.LargeFile(path) | |
90 } | |
91 } | |
92 queue.done() | |
93 } | |
94 } | |
95 | |
96 func WalkParallel(root string, smallfile_limit int64, obs WalkObserver) { | |
97 queue := fileQueue{queued: 0, finished: 0, items: channels.NewInfiniteCh annel(), waiton: make(chan bool)} | |
98 | |
99 for w := 0; w <= 10; w++ { | |
100 go examinePath(&queue, smallfile_limit, obs) | |
101 } | |
102 | |
103 queue.add(root) | |
104 queue.wait() | |
105 obs.Finished() | |
106 } | |
OLD | NEW |