1 /* Copyright 2018 The TensorFlow Authors. All Rights Reserved.
2
3 Licensed under the Apache License, Version 2.0 (the "License");
4 you may not use this file except in compliance with the License.
5 You may obtain a copy of the License at
6
7 http://www.apache.org/licenses/LICENSE-2.0
8
9 Unless required by applicable law or agreed to in writing, software
10 distributed under the License is distributed on an "AS IS" BASIS,
11 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 See the License for the specific language governing permissions and
13 limitations under the License.
14 ==============================================================================*/
15
16 #include "tensorflow/core/platform/file_system_helper.h"
17
18 #include <deque>
19 #include <string>
20 #include <vector>
21
22 #include "tensorflow/core/lib/core/status.h"
23 #include "tensorflow/core/lib/core/threadpool.h"
24 #include "tensorflow/core/lib/io/path.h"
25 #include "tensorflow/core/lib/strings/str_util.h"
26 #include "tensorflow/core/platform/env.h"
27 #include "tensorflow/core/platform/file_system.h"
28 #include "tensorflow/core/platform/platform.h"
29
30 namespace tensorflow {
31 namespace internal {
32
33 namespace {
34
35 constexpr int kNumThreads = 8;
36
37 // Run a function in parallel using a ThreadPool, but skip the ThreadPool
38 // on the iOS platform due to its problems with more than a few threads.
ForEach(int first,int last,const std::function<void (int)> & f)39 void ForEach(int first, int last, const std::function<void(int)>& f) {
40 #if TARGET_OS_IPHONE
41 for (int i = first; i < last; i++) {
42 f(i);
43 }
44 #else
45 int num_threads = std::min(kNumThreads, last - first);
46 thread::ThreadPool threads(Env::Default(), "ForEach", num_threads);
47 for (int i = first; i < last; i++) {
48 threads.Schedule([f, i] { f(i); });
49 }
50 #endif
51 }
52
53 } // namespace
54
GetMatchingPaths(FileSystem * fs,Env * env,const string & pattern,std::vector<string> * results)55 Status GetMatchingPaths(FileSystem* fs, Env* env, const string& pattern,
56 std::vector<string>* results) {
57 results->clear();
58 // Find the fixed prefix by looking for the first wildcard.
59 string fixed_prefix = pattern.substr(0, pattern.find_first_of("*?[\\"));
60 string eval_pattern = pattern;
61 std::vector<string> all_files;
62 string dir(io::Dirname(fixed_prefix));
63 // If dir is empty then we need to fix up fixed_prefix and eval_pattern to
64 // include . as the top level directory.
65 if (dir.empty()) {
66 dir = ".";
67 fixed_prefix = io::JoinPath(dir, fixed_prefix);
68 eval_pattern = io::JoinPath(dir, pattern);
69 }
70
71 // Setup a BFS to explore everything under dir.
72 std::deque<string> dir_q;
73 dir_q.push_back(dir);
74 Status ret; // Status to return.
75 // children_dir_status holds is_dir status for children. It can have three
76 // possible values: OK for true; FAILED_PRECONDITION for false; CANCELLED
77 // if we don't calculate IsDirectory (we might do that because there isn't
78 // any point in exploring that child path).
79 std::vector<Status> children_dir_status;
80 while (!dir_q.empty()) {
81 string current_dir = dir_q.front();
82 dir_q.pop_front();
83 std::vector<string> children;
84 Status s = fs->GetChildren(current_dir, &children);
85 // In case PERMISSION_DENIED is encountered, we bail here.
86 if (s.code() == tensorflow::error::PERMISSION_DENIED) {
87 continue;
88 }
89 ret.Update(s);
90 if (children.empty()) continue;
91 // This IsDirectory call can be expensive for some FS. Parallelizing it.
92 children_dir_status.resize(children.size());
93 ForEach(0, children.size(),
94 [fs, ¤t_dir, &children, &fixed_prefix,
95 &children_dir_status](int i) {
96 const string child_path = io::JoinPath(current_dir, children[i]);
97 // In case the child_path doesn't start with the fixed_prefix then
98 // we don't need to explore this path.
99 if (!str_util::StartsWith(child_path, fixed_prefix)) {
100 children_dir_status[i] = Status(tensorflow::error::CANCELLED,
101 "Operation not needed");
102 } else {
103 children_dir_status[i] = fs->IsDirectory(child_path);
104 }
105 });
106 for (int i = 0; i < children.size(); ++i) {
107 const string child_path = io::JoinPath(current_dir, children[i]);
108 // If the IsDirectory call was cancelled we bail.
109 if (children_dir_status[i].code() == tensorflow::error::CANCELLED) {
110 continue;
111 }
112 // If the child is a directory add it to the queue.
113 if (children_dir_status[i].ok()) {
114 dir_q.push_back(child_path);
115 }
116 all_files.push_back(child_path);
117 }
118 }
119
120 // Match all obtained files to the input pattern.
121 for (const auto& f : all_files) {
122 if (env->MatchPath(f, eval_pattern)) {
123 results->push_back(f);
124 }
125 }
126 return ret;
127 }
128
129 } // namespace internal
130 } // namespace tensorflow
131