io.cc

// Copyright by Contributors
 
#include <dmlc/base.h>
#include <dmlc/io.h>
#include <dmlc/logging.h>
#include <cstring>
#include "io/uri_spec.h"
#include "io/line_split.h"
#include "io/recordio_split.h"
#include "io/indexed_recordio_split.h"
#include "io/single_file_split.h"
#include "io/local_filesys.h"
#include "io/cached_input_split.h"
#include "io/threaded_input_split.h"
 
#if DMLC_USE_HDFS
#include "io/hdfs_filesys.h"
#endif
 
#if DMLC_USE_S3
#include "io/s3_filesys.h"
#endif
 
#if DMLC_USE_AZURE
#include "io/azure_filesys.h"
#endif
 
namespace dmlc {
namespace io {
FileSystem *FileSystem::GetInstance(const URI &path) {
  if (path.protocol == "file://" || path.protocol.length() == 0) {
    return LocalFileSystem::GetInstance();
  }
  if (path.protocol == "hdfs://" || path.protocol == "viewfs://") {
#if DMLC_USE_HDFS
    if (path.host.length() == 0) {
      return HDFSFileSystem::GetInstance("default");
    } else if (path.protocol == "viewfs://") {
      char* defaultFS = nullptr;
      hdfsConfGetStr("fs.defaultFS", &defaultFS);
      if (path.host.length() != 0) {
        CHECK("viewfs://" + path.host == defaultFS)
            << "viewfs is only supported as a fs.defaultFS.";
      }
      return HDFSFileSystem::GetInstance("default");
    } else {
      return HDFSFileSystem::GetInstance(path.host);
    }
#else
    LOG(FATAL) << "Please compile with DMLC_USE_HDFS=1 to use hdfs";
#endif
  }
  if (path.protocol == "s3://" || path.protocol == "http://" || path.protocol == "https://") {
#if DMLC_USE_S3
    return S3FileSystem::GetInstance();
#else
    LOG(FATAL) << "Please compile with DMLC_USE_S3=1 to use S3";
#endif
  }
 
  if (path.protocol == "azure://") {
#if DMLC_USE_AZURE
    return AzureFileSystem::GetInstance();
#else
    LOG(FATAL) << "Please compile with DMLC_USE_AZURE=1 to use Azure";
#endif
  }
 
  LOG(FATAL) << "unknown filesystem protocol " + path.protocol;
  return NULL;
}
}  // namespace io
 
InputSplit* InputSplit::Create(const char *uri_,
                               unsigned part,
                               unsigned nsplit,
                               const char *type) {
    return Create(uri_, nullptr, part, nsplit, type);
}
 
InputSplit* InputSplit::Create(const char *uri_,
                               const char *index_uri_,
                               unsigned part,
                               unsigned nsplit,
                               const char *type,
                               const bool shuffle,
                               const int seed,
                               const size_t batch_size,
                               const bool recurse_directories) {
  using namespace std;
  using namespace dmlc::io;
  // allow cachefile in format path#cachefile
  io::URISpec spec(uri_, part, nsplit);
  if (!strcmp(spec.uri.c_str(), "stdin")) {
    return new SingleFileSplit(spec.uri.c_str());
  }
  CHECK(part < nsplit) << "invalid input parameter for InputSplit::Create";
  URI path(spec.uri.c_str());
  InputSplitBase *split = NULL;
  if (!strcmp(type, "text")) {
    split =  new LineSplitter(FileSystem::GetInstance(path),
                              spec.uri.c_str(), part, nsplit);
  } else if (!strcmp(type, "indexed_recordio")) {
      if (index_uri_ != nullptr) {
      io::URISpec index_spec(index_uri_, part, nsplit);
    split =  new IndexedRecordIOSplitter(FileSystem::GetInstance(path),
                                  spec.uri.c_str(), index_spec.uri.c_str(), part, nsplit,
                                  batch_size, shuffle, seed);
      } else {
        LOG(FATAL) << "need to pass index file to use IndexedRecordIO";
      }
  } else if (!strcmp(type, "recordio")) {
    split =  new RecordIOSplitter(FileSystem::GetInstance(path),
                                  spec.uri.c_str(), part, nsplit,
                                  recurse_directories);
  } else {
    LOG(FATAL) << "unknown input split type " << type;
  }
#if DMLC_ENABLE_STD_THREAD
  if (spec.cache_file.length() == 0) {
    return new ThreadedInputSplit(split, batch_size);
  } else {
    return new CachedInputSplit(split, spec.cache_file.c_str());
  }
#else
  CHECK(spec.cache_file.length() == 0)
      << "to enable cached file, compile with c++11";
  return split;
#endif
}
 
Stream *Stream::Create(const char *uri,
                       const char * const flag,
                       bool try_create) {
  io::URI path(uri);
  return io::FileSystem::
      GetInstance(path)->Open(path, flag, try_create);
}
 
SeekStream *SeekStream::CreateForRead(const char *uri, bool try_create) {
  io::URI path(uri);
  return io::FileSystem::
      GetInstance(path)->OpenForRead(path, try_create);
}
}  // namespace dmlc