+++ /dev/null
-// Copyright (c) 2011-present, Facebook, Inc. All rights reserved.
-// This source code is licensed under both the GPLv2 (found in the
-// COPYING file in the root directory) and Apache 2.0 License
-// (found in the LICENSE.Apache file in the root directory).
-//
-
-#pragma once
-#include <algorithm>
-#include <stdio.h>
-#include <time.h>
-#include <iostream>
-#include "port/sys_time.h"
-#include "rocksdb/env.h"
-#include "rocksdb/status.h"
-
-#ifdef USE_HDFS
-#include <hdfs.h>
-
-namespace ROCKSDB_NAMESPACE {
-
-// Thrown during execution when there is an issue with the supplied
-// arguments.
-class HdfsUsageException : public std::exception { };
-
-// A simple exception that indicates something went wrong that is not
-// recoverable. The intention is for the message to be printed (with
-// nothing else) and the process terminate.
-class HdfsFatalException : public std::exception {
-public:
- explicit HdfsFatalException(const std::string& s) : what_(s) { }
- virtual ~HdfsFatalException() throw() { }
- virtual const char* what() const throw() {
- return what_.c_str();
- }
-private:
- const std::string what_;
-};
-
-//
-// The HDFS environment for rocksdb. This class overrides all the
-// file/dir access methods and delegates the thread-mgmt methods to the
-// default posix environment.
-//
-class HdfsEnv : public Env {
-
- public:
- explicit HdfsEnv(const std::string& fsname) : fsname_(fsname) {
- posixEnv = Env::Default();
- fileSys_ = connectToPath(fsname_);
- }
-
- virtual ~HdfsEnv() {
- fprintf(stderr, "Destroying HdfsEnv::Default()\n");
- hdfsDisconnect(fileSys_);
- }
-
- Status NewSequentialFile(const std::string& fname,
- std::unique_ptr<SequentialFile>* result,
- const EnvOptions& options) override;
-
- Status NewRandomAccessFile(const std::string& fname,
- std::unique_ptr<RandomAccessFile>* result,
- const EnvOptions& options) override;
-
- Status NewWritableFile(const std::string& fname,
- std::unique_ptr<WritableFile>* result,
- const EnvOptions& options) override;
-
- Status NewDirectory(const std::string& name,
- std::unique_ptr<Directory>* result) override;
-
- Status FileExists(const std::string& fname) override;
-
- Status GetChildren(const std::string& path,
- std::vector<std::string>* result) override;
-
- Status DeleteFile(const std::string& fname) override;
-
- Status CreateDir(const std::string& name) override;
-
- Status CreateDirIfMissing(const std::string& name) override;
-
- Status DeleteDir(const std::string& name) override;
-
- Status GetFileSize(const std::string& fname, uint64_t* size) override;
-
- Status GetFileModificationTime(const std::string& fname,
- uint64_t* file_mtime) override;
-
- Status RenameFile(const std::string& src, const std::string& target) override;
-
- Status LinkFile(const std::string& /*src*/,
- const std::string& /*target*/) override {
- return Status::NotSupported(); // not supported
- }
-
- Status LockFile(const std::string& fname, FileLock** lock) override;
-
- Status UnlockFile(FileLock* lock) override;
-
- Status NewLogger(const std::string& fname,
- std::shared_ptr<Logger>* result) override;
-
- Status IsDirectory(const std::string& path, bool* is_dir) override;
-
- void Schedule(void (*function)(void* arg), void* arg, Priority pri = LOW,
- void* tag = nullptr,
- void (*unschedFunction)(void* arg) = 0) override {
- posixEnv->Schedule(function, arg, pri, tag, unschedFunction);
- }
-
- int UnSchedule(void* tag, Priority pri) override {
- return posixEnv->UnSchedule(tag, pri);
- }
-
- void StartThread(void (*function)(void* arg), void* arg) override {
- posixEnv->StartThread(function, arg);
- }
-
- void WaitForJoin() override { posixEnv->WaitForJoin(); }
-
- unsigned int GetThreadPoolQueueLen(Priority pri = LOW) const override {
- return posixEnv->GetThreadPoolQueueLen(pri);
- }
-
- Status GetTestDirectory(std::string* path) override {
- return posixEnv->GetTestDirectory(path);
- }
-
- uint64_t NowMicros() override { return posixEnv->NowMicros(); }
-
- void SleepForMicroseconds(int micros) override {
- posixEnv->SleepForMicroseconds(micros);
- }
-
- Status GetHostName(char* name, uint64_t len) override {
- return posixEnv->GetHostName(name, len);
- }
-
- Status GetCurrentTime(int64_t* unix_time) override {
- return posixEnv->GetCurrentTime(unix_time);
- }
-
- Status GetAbsolutePath(const std::string& db_path,
- std::string* output_path) override {
- return posixEnv->GetAbsolutePath(db_path, output_path);
- }
-
- void SetBackgroundThreads(int number, Priority pri = LOW) override {
- posixEnv->SetBackgroundThreads(number, pri);
- }
-
- int GetBackgroundThreads(Priority pri = LOW) override {
- return posixEnv->GetBackgroundThreads(pri);
- }
-
- void IncBackgroundThreadsIfNeeded(int number, Priority pri) override {
- posixEnv->IncBackgroundThreadsIfNeeded(number, pri);
- }
-
- std::string TimeToString(uint64_t number) override {
- return posixEnv->TimeToString(number);
- }
-
- static uint64_t gettid() { return Env::Default()->GetThreadID(); }
-
- uint64_t GetThreadID() const override { return HdfsEnv::gettid(); }
-
- private:
- std::string fsname_; // string of the form "hdfs://hostname:port/"
- hdfsFS fileSys_; // a single FileSystem object for all files
- Env* posixEnv; // This object is derived from Env, but not from
- // posixEnv. We have posixnv as an encapsulated
- // object here so that we can use posix timers,
- // posix threads, etc.
-
- static const std::string kProto;
- static const std::string pathsep;
-
- /**
- * If the URI is specified of the form hdfs://server:port/path,
- * then connect to the specified cluster
- * else connect to default.
- */
- hdfsFS connectToPath(const std::string& uri) {
- if (uri.empty()) {
- return nullptr;
- }
- if (uri.find(kProto) != 0) {
- // uri doesn't start with hdfs:// -> use default:0, which is special
- // to libhdfs.
- return hdfsConnectNewInstance("default", 0);
- }
- const std::string hostport = uri.substr(kProto.length());
-
- std::vector <std::string> parts;
- split(hostport, ':', parts);
- if (parts.size() != 2) {
- throw HdfsFatalException("Bad uri for hdfs " + uri);
- }
- // parts[0] = hosts, parts[1] = port/xxx/yyy
- std::string host(parts[0]);
- std::string remaining(parts[1]);
-
- int rem = static_cast<int>(remaining.find(pathsep));
- std::string portStr = (rem == 0 ? remaining :
- remaining.substr(0, rem));
-
- tPort port = static_cast<tPort>(atoi(portStr.c_str()));
- if (port == 0) {
- throw HdfsFatalException("Bad host-port for hdfs " + uri);
- }
- hdfsFS fs = hdfsConnectNewInstance(host.c_str(), port);
- return fs;
- }
-
- void split(const std::string &s, char delim,
- std::vector<std::string> &elems) {
- elems.clear();
- size_t prev = 0;
- size_t pos = s.find(delim);
- while (pos != std::string::npos) {
- elems.push_back(s.substr(prev, pos));
- prev = pos + 1;
- pos = s.find(delim, prev);
- }
- elems.push_back(s.substr(prev, s.size()));
- }
-};
-
-} // namespace ROCKSDB_NAMESPACE
-
-#else // USE_HDFS
-
-namespace ROCKSDB_NAMESPACE {
-
-class HdfsEnv : public Env {
-
- public:
- explicit HdfsEnv(const std::string& /*fsname*/) {
- fprintf(stderr, "You have not build rocksdb with HDFS support\n");
- fprintf(stderr, "Please see hdfs/README for details\n");
- abort();
- }
-
- virtual ~HdfsEnv() {
- }
-
- virtual Status NewSequentialFile(const std::string& fname,
- std::unique_ptr<SequentialFile>* result,
- const EnvOptions& options) override;
-
- virtual Status NewRandomAccessFile(
- const std::string& /*fname*/,
- std::unique_ptr<RandomAccessFile>* /*result*/,
- const EnvOptions& /*options*/) override {
- return Status::NotSupported();
- }
-
- virtual Status NewWritableFile(const std::string& /*fname*/,
- std::unique_ptr<WritableFile>* /*result*/,
- const EnvOptions& /*options*/) override {
- return Status::NotSupported();
- }
-
- virtual Status NewDirectory(const std::string& /*name*/,
- std::unique_ptr<Directory>* /*result*/) override {
- return Status::NotSupported();
- }
-
- virtual Status FileExists(const std::string& /*fname*/) override {
- return Status::NotSupported();
- }
-
- virtual Status GetChildren(const std::string& /*path*/,
- std::vector<std::string>* /*result*/) override {
- return Status::NotSupported();
- }
-
- virtual Status DeleteFile(const std::string& /*fname*/) override {
- return Status::NotSupported();
- }
-
- virtual Status CreateDir(const std::string& /*name*/) override {
- return Status::NotSupported();
- }
-
- virtual Status CreateDirIfMissing(const std::string& /*name*/) override {
- return Status::NotSupported();
- }
-
- virtual Status DeleteDir(const std::string& /*name*/) override {
- return Status::NotSupported();
- }
-
- virtual Status GetFileSize(const std::string& /*fname*/,
- uint64_t* /*size*/) override {
- return Status::NotSupported();
- }
-
- virtual Status GetFileModificationTime(const std::string& /*fname*/,
- uint64_t* /*time*/) override {
- return Status::NotSupported();
- }
-
- virtual Status RenameFile(const std::string& /*src*/,
- const std::string& /*target*/) override {
- return Status::NotSupported();
- }
-
- virtual Status LinkFile(const std::string& /*src*/,
- const std::string& /*target*/) override {
- return Status::NotSupported();
- }
-
- virtual Status LockFile(const std::string& /*fname*/,
- FileLock** /*lock*/) override {
- return Status::NotSupported();
- }
-
- virtual Status UnlockFile(FileLock* /*lock*/) override {
- return Status::NotSupported();
- }
-
- virtual Status NewLogger(const std::string& /*fname*/,
- std::shared_ptr<Logger>* /*result*/) override {
- return Status::NotSupported();
- }
-
- Status IsDirectory(const std::string& /*path*/, bool* /*is_dir*/) override {
- return Status::NotSupported();
- }
-
- virtual void Schedule(void (* /*function*/)(void* arg), void* /*arg*/,
- Priority /*pri*/ = LOW, void* /*tag*/ = nullptr,
- void (* /*unschedFunction*/)(void* arg) = 0) override {}
-
- virtual int UnSchedule(void* /*tag*/, Priority /*pri*/) override { return 0; }
-
- virtual void StartThread(void (* /*function*/)(void* arg),
- void* /*arg*/) override {}
-
- virtual void WaitForJoin() override {}
-
- virtual unsigned int GetThreadPoolQueueLen(
- Priority /*pri*/ = LOW) const override {
- return 0;
- }
-
- virtual Status GetTestDirectory(std::string* /*path*/) override {
- return Status::NotSupported();
- }
-
- virtual uint64_t NowMicros() override { return 0; }
-
- virtual void SleepForMicroseconds(int /*micros*/) override {}
-
- virtual Status GetHostName(char* /*name*/, uint64_t /*len*/) override {
- return Status::NotSupported();
- }
-
- virtual Status GetCurrentTime(int64_t* /*unix_time*/) override {
- return Status::NotSupported();
- }
-
- virtual Status GetAbsolutePath(const std::string& /*db_path*/,
- std::string* /*outputpath*/) override {
- return Status::NotSupported();
- }
-
- virtual void SetBackgroundThreads(int /*number*/,
- Priority /*pri*/ = LOW) override {}
- virtual int GetBackgroundThreads(Priority /*pri*/ = LOW) override {
- return 0;
- }
- virtual void IncBackgroundThreadsIfNeeded(int /*number*/,
- Priority /*pri*/) override {}
- virtual std::string TimeToString(uint64_t /*number*/) override { return ""; }
-
- virtual uint64_t GetThreadID() const override {
- return 0;
- }
-};
-} // namespace ROCKSDB_NAMESPACE
-
-#endif // USE_HDFS
projects / ceph.git / blobdiff