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 ae...@apache.org on 2018/03/23 00:57:05 UTC
[45/51] [abbrv] [partial] hadoop git commit: Revert "Merge branch
'trunk' into HDFS-7240"
http://git-wip-us.apache.org/repos/asf/hadoop/blob/7a542fb3/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/namenode_info.cc
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/namenode_info.cc b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/namenode_info.cc
deleted file mode 100644
index d29f1e9..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/namenode_info.cc
+++ /dev/null
@@ -1,178 +0,0 @@
-/**
- * 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.
- */
-
-#include "namenode_info.h"
-
-#include "common/util.h"
-#include "common/logging.h"
-
-#include <sstream>
-#include <utility>
-#include <future>
-#include <memory>
-
-namespace hdfs {
-
-ResolvedNamenodeInfo& ResolvedNamenodeInfo::operator=(const NamenodeInfo &info) {
- nameservice = info.nameservice;
- name = info.name;
- uri = info.uri;
- return *this;
-}
-
-
-
-std::string ResolvedNamenodeInfo::str() const {
- std::stringstream ss;
- ss << "ResolvedNamenodeInfo {nameservice: " << nameservice << ", name: " << name << ", uri: " << uri.str();
- ss << ", host: " << uri.get_host();
-
- if(uri.has_port())
- ss << ", port: " << uri.get_port();
- else
- ss << ", invalid port (uninitialized)";
-
- ss << ", scheme: " << uri.get_scheme();
-
- ss << " [";
- for(unsigned int i=0;i<endpoints.size();i++)
- ss << endpoints[i] << " ";
- ss << "] }";
-
- return ss.str();
-}
-
-
-bool ResolveInPlace(::asio::io_service *ioservice, ResolvedNamenodeInfo &info) {
- // this isn't very memory friendly, but if it needs to be called often there are bigger issues at hand
- info.endpoints.clear();
- std::vector<ResolvedNamenodeInfo> resolved = BulkResolve(ioservice, {info});
- if(resolved.size() != 1)
- return false;
-
- info.endpoints = resolved[0].endpoints;
- if(info.endpoints.size() == 0)
- return false;
- return true;
-}
-
-typedef std::vector<asio::ip::tcp::endpoint> endpoint_vector;
-
-// RAII wrapper
-class ScopedResolver {
- private:
- ::asio::io_service *io_service_;
- std::string host_;
- std::string port_;
- ::asio::ip::tcp::resolver::query query_;
- ::asio::ip::tcp::resolver resolver_;
- endpoint_vector endpoints_;
-
- // Caller blocks on access if resolution isn't finished
- std::shared_ptr<std::promise<Status>> result_status_;
- public:
- ScopedResolver(::asio::io_service *service, const std::string &host, const std::string &port) :
- io_service_(service), host_(host), port_(port), query_(host, port), resolver_(*io_service_)
- {
- if(!io_service_)
- LOG_ERROR(kAsyncRuntime, << "ScopedResolver@" << this << " passed nullptr to io_service");
- }
-
- ~ScopedResolver() {
- resolver_.cancel();
- }
-
- bool BeginAsyncResolve() {
- // result_status_ would only exist if this was previously called. Invalid state.
- if(result_status_) {
- LOG_ERROR(kAsyncRuntime, << "ScopedResolver@" << this << "::BeginAsyncResolve invalid call: may only be called once per instance");
- return false;
- } else if(!io_service_) {
- LOG_ERROR(kAsyncRuntime, << "ScopedResolver@" << this << "::BeginAsyncResolve invalid call: null io_service");
- return false;
- }
-
- // Now set up the promise, set it in async_resolve's callback
- result_status_ = std::make_shared<std::promise<Status>>();
- std::shared_ptr<std::promise<Status>> shared_result = result_status_;
-
- // Callback to pull a copy of endpoints out of resolver and set promise
- auto callback = [this, shared_result](const asio::error_code &ec, ::asio::ip::tcp::resolver::iterator out) {
- if(!ec) {
- std::copy(out, ::asio::ip::tcp::resolver::iterator(), std::back_inserter(endpoints_));
- }
- shared_result->set_value( ToStatus(ec) );
- };
- resolver_.async_resolve(query_, callback);
- return true;
- }
-
- Status Join() {
- if(!result_status_) {
- std::ostringstream errmsg;
- errmsg << "ScopedResolver@" << this << "Join invalid call: promise never set";
- return Status::InvalidArgument(errmsg.str().c_str());
- }
-
- std::future<Status> future_result = result_status_->get_future();
- Status res = future_result.get();
- return res;
- }
-
- endpoint_vector GetEndpoints() {
- // Explicitly return by value to decouple lifecycles.
- return endpoints_;
- }
-};
-
-std::vector<ResolvedNamenodeInfo> BulkResolve(::asio::io_service *ioservice, const std::vector<NamenodeInfo> &nodes) {
- std::vector< std::unique_ptr<ScopedResolver> > resolvers;
- resolvers.reserve(nodes.size());
-
- std::vector<ResolvedNamenodeInfo> resolved_info;
- resolved_info.reserve(nodes.size());
-
- for(unsigned int i=0; i<nodes.size(); i++) {
- std::string host = nodes[i].get_host();
- std::string port = nodes[i].get_port();
-
- resolvers.emplace_back(new ScopedResolver(ioservice, host, port));
- resolvers[i]->BeginAsyncResolve();
- }
-
- // Join all async operations
- for(unsigned int i=0; i < resolvers.size(); i++) {
- Status asyncReturnStatus = resolvers[i]->Join();
-
- ResolvedNamenodeInfo info;
- info = nodes[i];
-
- if(asyncReturnStatus.ok()) {
- // Copy out endpoints if things went well
- info.endpoints = resolvers[i]->GetEndpoints();
- } else {
- LOG_ERROR(kAsyncRuntime, << "Unabled to resolve endpoints for host: " << nodes[i].get_host()
- << " port: " << nodes[i].get_port());
- }
-
- resolved_info.push_back(info);
- }
- return resolved_info;
-}
-
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/7a542fb3/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/namenode_info.h
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/namenode_info.h b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/namenode_info.h
deleted file mode 100644
index fdee8d7..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/namenode_info.h
+++ /dev/null
@@ -1,49 +0,0 @@
-/**
- * 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.
- */
-
-#ifndef COMMON_HDFS_NAMENODE_INFO_H_
-#define COMMON_HDFS_NAMENODE_INFO_H_
-
-#include <asio.hpp>
-#include <hdfspp/options.h>
-
-#include <string>
-#include <vector>
-
-namespace hdfs {
-
-// Internal representation of namenode info that keeps track
-// of its endpoints.
-struct ResolvedNamenodeInfo : public NamenodeInfo {
- ResolvedNamenodeInfo& operator=(const NamenodeInfo &info);
- std::string str() const;
-
- std::vector<::asio::ip::tcp::endpoint> endpoints;
-};
-
-// Clear endpoints if set and resolve all of them in parallel.
-// Only successful lookups will be placed in the result set.
-std::vector<ResolvedNamenodeInfo> BulkResolve(::asio::io_service *ioservice, const std::vector<NamenodeInfo> &nodes);
-
-// Clear endpoints, if any, and resolve them again
-// Return true if endpoints were resolved
-bool ResolveInPlace(::asio::io_service *ioservice, ResolvedNamenodeInfo &info);
-
-}
-
-#endif
http://git-wip-us.apache.org/repos/asf/hadoop/blob/7a542fb3/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/new_delete.h
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/new_delete.h b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/new_delete.h
deleted file mode 100644
index 48f97ca..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/new_delete.h
+++ /dev/null
@@ -1,52 +0,0 @@
-/**
- * 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.
- */
-
-#ifndef COMMON_HDFS_NEW_DELETE_H_
-#define COMMON_HDFS_NEW_DELETE_H_
-
-#include <cstring>
-
-struct mem_struct {
- size_t mem_size;
-};
-
-#ifndef NDEBUG
-#define MEMCHECKED_CLASS(clazz) \
-static void* operator new(size_t size) { \
- void* p = ::malloc(size); \
- return p; \
-} \
-static void* operator new[](size_t size) { \
- mem_struct* p = (mem_struct*)::malloc(sizeof(mem_struct) + size); \
- p->mem_size = size; \
- return (void*)++p; \
-} \
-static void operator delete(void* p) { \
- ::memset(p, 0, sizeof(clazz)); \
- ::free(p); \
-} \
-static void operator delete[](void* p) { \
- mem_struct* header = (mem_struct*)p; \
- size_t size = (--header)->mem_size; \
- ::memset(p, 0, size); \
- ::free(header); \
-}
-#else
-#define MEMCHECKED_CLASS(clazz)
-#endif
-#endif
http://git-wip-us.apache.org/repos/asf/hadoop/blob/7a542fb3/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/optional_wrapper.h
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/optional_wrapper.h b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/optional_wrapper.h
deleted file mode 100644
index 2d15dc3..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/optional_wrapper.h
+++ /dev/null
@@ -1,43 +0,0 @@
-/**
- * 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.
- */
-
-#ifndef COMMON_OPTIONAL_WRAPPER_H_
-#define COMMON_OPTIONAL_WRAPPER_H_
-
-#ifdef __clang__
- #pragma clang diagnostic push
- #if __has_warning("-Wweak-vtables")
- #pragma clang diagnostic ignored "-Wweak-vtables"
- #endif
- #if __has_warning("-Wreserved-id-macro")
- #pragma clang diagnostic ignored "-Wreserved-id-macro"
- #endif
- #if __has_warning("-Wextra-semi")
- #pragma clang diagnostic ignored "-Wextra-semi"
- #endif
- #define TR2_OPTIONAL_DISABLE_EMULATION_OF_TYPE_TRAITS //For Clang < 3_4_2
-#endif
-
-#include <optional.hpp>
-
-#ifdef __clang__
- #undef TR2_OPTIONAL_DISABLE_EMULATION_OF_TYPE_TRAITS //For Clang < 3_4_2
- #pragma clang diagnostic pop
-#endif
-
-#endif //COMMON_OPTIONAL_WRAPPER_H_
http://git-wip-us.apache.org/repos/asf/hadoop/blob/7a542fb3/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/options.cc
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/options.cc b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/options.cc
deleted file mode 100644
index 48c83f1..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/options.cc
+++ /dev/null
@@ -1,61 +0,0 @@
-/**
- * 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.
- */
-
-#include "hdfspp/options.h"
-
-namespace hdfs {
-
-// The linker needs a place to put all of those constants
-const int Options::kDefaultRpcTimeout;
-const int Options::kNoRetry;
-const int Options::kDefaultMaxRpcRetries;
-const int Options::kDefaultRpcRetryDelayMs;
-const unsigned int Options::kDefaultHostExclusionDuration;
-const unsigned int Options::kDefaultFailoverMaxRetries;
-const unsigned int Options::kDefaultFailoverConnectionMaxRetries;
-const long Options::kDefaultBlockSize;
-
-Options::Options() : rpc_timeout(kDefaultRpcTimeout),
- rpc_connect_timeout(kDefaultRpcConnectTimeout),
- max_rpc_retries(kDefaultMaxRpcRetries),
- rpc_retry_delay_ms(kDefaultRpcRetryDelayMs),
- host_exclusion_duration(kDefaultHostExclusionDuration),
- defaultFS(),
- failover_max_retries(kDefaultFailoverMaxRetries),
- failover_connection_max_retries(kDefaultFailoverConnectionMaxRetries),
- authentication(kDefaultAuthentication),
- block_size(kDefaultBlockSize),
- io_threads_(kDefaultIoThreads)
-{
-
-}
-
-std::string NamenodeInfo::get_host() const {
- return uri.get_host();
-}
-
-std::string NamenodeInfo::get_port() const {
- if(uri.has_port()) {
- return std::to_string(uri.get_port());
- }
- return "-1";
-}
-
-
-
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/7a542fb3/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/retry_policy.cc
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/retry_policy.cc b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/retry_policy.cc
deleted file mode 100644
index dca49fb..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/retry_policy.cc
+++ /dev/null
@@ -1,87 +0,0 @@
-/**
- * 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.
- */
-
-#include "common/retry_policy.h"
-#include "common/logging.h"
-
-#include <sstream>
-
-namespace hdfs {
-
-RetryAction FixedDelayRetryPolicy::ShouldRetry(
- const Status &s, uint64_t retries, uint64_t failovers,
- bool isIdempotentOrAtMostOnce) const {
- LOG_TRACE(kRPC, << "FixedDelayRetryPolicy::ShouldRetry(retries=" << retries << ", failovers=" << failovers << ")");
- (void)isIdempotentOrAtMostOnce;
- if (retries + failovers >= max_retries_) {
- return RetryAction::fail(
- "Failovers and retries(" + std::to_string(retries + failovers) +
- ") exceeded maximum retries (" + std::to_string(max_retries_) + "), Status: " +
- s.ToString());
- } else {
- return RetryAction::retry(delay_);
- }
-}
-
-
-RetryAction NoRetryPolicy::ShouldRetry(
- const Status &s, uint64_t retries, uint64_t failovers,
- bool isIdempotentOrAtMostOnce) const {
- LOG_TRACE(kRPC, << "NoRetryPolicy::ShouldRetry(retries=" << retries << ", failovers=" << failovers << ")");
- (void)retries;
- (void)failovers;
- (void)isIdempotentOrAtMostOnce;
- return RetryAction::fail("No retry, Status: " + s.ToString());
-}
-
-
-RetryAction FixedDelayWithFailover::ShouldRetry(const Status &s, uint64_t retries,
- uint64_t failovers,
- bool isIdempotentOrAtMostOnce) const {
- (void)isIdempotentOrAtMostOnce;
- (void)max_failover_conn_retries_;
- LOG_TRACE(kRPC, << "FixedDelayWithFailover::ShouldRetry(retries=" << retries << ", failovers=" << failovers << ")");
-
- if(failovers < max_failover_retries_ && (s.code() == ::asio::error::timed_out || s.get_server_exception_type() == Status::kStandbyException) )
- {
- // Try connecting to another NN in case this one keeps timing out
- // Can add the backoff wait specified by dfs.client.failover.sleep.base.millis here
- if(failovers == 0) {
- // No delay on first failover if it looks like the NN was bad.
- return RetryAction::failover(0);
- } else {
- return RetryAction::failover(delay_);
- }
- }
-
- if(retries < max_retries_ && failovers < max_failover_retries_) {
- LOG_TRACE(kRPC, << "FixedDelayWithFailover::ShouldRetry: retries < max_retries_ && failovers < max_failover_retries_");
- return RetryAction::retry(delay_);
- } else if (retries >= max_retries_ && failovers < max_failover_retries_) {
- LOG_TRACE(kRPC, << "FixedDelayWithFailover::ShouldRetry: retries >= max_retries_ && failovers < max_failover_retries_");
- return RetryAction::failover(delay_);
- } else if (retries <= max_retries_ && failovers == max_failover_retries_) {
- LOG_TRACE(kRPC, << "FixedDelayWithFailover::ShouldRetry: retries <= max_retries_ && failovers == max_failover_retries_");
- // 1 last retry on new connection
- return RetryAction::retry(delay_);
- }
-
- return RetryAction::fail("Retry and failover didn't work, Status: " + s.ToString());
-}
-
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/7a542fb3/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/retry_policy.h
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/retry_policy.h b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/retry_policy.h
deleted file mode 100644
index 0b5bd80..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/retry_policy.h
+++ /dev/null
@@ -1,160 +0,0 @@
-/**
- * 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.
- */
-#ifndef LIB_COMMON_RETRY_POLICY_H_
-#define LIB_COMMON_RETRY_POLICY_H_
-
-#include "common/util.h"
-
-#include <string>
-#include <stdint.h>
-
-namespace hdfs {
-
-class RetryAction {
- public:
- enum RetryDecision { FAIL, RETRY, FAILOVER_AND_RETRY };
-
- RetryDecision action;
- uint64_t delayMillis;
- std::string reason;
-
- RetryAction(RetryDecision in_action, uint64_t in_delayMillis,
- const std::string &in_reason)
- : action(in_action), delayMillis(in_delayMillis), reason(in_reason) {}
-
- static RetryAction fail(const std::string &reason) {
- return RetryAction(FAIL, 0, reason);
- }
- static RetryAction retry(uint64_t delay) {
- return RetryAction(RETRY, delay, "");
- }
- static RetryAction failover(uint64_t delay) {
- return RetryAction(FAILOVER_AND_RETRY, delay, "");
- }
-
- std::string decision_str() const {
- switch(action) {
- case FAIL: return "FAIL";
- case RETRY: return "RETRY";
- case FAILOVER_AND_RETRY: return "FAILOVER_AND_RETRY";
- default: return "UNDEFINED ACTION";
- }
- };
-};
-
-class RetryPolicy {
- protected:
- uint64_t delay_;
- uint64_t max_retries_;
- RetryPolicy(uint64_t delay, uint64_t max_retries) :
- delay_(delay), max_retries_(max_retries) {}
-
- public:
- RetryPolicy() {};
-
- virtual ~RetryPolicy() {}
- /*
- * If there was an error in communications, responds with the configured
- * action to take.
- */
- virtual RetryAction ShouldRetry(const Status &s, uint64_t retries,
- uint64_t failovers,
- bool isIdempotentOrAtMostOnce) const = 0;
-
- virtual std::string str() const { return "Base RetryPolicy"; }
-};
-
-
-/*
- * Overview of how the failover retry policy works:
- *
- * 1) Acts the same as FixedDelayRetryPolicy in terms of connection retries against a single NN
- * with two differences:
- * a) If we have retried more than the maximum number of retries we will failover to the
- * other node and reset the retry counter rather than error out. It will begin the same
- * routine on the other node.
- * b) If an attempted connection times out and max_failover_conn_retries_ is less than the
- * normal number of retries it will failover sooner. The connection timeout retry limit
- * defaults to zero; the idea being that if a node is unresponsive it's better to just
- * try the secondary rather than incur the timeout cost multiple times.
- *
- * 2) Keeps track of the failover count in the same way that the retry count is tracked. If failover
- * is triggered more than a set number (dfs.client.failover.max.attempts) of times then the operation
- * will error out in the same way that a non-HA operation would error if it ran out of retries.
- *
- * 3) Failover between namenodes isn't instantaneous so the RPC retry delay is reused to add a small
- * delay between failover attempts. This helps prevent the client from quickly using up all of
- * its failover attempts while thrashing between namenodes that are both temporarily marked standby.
- * Note: The java client implements exponential backoff here with a base other than the rpc delay,
- * and this will do the same here in the future. This doesn't do any sort of exponential backoff
- * and the name can be changed to ExponentialDelayWithFailover when backoff is implemented.
- */
-class FixedDelayWithFailover : public RetryPolicy {
- public:
- FixedDelayWithFailover(uint64_t delay, uint64_t max_retries,
- uint64_t max_failover_retries,
- uint64_t max_failover_conn_retries)
- : RetryPolicy(delay, max_retries), max_failover_retries_(max_failover_retries),
- max_failover_conn_retries_(max_failover_conn_retries) {}
-
- RetryAction ShouldRetry(const Status &s, uint64_t retries,
- uint64_t failovers,
- bool isIdempotentOrAtMostOnce) const override;
-
- std::string str() const override { return "FixedDelayWithFailover"; }
-
- private:
- // Attempts to fail over
- uint64_t max_failover_retries_;
- // Attempts to fail over if connection times out rather than
- // tring to connect and wait for the timeout delay failover_retries_
- // times.
- uint64_t max_failover_conn_retries_;
-};
-
-
-/*
- * Returns a fixed delay up to a certain number of retries
- */
-class FixedDelayRetryPolicy : public RetryPolicy {
- public:
- FixedDelayRetryPolicy(uint64_t delay, uint64_t max_retries)
- : RetryPolicy(delay, max_retries) {}
-
- RetryAction ShouldRetry(const Status &s, uint64_t retries,
- uint64_t failovers,
- bool isIdempotentOrAtMostOnce) const override;
-
- std::string str() const override { return "FixedDelayRetryPolicy"; }
-};
-
-/*
- * Never retries
- */
-class NoRetryPolicy : public RetryPolicy {
- public:
- NoRetryPolicy() {};
- RetryAction ShouldRetry(const Status &s, uint64_t retries,
- uint64_t failovers,
- bool isIdempotentOrAtMostOnce) const override;
-
- std::string str() const override { return "NoRetryPolicy"; }
-};
-}
-
-#endif
http://git-wip-us.apache.org/repos/asf/hadoop/blob/7a542fb3/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/sasl_authenticator.h
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/sasl_authenticator.h b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/sasl_authenticator.h
deleted file mode 100644
index 78b2a55..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/sasl_authenticator.h
+++ /dev/null
@@ -1,66 +0,0 @@
-/**
- * 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.
- */
-#ifndef LIB_COMMON_SASL_AUTHENTICATOR_H_
-#define LIB_COMMON_SASL_AUTHENTICATOR_H_
-
-#include "hdfspp/status.h"
-
-namespace hdfs {
-
-class DigestMD5AuthenticatorTest_TestResponse_Test;
-
-/**
- * A specialized implementation of RFC 2831 for the HDFS
- * DataTransferProtocol.
- *
- * The current lacks the following features:
- * * Encoding the username, realm, and password in ISO-8859-1 when
- * it is required by the RFC. They are always encoded in UTF-8.
- * * Checking whether the challenges from the server are
- * well-formed.
- * * Specifying authzid, digest-uri and maximum buffer size.
- * * Supporting QOP other than the auth level.
- **/
-class DigestMD5Authenticator {
-public:
- Status EvaluateResponse(const std::string &payload, std::string *result);
- DigestMD5Authenticator(const std::string &username,
- const std::string &password, bool mock_nonce = false);
-
-private:
- Status GenerateFirstResponse(std::string *result);
- Status GenerateResponseValue(std::string *response_value);
- Status ParseFirstChallenge(const std::string &payload);
-
- static size_t NextToken(const std::string &payload, size_t off,
- std::string *tok);
- void GenerateCNonce();
- std::string username_;
- std::string password_;
- std::string nonce_;
- std::string cnonce_;
- std::string realm_;
- std::string qop_;
- unsigned nonce_count_;
-
- const bool TEST_mock_cnonce_;
- friend class DigestMD5AuthenticatorTest_TestResponse_Test;
-};
-}
-
-#endif
http://git-wip-us.apache.org/repos/asf/hadoop/blob/7a542fb3/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/sasl_digest_md5.cc
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/sasl_digest_md5.cc b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/sasl_digest_md5.cc
deleted file mode 100644
index 3ca8578..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/sasl_digest_md5.cc
+++ /dev/null
@@ -1,240 +0,0 @@
-/**
- * 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.
- */
-#include "sasl_authenticator.h"
-
-#include "common/util.h"
-
-#include <openssl/rand.h>
-#include <openssl/md5.h>
-
-#include <iomanip>
-#include <map>
-#include <sstream>
-
-namespace hdfs {
-
-static std::string QuoteString(const std::string &src);
-static std::string GetMD5Digest(const std::string &src);
-static std::string BinaryToHex(const std::string &src);
-
-static const char kDigestUri[] = "hdfs/0";
-static const size_t kMaxBufferSize = 65536;
-
-DigestMD5Authenticator::DigestMD5Authenticator(const std::string &username,
- const std::string &password,
- bool mock_nonce)
- : username_(username), password_(password), nonce_count_(0),
- TEST_mock_cnonce_(mock_nonce) {}
-
-Status DigestMD5Authenticator::EvaluateResponse(const std::string &payload,
- std::string *result) {
- Status status = ParseFirstChallenge(payload);
- if (status.ok()) {
- status = GenerateFirstResponse(result);
- }
- return status;
-}
-
-size_t DigestMD5Authenticator::NextToken(const std::string &payload, size_t off,
- std::string *tok) {
- tok->clear();
- if (off >= payload.size()) {
- return std::string::npos;
- }
-
- char c = payload[off];
- if (c == '=' || c == ',') {
- *tok = c;
- return off + 1;
- }
-
- int quote_count = 0;
- for (; off < payload.size(); ++off) {
- char c = payload[off];
- if (c == '"') {
- ++quote_count;
- if (quote_count == 2) {
- return off + 1;
- }
- continue;
- }
-
- if (c == '=') {
- if (quote_count) {
- tok->append(&c, 1);
- } else {
- break;
- }
- } else if (('0' <= c && c <= '9') || ('a' <= c && c <= 'z') ||
- ('A' <= c && c <= 'Z') || c == '+' || c == '/' || c == '-' ||
- c == '_' || c == '@') {
- tok->append(&c, 1);
- } else {
- break;
- }
- }
- return off;
-}
-
-void DigestMD5Authenticator::GenerateCNonce() {
- if (!TEST_mock_cnonce_) {
- char buf[8] = {0,};
- RAND_pseudo_bytes(reinterpret_cast<unsigned char *>(buf), sizeof(buf));
- cnonce_ = Base64Encode(std::string(buf, sizeof(buf)));
- }
-}
-
-Status DigestMD5Authenticator::ParseFirstChallenge(const std::string &payload) {
- std::map<std::string, std::string> props;
- std::string token;
- enum {
- kStateLVal,
- kStateEqual,
- kStateRVal,
- kStateCommaOrEnd,
- };
-
- int state = kStateLVal;
-
- std::string lval, rval;
- size_t off = 0;
- while (true) {
- off = NextToken(payload, off, &token);
- if (off == std::string::npos) {
- break;
- }
-
- switch (state) {
- case kStateLVal:
- lval = token;
- state = kStateEqual;
- break;
- case kStateEqual:
- state = kStateRVal;
- break;
- case kStateRVal:
- rval = token;
- props[lval] = rval;
- state = kStateCommaOrEnd;
- break;
- case kStateCommaOrEnd:
- state = kStateLVal;
- break;
- }
- }
-
- if (props["algorithm"] != "md5-sess" || props["charset"] != "utf-8" ||
- props.find("nonce") == props.end()) {
- return Status::Error("Invalid challenge");
- }
- realm_ = props["realm"];
- nonce_ = props["nonce"];
- qop_ = props["qop"];
- return Status::OK();
-}
-
-Status DigestMD5Authenticator::GenerateFirstResponse(std::string *result) {
- // TODO: Support auth-int and auth-conf
- // Handle cipher
- if (qop_ != "auth") {
- return Status::Unimplemented();
- }
-
- std::stringstream ss;
- GenerateCNonce();
- ss << "charset=utf-8,username=\"" << QuoteString(username_) << "\""
- << ",authzid=\"" << QuoteString(username_) << "\""
- << ",nonce=\"" << QuoteString(nonce_) << "\""
- << ",digest-uri=\"" << kDigestUri << "\""
- << ",maxbuf=" << kMaxBufferSize << ",cnonce=\"" << cnonce_ << "\"";
-
- if (realm_.size()) {
- ss << ",realm=\"" << QuoteString(realm_) << "\"";
- }
-
- ss << ",nc=" << std::hex << std::setw(8) << std::setfill('0')
- << ++nonce_count_;
- std::string response_value;
- GenerateResponseValue(&response_value);
- ss << ",response=" << response_value;
- *result = ss.str();
- return result->size() > 4096 ? Status::Error("Response too big")
- : Status::OK();
-}
-
-/**
- * Generate the response value specified in S 2.1.2.1 in RFC2831.
- **/
-Status
-DigestMD5Authenticator::GenerateResponseValue(std::string *response_value) {
- std::stringstream begin_a1, a1_ss;
- std::string a1, a2;
-
- if (qop_ == "auth") {
- a2 = std::string("AUTHENTICATE:") + kDigestUri;
- } else {
- a2 = std::string("AUTHENTICATE:") + kDigestUri +
- ":00000000000000000000000000000000";
- }
-
- begin_a1 << username_ << ":" << realm_ << ":" << password_;
- a1_ss << GetMD5Digest(begin_a1.str()) << ":" << nonce_ << ":" << cnonce_
- << ":" << username_;
-
- std::stringstream combine_ss;
- combine_ss << BinaryToHex(GetMD5Digest(a1_ss.str())) << ":" << nonce_ << ":"
- << std::hex << std::setw(8) << std::setfill('0') << nonce_count_
- << ":" << cnonce_ << ":" << qop_ << ":"
- << BinaryToHex(GetMD5Digest(a2));
- *response_value = BinaryToHex(GetMD5Digest(combine_ss.str()));
- return Status::OK();
-}
-
-static std::string QuoteString(const std::string &src) {
- std::string dst;
- dst.resize(2 * src.size());
- size_t j = 0;
- for (size_t i = 0; i < src.size(); ++i) {
- if (src[i] == '"') {
- dst[j++] = '\\';
- }
- dst[j++] = src[i];
- }
- dst.resize(j);
- return dst;
-}
-
-static std::string GetMD5Digest(const std::string &src) {
- MD5_CTX ctx;
- unsigned long long res[2];
- MD5_Init(&ctx);
- MD5_Update(&ctx, src.c_str(), src.size());
- MD5_Final(reinterpret_cast<unsigned char *>(res), &ctx);
- return std::string(reinterpret_cast<char *>(res), sizeof(res));
-}
-
-static std::string BinaryToHex(const std::string &src) {
- std::stringstream ss;
- ss << std::hex << std::setfill('0');
- for (size_t i = 0; i < src.size(); ++i) {
- unsigned c = (unsigned)(static_cast<unsigned char>(src[i]));
- ss << std::setw(2) << c;
- }
- return ss.str();
-}
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/7a542fb3/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/statinfo.cc
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/statinfo.cc b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/statinfo.cc
deleted file mode 100644
index 2fb744f..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/statinfo.cc
+++ /dev/null
@@ -1,74 +0,0 @@
-/**
- * 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.
- */
-
-#include <hdfspp/statinfo.h>
-#include <sys/stat.h>
-#include <sstream>
-#include <iomanip>
-
-namespace hdfs {
-
-StatInfo::StatInfo()
- : file_type(0),
- length(0),
- permissions(0),
- modification_time(0),
- access_time(0),
- block_replication(0),
- blocksize(0),
- fileid(0),
- children_num(0) {
-}
-
-std::string StatInfo::str() const {
- char perms[11];
- perms[0] = file_type == StatInfo::IS_DIR ? 'd' : '-';
- perms[1] = permissions & S_IRUSR? 'r' : '-';
- perms[2] = permissions & S_IWUSR? 'w': '-';
- perms[3] = permissions & S_IXUSR? 'x': '-';
- perms[4] = permissions & S_IRGRP? 'r' : '-';
- perms[5] = permissions & S_IWGRP? 'w': '-';
- perms[6] = permissions & S_IXGRP? 'x': '-';
- perms[7] = permissions & S_IROTH? 'r' : '-';
- perms[8] = permissions & S_IWOTH? 'w': '-';
- perms[9] = permissions & S_IXOTH? 'x': '-';
- perms[10] = 0;
-
- //Convert to seconds from milliseconds
- const int time_field_length = 17;
- time_t rawtime = modification_time/1000;
- struct tm * timeinfo;
- char buffer[time_field_length];
- timeinfo = localtime(&rawtime);
-
- strftime(buffer,time_field_length,"%Y-%m-%d %H:%M",timeinfo);
- buffer[time_field_length-1] = 0; //null terminator
- std::string time(buffer);
-
- std::stringstream ss;
- ss << std::left << std::setw(12) << perms
- << std::left << std::setw(3) << (!block_replication ? "-" : std::to_string(block_replication))
- << std::left << std::setw(15) << owner
- << std::left << std::setw(15) << group
- << std::right << std::setw(5) << length
- << std::right << std::setw(time_field_length + 2) << time//modification_time
- << " " << full_path;
- return ss.str();
-}
-
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/7a542fb3/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/status.cc
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/status.cc b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/status.cc
deleted file mode 100644
index 4c5c7be..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/status.cc
+++ /dev/null
@@ -1,192 +0,0 @@
-/**
- * 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.
- */
-
-#include "hdfspp/status.h"
-
-#include <cassert>
-#include <sstream>
-#include <cstring>
-#include <map>
-#include <set>
-
-namespace hdfs {
-
-// Server side exceptions that we capture from the RpcResponseHeaderProto
-const char * kStatusAccessControlException = "org.apache.hadoop.security.AccessControlException";
-const char * kPathIsNotDirectoryException = "org.apache.hadoop.fs.PathIsNotDirectoryException";
-const char * kSnapshotException = "org.apache.hadoop.hdfs.protocol.SnapshotException";
-const char * kStatusStandbyException = "org.apache.hadoop.ipc.StandbyException";
-const char * kStatusSaslException = "javax.security.sasl.SaslException";
-const char * kPathNotFoundException = "org.apache.hadoop.fs.InvalidPathException";
-const char * kPathNotFoundException2 = "java.io.FileNotFoundException";
-const char * kFileAlreadyExistsException = "org.apache.hadoop.fs.FileAlreadyExistsException";
-const char * kPathIsNotEmptyDirectoryException = "org.apache.hadoop.fs.PathIsNotEmptyDirectoryException";
-
-
-const static std::map<std::string, int> kKnownServerExceptionClasses = {
- {kStatusAccessControlException, Status::kAccessControlException},
- {kPathIsNotDirectoryException, Status::kNotADirectory},
- {kSnapshotException, Status::kSnapshotProtocolException},
- {kStatusStandbyException, Status::kStandbyException},
- {kStatusSaslException, Status::kAuthenticationFailed},
- {kPathNotFoundException, Status::kPathNotFound},
- {kPathNotFoundException2, Status::kPathNotFound},
- {kFileAlreadyExistsException, Status::kFileAlreadyExists},
- {kPathIsNotEmptyDirectoryException, Status::kPathIsNotEmptyDirectory}
- };
-
-// Errors that retry cannot fix. TODO: complete the list.
-const static std::set<int> noRetryExceptions = {
- Status::kPermissionDenied,
- Status::kAuthenticationFailed,
- Status::kAccessControlException
-};
-
-Status::Status(int code, const char *msg1)
- : code_(code) {
- if(msg1) {
- msg_ = msg1;
- }
-}
-
-Status::Status(int code, const char *exception_class_name, const char *exception_details)
- : code_(code) {
- // If we can assure this never gets nullptr args this can be
- // in the initializer list.
- if(exception_class_name)
- exception_class_ = exception_class_name;
- if(exception_details)
- msg_ = exception_details;
-
- std::map<std::string, int>::const_iterator it = kKnownServerExceptionClasses.find(exception_class_);
- if(it != kKnownServerExceptionClasses.end()) {
- code_ = it->second;
- }
-}
-
-
-Status Status::OK() {
- return Status();
-}
-
-Status Status::InvalidArgument(const char *msg) {
- return Status(kInvalidArgument, msg);
-}
-
-Status Status::PathNotFound(const char *msg){
- return Status(kPathNotFound, msg);
-}
-
-Status Status::ResourceUnavailable(const char *msg) {
- return Status(kResourceUnavailable, msg);
-}
-
-Status Status::PathIsNotDirectory(const char *msg) {
- return Status(kNotADirectory, msg);
-}
-
-Status Status::Unimplemented() {
- return Status(kUnimplemented, "");
-}
-
-Status Status::Exception(const char *exception_class_name, const char *error_message) {
- // Server side exception but can be represented by std::errc codes
- if (exception_class_name && (strcmp(exception_class_name, kStatusAccessControlException) == 0) )
- return Status(kPermissionDenied, error_message);
- else if (exception_class_name && (strcmp(exception_class_name, kStatusSaslException) == 0))
- return AuthenticationFailed();
- else if (exception_class_name && (strcmp(exception_class_name, kPathNotFoundException) == 0))
- return Status(kPathNotFound, error_message);
- else if (exception_class_name && (strcmp(exception_class_name, kPathNotFoundException2) == 0))
- return Status(kPathNotFound, error_message);
- else if (exception_class_name && (strcmp(exception_class_name, kPathIsNotDirectoryException) == 0))
- return Status(kNotADirectory, error_message);
- else if (exception_class_name && (strcmp(exception_class_name, kSnapshotException) == 0))
- return Status(kInvalidArgument, error_message);
- else if (exception_class_name && (strcmp(exception_class_name, kFileAlreadyExistsException) == 0))
- return Status(kFileAlreadyExists, error_message);
- else if (exception_class_name && (strcmp(exception_class_name, kPathIsNotEmptyDirectoryException) == 0))
- return Status(kPathIsNotEmptyDirectory, error_message);
- else
- return Status(kException, exception_class_name, error_message);
-}
-
-Status Status::Error(const char *error_message) {
- return Exception("Exception", error_message);
-}
-
-Status Status::AuthenticationFailed() {
- return Status::AuthenticationFailed(nullptr);
-}
-
-Status Status::AuthenticationFailed(const char *msg) {
- std::string formatted = "AuthenticationFailed";
- if(msg) {
- formatted += ": ";
- formatted += msg;
- }
- return Status(kAuthenticationFailed, formatted.c_str());
-}
-
-Status Status::AuthorizationFailed() {
- return Status::AuthorizationFailed(nullptr);
-}
-
-Status Status::AuthorizationFailed(const char *msg) {
- std::string formatted = "AuthorizationFailed";
- if(msg) {
- formatted += ": ";
- formatted += msg;
- }
- return Status(kPermissionDenied, formatted.c_str());
-}
-
-Status Status::Canceled() {
- return Status(kOperationCanceled, "Operation canceled");
-}
-
-Status Status::InvalidOffset(const char *msg){
- return Status(kInvalidOffset, msg);
-}
-
-std::string Status::ToString() const {
- if (code_ == kOk) {
- return "OK";
- }
- std::stringstream ss;
- if(!exception_class_.empty()) {
- ss << exception_class_ << ":";
- }
- ss << msg_;
- return ss.str();
-}
-
-bool Status::notWorthRetry() const {
- return noRetryExceptions.find(code_) != noRetryExceptions.end();
-}
-
-Status Status::MutexError(const char *msg) {
- std::string formatted = "MutexError";
- if(msg) {
- formatted += ": ";
- formatted += msg;
- }
- return Status(kBusy/*try_lock failure errno*/, msg);
-}
-
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/7a542fb3/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/uri.cc
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/uri.cc b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/uri.cc
deleted file mode 100644
index 9e9319b..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/uri.cc
+++ /dev/null
@@ -1,454 +0,0 @@
-/**
- * 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.
- */
-
-
-#include <hdfspp/uri.h>
-
-#include <uriparser2/uriparser/Uri.h>
-
-#include <string.h>
-#include <sstream>
-#include <cstdlib>
-#include <cassert>
-#include <limits>
-
-namespace hdfs
-{
-
-///////////////////////////////////////////////////////////////////////////////
-//
-// Internal utilities
-//
-///////////////////////////////////////////////////////////////////////////////
-
-const char kReserved[] = ":/?#[]@%+";
-
-std::string URI::encode(const std::string & decoded)
-{
- bool hasCharactersToEncode = false;
- for (auto c : decoded)
- {
- if (isalnum(c) || (strchr(kReserved, c) == NULL))
- {
- continue;
- }
- else
- {
- hasCharactersToEncode = true;
- break;
- }
- }
-
- if (hasCharactersToEncode)
- {
- std::vector<char> buf(decoded.size() * 3 + 1);
- uriEscapeA(decoded.c_str(), &buf[0], true, URI_BR_DONT_TOUCH);
- return std::string(&buf[0]);
- }
- else
- {
- return decoded;
- }
-}
-
-std::string URI::decode(const std::string & encoded)
-{
- bool hasCharactersToDecode = false;
- for (auto c : encoded)
- {
- switch (c)
- {
- case '%':
- case '+':
- hasCharactersToDecode = true;
- break;
- default:
- continue;
- }
- }
-
- if (hasCharactersToDecode)
- {
- std::vector<char> buf(encoded.size() + 1);
- strncpy(&buf[0], encoded.c_str(), buf.size());
- uriUnescapeInPlaceExA(&buf[0], true, URI_BR_DONT_TOUCH);
- return std::string(&buf[0]);
- }
- else
- {
- return encoded;
- }
-}
-
-std::vector<std::string> split(const std::string input, char separator)
-{
- std::vector<std::string> result;
-
- if (!input.empty())
- {
- const char * remaining = input.c_str();
- if (*remaining == '/')
- remaining++;
-
- const char * next_end = strchr(remaining, separator);
- while (next_end) {
- int len = next_end - remaining;
- if (len)
- result.push_back(std::string(remaining, len));
- else
- result.push_back("");
- remaining = next_end + 1;
- next_end = strchr(remaining, separator);
- }
- result.push_back(std::string(remaining));
- }
-
- return result;
-}
-
-
-
-///////////////////////////////////////////////////////////////////////////////
-//
-// Parsing
-//
-///////////////////////////////////////////////////////////////////////////////
-
-
-
-std::string copy_range(const UriTextRangeA *r) {
- const int size = r->afterLast - r->first;
- if (size) {
- return std::string(r->first, size);
- }
- return "";
-}
-
-bool parse_int(const UriTextRangeA *r, int32_t& result)
-{
- std::string int_str = copy_range(r);
- if(!int_str.empty()) {
- errno = 0;
- unsigned long val = ::strtoul(int_str.c_str(), nullptr, 10);
- if(errno == 0 && val < std::numeric_limits<uint16_t>::max()) {
- result = val;
- return true;
- } else {
- return false;
- }
- }
- return true;
-}
-
-
-std::vector<std::string> copy_path(const UriPathSegmentA *ps) {
- std::vector<std::string> result;
- if (nullptr == ps)
- return result;
-
- for (; ps != 0; ps = ps->next) {
- result.push_back(copy_range(&ps->text));
- }
-
- return result;
-}
-
-void parse_user_info(const UriTextRangeA *r, std::string * user, std::string * pass) {
- // Output parameters
- assert(user);
- assert(pass);
-
- std::string user_and_password = copy_range(r);
- if (!user_and_password.empty()) {
- const char * begin = user_and_password.c_str();
- const char * colon_loc = strchr(begin, ':');
- if (colon_loc) {
- *user = std::string(begin, colon_loc - begin - 1);
- *pass = colon_loc + 1;
- } else {
- *user = user_and_password;
- }
- }
-}
-
-
-std::vector<URI::Query> parse_queries(const char *first, const char * afterLast) {
- std::vector<URI::Query> result;
- UriQueryListA * query;
- int count;
- int dissect_result = uriDissectQueryMallocExA(&query, &count, first, afterLast, false, URI_BR_DONT_TOUCH);
- if (URI_SUCCESS == dissect_result) {
- for (auto ps = query; ps != nullptr; ps = ps->next) {
- std::string key = ps->key ? URI::encode(ps->key) : "";
- std::string value = ps->value ? URI::encode(ps->value) : "";
- result.emplace_back(key, value);
- }
- uriFreeQueryListA(query);
- }
-
- return result;
-}
-
-// Parse a string into a URI. Throw a hdfs::uri_parse_error if URI is malformed.
-URI URI::parse_from_string(const std::string &str)
-{
- URI ret;
- bool ok = true;
-
- UriParserStateA state;
- memset(&state, 0, sizeof(state));
- UriUriA uu;
-
- state.uri = &uu;
- int parseResult = uriParseUriA(&state, str.c_str());
- ok &= (parseResult == URI_SUCCESS);
-
- if (ok) {
- ret.scheme = copy_range(&uu.scheme);
- ret.host = copy_range(&uu.hostText);
- ok &= parse_int(&uu.portText, ret._port);
- ret.path = copy_path(uu.pathHead);
- ret.queries = parse_queries(uu.query.first, uu.query.afterLast);
- ret.fragment = copy_range(&uu.fragment);
- parse_user_info(&uu.userInfo, &ret.user, &ret.pass);
- uriFreeUriMembersA(&uu);
- }
- uriFreeUriMembersA(&uu);
-
- if (ok) {
- return ret;
- } else {
- throw uri_parse_error(str);
- }
-}
-
-///////////////////////////////////////////////////////////////////////////////
-//
-// Getters and setters
-//
-///////////////////////////////////////////////////////////////////////////////
-
-URI::URI() : _port(-1) {}
-
-URI::Query::Query(const std::string& k, const std::string& v) : key(k), value(v) {}
-
-std::string URI::str(bool encoded_output) const
-{
- std::stringstream ss;
- if (!scheme.empty()) ss << from_encoded(encoded_output, scheme) << "://";
- if (!user.empty() || !pass.empty()) {
- if (!user.empty()) ss << from_encoded(encoded_output, user);
- if (!pass.empty()) ss << ":" << from_encoded(encoded_output, pass);
- ss << "@";
- }
- if (has_authority()) ss << build_authority(encoded_output);
- if (!path.empty()) ss << get_path(encoded_output);
- if (!queries.empty()) ss << "?" << get_query(encoded_output);
- if (!fragment.empty()) ss << "#" << from_encoded(encoded_output, fragment);
-
- return ss.str();
-}
-
-bool URI::has_authority() const
-{
- return (!host.empty()) || (has_port());
-}
-
-std::string URI::build_authority(bool encoded_output) const
-{
- std::stringstream ss;
- ss << URI::from_encoded(encoded_output, host);
- if (has_port())
- {
- ss << ":" << _port;
- }
- return ss.str();
-}
-
-std::string URI::get_scheme(bool encoded_output) const {
- return from_encoded(encoded_output,scheme);
-}
-
-void URI::set_scheme(const std::string &s, bool encoded_input) {
- scheme = to_encoded(encoded_input,s);
-}
-
-std::string URI::get_host(bool encoded_output) const {
- return from_encoded(encoded_output,host);
-}
-
-void URI::set_host(const std::string& h, bool encoded_input) {
- host = to_encoded(encoded_input,h);
-}
-
-bool URI::has_port() const {
- return _port != -1;
-}
-
-uint16_t URI::get_port() const {
- return (uint16_t)_port;
-}
-
-uint16_t URI::get_port_or_default(uint16_t val) const {
- return has_port() ? (uint16_t)_port : val;
-}
-
-void URI::set_port(uint16_t p)
-{
- _port = (int32_t)p & 0xFFFF;
-}
-
-void URI::clear_port()
-{
- _port = -1;
-}
-
-std::string URI::get_path(bool encoded_output) const
-{
- std::ostringstream out;
- for (const std::string& s: path) {
- out << "/" << from_encoded(encoded_output, s);
- }
- return out.str();
-}
-
-std::vector<std::string> URI::get_path_elements(bool encoded_output) const
-{
- std::vector<std::string> result;
- for (const std::string& path_elem: path) {
- result.push_back(from_encoded(encoded_output, path_elem));
- }
-
- return result;
-}
-
-void URI::parse_path(bool input_encoded, const std::string &input_path)
-{
- std::vector<std::string> split_path = split(input_path, '/');
- for (const std::string& s: split_path) {
- path.push_back(to_encoded(input_encoded, s));
- }
-}
-
-// Mostly copied and modified from uriparser2.c
-
-void URI::set_path(const std::string &p, bool encoded_input) {
- parse_path(encoded_input, p);
-}
-
-void URI::add_path(const std::string &p, bool encoded_input)
-{
- path.push_back(to_encoded(encoded_input, p));
-}
-
-std::string URI::get_query(bool encoded_output) const {
- bool first = true;
- std::stringstream ss;
- for (const Query& q: queries) {
- if (!first) {
- ss << "&";
- }
- ss << from_encoded(encoded_output, q.key) << "=" << from_encoded(encoded_output, q.value);
- first = false;
- }
-
- return ss.str();
-}
-
-std::vector<URI::Query> URI::get_query_elements(bool encoded_output) const
-{
- std::vector<Query> result;
- for (const Query& q: queries) {
- std::string key = from_encoded(encoded_output, q.key);
- std::string value = from_encoded(encoded_output, q.value);
- result.emplace_back(key, value);
- }
-
- return result;
-}
-
-void URI::set_query(const std::string &q) {
- queries = parse_queries(q.c_str(), q.c_str() + q.size() + 1);
-}
-
-
-void URI::add_query(const std::string &name, const std::string & value, bool encoded_input)
-{
- queries.emplace_back(to_encoded(encoded_input, name), to_encoded(encoded_input, value));
-}
-
-void URI::remove_query(const std::string &q_name, bool encoded_input)
-{
- if (queries.empty())
- return;
-
- // This is the one place we need to do decoded comparisons
- std::string decoded_key = encoded_input ? decode(q_name) : q_name;
-
- for (int i = queries.size() - 1; i >= 0; i--) {
- if (decode(queries[i].key) == decoded_key) {
- queries.erase(queries.begin() + i);
- }
- }
-}
-
-std::string URI::get_fragment(bool encoded_output) const {
- return from_encoded(encoded_output, fragment);
-}
-
-void URI::set_fragment(const std::string &f, bool encoded_input) {
- fragment = to_encoded(encoded_input,f);
-}
-
-std::string URI::from_encoded(bool encoded_output, const std::string & input) {
- return encoded_output ? input : decode(input);
-}
-
-std::string URI::to_encoded(bool encoded_input, const std::string & input) {
- return encoded_input ? input : encode(input);
-}
-
-std::string URI::GetDebugString() const {
- std::stringstream ss;
- ss << std::endl;
- ss << "\t" << "uri.str() = \"" << str() << "\"" << std::endl;
- ss << "\t" << "uri.get_scheme() = \"" << get_scheme() << "\"" << std::endl;
- ss << "\t" << "uri.get_host() = \"" << get_host() << "\"" << std::endl;
-
- if(_port == -1)
- ss << "\t" << "uri.get_port() = invalid (uninitialized)" << std::endl;
- else
- ss << "\t" << "uri.get_port() = \"" << _port << "\"" << std::endl;
-
- ss << "\t" << "uri.get_path() = \"" << get_path() << "\"" << std::endl;
- ss << "\t" << "uri.get_fragment() = \"" << get_fragment() << "\"" << std::endl;
-
-
- std::vector<Query> query_elems = get_query_elements();
-
- if(query_elems.size() > 0)
- ss << "\t" << "Query elements:" << std::endl;
-
- for(auto qry = query_elems.begin(); qry != query_elems.end(); qry++) {
- ss << "\t\t" << qry->key << " -> " << qry->value << std::endl;
- }
-
- return ss.str();
-}
-
-} // end namespace hdfs
http://git-wip-us.apache.org/repos/asf/hadoop/blob/7a542fb3/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/util.cc
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/util.cc b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/util.cc
deleted file mode 100644
index 375f951..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/util.cc
+++ /dev/null
@@ -1,170 +0,0 @@
-/**
- * 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.
- */
-
-#include "common/util.h"
-#include "common/util_c.h"
-
-#include <google/protobuf/io/zero_copy_stream_impl_lite.h>
-
-#include <exception>
-#include <sstream>
-#include <iostream>
-#include <iomanip>
-#include <thread>
-
-
-namespace hdfs {
-
-bool ReadDelimitedPBMessage(::google::protobuf::io::CodedInputStream *in,
- ::google::protobuf::MessageLite *msg) {
- uint32_t size = 0;
- in->ReadVarint32(&size);
- auto limit = in->PushLimit(size);
- bool res = msg->ParseFromCodedStream(in);
- in->PopLimit(limit);
-
- return res;
-}
-
-
-std::string SerializeDelimitedProtobufMessage(const ::google::protobuf::MessageLite *msg,
- bool *err) {
- namespace pbio = ::google::protobuf::io;
-
- std::string buf;
-
- int size = msg->ByteSize();
- buf.reserve(pbio::CodedOutputStream::VarintSize32(size) + size);
- pbio::StringOutputStream ss(&buf);
- pbio::CodedOutputStream os(&ss);
- os.WriteVarint32(size);
-
- if(err)
- *err = msg->SerializeToCodedStream(&os);
-
- return buf;
-}
-
-
-std::string GetRandomClientName() {
- std::vector<unsigned char>buf(8);
- RAND_pseudo_bytes(&buf[0], 8);
-
- std::ostringstream oss;
- oss << "DFSClient_" << getpid() << "_" <<
- std::this_thread::get_id() << "_" <<
- std::setw(2) << std::hex << std::uppercase << std::setfill('0');
- for (unsigned char b: buf)
- oss << static_cast<unsigned>(b);
-
- return oss.str();
-}
-
-std::string Base64Encode(const std::string &src) {
- //encoded size is (sizeof(buf) + 2) / 3 * 4
- static const std::string base64_chars =
- "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
- "abcdefghijklmnopqrstuvwxyz"
- "0123456789+/";
- std::string ret;
- int i = 0;
- int j = 0;
- unsigned char char_array_3[3];
- unsigned char char_array_4[4];
- unsigned const char *bytes_to_encode = reinterpret_cast<unsigned const char *>(&src[i]);
- unsigned int in_len = src.size();
-
- while (in_len--) {
- char_array_3[i++] = *(bytes_to_encode++);
- if (i == 3) {
- char_array_4[0] = (char_array_3[0] & 0xfc) >> 2;
- char_array_4[1] = ((char_array_3[0] & 0x03) << 4) + ((char_array_3[1] & 0xf0) >> 4);
- char_array_4[2] = ((char_array_3[1] & 0x0f) << 2) + ((char_array_3[2] & 0xc0) >> 6);
- char_array_4[3] = char_array_3[2] & 0x3f;
-
- for(i = 0; (i <4) ; i++)
- ret += base64_chars[char_array_4[i]];
- i = 0;
- }
- }
-
- if (i) {
- for(j = i; j < 3; j++)
- char_array_3[j] = '\0';
-
- char_array_4[0] = (char_array_3[0] & 0xfc) >> 2;
- char_array_4[1] = ((char_array_3[0] & 0x03) << 4) + ((char_array_3[1] & 0xf0) >> 4);
- char_array_4[2] = ((char_array_3[1] & 0x0f) << 2) + ((char_array_3[2] & 0xc0) >> 6);
- char_array_4[3] = char_array_3[2] & 0x3f;
-
- for (j = 0; (j < i + 1); j++)
- ret += base64_chars[char_array_4[j]];
-
- while((i++ < 3))
- ret += '=';
- }
- return ret;
-}
-
-
-std::string SafeDisconnect(asio::ip::tcp::socket *sock) {
- std::string err;
- if(sock && sock->is_open()) {
- /**
- * Even though we just checked that the socket is open it's possible
- * it isn't in a state where it can properly send or receive. If that's
- * the case asio will turn the underlying error codes from shutdown()
- * and close() into unhelpfully named std::exceptions. Due to the
- * relatively innocuous nature of most of these error codes it's better
- * to just catch and return a flag so the caller can log failure.
- **/
-
- try {
- sock->shutdown(asio::ip::tcp::socket::shutdown_both);
- } catch (const std::exception &e) {
- err = std::string("shutdown() threw") + e.what();
- }
-
- try {
- sock->close();
- } catch (const std::exception &e) {
- // don't append if shutdown() already failed, first failure is the useful one
- if(err.empty())
- err = std::string("close() threw") + e.what();
- }
-
- }
- return err;
-}
-
-bool IsHighBitSet(uint64_t num) {
- uint64_t firstBit = (uint64_t) 1 << 63;
- if (num & firstBit) {
- return true;
- } else {
- return false;
- }
-}
-
-}
-
-void ShutdownProtobufLibrary_C() {
- google::protobuf::ShutdownProtobufLibrary();
-}
-
-
http://git-wip-us.apache.org/repos/asf/hadoop/blob/7a542fb3/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/util.h
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/util.h b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/util.h
deleted file mode 100644
index be902bd..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/util.h
+++ /dev/null
@@ -1,185 +0,0 @@
-/**
- * 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.
- */
-#ifndef LIB_COMMON_UTIL_H_
-#define LIB_COMMON_UTIL_H_
-
-#include "hdfspp/status.h"
-#include "common/logging.h"
-
-#include <sstream>
-#include <mutex>
-#include <string>
-
-#include <asio/error_code.hpp>
-#include <openssl/rand.h>
-
-#include <google/protobuf/message_lite.h>
-#include <google/protobuf/io/coded_stream.h>
-#include <asio.hpp>
-
-namespace hdfs {
-
-// typedefs based on code that's repeated everywhere
-typedef std::lock_guard<std::mutex> mutex_guard;
-
-
-static inline Status ToStatus(const ::asio::error_code &ec) {
- if (ec) {
- return Status(ec.value(), ec.message().c_str());
- } else {
- return Status::OK();
- }
-}
-
-// Determine size of buffer that needs to be allocated in order to serialize msg
-// in delimited format
-static inline int DelimitedPBMessageSize(const ::google::protobuf::MessageLite *msg) {
- size_t size = msg->ByteSize();
- return ::google::protobuf::io::CodedOutputStream::VarintSize32(size) + size;
-}
-
-// Construct msg from the input held in the CodedInputStream
-// return false on failure, otherwise return true
-bool ReadDelimitedPBMessage(::google::protobuf::io::CodedInputStream *in,
- ::google::protobuf::MessageLite *msg);
-
-// Serialize msg into a delimited form (java protobuf compatible)
-// err, if not null, will be set to false on failure
-std::string SerializeDelimitedProtobufMessage(const ::google::protobuf::MessageLite *msg,
- bool *err);
-
-std::string Base64Encode(const std::string &src);
-
-// Return a new high-entropy client name
-std::string GetRandomClientName();
-
-// Returns true if _someone_ is holding the lock (not necessarily this thread,
-// but a std::mutex doesn't track which thread is holding the lock)
-template<class T>
-bool lock_held(T & mutex) {
- bool result = !mutex.try_lock();
- if (!result)
- mutex.unlock();
- return result;
-}
-
-// Shutdown and close a socket safely; will check if the socket is open and
-// catch anything thrown by asio.
-// Returns a string containing error message on failure, otherwise an empty string.
-std::string SafeDisconnect(asio::ip::tcp::socket *sock);
-
-
-
-// The following helper function is used for classes that look like the following:
-//
-// template <typename socket_like_object>
-// class ObjectThatHoldsSocket {
-// socket_like_object sock_;
-// void DoSomethingWithAsioTcpSocket();
-// }
-//
-// The trick here is that ObjectThatHoldsSocket may be templated on a mock socket
-// in mock tests. If you have a method that explicitly needs to call some asio
-// method unrelated to the mock test you need a way of making sure socket_like_object
-// is, in fact, an asio::ip::tcp::socket. Otherwise the mocks need to implement
-// lots of no-op boilerplate. This will return the value of the input param if
-// it's a asio socket, and nullptr if it's anything else.
-
-template <typename sock_t>
-inline asio::ip::tcp::socket *get_asio_socket_ptr(sock_t *s) {
- (void)s;
- return nullptr;
-}
-template<>
-inline asio::ip::tcp::socket *get_asio_socket_ptr<asio::ip::tcp::socket>
- (asio::ip::tcp::socket *s) {
- return s;
-}
-
-//Check if the high bit is set
-bool IsHighBitSet(uint64_t num);
-
-
-// Provide a way to do an atomic swap on a callback.
-// SetCallback, AtomicSwapCallback, and GetCallback can only be called once each.
-// AtomicSwapCallback and GetCallback must only be called after SetCallback.
-//
-// We can't throw on error, and since the callback is templated it's tricky to
-// generate generic dummy callbacks. Complain loudly in the log and get good
-// test coverage. It shouldn't be too hard to avoid invalid states.
-template <typename CallbackType>
-class SwappableCallbackHolder {
- private:
- std::mutex state_lock_;
- CallbackType callback_;
- bool callback_set_ = false;
- bool callback_swapped_ = false;
- bool callback_accessed_ = false;
- public:
- bool IsCallbackSet() {
- mutex_guard swap_lock(state_lock_);
- return callback_set_;
- }
-
- bool IsCallbackAccessed() {
- mutex_guard swap_lock(state_lock_);
- return callback_accessed_;
- }
-
- bool SetCallback(const CallbackType& callback) {
- mutex_guard swap_lock(state_lock_);
- if(callback_set_ || callback_swapped_ || callback_accessed_) {
- LOG_ERROR(kAsyncRuntime, << "SetCallback violates access invariants.")
- return false;
- }
- callback_ = callback;
- callback_set_ = true;
- return true;
- }
-
- CallbackType AtomicSwapCallback(const CallbackType& replacement, bool& swapped) {
- mutex_guard swap_lock(state_lock_);
- if(!callback_set_ || callback_swapped_) {
- LOG_ERROR(kAsyncRuntime, << "AtomicSwapCallback violates access invariants.")
- swapped = false;
- } else if (callback_accessed_) {
- // Common case where callback has been invoked but caller may not know
- LOG_DEBUG(kAsyncRuntime, << "AtomicSwapCallback called after callback has been accessed");
- return callback_;
- }
-
- CallbackType old = callback_;
- callback_ = replacement;
- callback_swapped_ = true;
- swapped = true;
- return old;
- }
- CallbackType GetCallback() {
- mutex_guard swap_lock(state_lock_);
- if(!callback_set_ || callback_accessed_) {
- LOG_ERROR(kAsyncRuntime, << "GetCallback violates access invariants.")
- }
- callback_accessed_ = true;
- return callback_;
- }
-};
-
-
-}
-
-#endif
http://git-wip-us.apache.org/repos/asf/hadoop/blob/7a542fb3/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/util_c.h
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/util_c.h b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/util_c.h
deleted file mode 100644
index c7db7d2..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/common/util_c.h
+++ /dev/null
@@ -1,31 +0,0 @@
-/**
- * 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.
- */
-#ifndef LIB_COMMON_UTIL_C_H_
-#define LIB_COMMON_UTIL_C_H_
-
-#ifdef __cplusplus
-extern "C" {
-#endif
-
- void ShutdownProtobufLibrary_C();
-
-#ifdef __cplusplus
-} /* end extern "C" */
-#endif
-
-#endif
http://git-wip-us.apache.org/repos/asf/hadoop/blob/7a542fb3/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/connection/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/connection/CMakeLists.txt b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/connection/CMakeLists.txt
deleted file mode 100644
index fda51a1..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/connection/CMakeLists.txt
+++ /dev/null
@@ -1,21 +0,0 @@
-#
-# 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.
-#
-
-add_library(connection_obj OBJECT datanodeconnection.cc)
-add_dependencies(connection_obj proto)
-add_library(connection $<TARGET_OBJECTS:connection_obj>)
http://git-wip-us.apache.org/repos/asf/hadoop/blob/7a542fb3/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/connection/datanodeconnection.cc
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/connection/datanodeconnection.cc b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/connection/datanodeconnection.cc
deleted file mode 100644
index 27cd666..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/connection/datanodeconnection.cc
+++ /dev/null
@@ -1,72 +0,0 @@
-/**
- * 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.
- */
-
-#include "datanodeconnection.h"
-#include "common/util.h"
-
-namespace hdfs {
-
-DataNodeConnection::~DataNodeConnection(){}
-DataNodeConnectionImpl::~DataNodeConnectionImpl(){}
-
-DataNodeConnectionImpl::DataNodeConnectionImpl(asio::io_service * io_service,
- const ::hadoop::hdfs::DatanodeInfoProto &dn_proto,
- const hadoop::common::TokenProto *token,
- LibhdfsEvents *event_handlers) : event_handlers_(event_handlers)
-{
- using namespace ::asio::ip;
-
- conn_.reset(new tcp::socket(*io_service));
- auto datanode_addr = dn_proto.id();
- endpoints_[0] = tcp::endpoint(address::from_string(datanode_addr.ipaddr()),
- datanode_addr.xferport());
- uuid_ = dn_proto.id().datanodeuuid();
-
- if (token) {
- token_.reset(new hadoop::common::TokenProto());
- token_->CheckTypeAndMergeFrom(*token);
- }
-}
-
-
-void DataNodeConnectionImpl::Connect(
- std::function<void(Status status, std::shared_ptr<DataNodeConnection> dn)> handler) {
- // Keep the DN from being freed until we're done
- mutex_guard state_lock(state_lock_);
- auto shared_this = shared_from_this();
- asio::async_connect(*conn_, endpoints_.begin(), endpoints_.end(),
- [shared_this, handler](const asio::error_code &ec, std::array<asio::ip::tcp::endpoint, 1>::iterator it) {
- (void)it;
- handler(ToStatus(ec), shared_this); });
-}
-
-void DataNodeConnectionImpl::Cancel() {
- std::string err;
-
- { // scope the lock for disconnect only, log has it's own lock
- mutex_guard state_lock(state_lock_);
- err = SafeDisconnect(conn_.get());
- }
-
- if(!err.empty()) {
- LOG_WARN(kBlockReader, << "Error disconnecting socket in DataNodeConnectionImpl::Cancel, " << err);
- }
-}
-
-
-}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/7a542fb3/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/connection/datanodeconnection.h
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/connection/datanodeconnection.h b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/connection/datanodeconnection.h
deleted file mode 100644
index 21193b3..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/connection/datanodeconnection.h
+++ /dev/null
@@ -1,97 +0,0 @@
-/**
- * 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.
- */
-#ifndef LIBHDFSPP_LIB_CONNECTION_DATANODECONNECTION_H_
-#define LIBHDFSPP_LIB_CONNECTION_DATANODECONNECTION_H_
-
-#include "common/hdfs_ioservice.h"
-#include "common/async_stream.h"
-#include "ClientNamenodeProtocol.pb.h"
-#include "common/libhdfs_events_impl.h"
-#include "common/logging.h"
-#include "common/util.h"
-#include "common/new_delete.h"
-
-#include "asio.hpp"
-
-namespace hdfs {
-
-class DataNodeConnection : public AsyncStream {
-public:
- MEMCHECKED_CLASS(DataNodeConnection)
- std::string uuid_;
- std::unique_ptr<hadoop::common::TokenProto> token_;
-
- virtual ~DataNodeConnection();
- virtual void Connect(std::function<void(Status status, std::shared_ptr<DataNodeConnection> dn)> handler) = 0;
- virtual void Cancel() = 0;
-};
-
-
-struct SocketDeleter {
- inline void operator()(asio::ip::tcp::socket *sock) {
- // Cancel may have already closed the socket.
- std::string err = SafeDisconnect(sock);
- if(!err.empty()) {
- LOG_WARN(kBlockReader, << "Error disconnecting socket: " << err);
- }
- delete sock;
- }
-};
-
-class DataNodeConnectionImpl : public DataNodeConnection, public std::enable_shared_from_this<DataNodeConnectionImpl>{
-private:
- // held (briefly) while posting async ops to the asio task queue
- std::mutex state_lock_;
-public:
- std::unique_ptr<asio::ip::tcp::socket, SocketDeleter> conn_;
- std::array<asio::ip::tcp::endpoint, 1> endpoints_;
- std::string uuid_;
- LibhdfsEvents *event_handlers_;
-
- virtual ~DataNodeConnectionImpl();
- DataNodeConnectionImpl(asio::io_service * io_service, const ::hadoop::hdfs::DatanodeInfoProto &dn_proto,
- const hadoop::common::TokenProto *token,
- LibhdfsEvents *event_handlers);
-
- void Connect(std::function<void(Status status, std::shared_ptr<DataNodeConnection> dn)> handler) override;
-
- void Cancel() override;
-
- void async_read_some(const MutableBuffers &buf,
- std::function<void (const asio::error_code & error, std::size_t bytes_transferred) > handler)
- override {
- event_handlers_->call("DN_read_req", "", "", buf.end() - buf.begin());
-
-
- mutex_guard state_lock(state_lock_);
- conn_->async_read_some(buf, handler);
- };
-
- void async_write_some(const ConstBuffers &buf,
- std::function<void (const asio::error_code & error, std::size_t bytes_transferred) > handler)
- override {
- event_handlers_->call("DN_write_req", "", "", buf.end() - buf.begin());
-
- mutex_guard state_lock(state_lock_);
- conn_->async_write_some(buf, handler);
- }
-};
-
-}
-
-#endif
http://git-wip-us.apache.org/repos/asf/hadoop/blob/7a542fb3/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/fs/CMakeLists.txt
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/fs/CMakeLists.txt b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/fs/CMakeLists.txt
deleted file mode 100644
index 624cda5..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/fs/CMakeLists.txt
+++ /dev/null
@@ -1,21 +0,0 @@
-#
-# 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.
-#
-
-add_library(fs_obj OBJECT filesystem.cc filesystem_sync.cc filehandle.cc bad_datanode_tracker.cc namenode_operations.cc)
-add_dependencies(fs_obj proto)
-add_library(fs $<TARGET_OBJECTS:fs_obj>)
http://git-wip-us.apache.org/repos/asf/hadoop/blob/7a542fb3/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/fs/bad_datanode_tracker.cc
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/fs/bad_datanode_tracker.cc b/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/fs/bad_datanode_tracker.cc
deleted file mode 100644
index be9020f..0000000
--- a/hadoop-hdfs-project/hadoop-hdfs-native-client/src/main/native/libhdfspp/lib/fs/bad_datanode_tracker.cc
+++ /dev/null
@@ -1,71 +0,0 @@
-/**
- * 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.
- **/
-
-#include "bad_datanode_tracker.h"
-
-namespace hdfs {
-
-NodeExclusionRule::~NodeExclusionRule() {}
-
-BadDataNodeTracker::BadDataNodeTracker(const Options& options)
- : timeout_duration_(options.host_exclusion_duration),
- test_clock_shift_(0) {}
-
-BadDataNodeTracker::~BadDataNodeTracker() {}
-
-void BadDataNodeTracker::AddBadNode(const std::string& dn) {
- std::lock_guard<std::mutex> update_lock(datanodes_update_lock_);
- datanodes_[dn] = Clock::now();
-}
-
-bool BadDataNodeTracker::IsBadNode(const std::string& dn) {
- std::lock_guard<std::mutex> update_lock(datanodes_update_lock_);
-
- if (datanodes_.count(dn) == 1) {
- const TimePoint& entered_time = datanodes_[dn];
- if (TimeoutExpired(entered_time)) {
- datanodes_.erase(dn);
- return false;
- }
- /* node in set and still marked bad */
- return true;
- }
- return false;
-}
-
-void BadDataNodeTracker::TEST_set_clock_shift(int t) { test_clock_shift_ = t; }
-
-bool BadDataNodeTracker::TimeoutExpired(const TimePoint& t) {
- TimePoint threshold = Clock::now() -
- std::chrono::milliseconds(timeout_duration_) +
- std::chrono::milliseconds(test_clock_shift_);
- if (t < threshold) {
- return true;
- }
- return false;
-}
-
-ExclusionSet::ExclusionSet(const std::set<std::string>& excluded)
- : excluded_(excluded) {}
-
-ExclusionSet::~ExclusionSet() {}
-
-bool ExclusionSet::IsBadNode(const std::string& node_uuid) {
- return excluded_.count(node_uuid) == 1;
-}
-}
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org