You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by jh...@apache.org on 2018/03/22 20:43:01 UTC
[15/50] [abbrv] hadoop git commit: HDFS-10754: libhdfs++: Create
tools directory and implement hdfs_cat, hdfs_chgrp, hdfs_chown,
hdfs_chmod and hdfs_find. Contributed by Anatoli Shein.
HDFS-10754: libhdfs++: Create tools directory and implement hdfs_cat, hdfs_chgrp, hdfs_chown, hdfs_chmod and hdfs_find. Contributed by Anatoli Shein.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/a5a8a1f5
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/a5a8a1f5
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/a5a8a1f5
Branch: refs/heads/HDFS-8707
Commit: a5a8a1f5d8b163d26c81b6bb9c06d0cdd713e963
Parents: 63b9d31
Author: James <jh...@apache.org>
Authored: Fri Aug 26 12:52:05 2016 -0400
Committer: James Clampffer <ja...@hp.com>
Committed: Thu Mar 22 16:20:45 2018 -0400
----------------------------------------------------------------------
.../src/main/native/libhdfspp/CMakeLists.txt | 1 +
.../libhdfspp/examples/cpp/CMakeLists.txt | 3 +-
.../native/libhdfspp/examples/cpp/cat/cat.cpp | 126 +++++-----
.../libhdfspp/examples/cpp/find/CMakeLists.txt | 35 +++
.../native/libhdfspp/examples/cpp/find/find.cpp | 162 ++++++++++++
.../libhdfspp/examples/cpp/gendirs/gendirs.cpp | 116 ++++-----
.../native/libhdfspp/include/hdfspp/hdfs_ext.h | 14 ++
.../native/libhdfspp/include/hdfspp/hdfspp.h | 45 +++-
.../native/libhdfspp/include/hdfspp/statinfo.h | 1 +
.../native/libhdfspp/lib/bindings/c/hdfs.cc | 64 +++--
.../lib/common/configuration_loader.cc | 8 +-
.../libhdfspp/lib/common/configuration_loader.h | 8 +-
.../lib/common/configuration_loader_impl.h | 2 +-
.../main/native/libhdfspp/lib/fs/filesystem.cc | 244 +++++++++++++++++--
.../main/native/libhdfspp/lib/fs/filesystem.h | 68 +++++-
.../libhdfspp/lib/fs/namenode_operations.cc | 82 ++-----
.../libhdfspp/lib/fs/namenode_operations.h | 8 +-
.../libhdfspp/tests/configuration_test.cc | 84 +++++--
.../native/libhdfspp/tests/configuration_test.h | 4 +-
.../libhdfspp/tests/hdfs_configuration_test.cc | 9 +-
.../src/main/native/libhdfspp/tests/hdfs_shim.c | 4 +
.../libhdfspp/tests/libhdfs_wrapper_undefs.h | 1 +
.../libhdfspp/tests/libhdfspp_wrapper_defines.h | 1 +
.../main/native/libhdfspp/tools/CMakeLists.txt | 42 ++++
.../main/native/libhdfspp/tools/hdfs_cat.cpp | 120 +++++++++
.../main/native/libhdfspp/tools/hdfs_chgrp.cpp | 196 +++++++++++++++
.../main/native/libhdfspp/tools/hdfs_chmod.cpp | 194 +++++++++++++++
.../main/native/libhdfspp/tools/hdfs_chown.cpp | 206 ++++++++++++++++
.../main/native/libhdfspp/tools/hdfs_find.cpp | 156 ++++++++++++
.../native/libhdfspp/tools/tools_common.cpp | 70 ++++++
.../main/native/libhdfspp/tools/tools_common.h | 39 +++
31 files changed, 1829 insertions(+), 284 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/a5a8a1f5/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/CMakeLists.txt b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/CMakeLists.txt
index a663e60..da56403 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/CMakeLists.txt
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/CMakeLists.txt
@@ -163,6 +163,7 @@ add_subdirectory(third_party/uriparser2)
add_subdirectory(lib)
add_subdirectory(tests)
add_subdirectory(examples)
+add_subdirectory(tools)
# create an empty file; hadoop_add_dual_library wraps add_library which
# requires at least one file as an argument
http://git-wip-us.apache.org/repos/asf/hadoop/blob/a5a8a1f5/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/examples/cpp/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/examples/cpp/CMakeLists.txt b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/examples/cpp/CMakeLists.txt
index 183299a..9e16b0b 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/examples/cpp/CMakeLists.txt
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/examples/cpp/CMakeLists.txt
@@ -17,4 +17,5 @@
#
add_subdirectory(cat)
-add_subdirectory(gendirs)
\ No newline at end of file
+add_subdirectory(gendirs)
+add_subdirectory(find)
http://git-wip-us.apache.org/repos/asf/hadoop/blob/a5a8a1f5/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/examples/cpp/cat/cat.cpp
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/examples/cpp/cat/cat.cpp b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/examples/cpp/cat/cat.cpp
index bfab507..17626ea 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/examples/cpp/cat/cat.cpp
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/examples/cpp/cat/cat.cpp
@@ -17,105 +17,91 @@
under the License.
*/
-/*
- A a stripped down version of unix's "cat".
- Doesn't deal with any flags for now, will just attempt to read the whole file.
-*/
+/**
+ * Unix-like cat tool example.
+ *
+ * Reads the specified file from HDFS and outputs to stdout.
+ *
+ * Usage: cat /<path-to-file>
+ *
+ * Example: cat /dir/file
+ *
+ * @param path-to-file Absolute path to the file to read.
+ *
+ **/
#include "hdfspp/hdfspp.h"
#include "common/hdfs_configuration.h"
#include "common/configuration_loader.h"
-#include "common/uri.h"
-
-#include <google/protobuf/io/coded_stream.h>
-using namespace std;
-using namespace hdfs;
+#include <google/protobuf/stubs/common.h>
-#define SCHEME "hdfs"
+#define BUF_SIZE 4096
int main(int argc, char *argv[]) {
if (argc != 2) {
- cerr << "usage: cat [hdfs://[<hostname>:<port>]]/<path-to-file>" << endl;
- return 1;
- }
-
- optional<URI> uri;
- const string uri_path = argv[1];
-
- //Separate check for scheme is required, otherwise common/uri.h library causes memory issues under valgrind
- size_t scheme_end = uri_path.find("://");
- if (scheme_end != string::npos) {
- if(uri_path.substr(0, string(SCHEME).size()).compare(SCHEME) != 0) {
- cerr << "Scheme " << uri_path.substr(0, scheme_end) << ":// is not supported" << endl;
- return 1;
- } else {
- uri = URI::parse_from_string(uri_path);
- }
- }
- if (!uri) {
- cerr << "Malformed URI: " << uri_path << endl;
- return 1;
+ std::cerr << "usage: cat /<path-to-file>" << std::endl;
+ exit(EXIT_FAILURE);
}
-
- ConfigurationLoader loader;
- optional<HdfsConfiguration> config = loader.LoadDefaultResources<HdfsConfiguration>();
- const char * envHadoopConfDir = getenv("HADOOP_CONF_DIR");
- if (envHadoopConfDir && (*envHadoopConfDir != 0) ) {
- config = loader.OverlayResourceFile(*config, string(envHadoopConfDir) + "/core-site.xml");
- }
-
- Options options;
+ std::string path = argv[1];
+
+ hdfs::Options options;
+ //Setting the config path to the default: "$HADOOP_CONF_DIR" or "/etc/hadoop/conf"
+ hdfs::ConfigurationLoader loader;
+ //Loading default config files core-site.xml and hdfs-site.xml from the config path
+ hdfs::optional<hdfs::HdfsConfiguration> config = loader.LoadDefaultResources<hdfs::HdfsConfiguration>();
+ //TODO: HDFS-9539 - after this is resolved, valid config will always be returned.
if(config){
+ //Loading options from the config
options = config->GetOptions();
}
-
- IoService * io_service = IoService::New();
-
- FileSystem *fs_raw = FileSystem::New(io_service, "", options);
- if (!fs_raw) {
- cerr << "Could not create FileSystem object" << endl;
- return 1;
+ hdfs::IoService * io_service = hdfs::IoService::New();
+ //Wrapping fs into a shared pointer to guarantee deletion
+ std::shared_ptr<hdfs::FileSystem> fs(hdfs::FileSystem::New(io_service, "", options));
+ if (!fs) {
+ std::cerr << "Could not connect the file system." << std::endl;
+ exit(EXIT_FAILURE);
}
- //wrapping fs_raw into a unique pointer to guarantee deletion
- unique_ptr<FileSystem> fs(fs_raw);
-
- Status stat = fs->Connect(uri->get_host(), to_string(*(uri->get_port())));
- if (!stat.ok()) {
- cerr << "Could not connect to " << uri->get_host() << ":" << *(uri->get_port()) << endl;
- return 1;
+ hdfs::Status status = fs->ConnectToDefaultFs();
+ if (!status.ok()) {
+ if(!options.defaultFS.get_host().empty()){
+ std::cerr << "Error connecting to " << options.defaultFS << ". " << status.ToString() << std::endl;
+ } else {
+ std::cerr << "Error connecting to the cluster: defaultFS is empty. " << status.ToString() << std::endl;
+ }
+ exit(EXIT_FAILURE);
}
- FileHandle *file_raw = nullptr;
- stat = fs->Open(uri->get_path(), &file_raw);
- if (!stat.ok()) {
- cerr << "Could not open file " << uri->get_path() << endl;
- return 1;
+ hdfs::FileHandle *file_raw = nullptr;
+ status = fs->Open(path, &file_raw);
+ if (!status.ok()) {
+ std::cerr << "Could not open file " << path << ". " << status.ToString() << std::endl;
+ exit(EXIT_FAILURE);
}
//wrapping file_raw into a unique pointer to guarantee deletion
- unique_ptr<FileHandle> file(file_raw);
+ std::unique_ptr<hdfs::FileHandle> file(file_raw);
- char input_buffer[4096];
- ssize_t read_bytes_count = 0;
- size_t last_read_bytes = 0;
+ char input_buffer[BUF_SIZE];
+ ssize_t total_bytes_read = 0;
+ size_t last_bytes_read = 0;
do{
//Reading file chunks
- Status stat = file->PositionRead(input_buffer, sizeof(input_buffer), read_bytes_count, &last_read_bytes);
- if(stat.ok()) {
+ status = file->Read(input_buffer, sizeof(input_buffer), &last_bytes_read);
+ if(status.ok()) {
//Writing file chunks to stdout
- fwrite(input_buffer, last_read_bytes, 1, stdout);
- read_bytes_count += last_read_bytes;
+ fwrite(input_buffer, last_bytes_read, 1, stdout);
+ total_bytes_read += last_bytes_read;
} else {
- if(stat.is_invalid_offset()){
+ if(status.is_invalid_offset()){
//Reached the end of the file
break;
} else {
- cerr << "Error reading the file: " << stat.ToString() << endl;
- return 1;
+ std::cerr << "Error reading the file: " << status.ToString() << std::endl;
+ exit(EXIT_FAILURE);
}
}
- } while (last_read_bytes > 0);
+ } while (last_bytes_read > 0);
// Clean up static data and prevent valgrind memory leaks
google::protobuf::ShutdownProtobufLibrary();
http://git-wip-us.apache.org/repos/asf/hadoop/blob/a5a8a1f5/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/examples/cpp/find/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/examples/cpp/find/CMakeLists.txt b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/examples/cpp/find/CMakeLists.txt
new file mode 100644
index 0000000..7ae27c5
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/examples/cpp/find/CMakeLists.txt
@@ -0,0 +1,35 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# Default LIBHDFSPP_DIR to the default install location. You can override
+# it by add -DLIBHDFSPP_DIR=... to your cmake invocation
+set(LIBHDFSPP_DIR CACHE STRING ${CMAKE_INSTALL_PREFIX})
+
+include_directories( ${LIBHDFSPP_DIR}/include )
+link_directories( ${LIBHDFSPP_DIR}/lib )
+
+add_executable(find_cpp find.cpp)
+target_link_libraries(find_cpp hdfspp)
+
+# Several examples in different languages need to produce executables with
+# same names. To allow executables with same names we keep their CMake
+# names different, but specify their executable names as follows:
+set_target_properties( find_cpp
+ PROPERTIES
+ OUTPUT_NAME "find"
+)
http://git-wip-us.apache.org/repos/asf/hadoop/blob/a5a8a1f5/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/examples/cpp/find/find.cpp
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/examples/cpp/find/find.cpp b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/examples/cpp/find/find.cpp
new file mode 100644
index 0000000..21a731b
--- /dev/null
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/examples/cpp/find/find.cpp
@@ -0,0 +1,162 @@
+/*
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+*/
+
+/**
+ * A parallel find tool example.
+ *
+ * Finds all files matching the specified name recursively starting from the
+ * specified directory and prints their filepaths. Works either synchronously
+ * or asynchronously.
+ *
+ * Usage: find /<path-to-file> <file-name> <use_async>
+ *
+ * Example: find /dir?/tree* some?file*name 1
+ *
+ * @param path-to-file Absolute path at which to begin search, can have wild
+ * cards and must be non-blank
+ * @param file-name Name to find, can have wild cards and must be non-blank
+ * @param use_async If set to 1 it prints out results asynchronously as
+ * they arrive. If set to 0 results are printed in one
+ * big chunk when it becomes available.
+ *
+ **/
+
+#include "hdfspp/hdfspp.h"
+#include "common/hdfs_configuration.h"
+#include "common/configuration_loader.h"
+
+#include <google/protobuf/stubs/common.h>
+#include <future>
+
+void SyncFind(std::shared_ptr<hdfs::FileSystem> fs, const std::string &path, const std::string &name){
+ std::vector<hdfs::StatInfo> results;
+ //Synchronous call to Find
+ hdfs::Status stat = fs->Find(path, name, hdfs::FileSystem::GetDefaultFindMaxDepth(), &results);
+
+ if (!stat.ok()) {
+ std::cerr << "Error: " << stat.ToString() << std::endl;
+ }
+
+ if(results.empty()){
+ std::cout << "Nothing Found" << std::endl;
+ } else {
+ //Printing out the results
+ for (hdfs::StatInfo const& si : results) {
+ std::cout << si.full_path << std::endl;
+ }
+ }
+}
+
+void AsyncFind(std::shared_ptr<hdfs::FileSystem> fs, const std::string &path, const std::string &name){
+ std::promise<void> promise;
+ std::future<void> future(promise.get_future());
+ bool something_found = false;
+ hdfs::Status status = hdfs::Status::OK();
+
+ /**
+ * Keep requesting more until we get the entire listing. Set the promise
+ * when we have the entire listing to stop.
+ *
+ * Find guarantees that the handler will only be called once at a time,
+ * so we do not need any locking here
+ */
+ auto handler = [&promise, &status, &something_found]
+ (const hdfs::Status &s, const std::vector<hdfs::StatInfo> & si, bool has_more_results) -> bool {
+ //Print result chunks as they arrive
+ if(!si.empty()) {
+ something_found = true;
+ for (hdfs::StatInfo const& s : si) {
+ std::cout << s.full_path << std::endl;
+ }
+ }
+ if(!s.ok() && status.ok()){
+ //We make sure we set 'status' only on the first error.
+ status = s;
+ }
+ if (!has_more_results) {
+ promise.set_value(); //set promise
+ return false; //request stop sending results
+ }
+ return true; //request more results
+ };
+
+ //Asynchronous call to Find
+ fs->Find(path, name, hdfs::FileSystem::GetDefaultFindMaxDepth(), handler);
+
+ //block until promise is set
+ future.get();
+ if(!status.ok()) {
+ std::cerr << "Error: " << status.ToString() << std::endl;
+ }
+ if(!something_found){
+ std::cout << "Nothing Found" << std::endl;
+ }
+}
+
+int main(int argc, char *argv[]) {
+ if (argc != 4) {
+ std::cerr << "usage: find /<path-to-file> <file-name> <use_async>" << std::endl;
+ exit(EXIT_FAILURE);
+ }
+
+ std::string path = argv[1];
+ std::string name = argv[2];
+ bool use_async = (std::stoi(argv[3]) != 0);
+
+ hdfs::Options options;
+ //Setting the config path to the default: "$HADOOP_CONF_DIR" or "/etc/hadoop/conf"
+ hdfs::ConfigurationLoader loader;
+ //Loading default config files core-site.xml and hdfs-site.xml from the config path
+ hdfs::optional<hdfs::HdfsConfiguration> config = loader.LoadDefaultResources<hdfs::HdfsConfiguration>();
+ //TODO: HDFS-9539 - after this is resolved, valid config will always be returned.
+ if(config){
+ //Loading options from the config
+ options = config->GetOptions();
+ }
+ //TODO: HDFS-9539 - until then we increase the time-out to allow all recursive async calls to finish
+ options.rpc_timeout = std::numeric_limits<int>::max();
+ hdfs::IoService * io_service = hdfs::IoService::New();
+ //Wrapping fs into a unique pointer to guarantee deletion
+ std::shared_ptr<hdfs::FileSystem> fs(hdfs::FileSystem::New(io_service, "", options));
+ if (!fs) {
+ std::cerr << "Could not connect the file system." << std::endl;
+ exit(EXIT_FAILURE);
+ }
+ hdfs::Status status = fs->ConnectToDefaultFs();
+ if (!status.ok()) {
+ if(!options.defaultFS.get_host().empty()){
+ std::cerr << "Error connecting to " << options.defaultFS << ". " << status.ToString() << std::endl;
+ } else {
+ std::cerr << "Error connecting to the cluster: defaultFS is empty. " << status.ToString() << std::endl;
+ }
+ exit(EXIT_FAILURE);
+ }
+
+ if (use_async){
+ //Example of Async find
+ AsyncFind(fs, path, name);
+ } else {
+ //Example of Sync find
+ SyncFind(fs, path, name);
+ }
+
+ // Clean up static data and prevent valgrind memory leaks
+ google::protobuf::ShutdownProtobufLibrary();
+ return 0;
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/a5a8a1f5/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/examples/cpp/gendirs/gendirs.cpp
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/examples/cpp/gendirs/gendirs.cpp b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/examples/cpp/gendirs/gendirs.cpp
index 9f5ae5a..c90abbd 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/examples/cpp/gendirs/gendirs.cpp
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/examples/cpp/gendirs/gendirs.cpp
@@ -23,9 +23,9 @@
* Generates a directory tree with specified depth and fanout starting from
* a given path. Generation is asynchronous.
*
- * Usage: gendirs [hdfs://[<hostname>:<port>]]/<path-to-dir> <depth> <fanout>
+ * Usage: gendirs /<path-to-dir> <depth> <fanout>
*
- * Example: gendirs hdfs://localhost.localdomain:9433/dir0 3 10
+ * Example: gendirs /dir0 3 10
*
* @param path-to-dir Absolute path to the directory tree root where the
* directory tree will be generated
@@ -37,99 +37,76 @@
**/
#include "hdfspp/hdfspp.h"
-#include "fs/namenode_operations.h"
#include "common/hdfs_configuration.h"
#include "common/configuration_loader.h"
-#include "common/uri.h"
-#include <google/protobuf/io/coded_stream.h>
+#include <google/protobuf/stubs/common.h>
#include <future>
-using namespace std;
-using namespace hdfs;
+#define DEFAULT_PERMISSIONS 0755
-#define SCHEME "hdfs"
-
-void GenerateDirectories (shared_ptr<FileSystem> fs, int depth, int level, int fanout, string path, vector<future<Status>> & futures) {
+void GenerateDirectories (std::shared_ptr<hdfs::FileSystem> fs, int depth, int level, int fanout, std::string path, std::vector<std::future<hdfs::Status>> & futures) {
//Level contains our current depth in the directory tree
if(level < depth) {
for(int i = 0; i < fanout; i++){
//Recursive calls to cover all possible paths from the root to the leave nodes
- GenerateDirectories(fs, depth, level+1, fanout, path + "dir" + to_string(i) + "/", futures);
+ GenerateDirectories(fs, depth, level+1, fanout, path + "dir" + std::to_string(i) + "/", futures);
}
} else {
//We have reached the leaf nodes and now start making calls to create directories
//We make a promise which will be set when the call finishes and executes our handler
- auto callstate = make_shared<promise<Status>>();
+ auto callstate = std::make_shared<std::promise<hdfs::Status>>();
//Extract a future from this promise
- future<Status> future(callstate->get_future());
+ std::future<hdfs::Status> future(callstate->get_future());
//Save this future to the vector of futures which will be used to wait on all promises
//after the whole recursion is done
- futures.push_back(move(future));
+ futures.push_back(std::move(future));
//Create a handler that will be executed when Mkdirs is done
- auto handler = [callstate](const Status &s) {
+ auto handler = [callstate](const hdfs::Status &s) {
callstate->set_value(s);
};
//Asynchronous call to create this directory along with all missing parent directories
- fs->Mkdirs(path, NameNodeOperations::GetDefaultPermissionMask(), true, handler);
+ fs->Mkdirs(path, DEFAULT_PERMISSIONS, true, handler);
}
}
int main(int argc, char *argv[]) {
if (argc != 4) {
- cerr << "usage: gendirs [hdfs://[<hostname>:<port>]]/<path-to-dir> <depth> <fanout>" << endl;
- return 1;
+ std::cerr << "usage: gendirs /<path-to-dir> <depth> <fanout>" << std::endl;
+ exit(EXIT_FAILURE);
}
- optional<URI> uri;
- const string uri_path = argv[1];
- const int depth = stoi(argv[2]);
- const int fanout = stoi(argv[3]);
+ std::string path = argv[1];
+ int depth = std::stoi(argv[2]);
+ int fanout = std::stoi(argv[3]);
- //Separate check for scheme is required, otherwise common/uri.h library causes memory issues under valgrind
- size_t scheme_end = uri_path.find("://");
- if (scheme_end != string::npos) {
- if(uri_path.substr(0, string(SCHEME).size()).compare(SCHEME) != 0) {
- cerr << "Scheme " << uri_path.substr(0, scheme_end) << ":// is not supported" << endl;
- return 1;
- } else {
- uri = URI::parse_from_string(uri_path);
- }
- }
- if (!uri) {
- cerr << "Malformed URI: " << uri_path << endl;
- return 1;
- }
-
- ConfigurationLoader loader;
- optional<HdfsConfiguration> config = loader.LoadDefaultResources<HdfsConfiguration>();
- const char * envHadoopConfDir = getenv("HADOOP_CONF_DIR");
- if (envHadoopConfDir && (*envHadoopConfDir != 0) ) {
- config = loader.OverlayResourceFile(*config, string(envHadoopConfDir) + "/core-site.xml");
- }
-
- Options options;
- options.rpc_timeout = numeric_limits<int>::max();
+ hdfs::Options options;
+ //Setting the config path to the default: "$HADOOP_CONF_DIR" or "/etc/hadoop/conf"
+ hdfs::ConfigurationLoader loader;
+ //Loading default config files core-site.xml and hdfs-site.xml from the config path
+ hdfs::optional<hdfs::HdfsConfiguration> config = loader.LoadDefaultResources<hdfs::HdfsConfiguration>();
+ //TODO: HDFS-9539 - after this is resolved, valid config will always be returned.
if(config){
+ //Loading options from the config
options = config->GetOptions();
}
-
- IoService * io_service = IoService::New();
-
- FileSystem *fs_raw = FileSystem::New(io_service, "", options);
- if (!fs_raw) {
- cerr << "Could not create FileSystem object" << endl;
- return 1;
+ //TODO: HDFS-9539 - until then we increase the time-out to allow all recursive async calls to finish
+ options.rpc_timeout = std::numeric_limits<int>::max();
+ hdfs::IoService * io_service = hdfs::IoService::New();
+ //Wrapping fs into a unique pointer to guarantee deletion
+ std::shared_ptr<hdfs::FileSystem> fs(hdfs::FileSystem::New(io_service, "", options));
+ if (!fs) {
+ std::cerr << "Could not connect the file system." << std::endl;
+ exit(EXIT_FAILURE);
}
- //Wrapping fs_raw into a unique pointer to guarantee deletion
- shared_ptr<FileSystem> fs(fs_raw);
-
- //Get port from the uri, otherwise use the default port
- string port = to_string(uri->get_port().value_or(8020));
- Status stat = fs->Connect(uri->get_host(), port);
- if (!stat.ok()) {
- cerr << "Could not connect to " << uri->get_host() << ":" << port << endl;
- return 1;
+ hdfs::Status status = fs->ConnectToDefaultFs();
+ if (!status.ok()) {
+ if(!options.defaultFS.get_host().empty()){
+ std::cerr << "Error connecting to " << options.defaultFS << ". " << status.ToString() << std::endl;
+ } else {
+ std::cerr << "Error connecting to the cluster: defaultFS is empty. " << status.ToString() << std::endl;
+ }
+ exit(EXIT_FAILURE);
}
/**
@@ -143,22 +120,23 @@ int main(int argc, char *argv[]) {
* processed. After the whole recursion is complete we will need to wait until
* all promises are set before we can exit.
**/
- vector<future<Status>> futures;
+ std::vector<std::future<hdfs::Status>> futures;
- GenerateDirectories(fs, depth, 0, fanout, uri->get_path() + "/", futures);
+ GenerateDirectories(fs, depth, 0, fanout, path + "/", futures);
/**
* We are waiting here until all promises are set, and checking whether
* the returned statuses contained any errors.
**/
- for(future<Status> &fs : futures){
- Status stat = fs.get();
- if (!stat.ok()) {
- cerr << "Error: " << stat.ToString() << endl;
+ for(std::future<hdfs::Status> &fs : futures){
+ hdfs::Status status = fs.get();
+ if (!status.ok()) {
+ std::cerr << "Error: " << status.ToString() << std::endl;
+ exit(EXIT_FAILURE);
}
}
- cout << "All done!" << endl;
+ std::cout << "All done!" << std::endl;
// Clean up static data and prevent valgrind memory leaks
google::protobuf::ShutdownProtobufLibrary();
http://git-wip-us.apache.org/repos/asf/hadoop/blob/a5a8a1f5/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/include/hdfspp/hdfs_ext.h
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/include/hdfspp/hdfs_ext.h b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/include/hdfspp/hdfs_ext.h
index ce9f0f5..b41857c 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/include/hdfspp/hdfs_ext.h
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/include/hdfspp/hdfs_ext.h
@@ -287,6 +287,20 @@ LIBHDFS_EXTERNAL
int hdfsPreAttachFileMonitor(libhdfspp_file_event_callback handler, int64_t cookie);
+/**
+ * Finds file name on the file system. hdfsFreeFileInfo should be called to deallocate memory.
+ *
+ * @param fs The filesystem (required)
+ * @param path Path at which to begin search, can have wild cards (must be non-blank)
+ * @param name Name to find, can have wild cards (must be non-blank)
+ * @param numEntries Set to the number of files/directories in the result.
+ * @return Returns a dynamically-allocated array of hdfsFileInfo
+ * objects; NULL on error or empty result.
+ * errno is set to non-zero on error or zero on success.
+ **/
+hdfsFileInfo * hdfsFind(hdfsFS fs, const char* path, const char* name, uint32_t * numEntries);
+
+
/*****************************************************************************
* HDFS SNAPSHOT FUNCTIONS
****************************************************************************/
http://git-wip-us.apache.org/repos/asf/hadoop/blob/a5a8a1f5/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/include/hdfspp/hdfspp.h
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/include/hdfspp/hdfspp.h b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/include/hdfspp/hdfspp.h
index 46fe8e9..4b88fe5 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/include/hdfspp/hdfspp.h
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/include/hdfspp/hdfspp.h
@@ -138,6 +138,18 @@ public:
**/
class FileSystem {
public:
+ //Returns the default maximum depth for recursive Find tool
+ static uint32_t GetDefaultFindMaxDepth();
+
+ //Returns the default permission mask
+ static uint16_t GetDefaultPermissionMask();
+
+ //Checks if the given permission mask is valid
+ static Status CheckValidPermissionMask(uint16_t permissions);
+
+ //Checks if replication value is valid
+ static Status CheckValidReplication(uint16_t replication);
+
/**
* Create a new instance of the FileSystem object. The call
* initializes the RPC connections to the NameNode and returns an
@@ -236,7 +248,7 @@ class FileSystem {
*
* The asynchronous method will return batches of files; the consumer must
* return true if they want more files to be delivered. The final bool
- * parameter in the callback will be set to true if this is the final
+ * parameter in the callback will be set to false if this is the final
* batch of files.
*
* The synchronous method will return all files in the directory.
@@ -245,9 +257,8 @@ class FileSystem {
**/
virtual void
GetListing(const std::string &path,
- const std::function<bool(const Status &, std::shared_ptr<std::vector<StatInfo>> &, bool)> &handler) = 0;
- virtual Status GetListing(const std::string &path,
- std::shared_ptr<std::vector<StatInfo>> & stat_infos) = 0;
+ const std::function<bool(const Status &, const std::vector<StatInfo> &, bool)> &handler) = 0;
+ virtual Status GetListing(const std::string &path, std::vector<StatInfo> * stat_infos) = 0;
/**
* Returns the locations of all known blocks for the indicated file (or part of it), or an error
@@ -297,8 +308,8 @@ class FileSystem {
* @param path the path to the file or directory
* @param permissions the bitmask to set it to (should be between 0 and 01777)
*/
- virtual void SetPermission(const std::string & path,
- uint16_t permissions, const std::function<void(const Status &)> &handler) = 0;
+ virtual void SetPermission(const std::string & path, uint16_t permissions,
+ const std::function<void(const Status &)> &handler) = 0;
virtual Status SetPermission(const std::string & path, uint16_t permissions) = 0;
/**
@@ -307,12 +318,34 @@ class FileSystem {
* @param path file path
* @param username If it is empty, the original username remains unchanged.
* @param groupname If it is empty, the original groupname remains unchanged.
+ * @param recursive If true, the change will be propagated recursively.
*/
virtual void SetOwner(const std::string & path, const std::string & username,
const std::string & groupname, const std::function<void(const Status &)> &handler) = 0;
virtual Status SetOwner(const std::string & path,
const std::string & username, const std::string & groupname) = 0;
+ /**
+ * Finds all files matching the specified name recursively starting from the
+ * specified directory. Returns metadata for each of them.
+ *
+ * Example: Find("/dir?/tree*", "some?file*name")
+ *
+ * @param path Absolute path at which to begin search, can have wild cards (must be non-blank)
+ * @param name Name to find, can also have wild cards (must be non-blank)
+ *
+ * The asynchronous method will return batches of files; the consumer must
+ * return true if they want more files to be delivered. The final bool
+ * parameter in the callback will be set to false if this is the final
+ * batch of files.
+ *
+ * The synchronous method will return matching files.
+ **/
+ virtual void
+ Find(const std::string &path, const std::string &name, const uint32_t maxdepth,
+ const std::function<bool(const Status &, const std::vector<StatInfo> & , bool)> &handler) = 0;
+ virtual Status Find(const std::string &path, const std::string &name,
+ const uint32_t maxdepth, std::vector<StatInfo> * stat_infos) = 0;
/*****************************************************************************
http://git-wip-us.apache.org/repos/asf/hadoop/blob/a5a8a1f5/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/include/hdfspp/statinfo.h
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/include/hdfspp/statinfo.h b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/include/hdfspp/statinfo.h
index a53ab8b..e077dda 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/include/hdfspp/statinfo.h
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/include/hdfspp/statinfo.h
@@ -33,6 +33,7 @@ struct StatInfo {
int file_type;
::std::string path;
+ ::std::string full_path;
unsigned long int length;
unsigned long int permissions; //Octal number as in POSIX permissions; e.g. 0777
::std::string owner;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/a5a8a1f5/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/bindings/c/hdfs.cc
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/bindings/c/hdfs.cc b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/bindings/c/hdfs.cc
index 4003358..a43d94f 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/bindings/c/hdfs.cc
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/bindings/c/hdfs.cc
@@ -731,22 +731,21 @@ hdfsFileInfo *hdfsListDirectory(hdfsFS fs, const char* path, int *numEntries) {
if(!abs_path) {
return nullptr;
}
- std::shared_ptr<std::vector<StatInfo>> stat_infos;
- Status stat = fs->get_impl()->GetListing(*abs_path, stat_infos);
+ std::vector<StatInfo> stat_infos;
+ Status stat = fs->get_impl()->GetListing(*abs_path, &stat_infos);
if (!stat.ok()) {
Error(stat);
*numEntries = 0;
return nullptr;
}
- //Existing API expects nullptr if size is 0
- if(!stat_infos || stat_infos->size()==0){
+ if(stat_infos.empty()){
*numEntries = 0;
return nullptr;
}
- *numEntries = stat_infos->size();
- hdfsFileInfo *file_infos = new hdfsFileInfo[stat_infos->size()];
- for(std::vector<StatInfo>::size_type i = 0; i < stat_infos->size(); i++) {
- StatInfoToHdfsFileInfo(&file_infos[i], stat_infos->at(i));
+ *numEntries = stat_infos.size();
+ hdfsFileInfo *file_infos = new hdfsFileInfo[stat_infos.size()];
+ for(std::vector<StatInfo>::size_type i = 0; i < stat_infos.size(); i++) {
+ StatInfoToHdfsFileInfo(&file_infos[i], stat_infos.at(i));
}
return file_infos;
@@ -785,7 +784,7 @@ int hdfsCreateDirectory(hdfsFS fs, const char* path) {
}
Status stat;
//Use default permissions and set true for creating all non-existant parent directories
- stat = fs->get_impl()->Mkdirs(*abs_path, NameNodeOperations::GetDefaultPermissionMask(), true);
+ stat = fs->get_impl()->Mkdirs(*abs_path, FileSystem::GetDefaultPermissionMask(), true);
if (!stat.ok()) {
return Error(stat);
}
@@ -854,7 +853,7 @@ int hdfsChmod(hdfsFS fs, const char* path, short mode){
if(!abs_path) {
return -1;
}
- Status stat = NameNodeOperations::CheckValidPermissionMask(mode);
+ Status stat = FileSystem::CheckValidPermissionMask(mode);
if (!stat.ok()) {
return Error(stat);
}
@@ -896,6 +895,44 @@ int hdfsChown(hdfsFS fs, const char* path, const char *owner, const char *group)
}
}
+hdfsFileInfo * hdfsFind(hdfsFS fs, const char* path, const char* name, uint32_t * numEntries){
+ try {
+ errno = 0;
+ if (!CheckSystem(fs)) {
+ *numEntries = 0;
+ return nullptr;
+ }
+
+ std::vector<StatInfo> stat_infos;
+ Status stat = fs->get_impl()->Find(path, name, hdfs::FileSystem::GetDefaultFindMaxDepth(), &stat_infos);
+ if (!stat.ok()) {
+ Error(stat);
+ *numEntries = 0;
+ return nullptr;
+ }
+ //Existing API expects nullptr if size is 0
+ if(stat_infos.empty()){
+ *numEntries = 0;
+ return nullptr;
+ }
+ *numEntries = stat_infos.size();
+ hdfsFileInfo *file_infos = new hdfsFileInfo[stat_infos.size()];
+ for(std::vector<StatInfo>::size_type i = 0; i < stat_infos.size(); i++) {
+ StatInfoToHdfsFileInfo(&file_infos[i], stat_infos.at(i));
+ }
+
+ return file_infos;
+ } catch (const std::exception & e) {
+ ReportException(e);
+ *numEntries = 0;
+ return nullptr;
+ } catch (...) {
+ ReportCaughtNonException();
+ *numEntries = 0;
+ return nullptr;
+ }
+}
+
int hdfsCreateSnapshot(hdfsFS fs, const char* path, const char* name) {
try {
errno = 0;
@@ -1373,19 +1410,18 @@ HdfsConfiguration LoadDefault(ConfigurationLoader & loader)
}
else
{
- return loader.New<HdfsConfiguration>();
+ return loader.NewConfig<HdfsConfiguration>();
}
}
-hdfsBuilder::hdfsBuilder() : config(loader.New<HdfsConfiguration>())
+hdfsBuilder::hdfsBuilder() : config(loader.NewConfig<HdfsConfiguration>())
{
errno = 0;
- loader.SetDefaultSearchPath();
config = LoadDefault(loader);
}
hdfsBuilder::hdfsBuilder(const char * directory) :
- config(loader.New<HdfsConfiguration>())
+ config(loader.NewConfig<HdfsConfiguration>())
{
errno = 0;
loader.SetSearchPath(directory);
http://git-wip-us.apache.org/repos/asf/hadoop/blob/a5a8a1f5/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/configuration_loader.cc
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/configuration_loader.cc b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/configuration_loader.cc
index 1eb70c3..e1434dad 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/configuration_loader.cc
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/configuration_loader.cc
@@ -62,6 +62,12 @@ bool str_to_bool(const std::string& raw) {
return false;
}
+ConfigurationLoader::ConfigurationLoader() {
+ //In order to creates a configuration loader with the default search path
+ //("$HADOOP_CONF_DIR" or "/etc/hadoop/conf") we call SetDefaultSearchPath().
+ ConfigurationLoader::SetDefaultSearchPath();
+}
+
void ConfigurationLoader::SetDefaultSearchPath() {
// Try (in order, taking the first valid one):
// $HADOOP_CONF_DIR
@@ -257,4 +263,4 @@ bool ConfigurationLoader::UpdateMapWithValue(ConfigMap& map,
return true;
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/a5a8a1f5/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/configuration_loader.h
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/configuration_loader.h b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/configuration_loader.h
index 059e48b..51ac23a 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/configuration_loader.h
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/configuration_loader.h
@@ -27,9 +27,9 @@ namespace hdfs {
class ConfigurationLoader {
public:
// Creates a new, empty Configuration object
- // T must be Configuration or a subclass
+ // T must be Configuration or a subclass
template<class T>
- T New();
+ T NewConfig();
/****************************************************************************
* LOADING CONFIG FILES
@@ -79,6 +79,10 @@ public:
* SEARCH PATH METHODS
***************************************************************************/
+ //Creates a configuration loader with the default search path ("$HADOOP_CONF_DIR" or "/etc/hadoop/conf").
+ //If you want to explicitly set the entire search path, call ClearSearchPath() first
+ ConfigurationLoader();
+
// Sets the search path to the default search path (namely, "$HADOOP_CONF_DIR" or "/etc/hadoop/conf")
void SetDefaultSearchPath();
http://git-wip-us.apache.org/repos/asf/hadoop/blob/a5a8a1f5/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/configuration_loader_impl.h
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/configuration_loader_impl.h b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/configuration_loader_impl.h
index 9e18878..6258450 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/configuration_loader_impl.h
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/configuration_loader_impl.h
@@ -23,7 +23,7 @@ namespace hdfs {
template<class T>
-T ConfigurationLoader::New() {
+T ConfigurationLoader::NewConfig() {
return T();
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/a5a8a1f5/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/fs/filesystem.cc
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/fs/filesystem.cc b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/fs/filesystem.cc
index de6ebb7..d75939f 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/fs/filesystem.cc
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/fs/filesystem.cc
@@ -26,6 +26,7 @@
#include <tuple>
#include <iostream>
#include <pwd.h>
+#include <fnmatch.h>
#define FMT_THIS_ADDR "this=" << (void*)this
@@ -39,6 +40,34 @@ using ::asio::ip::tcp;
static constexpr uint16_t kDefaultPort = 8020;
+uint32_t FileSystem::GetDefaultFindMaxDepth() {
+ return std::numeric_limits<uint32_t>::max();
+}
+
+uint16_t FileSystem::GetDefaultPermissionMask() {
+ return 0755;
+}
+
+Status FileSystem::CheckValidPermissionMask(uint16_t permissions) {
+ if (permissions > 01777) {
+ std::stringstream errormsg;
+ errormsg << "CheckValidPermissionMask: argument 'permissions' is " << std::oct
+ << std::showbase << permissions << " (should be between 0 and 01777)";
+ return Status::InvalidArgument(errormsg.str().c_str());
+ }
+ return Status::OK();
+}
+
+Status FileSystem::CheckValidReplication(uint16_t replication) {
+ if (replication < 1 || replication > 512) {
+ std::stringstream errormsg;
+ errormsg << "CheckValidReplication: argument 'replication' is "
+ << replication << " (should be between 1 and 512)";
+ return Status::InvalidArgument(errormsg.str().c_str());
+ }
+ return Status::OK();
+}
+
/*****************************************************************************
* FILESYSTEM BASE CLASS
****************************************************************************/
@@ -446,7 +475,7 @@ void FileSystemImpl::SetReplication(const std::string & path, int16_t replicatio
handler(Status::InvalidArgument("SetReplication: argument 'path' cannot be empty"));
return;
}
- Status replStatus = NameNodeOperations::CheckValidReplication(replication);
+ Status replStatus = FileSystem::CheckValidReplication(replication);
if (!replStatus.ok()) {
handler(replStatus);
return;
@@ -593,44 +622,43 @@ Status FileSystemImpl::GetFsStats(FsInfo & fs_info) {
* Some compilers don't like recursive lambdas, so we make the lambda call a
* method, which in turn creates a lambda calling itself.
*/
-void FileSystemImpl::GetListingShim(const Status &stat, std::shared_ptr<std::vector<StatInfo>> &stat_infos, bool has_more,
- std::string path,
- const std::function<bool(const Status &, std::shared_ptr<std::vector<StatInfo>>&, bool)> &handler) {
- bool has_next = stat_infos && stat_infos->size() > 0;
+void FileSystemImpl::GetListingShim(const Status &stat, const std::vector<StatInfo> & stat_infos, bool has_more,
+ std::string path, const std::function<bool(const Status &, const std::vector<StatInfo> &, bool)> &handler) {
+ bool has_next = !stat_infos.empty();
bool get_more = handler(stat, stat_infos, has_more && has_next);
if (get_more && has_more && has_next ) {
- auto callback = [this, path, handler](const Status &stat, std::shared_ptr<std::vector<StatInfo>> &stat_infos, bool has_more) {
+ auto callback = [this, path, handler](const Status &stat, const std::vector<StatInfo> & stat_infos, bool has_more) {
GetListingShim(stat, stat_infos, has_more, path, handler);
};
- std::string last = stat_infos->back().path;
+ std::string last = stat_infos.back().path;
nn_.GetListing(path, callback, last);
}
}
void FileSystemImpl::GetListing(
const std::string &path,
- const std::function<bool(const Status &, std::shared_ptr<std::vector<StatInfo>>&, bool)> &handler) {
+ const std::function<bool(const Status &, const std::vector<StatInfo> &, bool)> &handler) {
LOG_INFO(kFileSystem, << "FileSystemImpl::GetListing("
<< FMT_THIS_ADDR << ", path="
<< path << ") called");
// Caputure the state and push it into the shim
- auto callback = [this, path, handler](const Status &stat, std::shared_ptr<std::vector<StatInfo>> &stat_infos, bool has_more) {
+ auto callback = [this, path, handler](const Status &stat, const std::vector<StatInfo> & stat_infos, bool has_more) {
GetListingShim(stat, stat_infos, has_more, path, handler);
};
nn_.GetListing(path, callback);
}
-Status FileSystemImpl::GetListing(const std::string &path, std::shared_ptr<std::vector<StatInfo>> &stat_infos) {
+Status FileSystemImpl::GetListing(const std::string &path, std::vector<StatInfo> * stat_infos) {
LOG_INFO(kFileSystem, << "FileSystemImpl::[sync]GetListing("
<< FMT_THIS_ADDR << ", path="
<< path << ") called");
- // In this case, we're going to allocate the result on the heap and have the
- // async code populate it.
- auto results = std::make_shared<std::vector<StatInfo>>();
+ if (!stat_infos) {
+ return Status::InvalidArgument("FileSystemImpl::GetListing: argument 'stat_infos' cannot be NULL");
+ }
auto callstate = std::make_shared<std::promise<Status>>();
std::future<Status> future(callstate->get_future());
@@ -640,9 +668,9 @@ Status FileSystemImpl::GetListing(const std::string &path, std::shared_ptr<std::
Keep requesting more until we get the entire listing, and don't set the promise
* until we have the entire listing.
*/
- auto h = [callstate, results](const Status &s, std::shared_ptr<std::vector<StatInfo>> si, bool has_more) -> bool {
- if (si) {
- results->insert(results->end(), si->begin(), si->end());
+ auto h = [callstate, stat_infos](const Status &s, const std::vector<StatInfo> & si, bool has_more) -> bool {
+ if (!si.empty()) {
+ stat_infos->insert(stat_infos->end(), si.begin(), si.end());
}
bool done = !s.ok() || !has_more;
@@ -658,11 +686,6 @@ Status FileSystemImpl::GetListing(const std::string &path, std::shared_ptr<std::
/* block until promise is set */
Status stat = future.get();
- if (!stat.ok()) {
- return stat;
- }
-
- stat_infos = results;
return stat;
}
@@ -677,7 +700,7 @@ void FileSystemImpl::Mkdirs(const std::string & path, uint16_t permissions, bool
return;
}
- Status permStatus = NameNodeOperations::CheckValidPermissionMask(permissions);
+ Status permStatus = FileSystem::CheckValidPermissionMask(permissions);
if (!permStatus.ok()) {
handler(permStatus);
return;
@@ -790,7 +813,7 @@ void FileSystemImpl::SetPermission(const std::string & path,
handler(Status::InvalidArgument("SetPermission: argument 'path' cannot be empty"));
return;
}
- Status permStatus = NameNodeOperations::CheckValidPermissionMask(permissions);
+ Status permStatus = FileSystem::CheckValidPermissionMask(permissions);
if (!permStatus.ok()) {
handler(permStatus);
return;
@@ -832,8 +855,8 @@ void FileSystemImpl::SetOwner(const std::string & path, const std::string & user
nn_.SetOwner(path, username, groupname, handler);
}
-Status FileSystemImpl::SetOwner(const std::string & path,
- const std::string & username, const std::string & groupname) {
+Status FileSystemImpl::SetOwner(const std::string & path, const std::string & username,
+ const std::string & groupname) {
LOG_DEBUG(kFileSystem,
<< "FileSystemImpl::[sync]SetOwner(" << FMT_THIS_ADDR << ", path=" << path << ", username=" << username << ", groupname=" << groupname << ") called");
@@ -849,10 +872,179 @@ Status FileSystemImpl::SetOwner(const std::string & path,
/* block until promise is set */
Status stat = future.get();
-
return stat;
}
+/**
+ * Helper function for recursive Find calls.
+ *
+ * Some compilers don't like recursive lambdas, so we make the lambda call a
+ * method, which in turn creates a lambda calling itself.
+ *
+ * ***High-level explanation***
+ *
+ * Since we are allowing to use wild cards in both path and name, we start by expanding the path first.
+ * Boolean search_path is set to true when we search for the path and false when we search for the name.
+ * When we search for the path we break the given path pattern into sub-directories. Starting from the
+ * first sub-directory we list them one-by-one and recursively continue into directories that matched the
+ * path pattern at the current depth. Directories that are large will be requested to continue sending
+ * the results. We keep track of the current depth within the path pattern in the 'depth' variable.
+ * This continues recursively until the depth reaches the end of the path. Next that we start matching
+ * the name pattern. All directories that we find we recurse now, and all names that match the given name
+ * pattern are being stored in outputs and later sent back to the user.
+ */
+void FileSystemImpl::FindShim(const Status &stat, const std::vector<StatInfo> & stat_infos, bool directory_has_more,
+ std::shared_ptr<FindOperationalState> operational_state, std::shared_ptr<FindSharedState> shared_state) {
+ //We buffer the outputs then send them back at the end
+ std::vector<StatInfo> outputs;
+ //Return on error
+ if(!stat.ok()){
+ std::lock_guard<std::mutex> find_lock(shared_state->lock);
+ //We send true becuase we do not want the user code to exit before all our requests finished
+ shared_state->handler(stat, outputs, true);
+ shared_state->aborted = true;
+ }
+ if(!shared_state->aborted){
+ //User did not abort the operation
+ if (directory_has_more) {
+ //Directory is large and has more results
+ //We launch another async call to get more results
+ shared_state->outstanding_requests++;
+ auto callback = [this, operational_state, shared_state](const Status &stat, const std::vector<StatInfo> & stat_infos, bool has_more) {
+ FindShim(stat, stat_infos, has_more, operational_state, shared_state);
+ };
+ std::string last = stat_infos.back().path;
+ nn_.GetListing(operational_state->path, callback, last);
+ }
+ if(operational_state->search_path && operational_state->depth < shared_state->dirs.size() - 1){
+ //We are searching for the path and did not reach the end of the path yet
+ for (StatInfo const& si : stat_infos) {
+ //If we are at the last depth and it matches both path and name, we need to output it.
+ if (operational_state->depth == shared_state->dirs.size() - 2
+ && !fnmatch(shared_state->dirs[operational_state->depth + 1].c_str(), si.path.c_str(), 0)
+ && !fnmatch(shared_state->name.c_str(), si.path.c_str(), 0)) {
+ outputs.push_back(si);
+ }
+ //Skip if not directory
+ if(si.file_type != StatInfo::IS_DIR) {
+ continue;
+ }
+ //Checking for a match with the path at the current depth
+ if(!fnmatch(shared_state->dirs[operational_state->depth + 1].c_str(), si.path.c_str(), 0)){
+ //Launch a new requests for every matched directory
+ shared_state->outstanding_requests++;
+ auto callback = [this, si, operational_state, shared_state](const Status &stat, const std::vector<StatInfo> & stat_infos, bool has_more) {
+ std::shared_ptr<FindOperationalState> new_current_state = std::make_shared<FindOperationalState>(si.full_path, operational_state->depth + 1, true); //true because searching for the path
+ FindShim(stat, stat_infos, has_more, new_current_state, shared_state);
+ };
+ nn_.GetListing(si.full_path, callback);
+ }
+ }
+ }
+ else if(shared_state->maxdepth > operational_state->depth - shared_state->dirs.size() + 1){
+ //We are searching for the name now and maxdepth has not been reached
+ for (StatInfo const& si : stat_infos) {
+ //Launch a new request for every directory
+ if(si.file_type == StatInfo::IS_DIR) {
+ shared_state->outstanding_requests++;
+ auto callback = [this, si, operational_state, shared_state](const Status &stat, const std::vector<StatInfo> & stat_infos, bool has_more) {
+ std::shared_ptr<FindOperationalState> new_current_state = std::make_shared<FindOperationalState>(si.full_path, operational_state->depth + 1, false); //false because searching for the name
+ FindShim(stat, stat_infos, has_more, new_current_state, shared_state);
+ };
+ nn_.GetListing(si.full_path, callback);
+ }
+ //All names that match the specified name are saved to outputs
+ if(!fnmatch(shared_state->name.c_str(), si.path.c_str(), 0)){
+ outputs.push_back(si);
+ }
+ }
+ }
+ }
+ //This section needs a lock to make sure we return the final chunk only once
+ //and no results are sent after aborted is set
+ std::lock_guard<std::mutex> find_lock(shared_state->lock);
+ //Decrement the counter once since we are done with this chunk
+ shared_state->outstanding_requests--;
+ if(shared_state->outstanding_requests == 0){
+ //Send the outputs back to the user and notify that this is the final chunk
+ shared_state->handler(stat, outputs, false);
+ } else {
+ //There will be more results and we are not aborting
+ if (outputs.size() > 0 && !shared_state->aborted){
+ //Send the outputs back to the user and notify that there is more
+ bool user_wants_more = shared_state->handler(stat, outputs, true);
+ if(!user_wants_more) {
+ //Abort if user doesn't want more
+ shared_state->aborted = true;
+ }
+ }
+ }
+}
+
+void FileSystemImpl::Find(
+ const std::string &path, const std::string &name, const uint32_t maxdepth,
+ const std::function<bool(const Status &, const std::vector<StatInfo> &, bool)> &handler) {
+ LOG_INFO(kFileSystem, << "FileSystemImpl::Find("
+ << FMT_THIS_ADDR << ", path="
+ << path << ", name="
+ << name << ") called");
+
+ //Populating the operational state, which includes:
+ //current search path, depth within the path, and the indication that we are currently searching for a path (not name yet).
+ std::shared_ptr<FindOperationalState> operational_state = std::make_shared<FindOperationalState>(path, 0, true);
+ //Populating the shared state, which includes:
+ //vector of sub-directories constructed from path, name to search, handler to use for result returning, outstanding_requests counter, and aborted flag.
+ std::shared_ptr<FindSharedState> shared_state = std::make_shared<FindSharedState>(path, name, maxdepth, handler, 1, false);
+ auto callback = [this, operational_state, shared_state](const Status &stat, const std::vector<StatInfo> & stat_infos, bool directory_has_more) {
+ FindShim(stat, stat_infos, directory_has_more, operational_state, shared_state);
+ };
+ nn_.GetListing("/", callback);
+}
+
+Status FileSystemImpl::Find(const std::string &path, const std::string &name, const uint32_t maxdepth, std::vector<StatInfo> * stat_infos) {
+ LOG_INFO(kFileSystem, << "FileSystemImpl::[sync]Find("
+ << FMT_THIS_ADDR << ", path="
+ << path << ", name="
+ << name << ") called");
+
+ if (!stat_infos) {
+ return Status::InvalidArgument("FileSystemImpl::Find: argument 'stat_infos' cannot be NULL");
+ }
+
+ // In this case, we're going to have the async code populate stat_infos.
+
+ std::promise<void> promise = std::promise<void>();
+ std::future<void> future(promise.get_future());
+ Status status = Status::OK();
+
+ /**
+ * Keep requesting more until we get the entire listing. Set the promise
+ * when we have the entire listing to stop.
+ *
+ * Find guarantees that the handler will only be called once at a time,
+ * so we do not need any locking here
+ */
+ auto h = [&status, &promise, stat_infos](const Status &s, const std::vector<StatInfo> & si, bool has_more_results) -> bool {
+ if (!si.empty()) {
+ stat_infos->insert(stat_infos->end(), si.begin(), si.end());
+ }
+ if (!s.ok() && status.ok()){
+ //We make sure we set 'status' only on the first error.
+ status = s;
+ }
+ if (!has_more_results) {
+ promise.set_value();
+ return false;
+ }
+ return true;
+ };
+
+ Find(path, name, maxdepth, h);
+
+ /* block until promise is set */
+ future.get();
+ return status;
+}
void FileSystemImpl::CreateSnapshot(const std::string &path,
const std::string &name,
http://git-wip-us.apache.org/repos/asf/hadoop/blob/a5a8a1f5/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/fs/filesystem.h
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/fs/filesystem.h b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/fs/filesystem.h
index 75a43f3..0e9cedd 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/fs/filesystem.h
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/fs/filesystem.h
@@ -94,9 +94,9 @@ public:
void GetListing(
const std::string &path,
- const std::function<bool(const Status &, std::shared_ptr<std::vector<StatInfo>> &, bool)> &handler) override;
+ const std::function<bool(const Status &, const std::vector<StatInfo> &, bool)> &handler) override;
- Status GetListing(const std::string &path, std::shared_ptr<std::vector<StatInfo>> &stat_infos) override;
+ Status GetListing(const std::string &path, std::vector<StatInfo> * stat_infos) override;
virtual void GetBlockLocations(const std::string & path, uint64_t offset, uint64_t length,
const std::function<void(const Status &, std::shared_ptr<FileBlockLocation> locations)> ) override;
@@ -115,8 +115,8 @@ public:
const std::function<void(const Status &)> &handler) override;
virtual Status Rename(const std::string &oldPath, const std::string &newPath) override;
- virtual void SetPermission(const std::string & path,
- uint16_t permissions, const std::function<void(const Status &)> &handler) override;
+ virtual void SetPermission(const std::string & path, uint16_t permissions,
+ const std::function<void(const Status &)> &handler) override;
virtual Status SetPermission(const std::string & path, uint16_t permissions) override;
virtual void SetOwner(const std::string & path, const std::string & username,
@@ -124,6 +124,11 @@ public:
virtual Status SetOwner(const std::string & path,
const std::string & username, const std::string & groupname) override;
+ void Find(
+ const std::string &path, const std::string &name, const uint32_t maxdepth,
+ const std::function<bool(const Status &, const std::vector<StatInfo> &, bool)> &handler) override;
+ Status Find(const std::string &path, const std::string &name, const uint32_t maxdepth, std::vector<StatInfo> * stat_infos) override;
+
/*****************************************************************************
* FILE SYSTEM SNAPSHOT FUNCTIONS
****************************************************************************/
@@ -204,9 +209,58 @@ private:
**/
std::shared_ptr<LibhdfsEvents> event_handlers_;
- void GetListingShim(const Status &stat, std::shared_ptr<std::vector<StatInfo>> &stat_infos, bool has_more,
- std::string path,
- const std::function<bool(const Status &, std::shared_ptr<std::vector<StatInfo>>&, bool)> &handler);
+ void GetListingShim(const Status &stat, const std::vector<StatInfo> & stat_infos, bool has_more,
+ std::string path, const std::function<bool(const Status &, const std::vector<StatInfo> &, bool)> &handler);
+
+ struct FindSharedState {
+ //Name pattern (can have wild-cards) to find
+ const std::string name;
+ //Maximum depth to recurse after the end of path is reached.
+ //Can be set to 0 for pure path globbing and ignoring name pattern entirely.
+ const uint32_t maxdepth;
+ //Vector of all sub-directories from the path argument (each can have wild-cards)
+ std::vector<std::string> dirs;
+ //Callback from Find
+ const std::function<bool(const Status &, const std::vector<StatInfo> &, bool)> handler;
+ //outstanding_requests is incremented once for every GetListing call.
+ std::atomic<uint64_t> outstanding_requests;
+ //Boolean needed to abort all recursion on error or on user command
+ std::atomic<bool> aborted;
+ //Shared variables will need protection with a lock
+ std::mutex lock;
+ FindSharedState(const std::string path_, const std::string name_, const uint32_t maxdepth_,
+ const std::function<bool(const Status &, const std::vector<StatInfo> &, bool)> handler_,
+ uint64_t outstanding_recuests_, bool aborted_)
+ : name(name_),
+ maxdepth(maxdepth_),
+ handler(handler_),
+ outstanding_requests(outstanding_recuests_),
+ aborted(aborted_),
+ lock() {
+ //Constructing the list of sub-directories
+ std::stringstream ss(path_);
+ if(path_.back() != '/'){
+ ss << "/";
+ }
+ for (std::string token; std::getline(ss, token, '/'); ) {
+ dirs.push_back(token);
+ }
+ }
+ };
+
+ struct FindOperationalState {
+ const std::string path;
+ const uint32_t depth;
+ const bool search_path;
+ FindOperationalState(const std::string path_, const uint32_t depth_, const bool search_path_)
+ : path(path_),
+ depth(depth_),
+ search_path(search_path_) {
+ }
+ };
+
+ void FindShim(const Status &stat, const std::vector<StatInfo> & stat_infos,
+ bool directory_has_more, std::shared_ptr<FindOperationalState> current_state, std::shared_ptr<FindSharedState> shared_state);
};
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/a5a8a1f5/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/fs/namenode_operations.cc
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/fs/namenode_operations.cc b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/fs/namenode_operations.cc
index 27ccb5d..89acac3 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/fs/namenode_operations.cc
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/fs/namenode_operations.cc
@@ -39,30 +39,6 @@ namespace hdfs {
* NAMENODE OPERATIONS
****************************************************************************/
-uint16_t NameNodeOperations::GetDefaultPermissionMask() {
- return 0755;
-}
-
-Status NameNodeOperations::CheckValidPermissionMask(uint16_t permissions) {
- if (permissions > 01777) {
- std::stringstream errormsg;
- errormsg << "CheckValidPermissionMask: argument 'permissions' is " << std::oct
- << std::showbase << permissions << " (should be between 0 and 01777)";
- return Status::InvalidArgument(errormsg.str().c_str());
- }
- return Status::OK();
-}
-
-Status NameNodeOperations::CheckValidReplication(uint16_t replication) {
- if (replication < 1 || replication > 512) {
- std::stringstream errormsg;
- errormsg << "CheckValidReplication: argument 'replication' is "
- << replication << " (should be between 1 and 512)";
- return Status::InvalidArgument(errormsg.str().c_str());
- }
- return Status::OK();
-}
-
void NameNodeOperations::Connect(const std::string &cluster_name,
const std::vector<ResolvedNamenodeInfo> &servers,
std::function<void(const Status &)> &&handler) {
@@ -170,7 +146,7 @@ void NameNodeOperations::SetReplication(const std::string & path, int16_t replic
handler(Status::InvalidArgument("SetReplication: argument 'path' cannot be empty"));
return;
}
- Status replStatus = CheckValidReplication(replication);
+ Status replStatus = FileSystemImpl::CheckValidReplication(replication);
if (!replStatus.ok()) {
handler(replStatus);
return;
@@ -252,7 +228,8 @@ void NameNodeOperations::GetFileInfo(const std::string & path,
// no fs in the protobuf.
if(resp -> has_fs()){
struct StatInfo stat_info;
- stat_info.path=path;
+ stat_info.path = path;
+ stat_info.full_path = path;
HdfsFileStatusProtoToStatInfo(stat_info, resp->fs());
handler(stat, stat_info);
} else {
@@ -290,7 +267,7 @@ void NameNodeOperations::GetFsStats(
void NameNodeOperations::GetListing(
const std::string & path,
- std::function<void(const Status &, std::shared_ptr<std::vector<StatInfo>> &, bool)> handler,
+ std::function<void(const Status &, const std::vector<StatInfo> &, bool)> handler,
const std::string & start_after) {
using ::hadoop::hdfs::GetListingRequestProto;
using ::hadoop::hdfs::GetListingResponseProto;
@@ -300,8 +277,8 @@ void NameNodeOperations::GetListing(
<< "NameNodeOperations::GetListing(" << FMT_THIS_ADDR << ", path=" << path << ") called");
if (path.empty()) {
- std::shared_ptr<std::vector<StatInfo>> stat_infos;
- handler(Status::InvalidArgument("GetListing: argument 'path' cannot be empty"), stat_infos, false);
+ std::vector<StatInfo> empty;
+ handler(Status::InvalidArgument("GetListing: argument 'path' cannot be empty"), empty, false);
return;
}
@@ -312,31 +289,26 @@ void NameNodeOperations::GetListing(
auto resp = std::make_shared<GetListingResponseProto>();
- namenode_.GetListing(
- &req,
- resp,
- [resp, handler, path](const Status &stat) {
- if (stat.ok()) {
- if(resp -> has_dirlist()){
- std::shared_ptr<std::vector<StatInfo>> stat_infos(new std::vector<StatInfo>);
- for (::hadoop::hdfs::HdfsFileStatusProto const& fs : resp->dirlist().partiallisting()) {
- StatInfo si;
- si.path=fs.path();
- HdfsFileStatusProtoToStatInfo(si, fs);
- stat_infos->push_back(si);
- }
- handler(stat, stat_infos, resp->dirlist().remainingentries() > 0);
- } else {
- std::string errormsg = "No such file or directory: " + path;
- Status statNew = Status::PathNotFound(errormsg.c_str());
- std::shared_ptr<std::vector<StatInfo>> stat_infos;
- handler(statNew, stat_infos, false);
- }
- } else {
- std::shared_ptr<std::vector<StatInfo>> stat_infos;
- handler(stat, stat_infos, false);
+ namenode_.GetListing(&req, resp, [resp, handler, path](const Status &stat) {
+ std::vector<StatInfo> stat_infos;
+ if (stat.ok()) {
+ if(resp -> has_dirlist()){
+ for (::hadoop::hdfs::HdfsFileStatusProto const& fs : resp->dirlist().partiallisting()) {
+ StatInfo si;
+ si.path = fs.path();
+ si.full_path = path + fs.path() + "/";
+ HdfsFileStatusProtoToStatInfo(si, fs);
+ stat_infos.push_back(si);
}
- });
+ handler(stat, stat_infos, resp->dirlist().remainingentries() > 0);
+ } else {
+ std::string errormsg = "No such file or directory: " + path;
+ handler(Status::PathNotFound(errormsg.c_str()), stat_infos, false);
+ }
+ } else {
+ handler(stat, stat_infos, false);
+ }
+ });
}
void NameNodeOperations::Mkdirs(const std::string & path, uint16_t permissions, bool createparent,
@@ -355,7 +327,7 @@ void NameNodeOperations::Mkdirs(const std::string & path, uint16_t permissions,
}
MkdirsRequestProto req;
- Status permStatus = CheckValidPermissionMask(permissions);
+ Status permStatus = FileSystemImpl::CheckValidPermissionMask(permissions);
if (!permStatus.ok()) {
handler(permStatus);
return;
@@ -471,7 +443,7 @@ void NameNodeOperations::SetPermission(const std::string & path,
handler(Status::InvalidArgument("SetPermission: argument 'path' cannot be empty"));
return;
}
- Status permStatus = CheckValidPermissionMask(permissions);
+ Status permStatus = FileSystemImpl::CheckValidPermissionMask(permissions);
if (!permStatus.ok()) {
handler(permStatus);
return;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/a5a8a1f5/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/fs/namenode_operations.h
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/fs/namenode_operations.h b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/fs/namenode_operations.h
index 3afa2e9..60efacc 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/fs/namenode_operations.h
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/fs/namenode_operations.h
@@ -48,12 +48,6 @@ public:
engine_(io_service, options, client_name, user_name, protocol_name, protocol_version),
namenode_(& engine_), options_(options) {}
- static uint16_t GetDefaultPermissionMask();
-
- static Status CheckValidPermissionMask(uint16_t permissions);
-
- static Status CheckValidReplication(uint16_t replication);
-
void Connect(const std::string &cluster_name,
const std::vector<ResolvedNamenodeInfo> &servers,
std::function<void(const Status &)> &&handler);
@@ -77,7 +71,7 @@ public:
// start_after="" for initial call
void GetListing(const std::string & path,
- std::function<void(const Status &, std::shared_ptr<std::vector<StatInfo>>&, bool)> handler,
+ std::function<void(const Status &, const std::vector<StatInfo> &, bool)> handler,
const std::string & start_after = "");
void Mkdirs(const std::string & path, uint16_t permissions, bool createparent,
http://git-wip-us.apache.org/repos/asf/hadoop/blob/a5a8a1f5/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/configuration_test.cc
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/configuration_test.cc b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/configuration_test.cc
index 4d49728..9534204 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/configuration_test.cc
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/configuration_test.cc
@@ -33,21 +33,27 @@ TEST(ConfigurationTest, TestDegenerateInputs) {
/* Completely empty stream */
{
std::stringstream stream;
- optional<Configuration> config = ConfigurationLoader().Load<Configuration>("");
+ ConfigurationLoader config_loader;
+ config_loader.ClearSearchPath();
+ optional<Configuration> config = config_loader.Load<Configuration>("");
EXPECT_FALSE(config && "Empty stream");
}
/* No values */
{
std::string data = "<configuration></configuration>";
- optional<Configuration> config = ConfigurationLoader().Load<Configuration>(data);
+ ConfigurationLoader config_loader;
+ config_loader.ClearSearchPath();
+ optional<Configuration> config = config_loader.Load<Configuration>(data);
EXPECT_TRUE(config && "Blank config");
}
/* Extraneous values */
{
std::string data = "<configuration><spam></spam></configuration>";
- optional<Configuration> config = ConfigurationLoader().Load<Configuration>(data);
+ ConfigurationLoader config_loader;
+ config_loader.ClearSearchPath();
+ optional<Configuration> config = config_loader.Load<Configuration>(data);
EXPECT_TRUE(config && "Extraneous values");
}
}
@@ -57,7 +63,9 @@ TEST(ConfigurationTest, TestBasicOperations) {
{
std::stringstream stream;
simpleConfigStream(stream, "key1", "value1");
- optional<Configuration> config = ConfigurationLoader().Load<Configuration>(stream.str());
+ ConfigurationLoader config_loader;
+ config_loader.ClearSearchPath();
+ optional<Configuration> config = config_loader.Load<Configuration>(stream.str());
EXPECT_TRUE(config && "Parse single value");
EXPECT_EQ("value1", config->GetWithDefault("key1", ""));
}
@@ -74,7 +82,9 @@ TEST(ConfigurationTest, TestBasicOperations) {
{
std::stringstream stream;
simpleConfigStream(stream, "key1", "value1");
- optional<Configuration> config = ConfigurationLoader().Load<Configuration>(stream.str());
+ ConfigurationLoader config_loader;
+ config_loader.ClearSearchPath();
+ optional<Configuration> config = config_loader.Load<Configuration>(stream.str());
EXPECT_TRUE(config && "Parse single value");
EXPECT_EQ("value1", config->GetWithDefault("KEY1", ""));
}
@@ -83,7 +93,9 @@ TEST(ConfigurationTest, TestBasicOperations) {
{
std::stringstream stream;
simpleConfigStream(stream, "key1", "value1");
- optional<Configuration> config = ConfigurationLoader().Load<Configuration>(stream.str());
+ ConfigurationLoader config_loader;
+ config_loader.ClearSearchPath();
+ optional<Configuration> config = config_loader.Load<Configuration>(stream.str());
EXPECT_TRUE(config && "Parse single value");
optional<std::string> value = config->Get("key1");
EXPECT_TRUE((bool)value);
@@ -97,7 +109,9 @@ TEST(ConfigurationTest, TestCompactValues) {
std::stringstream stream;
stream << "<configuration><property name=\"key1\" "
"value=\"value1\"/></configuration>";
- optional<Configuration> config = ConfigurationLoader().Load<Configuration>(stream.str());
+ ConfigurationLoader config_loader;
+ config_loader.ClearSearchPath();
+ optional<Configuration> config = config_loader.Load<Configuration>(stream.str());
EXPECT_TRUE(config && "Compact value parse");
EXPECT_EQ("value1", config->GetWithDefault("key1", ""));
}
@@ -108,7 +122,9 @@ TEST(ConfigurationTest, TestMultipleResources) {
{
std::stringstream stream;
simpleConfigStream(stream, "key1", "value1");
- optional<Configuration> config = ConfigurationLoader().Load<Configuration>(stream.str());
+ ConfigurationLoader config_loader;
+ config_loader.ClearSearchPath();
+ optional<Configuration> config = config_loader.Load<Configuration>(stream.str());
EXPECT_TRUE(config && "Parse first stream");
EXPECT_EQ("value1", config->GetWithDefault("key1", ""));
@@ -129,7 +145,9 @@ TEST(ConfigurationTest, TestStringResource) {
simpleConfigStream(stream, "key1", "value1");
std::string str = stream.str();
- optional<Configuration> config = ConfigurationLoader().Load<Configuration>(stream.str());
+ ConfigurationLoader config_loader;
+ config_loader.ClearSearchPath();
+ optional<Configuration> config = config_loader.Load<Configuration>(stream.str());
EXPECT_TRUE(config && "Parse single value");
EXPECT_EQ("value1", config->GetWithDefault("key1", ""));
}
@@ -171,7 +189,9 @@ TEST(ConfigurationTest, TestFinal) {
std::stringstream stream;
stream << "<configuration><property><name>key1</name><value>value1</"
"value><final>false</final></property></configuration>";
- optional<Configuration> config = ConfigurationLoader().Load<Configuration>(stream.str());
+ ConfigurationLoader config_loader;
+ config_loader.ClearSearchPath();
+ optional<Configuration> config = config_loader.Load<Configuration>(stream.str());
EXPECT_TRUE(config && "Parse first stream");
EXPECT_EQ("value1", config->GetWithDefault("key1", ""));
@@ -187,7 +207,9 @@ TEST(ConfigurationTest, TestFinal) {
std::stringstream stream;
stream << "<configuration><property><name>key1</name><value>value1</"
"value><final>true</final></property></configuration>";
- optional<Configuration> config = ConfigurationLoader().Load<Configuration>(stream.str());
+ ConfigurationLoader config_loader;
+ config_loader.ClearSearchPath();
+ optional<Configuration> config = config_loader.Load<Configuration>(stream.str());
EXPECT_TRUE(config && "Parse first stream");
EXPECT_EQ("value1", config->GetWithDefault("key1", ""));
@@ -203,7 +225,9 @@ TEST(ConfigurationTest, TestFinal) {
std::stringstream stream;
stream << "<configuration><property name=\"key1\" value=\"value1\" "
"final=\"false\"/></configuration>";
- optional<Configuration> config = ConfigurationLoader().Load<Configuration>(stream.str());
+ ConfigurationLoader config_loader;
+ config_loader.ClearSearchPath();
+ optional<Configuration> config = config_loader.Load<Configuration>(stream.str());
EXPECT_TRUE(config && "Parse first stream");
EXPECT_EQ("value1", config->GetWithDefault("key1", ""));
@@ -219,7 +243,9 @@ TEST(ConfigurationTest, TestFinal) {
std::stringstream stream;
stream << "<configuration><property name=\"key1\" value=\"value1\" "
"final=\"true\"/></configuration>";
- optional<Configuration> config = ConfigurationLoader().Load<Configuration>(stream.str());
+ ConfigurationLoader config_loader;
+ config_loader.ClearSearchPath();
+ optional<Configuration> config = config_loader.Load<Configuration>(stream.str());
EXPECT_TRUE(config && "Parse first stream");
EXPECT_EQ("value1", config->GetWithDefault("key1", ""));
@@ -235,7 +261,9 @@ TEST(ConfigurationTest, TestFinal) {
std::stringstream stream;
stream << "<configuration><property><name>key1</name><value>value1</"
"value><final>spam</final></property></configuration>";
- optional<Configuration> config = ConfigurationLoader().Load<Configuration>(stream.str());
+ ConfigurationLoader config_loader;
+ config_loader.ClearSearchPath();
+ optional<Configuration> config = config_loader.Load<Configuration>(stream.str());
EXPECT_TRUE(config && "Parse first stream");
EXPECT_EQ("value1", config->GetWithDefault("key1", ""));
@@ -251,7 +279,9 @@ TEST(ConfigurationTest, TestFinal) {
std::stringstream stream;
stream << "<configuration><property><name>key1</name><value>value1</"
"value><final></final></property></configuration>";
- optional<Configuration> config = ConfigurationLoader().Load<Configuration>(stream.str());
+ ConfigurationLoader config_loader;
+ config_loader.ClearSearchPath();
+ optional<Configuration> config = config_loader.Load<Configuration>(stream.str());
EXPECT_TRUE(config && "Parse first stream");
EXPECT_EQ("value1", config->GetWithDefault("key1", ""));
@@ -271,7 +301,9 @@ TEST(ConfigurationTest, TestFileReads)
TempFile tempFile;
writeSimpleConfig(tempFile.filename, "key1", "value1");
- optional<Configuration> config = ConfigurationLoader().LoadFromFile<Configuration>(tempFile.filename);
+ ConfigurationLoader config_loader;
+ config_loader.ClearSearchPath();
+ optional<Configuration> config = config_loader.LoadFromFile<Configuration>(tempFile.filename);
EXPECT_TRUE(config && "Parse first stream");
EXPECT_EQ("value1", config->GetWithDefault("key1", ""));
}
@@ -298,7 +330,9 @@ TEST(ConfigurationTest, TestFileReads)
{
TempDir tempDir;
- optional<Configuration> config = ConfigurationLoader().LoadFromFile<Configuration>(tempDir.path);
+ ConfigurationLoader config_loader;
+ config_loader.ClearSearchPath();
+ optional<Configuration> config = config_loader.LoadFromFile<Configuration>(tempDir.path);
EXPECT_FALSE(config && "Add directory as file resource");
}
@@ -359,7 +393,9 @@ TEST(ConfigurationTest, TestIntConversions) {
{
std::stringstream stream;
simpleConfigStream(stream, "key1", "1");
- optional<Configuration> config = ConfigurationLoader().Load<Configuration>(stream.str());
+ ConfigurationLoader config_loader;
+ config_loader.ClearSearchPath();
+ optional<Configuration> config = config_loader.Load<Configuration>(stream.str());
EXPECT_TRUE(config && "Parse single value");
optional<int64_t> value = config->GetInt("key1");
EXPECT_TRUE((bool)value);
@@ -398,7 +434,9 @@ TEST(ConfigurationTest, TestDoubleConversions) {
{
std::stringstream stream;
simpleConfigStream(stream, "key1", "1");
- optional<Configuration> config = ConfigurationLoader().Load<Configuration>(stream.str());
+ ConfigurationLoader config_loader;
+ config_loader.ClearSearchPath();
+ optional<Configuration> config = config_loader.Load<Configuration>(stream.str());
EXPECT_TRUE(config && "Parse single value");
optional<double> value = config->GetDouble("key1");
EXPECT_TRUE((bool)value);
@@ -441,7 +479,9 @@ TEST(ConfigurationTest, TestBoolConversions) {
{
std::stringstream stream;
simpleConfigStream(stream, "key1", "true");
- optional<Configuration> config = ConfigurationLoader().Load<Configuration>(stream.str());
+ ConfigurationLoader config_loader;
+ config_loader.ClearSearchPath();
+ optional<Configuration> config = config_loader.Load<Configuration>(stream.str());
EXPECT_TRUE(config && "Parse single value");
optional<bool> value = config->GetBool("key1");
EXPECT_TRUE((bool)value);
@@ -488,7 +528,9 @@ TEST(ConfigurationTest, TestUriConversions) {
{
std::stringstream stream;
simpleConfigStream(stream, "key1", "hdfs:///");
- optional<Configuration> config = ConfigurationLoader().Load<Configuration>(stream.str());
+ ConfigurationLoader config_loader;
+ config_loader.ClearSearchPath();
+ optional<Configuration> config = config_loader.Load<Configuration>(stream.str());
EXPECT_TRUE(config && "Parse single value");
optional<URI> value = config->GetUri("key1");
EXPECT_TRUE((bool)value);
http://git-wip-us.apache.org/repos/asf/hadoop/blob/a5a8a1f5/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/configuration_test.h
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/configuration_test.h b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/configuration_test.h
index 669557f..7947ff5 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/configuration_test.h
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/configuration_test.h
@@ -54,7 +54,9 @@ template <typename... Args>
optional<Configuration> simpleConfig(Args... args) {
std::stringstream stream;
simpleConfigStream(stream, args...);
- optional<Configuration> parse = ConfigurationLoader().Load<Configuration>(stream.str());
+ ConfigurationLoader config_loader;
+ config_loader.ClearSearchPath();
+ optional<Configuration> parse = config_loader.Load<Configuration>(stream.str());
EXPECT_TRUE((bool)parse);
return parse;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/a5a8a1f5/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/hdfs_configuration_test.cc
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/hdfs_configuration_test.cc b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/hdfs_configuration_test.cc
index 7e9ca66..360f886 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/hdfs_configuration_test.cc
+++ b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/tests/hdfs_configuration_test.cc
@@ -30,7 +30,9 @@ TEST(HdfsConfigurationTest, TestDefaultOptions)
{
// Completely empty stream
{
- HdfsConfiguration empty_config = ConfigurationLoader().New<HdfsConfiguration>();
+ ConfigurationLoader config_loader;
+ config_loader.ClearSearchPath();
+ HdfsConfiguration empty_config = config_loader.NewConfig<HdfsConfiguration>();
Options options = empty_config.GetOptions();
EXPECT_EQ(Options::kDefaultRpcTimeout, options.rpc_timeout);
}
@@ -49,8 +51,9 @@ TEST(HdfsConfigurationTest, TestSetOptions)
HdfsConfiguration::kIpcClientConnectTimeoutKey, 103,
HdfsConfiguration::kHadoopSecurityAuthenticationKey, HdfsConfiguration::kHadoopSecurityAuthentication_kerberos
);
-
- optional<HdfsConfiguration> config = ConfigurationLoader().Load<HdfsConfiguration>(stream.str());
+ ConfigurationLoader config_loader;
+ config_loader.ClearSearchPath();
+ optional<HdfsConfiguration> config = config_loader.Load<HdfsConfiguration>(stream.str());
EXPECT_TRUE(config && "Read stream");
Options options = config->GetOptions();
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org