YugabyteDB (2.13.0.0-b42, bfc6a6643e7399ac8a0e81d06a3ee6d6571b33ab)

//  Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.

//  This source code is licensed under the BSD-style license found in the

//  LICENSE file in the root directory of this source tree. An additional grant

//  of patent rights can be found in the PATENTS file in the same directory.

//

// The following only applies to changes made to this file as part of YugaByte development.

//

// Portions Copyright (c) YugaByte, Inc.

//

// Licensed 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.

//

// Copyright (c) 2011 The LevelDB Authors. All rights reserved.

// Use of this source code is governed by a BSD-style license that can be

// found in the LICENSE file. See the AUTHORS file for names of contributors.

#include "yb/rocksdb/db/wal_manager.h"

#ifndef __STDC_FORMAT_MACROS

#define __STDC_FORMAT_MACROS

#endif

#include <inttypes.h>

#include <algorithm>

#include <vector>

#include <memory>

#include "yb/rocksdb/db/filename.h"

#include "yb/rocksdb/db/transaction_log_impl.h"

#include "yb/rocksdb/db/log_reader.h"

#include "yb/rocksdb/db/write_batch_internal.h"

#include "yb/rocksdb/port/port.h"

#include "yb/rocksdb/env.h"

#include "yb/rocksdb/options.h"

#include "yb/rocksdb/write_batch.h"

#include "yb/rocksdb/util/coding.h"

#include "yb/rocksdb/util/file_reader_writer.h"

#include "yb/rocksdb/util/logging.h"

#include "yb/rocksdb/util/mutexlock.h"

#include "yb/rocksdb/util/sync_point.h"

#include "yb/util/status_log.h"

#include "yb/util/string_util.h"

namespace rocksdb {

#ifndef ROCKSDB_LITE

Status WalManager::GetSortedWalFiles(VectorLogPtr* files) {

  // First get sorted files in db dir, then get sorted files from archived

  // dir, to avoid a race condition where a log file is moved to archived

  // dir in between.

  Status s;

  // list wal files in main db dir.

  VectorLogPtr logs;

  s = GetSortedWalsOfType(db_options_.wal_dir, &logs, kAliveLogFile);

  if (!s.ok()) {

    return s;

  }

  // Reproduce the race condition where a log file is moved

  // to archived dir, between these two sync points, used in

  // (DBTest,TransactionLogIteratorRace)

  TEST_SYNC_POINT("WalManager::GetSortedWalFiles:1");

  TEST_SYNC_POINT("WalManager::GetSortedWalFiles:2");

  files->clear();

  // list wal files in archive dir.

  std::string archivedir = ArchivalDirectory(db_options_.wal_dir);

  Status exists = env_->FileExists(archivedir);

  if (exists.ok()) {

    s = GetSortedWalsOfType(archivedir, files, kArchivedLogFile);

    if (!s.ok()) {

      return s;

    }

  } else if (!exists.IsNotFound()) {

    assert(s.IsIOError());

    return s;

  }

  uint64_t latest_archived_log_number = 0;

  if (!files->empty()) {

    latest_archived_log_number = files->back()->LogNumber();

    RLOG(InfoLogLevel::INFO_LEVEL, db_options_.info_log,

        "Latest Archived log: %" PRIu64,

        latest_archived_log_number);

  }

  files->reserve(files->size() + logs.size());

  for (auto& log : logs) {

    if (log->LogNumber() > latest_archived_log_number) {

      files->push_back(std::move(log));

    } else {

      // When the race condition happens, we could see the

      // same log in both db dir and archived dir. Simply

      // ignore the one in db dir. Note that, if we read

      // archived dir first, we would have missed the log file.

      RLOG(InfoLogLevel::WARN_LEVEL, db_options_.info_log,

          "%s already moved to archive", log->PathName().c_str());

    }

  }

  return s;

}

Status WalManager::GetUpdatesSince(

    SequenceNumber seq, std::unique_ptr<TransactionLogIterator>* iter,

    const TransactionLogIterator::ReadOptions& read_options,

    VersionSet* version_set) {

  //  Get all sorted Wal Files.

  //  Do binary search and open files and find the seq number.

  std::unique_ptr<VectorLogPtr> wal_files(new VectorLogPtr);

  Status s = GetSortedWalFiles(wal_files.get());

  if (!s.ok()) {

    return s;

  }

  s = RetainProbableWalFiles(wal_files.get(), seq);

  if (!s.ok()) {

    return s;

  }

  iter->reset(new TransactionLogIteratorImpl(

      db_options_.wal_dir, &db_options_, read_options, env_options_, seq,

      std::move(wal_files), version_set));

  return (*iter)->status();

}

// 1. Go through all archived files and

//    a. if ttl is enabled, delete outdated files

//    b. if archive size limit is enabled, delete empty files,

//        compute file number and size.

// 2. If size limit is enabled:

//    a. compute how many files should be deleted

//    b. get sorted non-empty archived logs

//    c. delete what should be deleted

void WalManager::PurgeObsoleteWALFiles() {

  bool const ttl_enabled = db_options_.WAL_ttl_seconds > 0;

  bool const size_limit_enabled = db_options_.WAL_size_limit_MB > 0;

  if (!ttl_enabled && !size_limit_enabled) {

    return;

  }

  int64_t current_time;

  Status s = env_->GetCurrentTime(&current_time);

  if (!s.ok()) {

    RLOG(InfoLogLevel::ERROR_LEVEL, db_options_.info_log,

        "Can't get current time: %s", s.ToString().c_str());

    assert(false);

    return;

  }

  uint64_t const now_seconds = static_cast<uint64_t>(current_time);

  uint64_t const time_to_check = (ttl_enabled && !size_limit_enabled)

                                     ? db_options_.WAL_ttl_seconds / 2

                                     : kDefaultIntervalToDeleteObsoleteWAL;

  if (purge_wal_files_last_run_ + time_to_check > now_seconds) {

    return;

  }

  purge_wal_files_last_run_ = now_seconds;

  std::string archival_dir = ArchivalDirectory(db_options_.wal_dir);

  std::vector<std::string> files;

  s = env_->GetChildren(archival_dir, &files);

  if (!s.ok()) {

    RLOG(InfoLogLevel::ERROR_LEVEL, db_options_.info_log,

        "Can't get archive files: %s", s.ToString().c_str());

    assert(false);

    return;

  }

  size_t log_files_num = 0;

  uint64_t log_file_size = 0;

  for (auto& f : files) {

    uint64_t number;

    FileType type;

    if (ParseFileName(f, &number, &type) && type == kLogFile) {

      std::string const file_path = archival_dir + "/" + f;

      if (ttl_enabled) {

        uint64_t file_m_time;

        s = env_->GetFileModificationTime(file_path, &file_m_time);

        if (!s.ok()) {

          RLOG(InfoLogLevel::WARN_LEVEL, db_options_.info_log,

              "Can't get file mod time: %s: %s",

              file_path.c_str(), s.ToString().c_str());

          continue;

        }

        if (now_seconds - file_m_time > db_options_.WAL_ttl_seconds) {

          s = env_->DeleteFile(file_path);

          if (!s.ok()) {

            RLOG(InfoLogLevel::WARN_LEVEL, db_options_.info_log,

                "Can't delete file: %s: %s",

                file_path.c_str(), s.ToString().c_str());

            continue;

          } else {

            MutexLock l(&read_first_record_cache_mutex_);

            read_first_record_cache_.erase(number);

          }

          continue;

        }

      }

      if (size_limit_enabled) {

        uint64_t file_size;

        s = env_->GetFileSize(file_path, &file_size);

        if (!s.ok()) {

          RLOG(InfoLogLevel::ERROR_LEVEL, db_options_.info_log,

              "Unable to get file size: %s: %s",

              file_path.c_str(), s.ToString().c_str());

          return;

        } else {

          if (file_size > 0) {

            log_file_size = std::max(log_file_size, file_size);

            ++log_files_num;

          } else {

            s = env_->DeleteFile(file_path);

            if (!s.ok()) {

              RLOG(InfoLogLevel::WARN_LEVEL, db_options_.info_log,

                  "Unable to delete file: %s: %s",

                  file_path.c_str(), s.ToString().c_str());

              continue;

            } else {

              MutexLock l(&read_first_record_cache_mutex_);

              read_first_record_cache_.erase(number);

            }

          }

        }

      }

    }

  }

  if (0 == log_files_num || !size_limit_enabled) {

    return;

  }

  size_t const files_keep_num =

      db_options_.WAL_size_limit_MB * 1024 * 1024 / log_file_size;

  if (log_files_num <= files_keep_num) {

    return;

  }

  size_t files_del_num = log_files_num - files_keep_num;

  VectorLogPtr archived_logs;

  CHECK_OK(GetSortedWalsOfType(archival_dir, &archived_logs, kArchivedLogFile));

  if (files_del_num > archived_logs.size()) {

    RLOG(InfoLogLevel::WARN_LEVEL, db_options_.info_log,

        "Trying to delete more archived log files than "

        "exist. Deleting all");

    files_del_num = archived_logs.size();

  }

  for (size_t i = 0; i < files_del_num; ++i) {

    std::string const file_path = archived_logs[i]->PathName();

    s = env_->DeleteFile(db_options_.wal_dir + "/" + file_path);

    if (!s.ok()) {

      RLOG(InfoLogLevel::WARN_LEVEL, db_options_.info_log,

          "Unable to delete file: %s: %s", file_path.c_str(),

          s.ToString().c_str());

      continue;

    } else {

      MutexLock l(&read_first_record_cache_mutex_);

      read_first_record_cache_.erase(archived_logs[i]->LogNumber());

    }

  }

}

void WalManager::ArchiveWALFile(const std::string& fname, uint64_t number) {

  auto archived_log_name = ArchivedLogFileName(db_options_.wal_dir, number);

  // The sync point below is used in (DBTest,TransactionLogIteratorRace)

  TEST_SYNC_POINT("WalManager::PurgeObsoleteFiles:1");

  Status s = env_->RenameFile(fname, archived_log_name);

  // The sync point below is used in (DBTest,TransactionLogIteratorRace)

  TEST_SYNC_POINT("WalManager::PurgeObsoleteFiles:2");

  RLOG(InfoLogLevel::INFO_LEVEL, db_options_.info_log,

      "Move log file %s to %s -- %s\n", fname.c_str(),

      archived_log_name.c_str(), s.ToString().c_str());

}

namespace {

struct CompareLogByPointer {

  bool operator()(const std::unique_ptr<LogFile>& a,

                  const std::unique_ptr<LogFile>& b) {

    LogFileImpl* a_impl = dynamic_cast<LogFileImpl*>(a.get());

    LogFileImpl* b_impl = dynamic_cast<LogFileImpl*>(b.get());

    return *a_impl < *b_impl;

  }

};

}

Status WalManager::GetSortedWalsOfType(const std::string& path,

                                       VectorLogPtr* log_files,

                                       WalFileType log_type) {

  std::vector<std::string> all_files;

  const Status status = env_->GetChildren(path, &all_files);

  if (!status.ok()) {

    return status;

  }

  log_files->reserve(all_files.size());

  for (const auto& f : all_files) {

    uint64_t number;

    FileType type;

    if (ParseFileName(f, &number, &type) && type == kLogFile) {

      SequenceNumber sequence;

      Status s = ReadFirstRecord(log_type, number, &sequence);

      if (!s.ok()) {

        return s;

      }

      if (sequence == 0) {

        // empty file

        continue;

      }

      // Reproduce the race condition where a log file is moved

      // to archived dir, between these two sync points, used in

      // (DBTest,TransactionLogIteratorRace)

      TEST_SYNC_POINT("WalManager::GetSortedWalsOfType:1");

      TEST_SYNC_POINT("WalManager::GetSortedWalsOfType:2");

      uint64_t size_bytes;

      s = env_->GetFileSize(LogFileName(path, number), &size_bytes);

      // re-try in case the alive log file has been moved to archive.

      std::string archived_file = ArchivedLogFileName(path, number);

      if (!s.ok() && log_type == kAliveLogFile &&

          env_->FileExists(archived_file).ok()) {

        s = env_->GetFileSize(archived_file, &size_bytes);

        if (!s.ok() && env_->FileExists(archived_file).IsNotFound()) {

          // oops, the file just got deleted from archived dir! move on

          s = Status::OK();

          continue;

        }

      }

      if (!s.ok()) {

        return s;

      }

      log_files->push_back(std::unique_ptr<LogFile>(

          new LogFileImpl(number, log_type, sequence, size_bytes)));

    }

  }

  CompareLogByPointer compare_log_files;

  std::sort(log_files->begin(), log_files->end(), compare_log_files);

  return status;

}

Status WalManager::RetainProbableWalFiles(VectorLogPtr* all_logs,

                                          const SequenceNumber target) {

  int64_t start = 0;  // signed to avoid overflow when target is < first file.

  int64_t end = static_cast<int64_t>(all_logs->size()) - 1;

  // Binary Search. avoid opening all files.

  while (end >= start) {

    int64_t mid = start + (end - start) / 2;  // Avoid overflow.

    SequenceNumber current_seq_num = (*all_logs)[mid]->StartSequence();

    if (current_seq_num == target) {

      end = mid;

      break;

    } else if (current_seq_num < target) {

      start = mid + 1;

    } else {

      end = mid - 1;

    }

  }

  // end could be -ve.

  size_t start_index = std::max<int64_t>(0, end);

  // The last wal file is always included

  all_logs->erase(all_logs->begin(), all_logs->begin() + start_index);

  return Status::OK();

}

Status WalManager::ReadFirstRecord(const WalFileType type,

                                   const uint64_t number,

                                   SequenceNumber* sequence) {

  *sequence = 0;

  if (type != kAliveLogFile && type != kArchivedLogFile) {

    RLOG(InfoLogLevel::ERROR_LEVEL, db_options_.info_log,

        "[WalManger] Unknown file type %s", ToString(type).c_str());

    return STATUS(NotSupported,

        "File Type Not Known " + ToString(type));

  }

  {

    MutexLock l(&read_first_record_cache_mutex_);

    auto itr = read_first_record_cache_.find(number);

    if (itr != read_first_record_cache_.end()) {

      *sequence = itr->second;

      return Status::OK();

    }

  }

  Status s;

  if (type == kAliveLogFile) {

    std::string fname = LogFileName(db_options_.wal_dir, number);

    s = ReadFirstLine(fname, sequence);

    if (env_->FileExists(fname).ok() && !s.ok()) {

      // return any error that is not caused by non-existing file

      return s;

    }

  }

  if (type == kArchivedLogFile || !s.ok()) {

    //  check if the file got moved to archive.

    std::string archived_file =

        ArchivedLogFileName(db_options_.wal_dir, number);

    s = ReadFirstLine(archived_file, sequence);

    // maybe the file was deleted from archive dir. If that's the case, return

    // Status::OK(). The caller with identify this as empty file because

    // *sequence == 0

    if (!s.ok() && env_->FileExists(archived_file).IsNotFound()) {

      return Status::OK();

    }

  }

  if (s.ok() && *sequence != 0) {

    MutexLock l(&read_first_record_cache_mutex_);

    read_first_record_cache_.insert({number, *sequence});

  }

  return s;

}

// the function returns status.ok() and sequence == 0 if the file exists, but is

// empty

Status WalManager::ReadFirstLine(const std::string& fname,

                                 SequenceNumber* sequence) {

  struct LogReporter : public log::Reader::Reporter {

    Env* env;

    Logger* info_log;

    const char* fname;

    Status* status;

    bool ignore_error;  // true if db_options_.paranoid_checks==false

    void Corruption(size_t bytes, const Status& s) override {

      RLOG(InfoLogLevel::WARN_LEVEL, info_log,

          "[WalManager] %s%s: dropping %d bytes; %s",

          (this->ignore_error ? "(ignoring error) " : ""), fname,

          static_cast<int>(bytes), s.ToString().c_str());

      if (this->status->ok()) {

        // only keep the first error

        *this->status = s;

      }

    }

  };

  std::unique_ptr<SequentialFile> file;

  Status status = env_->NewSequentialFile(fname, &file, env_options_);

  unique_ptr<SequentialFileReader> file_reader(

      new SequentialFileReader(std::move(file)));

  if (!status.ok()) {

    return status;

  }

  LogReporter reporter;

  reporter.env = env_;

  reporter.info_log = db_options_.info_log.get();

  reporter.fname = fname.c_str();

  reporter.status = &status;

  reporter.ignore_error = !db_options_.paranoid_checks;

  log::Reader reader(db_options_.info_log, std::move(file_reader), &reporter,

                     true /*checksum*/, 0 /*initial_offset*/, *sequence);

  std::string scratch;

  Slice record;

  if (reader.ReadRecord(&record, &scratch) &&

      (status.ok() || !db_options_.paranoid_checks)) {

    if (record.size() < 12) {

      reporter.Corruption(record.size(),

                          STATUS(Corruption, "log record too small"));

      // TODO read record's till the first no corrupt entry?

    } else {

      WriteBatch batch;

      WriteBatchInternal::SetContents(&batch, record);

      *sequence = WriteBatchInternal::Sequence(&batch);

      return Status::OK();

    }

  }

  // ReadRecord returns false on EOF, which means that the log file is empty. we

  // return status.ok() in that case and set sequence number to 0

  *sequence = 0;

  return status;

}

#endif  // ROCKSDB_LITE

}  // namespace rocksdb