// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
// vim: ts=8 sw=2 smarttab

#include "BlueRocksEnv.h"
#include "BlueFS.h"
#include "include/stringify.h"
#include "kv/RocksDBStore.h"

rocksdb::Status err_to_status(int r)
{
  switch (r) {
  case 0:
    return rocksdb::Status::OK();
  case -ENOENT:
    return rocksdb::Status::NotFound(rocksdb::Status::kNone);
  case -EINVAL:
    return rocksdb::Status::InvalidArgument(rocksdb::Status::kNone);
  case -EIO:
    return rocksdb::Status::IOError(rocksdb::Status::kNone);
  default:
    // FIXME :(
    assert(0 == "unrecognized error code");
    return rocksdb::Status::NotSupported(rocksdb::Status::kNone);
  }
}

// A file abstraction for reading sequentially through a file
class BlueRocksSequentialFile : public rocksdb::SequentialFile {
  BlueFS *fs;
  BlueFS::FileReader *h;
 public:
  BlueRocksSequentialFile(BlueFS *fs, BlueFS::FileReader *h) : fs(fs), h(h) {}
  ~BlueRocksSequentialFile() override {
    delete h;
  }

  // Read up to "n" bytes from the file.  "scratch[0..n-1]" may be
  // written by this routine.  Sets "*result" to the data that was
  // read (including if fewer than "n" bytes were successfully read).
  // May set "*result" to point at data in "scratch[0..n-1]", so
  // "scratch[0..n-1]" must be live when "*result" is used.
  // If an error was encountered, returns a non-OK status.
  //
  // REQUIRES: External synchronization
  rocksdb::Status Read(size_t n, rocksdb::Slice* result, char* scratch) override {
    int r = fs->read(h, &h->buf, h->buf.pos, n, NULL, scratch);
    assert(r >= 0);
    *result = rocksdb::Slice(scratch, r);
    return rocksdb::Status::OK();
  }

  // Skip "n" bytes from the file. This is guaranteed to be no
  // slower that reading the same data, but may be faster.
  //
  // If end of file is reached, skipping will stop at the end of the
  // file, and Skip will return OK.
  //
  // REQUIRES: External synchronization
  rocksdb::Status Skip(uint64_t n) override {
    h->buf.skip(n);
    return rocksdb::Status::OK();
  }

  // Remove any kind of caching of data from the offset to offset+length
  // of this file. If the length is 0, then it refers to the end of file.
  // If the system is not caching the file contents, then this is a noop.
  rocksdb::Status InvalidateCache(size_t offset, size_t length) override {
    fs->invalidate_cache(h->file, offset, length);
    return rocksdb::Status::OK();
  }
};

// A file abstraction for randomly reading the contents of a file.
class BlueRocksRandomAccessFile : public rocksdb::RandomAccessFile {
  BlueFS *fs;
  BlueFS::FileReader *h;
 public:
  BlueRocksRandomAccessFile(BlueFS *fs, BlueFS::FileReader *h) : fs(fs), h(h) {}
  ~BlueRocksRandomAccessFile() override {
    delete h;
  }

  // Read up to "n" bytes from the file starting at "offset".
  // "scratch[0..n-1]" may be written by this routine.  Sets "*result"
  // to the data that was read (including if fewer than "n" bytes were
  // successfully read).  May set "*result" to point at data in
  // "scratch[0..n-1]", so "scratch[0..n-1]" must be live when
  // "*result" is used.  If an error was encountered, returns a non-OK
  // status.
  //
  // Safe for concurrent use by multiple threads.
  rocksdb::Status Read(uint64_t offset, size_t n, rocksdb::Slice* result,
		       char* scratch) const override {
    int r = fs->read_random(h, offset, n, scratch);
    assert(r >= 0);
    *result = rocksdb::Slice(scratch, r);
    return rocksdb::Status::OK();
  }

  // Tries to get an unique ID for this file that will be the same each time
  // the file is opened (and will stay the same while the file is open).
  // Furthermore, it tries to make this ID at most "max_size" bytes. If such an
  // ID can be created this function returns the length of the ID and places it
  // in "id"; otherwise, this function returns 0, in which case "id"
  // may not have been modified.
  //
  // This function guarantees, for IDs from a given environment, two unique ids
  // cannot be made equal to eachother by adding arbitrary bytes to one of
  // them. That is, no unique ID is the prefix of another.
  //
  // This function guarantees that the returned ID will not be interpretable as
  // a single varint.
  //
  // Note: these IDs are only valid for the duration of the process.
  size_t GetUniqueId(char* id, size_t max_size) const override {
    return snprintf(id, max_size, "%016llx",
		    (unsigned long long)h->file->fnode.ino);
  };

  //enum AccessPattern { NORMAL, RANDOM, SEQUENTIAL, WILLNEED, DONTNEED };

  void Hint(AccessPattern pattern) override {
    if (pattern == RANDOM)
      h->buf.max_prefetch = 4096;
    else if (pattern == SEQUENTIAL)
      h->buf.max_prefetch = fs->cct->_conf->bluefs_max_prefetch;
  }

  // Remove any kind of caching of data from the offset to offset+length
  // of this file. If the length is 0, then it refers to the end of file.
  // If the system is not caching the file contents, then this is a noop.
  rocksdb::Status InvalidateCache(size_t offset, size_t length) override {
    fs->invalidate_cache(h->file, offset, length);
    return rocksdb::Status::OK();
  }
};


// A file abstraction for sequential writing.  The implementation
// must provide buffering since callers may append small fragments
// at a time to the file.
class BlueRocksWritableFile : public rocksdb::WritableFile {
  BlueFS *fs;
  BlueFS::FileWriter *h;
 public:
  BlueRocksWritableFile(BlueFS *fs, BlueFS::FileWriter *h) : fs(fs), h(h) {}
  ~BlueRocksWritableFile() override {
    fs->close_writer(h);
  }

  // Indicates if the class makes use of unbuffered I/O
  /*bool UseOSBuffer() const {
    return true;
    }*/

  // This is needed when you want to allocate
  // AlignedBuffer for use with file I/O classes
  // Used for unbuffered file I/O when UseOSBuffer() returns false
  /*size_t GetRequiredBufferAlignment() const {
    return c_DefaultPageSize;
    }*/

  rocksdb::Status Append(const rocksdb::Slice& data) override {
    h->append(data.data(), data.size());
    return rocksdb::Status::OK();
  }

  // Positioned write for unbuffered access default forward
  // to simple append as most of the tests are buffered by default
  rocksdb::Status PositionedAppend(
    const rocksdb::Slice& /* data */,
    uint64_t /* offset */) override {
    return rocksdb::Status::NotSupported();
  }

  // Truncate is necessary to trim the file to the correct size
  // before closing. It is not always possible to keep track of the file
  // size due to whole pages writes. The behavior is undefined if called
  // with other writes to follow.
  rocksdb::Status Truncate(uint64_t size) override {
    // we mirror the posix env, which does nothing here; instead, it
    // truncates to the final size on close.  whatever!
    return rocksdb::Status::OK();
    //int r = fs->truncate(h, size);
    //  return err_to_status(r);
  }

  rocksdb::Status Close() override {
    Flush();

    // mimic posix env, here.  shrug.
    size_t block_size;
    size_t last_allocated_block;
    GetPreallocationStatus(&block_size, &last_allocated_block);
    if (last_allocated_block > 0) {
      int r = fs->truncate(h, h->pos);
      if (r < 0)
	return err_to_status(r);
    }

    return rocksdb::Status::OK();
  }

  rocksdb::Status Flush() override {
    fs->flush(h);
    return rocksdb::Status::OK();
  }

  rocksdb::Status Sync() override { // sync data
    fs->fsync(h);
    return rocksdb::Status::OK();
  }

  // true if Sync() and Fsync() are safe to call concurrently with Append()
  // and Flush().
  bool IsSyncThreadSafe() const override {
    return true;
  }

  // Indicates the upper layers if the current WritableFile implementation
  // uses direct IO.
  bool UseDirectIO() const {
    return false;
  }

  /*
   * Get the size of valid data in the file.
   */
  uint64_t GetFileSize() override {
    return h->file->fnode.size + h->buffer.length();;
  }

  // For documentation, refer to RandomAccessFile::GetUniqueId()
  size_t GetUniqueId(char* id, size_t max_size) const override {
    return snprintf(id, max_size, "%016llx",
		    (unsigned long long)h->file->fnode.ino);
  }

  // Remove any kind of caching of data from the offset to offset+length
  // of this file. If the length is 0, then it refers to the end of file.
  // If the system is not caching the file contents, then this is a noop.
  // This call has no effect on dirty pages in the cache.
  rocksdb::Status InvalidateCache(size_t offset, size_t length) override {
    fs->invalidate_cache(h->file, offset, length);
    return rocksdb::Status::OK();
  }

  using rocksdb::WritableFile::RangeSync;
  // Sync a file range with disk.
  // offset is the starting byte of the file range to be synchronized.
  // nbytes specifies the length of the range to be synchronized.
  // This asks the OS to initiate flushing the cached data to disk,
  // without waiting for completion.
  // Default implementation does nothing.
  rocksdb::Status RangeSync(off_t offset, off_t nbytes) {
    // round down to page boundaries
    int partial = offset & 4095;
    offset -= partial;
    nbytes += partial;
    nbytes &= ~4095;
    if (nbytes)
      fs->flush_range(h, offset, nbytes);
    return rocksdb::Status::OK();
  }

 protected:
  using rocksdb::WritableFile::Allocate;
  /*
   * Pre-allocate space for a file.
   */
  rocksdb::Status Allocate(off_t offset, off_t len) {
    int r = fs->preallocate(h->file, offset, len);
    return err_to_status(r);
  }
};


// Directory object represents collection of files and implements
// filesystem operations that can be executed on directories.
class BlueRocksDirectory : public rocksdb::Directory {
  BlueFS *fs;
 public:
  explicit BlueRocksDirectory(BlueFS *f) : fs(f) {}

  // Fsync directory. Can be called concurrently from multiple threads.
  rocksdb::Status Fsync() override {
    // it is sufficient to flush the log.
    fs->sync_metadata();
    return rocksdb::Status::OK();
  }
};

// Identifies a locked file.
class BlueRocksFileLock : public rocksdb::FileLock {
 public:
  BlueFS *fs;
  BlueFS::FileLock *lock;
  BlueRocksFileLock(BlueFS *fs, BlueFS::FileLock *l) : fs(fs), lock(l) { }
  ~BlueRocksFileLock() override {
  }
};


// --------------------
// --- BlueRocksEnv ---
// --------------------

BlueRocksEnv::BlueRocksEnv(BlueFS *f)
  : EnvWrapper(Env::Default()),  // forward most of it to POSIX
    fs(f)
{

}

rocksdb::Status BlueRocksEnv::NewSequentialFile(
  const std::string& fname,
  std::unique_ptr<rocksdb::SequentialFile>* result,
  const rocksdb::EnvOptions& options)
{
  if (fname[0] == '/')
    return target()->NewSequentialFile(fname, result, options);
  std::string dir, file;
  split(fname, &dir, &file);
  BlueFS::FileReader *h;
  int r = fs->open_for_read(dir, file, &h, false);
  if (r < 0)
    return err_to_status(r);
  result->reset(new BlueRocksSequentialFile(fs, h));
  return rocksdb::Status::OK();
}

rocksdb::Status BlueRocksEnv::NewRandomAccessFile(
  const std::string& fname,
  std::unique_ptr<rocksdb::RandomAccessFile>* result,
  const rocksdb::EnvOptions& options)
{
  std::string dir, file;
  split(fname, &dir, &file);
  BlueFS::FileReader *h;
  int r = fs->open_for_read(dir, file, &h, true);
  if (r < 0)
    return err_to_status(r);
  result->reset(new BlueRocksRandomAccessFile(fs, h));
  return rocksdb::Status::OK();
}

rocksdb::Status BlueRocksEnv::NewWritableFile(
  const std::string& fname,
  std::unique_ptr<rocksdb::WritableFile>* result,
  const rocksdb::EnvOptions& options)
{
  std::string dir, file;
  split(fname, &dir, &file);
  BlueFS::FileWriter *h;
  int r = fs->open_for_write(dir, file, &h, false);
  if (r < 0)
    return err_to_status(r);
  result->reset(new BlueRocksWritableFile(fs, h));
  return rocksdb::Status::OK();
}

rocksdb::Status BlueRocksEnv::ReuseWritableFile(
  const std::string& new_fname,
  const std::string& old_fname,
  std::unique_ptr<rocksdb::WritableFile>* result,
  const rocksdb::EnvOptions& options)
{
  std::string old_dir, old_file;
  split(old_fname, &old_dir, &old_file);
  std::string new_dir, new_file;
  split(new_fname, &new_dir, &new_file);

  int r = fs->rename(old_dir, old_file, new_dir, new_file);
  if (r < 0)
    return err_to_status(r);

  BlueFS::FileWriter *h;
  r = fs->open_for_write(new_dir, new_file, &h, true);
  if (r < 0)
    return err_to_status(r);
  result->reset(new BlueRocksWritableFile(fs, h));
  return rocksdb::Status::OK();
}

rocksdb::Status BlueRocksEnv::NewDirectory(
  const std::string& name,
  std::unique_ptr<rocksdb::Directory>* result)
{
  if (!fs->dir_exists(name))
    return rocksdb::Status::IOError(name, strerror(ENOENT));
  result->reset(new BlueRocksDirectory(fs));
  return rocksdb::Status::OK();
}

rocksdb::Status BlueRocksEnv::FileExists(const std::string& fname)
{
  if (fname[0] == '/')
    return target()->FileExists(fname);
  std::string dir, file;
  split(fname, &dir, &file);
  if (fs->stat(dir, file, NULL, NULL) == 0)
    return rocksdb::Status::OK();
  return err_to_status(-ENOENT);
}

rocksdb::Status BlueRocksEnv::GetChildren(
  const std::string& dir,
  std::vector<std::string>* result)
{
  result->clear();
  int r = fs->readdir(dir, result);
  if (r < 0)
    return rocksdb::Status::IOError(dir, strerror(ENOENT));//    return err_to_status(r);
  return rocksdb::Status::OK();
}

rocksdb::Status BlueRocksEnv::DeleteFile(const std::string& fname)
{
  std::string dir, file;
  split(fname, &dir, &file);
  int r = fs->unlink(dir, file);
  if (r < 0)
    return err_to_status(r);
  return rocksdb::Status::OK();
}

rocksdb::Status BlueRocksEnv::CreateDir(const std::string& dirname)
{
  int r = fs->mkdir(dirname);
  if (r < 0)
    return err_to_status(r);
  return rocksdb::Status::OK();
}

rocksdb::Status BlueRocksEnv::CreateDirIfMissing(const std::string& dirname)
{
  int r = fs->mkdir(dirname);
  if (r < 0 && r != -EEXIST)
    return err_to_status(r);
  return rocksdb::Status::OK();
}

rocksdb::Status BlueRocksEnv::DeleteDir(const std::string& dirname)
{
  int r = fs->rmdir(dirname);
  if (r < 0)
    return err_to_status(r);
  return rocksdb::Status::OK();
}

rocksdb::Status BlueRocksEnv::GetFileSize(
  const std::string& fname,
  uint64_t* file_size)
{
  std::string dir, file;
  split(fname, &dir, &file);
  int r = fs->stat(dir, file, file_size, NULL);
  if (r < 0)
    return err_to_status(r);
  return rocksdb::Status::OK();
}

rocksdb::Status BlueRocksEnv::GetFileModificationTime(const std::string& fname,
						      uint64_t* file_mtime)
{
  std::string dir, file;
  split(fname, &dir, &file);
  utime_t mtime;
  int r = fs->stat(dir, file, NULL, &mtime);
  if (r < 0)
    return err_to_status(r);
  *file_mtime = mtime.sec();
  return rocksdb::Status::OK();
}

rocksdb::Status BlueRocksEnv::RenameFile(
  const std::string& src,
  const std::string& target)
{
  std::string old_dir, old_file;
  split(src, &old_dir, &old_file);
  std::string new_dir, new_file;
  split(target, &new_dir, &new_file);

  int r = fs->rename(old_dir, old_file, new_dir, new_file);
  if (r < 0)
    return err_to_status(r);
  return rocksdb::Status::OK();
}

rocksdb::Status BlueRocksEnv::LinkFile(
  const std::string& src,
  const std::string& target)
{
  ceph_abort();
}

rocksdb::Status BlueRocksEnv::LockFile(
  const std::string& fname,
  rocksdb::FileLock** lock)
{
  std::string dir, file;
  split(fname, &dir, &file);
  BlueFS::FileLock *l = NULL;
  int r = fs->lock_file(dir, file, &l);
  if (r < 0)
    return err_to_status(r);
  *lock = new BlueRocksFileLock(fs, l);
  return rocksdb::Status::OK();
}

rocksdb::Status BlueRocksEnv::UnlockFile(rocksdb::FileLock* lock)
{
  BlueRocksFileLock *l = static_cast<BlueRocksFileLock*>(lock);
  int r = fs->unlock_file(l->lock);
  if (r < 0)
    return err_to_status(r);
  delete lock;
  return rocksdb::Status::OK();
}

rocksdb::Status BlueRocksEnv::GetAbsolutePath(
  const std::string& db_path,
  std::string* output_path)
{
  // this is a lie...
  *output_path = "/" + db_path;
  return rocksdb::Status::OK();
}

rocksdb::Status BlueRocksEnv::NewLogger(
  const std::string& fname,
  std::shared_ptr<rocksdb::Logger>* result)
{
  // ignore the filename :)
  result->reset(create_rocksdb_ceph_logger());
  return rocksdb::Status::OK();
}

rocksdb::Status BlueRocksEnv::GetTestDirectory(std::string* path)
{
  static int foo = 0;
  *path = "temp_" + stringify(++foo);
  return rocksdb::Status::OK();
}