<commit_before>// Copyright (c) 2015, Baidu.com, Inc. All Rights Reserved
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.
//
// 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 "db/db_impl.h"

#include <iostream>

#include <algorithm>
#include <set>
#include <string>
#include <stdint.h>
#include <stdio.h>
#include <vector>
#include "db/builder.h"
#include "db/db_iter.h"
#include "db/dbformat.h"
#include "db/filename.h"
#include "db/log_reader.h"
#include "db/log_writer.h"
#include "db/memtable.h"
#include "db/memtable_on_leveldb.h"
#include "db/table_cache.h"
#include "db/version_set.h"
#include "db/write_batch_internal.h"
#include "leveldb/db.h"
#include "leveldb/compact_strategy.h"
#include "leveldb/env.h"
#include "leveldb/status.h"
#include "leveldb/table.h"
#include "leveldb/table_builder.h"
#include "leveldb/table_utils.h"
#include "port/port.h"
#include "table/block.h"
#include "table/merger.h"
#include "table/two_level_iterator.h"
#include "util/coding.h"
#include "util/logging.h"
#include "util/mutexlock.h"

namespace leveldb {

const int kNumNonTableCacheFiles = 10;

// Information kept for every waiting writer
struct DBImpl::Writer {
  Status status;
  WriteBatch* batch;
  bool sync;
  bool done;
  port::CondVar cv;

  explicit Writer(port::Mutex* mu) : cv(mu) { }
};

struct DBImpl::CompactionState {
  Compaction* const compaction;

  // Sequence numbers < smallest_snapshot are not significant since we
  // will never have to service a snapshot below smallest_snapshot.
  // Therefore if we have seen a sequence number S <= smallest_snapshot,
  // we can drop all entries for the same key with sequence numbers < S.
  SequenceNumber smallest_snapshot;

  // Files produced by compaction
  struct Output {
    uint64_t number;
    uint64_t file_size;
    InternalKey smallest, largest;
  };
  std::vector<Output> outputs;

  // State kept for output being generated
  WritableFile* outfile;
  TableBuilder* builder;

  uint64_t total_bytes;

  Output* current_output() { return &outputs[outputs.size()-1]; }

  explicit CompactionState(Compaction* c)
      : compaction(c),
        outfile(NULL),
        builder(NULL),
        total_bytes(0) {
  }
};

// Fix user-supplied options to be reasonable
template <class T,class V>
static void ClipToRange(T* ptr, V minvalue, V maxvalue) {
  if (static_cast<V>(*ptr) > maxvalue) *ptr = maxvalue;
  if (static_cast<V>(*ptr) < minvalue) *ptr = minvalue;
}
Options SanitizeOptions(const std::string& dbname,
                        const InternalKeyComparator* icmp,
                        const InternalFilterPolicy* ipolicy,
                        const Options& src) {
  Options result = src;
  result.comparator = icmp;
  result.filter_policy = (src.filter_policy != NULL) ? ipolicy : NULL;
  ClipToRange(&result.max_open_files,    64 + kNumNonTableCacheFiles, 50000);
  ClipToRange(&result.write_buffer_size, 64<<10,                      1<<30);
  ClipToRange(&result.block_size,        1<<10,                       4<<20);
  if (result.info_log == NULL) {
    // Open a log file in the same directory as the db
    src.env->CreateDir(dbname);  // In case it does not exist
    src.env->RenameFile(InfoLogFileName(dbname), OldInfoLogFileName(dbname));
    Status s = src.env->NewLogger(InfoLogFileName(dbname), &result.info_log);
    if (!s.ok()) {
      // No place suitable for logging
      result.info_log = NULL;
    }
  }
  if (result.block_cache == NULL) {
    result.block_cache = NewLRUCache(8 << 20);
  }
  return result;
}

DBImpl::DBImpl(const Options& options, const std::string& dbname)
    : key_start_(options.key_start), key_end_(options.key_end),
      env_(options.env),
      internal_comparator_(options.comparator),
      internal_filter_policy_(options.filter_policy),
      options_(SanitizeOptions(
          dbname, &internal_comparator_, &internal_filter_policy_, options)),
      owns_info_log_(options_.info_log != options.info_log),
      owns_block_cache_(options_.block_cache != options.block_cache),
      dbname_(dbname),
      table_cache_(options_.table_cache),
      owns_table_cache_(options_.table_cache == NULL),
      db_lock_(NULL),
      shutting_down_(NULL),
      bg_cv_(&mutex_),
      writting_mem_cv_(&mutex_),
      is_writting_mem_(false),
      mem_(NewMemTable()),
      imm_(NULL), recover_mem_(NULL),
      logfile_(NULL),
      logfile_number_(0),
      log_(NULL),
      bound_log_size_(0),
      tmp_batch_(new WriteBatch),
      bg_compaction_scheduled_(false),
      bg_compaction_score_(0),
      bg_schedule_id_(0),
      manual_compaction_(NULL),
      consecutive_compaction_errors_(0),
      flush_on_destroy_(false) {
  mem_->Ref();
  has_imm_.Release_Store(NULL);

  // Reserve ten files or so for other uses and give the rest to TableCache.
  if (owns_table_cache_) {
    Log(options_.info_log, "[%s] create new table cache.", dbname_.c_str());
    const int table_cache_size = options_.max_open_files - kNumNonTableCacheFiles;
    table_cache_ = new TableCache(table_cache_size);
  }
  versions_ = new VersionSet(dbname_, &options_, table_cache_,
                             &internal_comparator_);
}

Status DBImpl::Shutdown1() {
  MutexLock l(&mutex_);
  shutting_down_.Release_Store(this);  // Any non-NULL value is ok

  Log(options_.info_log, "[%s] wait bg compact finish", dbname_.c_str());
  while (bg_compaction_scheduled_) {
    bg_cv_.Wait();
  }

  Status s;
  if (!options_.dump_mem_on_shutdown) {
    return s;
  }
  Log(options_.info_log, "[%s] fg compact mem table", dbname_.c_str());
  if (imm_ != NULL) {
    s = CompactMemTable();
  }
  if (s.ok()) {
    assert(imm_ == NULL);
    while (is_writting_mem_) {
        writting_mem_cv_.Wait();
    }
    imm_ = mem_;
    has_imm_.Release_Store(imm_);
    mem_ = NewMemTable();
    mem_->Ref();
    bound_log_size_ = 0;
    s = CompactMemTable();
  }
  return s;
}

Status DBImpl::Shutdown2() {
  MutexLock l(&mutex_);<commit_msg>more faster unload<commit_after>// Copyright (c) 2015, Baidu.com, Inc. All Rights Reserved
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.
//
// 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 "db/db_impl.h"

#include <iostream>

#include <algorithm>
#include <set>
#include <string>
#include <stdint.h>
#include <stdio.h>
#include <vector>
#include "db/builder.h"
#include "db/db_iter.h"
#include "db/dbformat.h"
#include "db/filename.h"
#include "db/log_reader.h"
#include "db/log_writer.h"
#include "db/memtable.h"
#include "db/memtable_on_leveldb.h"
#include "db/table_cache.h"
#include "db/version_set.h"
#include "db/write_batch_internal.h"
#include "leveldb/db.h"
#include "leveldb/compact_strategy.h"
#include "leveldb/env.h"
#include "leveldb/status.h"
#include "leveldb/table.h"
#include "leveldb/table_builder.h"
#include "leveldb/table_utils.h"
#include "port/port.h"
#include "table/block.h"
#include "table/merger.h"
#include "table/two_level_iterator.h"
#include "util/coding.h"
#include "util/logging.h"
#include "util/mutexlock.h"

namespace leveldb {

const int kNumNonTableCacheFiles = 10;
const double kVeryHighScore = 1000.0;

// Information kept for every waiting writer
struct DBImpl::Writer {
  Status status;
  WriteBatch* batch;
  bool sync;
  bool done;
  port::CondVar cv;

  explicit Writer(port::Mutex* mu) : cv(mu) { }
};

struct DBImpl::CompactionState {
  Compaction* const compaction;

  // Sequence numbers < smallest_snapshot are not significant since we
  // will never have to service a snapshot below smallest_snapshot.
  // Therefore if we have seen a sequence number S <= smallest_snapshot,
  // we can drop all entries for the same key with sequence numbers < S.
  SequenceNumber smallest_snapshot;

  // Files produced by compaction
  struct Output {
    uint64_t number;
    uint64_t file_size;
    InternalKey smallest, largest;
  };
  std::vector<Output> outputs;

  // State kept for output being generated
  WritableFile* outfile;
  TableBuilder* builder;

  uint64_t total_bytes;

  Output* current_output() { return &outputs[outputs.size()-1]; }

  explicit CompactionState(Compaction* c)
      : compaction(c),
        outfile(NULL),
        builder(NULL),
        total_bytes(0) {
  }
};

// Fix user-supplied options to be reasonable
template <class T,class V>
static void ClipToRange(T* ptr, V minvalue, V maxvalue) {
  if (static_cast<V>(*ptr) > maxvalue) *ptr = maxvalue;
  if (static_cast<V>(*ptr) < minvalue) *ptr = minvalue;
}
Options SanitizeOptions(const std::string& dbname,
                        const InternalKeyComparator* icmp,
                        const InternalFilterPolicy* ipolicy,
                        const Options& src) {
  Options result = src;
  result.comparator = icmp;
  result.filter_policy = (src.filter_policy != NULL) ? ipolicy : NULL;
  ClipToRange(&result.max_open_files,    64 + kNumNonTableCacheFiles, 50000);
  ClipToRange(&result.write_buffer_size, 64<<10,                      1<<30);
  ClipToRange(&result.block_size,        1<<10,                       4<<20);
  if (result.info_log == NULL) {
    // Open a log file in the same directory as the db
    src.env->CreateDir(dbname);  // In case it does not exist
    src.env->RenameFile(InfoLogFileName(dbname), OldInfoLogFileName(dbname));
    Status s = src.env->NewLogger(InfoLogFileName(dbname), &result.info_log);
    if (!s.ok()) {
      // No place suitable for logging
      result.info_log = NULL;
    }
  }
  if (result.block_cache == NULL) {
    result.block_cache = NewLRUCache(8 << 20);
  }
  return result;
}

DBImpl::DBImpl(const Options& options, const std::string& dbname)
    : key_start_(options.key_start), key_end_(options.key_end),
      env_(options.env),
      internal_comparator_(options.comparator),
      internal_filter_policy_(options.filter_policy),
      options_(SanitizeOptions(
          dbname, &internal_comparator_, &internal_filter_policy_, options)),
      owns_info_log_(options_.info_log != options.info_log),
      owns_block_cache_(options_.block_cache != options.block_cache),
      dbname_(dbname),
      table_cache_(options_.table_cache),
      owns_table_cache_(options_.table_cache == NULL),
      db_lock_(NULL),
      shutting_down_(NULL),
      bg_cv_(&mutex_),
      writting_mem_cv_(&mutex_),
      is_writting_mem_(false),
      mem_(NewMemTable()),
      imm_(NULL), recover_mem_(NULL),
      logfile_(NULL),
      logfile_number_(0),
      log_(NULL),
      bound_log_size_(0),
      tmp_batch_(new WriteBatch),
      bg_compaction_scheduled_(false),
      bg_compaction_score_(0),
      bg_schedule_id_(0),
      manual_compaction_(NULL),
      consecutive_compaction_errors_(0),
      flush_on_destroy_(false) {
  mem_->Ref();
  has_imm_.Release_Store(NULL);

  // Reserve ten files or so for other uses and give the rest to TableCache.
  if (owns_table_cache_) {
    Log(options_.info_log, "[%s] create new table cache.", dbname_.c_str());
    const int table_cache_size = options_.max_open_files - kNumNonTableCacheFiles;
    table_cache_ = new TableCache(table_cache_size);
  }
  versions_ = new VersionSet(dbname_, &options_, table_cache_,
                             &internal_comparator_);
}

Status DBImpl::Shutdown1() {
  MutexLock l(&mutex_);
  shutting_down_.Release_Store(this);  // Any non-NULL value is ok

  Log(options_.info_log, "[%s] wait bg compact finish", dbname_.c_str());
  if (bg_compaction_scheduled_) {
    env_->ReSchedule(bg_schedule_id_, kVeryHighScore);
  }
  while (bg_compaction_scheduled_) {
    bg_cv_.Wait();
  }

  Status s;
  if (!options_.dump_mem_on_shutdown) {
    return s;
  }
  Log(options_.info_log, "[%s] fg compact mem table", dbname_.c_str());
  if (imm_ != NULL) {
    s = CompactMemTable();
  }
  if (s.ok()) {
    assert(imm_ == NULL);
    while (is_writting_mem_) {
        writting_mem_cv_.Wait();
    }
    imm_ = mem_;
    has_imm_.Release_Store(imm_);
    mem_ = NewMemTable();
    mem_->Ref();
    bound_log_size_ = 0;
    s = CompactMemTable();
  }
  return s;
}

Status DBImpl::Shutdown2() {
  MutexLock l(&mutex_);
{'fraction_non_alphanumeric': 0.12069517860003204, 'fraction_numerical': 0.008249239147845587, 'mean_word_length': 3.5440320232896654, 'pattern_counts': {'":': 0, '<': 53, '<?xml version=': 0, '>': 38, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1174792', 'n_tokens_mistral': 4188, 'n_tokens_neox': 4121, 'n_words': 993}