// Matrix Construct

//

// Copyright (C) Matrix Construct Developers, Authors & Contributors

// Copyright (C) 2016-2018 Jason Volk <jason@zemos.net>

//

// Permission to use, copy, modify, and/or distribute this software for any

// purpose with or without fee is hereby granted, provided that the above

// copyright notice and this permission notice is present in all copies. The

// full license for this software is available in the LICENSE file.

#include <rocksdb/version.h>

#include <rocksdb/status.h>

#include <rocksdb/db.h>

#include <rocksdb/cache.h>

#include <rocksdb/comparator.h>

#include <rocksdb/merge_operator.h>

#include <rocksdb/perf_level.h>

#include <rocksdb/perf_context.h>

#include <rocksdb/iostats_context.h>

#include <rocksdb/listener.h>

#include <rocksdb/statistics.h>

#include <rocksdb/convenience.h>

#include <rocksdb/env.h>

#include <rocksdb/slice_transform.h>

#include <rocksdb/utilities/checkpoint.h>

#include <rocksdb/filter_policy.h>

#include <rocksdb/table.h>

#include <rocksdb/sst_file_manager.h>

// ircd::db interfaces requiring complete RocksDB (frontside).

#include <ircd/db/database/comparator.h>

#include <ircd/db/database/prefix_transform.h>

#include <ircd/db/database/mergeop.h>

#include <ircd/db/database/events.h>

#include <ircd/db/database/stats.h>

#include <ircd/db/database/logs.h>

#include <ircd/db/database/column.h>

#include <ircd/db/database/txn.h>

// RocksDB embedding environment callback interfaces (backside).

#include <ircd/db/database/env/env.h>

#include <ircd/db/database/env/writable_file.h>

#include <ircd/db/database/env/sequential_file.h>

#include <ircd/db/database/env/random_access_file.h>

#include <ircd/db/database/env/random_rw_file.h>

#include <ircd/db/database/env/directory.h>

#include <ircd/db/database/env/file_lock.h>

#include <ircd/db/database/env/state.h>

#define IRCD_DB_PORT

// RocksDB port linktime-overriding interfaces (experimental).

#ifdef IRCD_DB_PORT

#include <ircd/db/database/env/port.h>

#endif

// Internal utility interface for this definition file.

#include "db.h"

//

// Misc / General linkages

//

/// Dedicated logging facility for the database subsystem

decltype(ircd::db::log)

ircd::db::log

{

"db", 'D'

};

/// Dedicated logging facility for rocksdb's log callbacks

decltype(ircd::db::rog)

ircd::db::rog

{

"rdb", 'R'

};

ircd::conf::item<size_t>

ircd::db::request_pool_stack_size

{

{ "name", "ircd.db.request_pool.stack_size" },

{ "default", long(128_KiB) },

};

ircd::conf::item<size_t>

ircd::db::request_pool_size

{

{

{ "name", "ircd.db.request_pool.size" },

{ "default", 32L },

}, []

{

request.set(size_t(request_pool_size));

}

};

/// Concurrent request pool. Requests to seek may be executed on this

/// pool in cases where a single context would find it advantageous.

/// Some examples are a db::row seek, or asynchronous prefetching.

///

/// The number of workers in this pool should upper bound at the

/// number of concurrent AIO requests which are effective on this

/// system. This is a static pool shared by all databases.

decltype(ircd::db::request)

ircd::db::request

{

"db req",

size_t(request_pool_stack_size),

0, // don't prespawn because this is static

};

/// This mutex is necessary to serialize entry into rocksdb's write impl

/// otherwise there's a risk of a deadlock if their internal pthread

/// mutexes are contended. This is because a few parts of rocksdb are

/// incorrectly using std::mutex directly when they ought to be using their

/// rocksdb::port wrapper.

decltype(ircd::db::write_mutex)

ircd::db::write_mutex;

///////////////////////////////////////////////////////////////////////////////

//

// init

//

namespace ircd::db

{

static void init_compressions();

static void init_directory();

static void init_version();

}

static char ircd_db_version[64];

const char *const ircd::db::version(ircd_db_version);

ircd::db::init::init()

{

init_compressions();

init_directory();

request.add(request_pool_size);

}

ircd::db::init::~init()

noexcept

{

if(request.active())

log::warning

{

log, "Terminating %zu active of %zu client request contexts; %zu pending; %zu queued",

request.active(),

request.size(),

request.pending(),

request.queued()

};

request.terminate();

log::debug

{

log, "Waiting for %zu active of %zu client request contexts; %zu pending; %zu queued",

request.active(),

request.size(),

request.pending(),

request.queued()

};

request.join();

log::debug

{

log, "All contexts joined; all requests are clear."

};

}

void

ircd::db::init_directory()

try

{

const auto dbdir

{

fs::get(fs::DB)

};

if(fs::mkdir(dbdir))

log::notice

{

log, "Created new database directory at `%s'", dbdir

};

else

log::info

{

log, "Using database directory at `%s'", dbdir

};

}

catch(const fs::error &e)

{

log::error

{

log, "Cannot start database system: %s", e.what()

};

if(ircd::debugmode)

throw;

}

void

ircd::db::init_compressions()

{

const auto compressions

{

rocksdb::GetSupportedCompressions()

};

if(compressions.empty())

log::warning

{

"No compression libraries have been linked with the DB."

" This is probably not what you want."

};

}

// Renders a version string from the defines included here.

__attribute__((constructor))

void

ircd::db::init_version()

{

snprintf(ircd_db_version, sizeof(ircd_db_version), "%d.%d.%d",

ROCKSDB_MAJOR,

ROCKSDB_MINOR,

ROCKSDB_PATCH);

}

///////////////////////////////////////////////////////////////////////////////

//

// database

//

void

ircd::db::sync(database &d)

{

const ctx::uninterruptible::nothrow ui;

const std::lock_guard<decltype(write_mutex)> lock{write_mutex};

log::debug

{

log, "'%s': @%lu SYNC WAL",

name(d),

sequence(d)

};

throw_on_error

{

d.d->SyncWAL()

};

}

/// Flushes all columns. Note that if blocking=true, blocking may occur for

/// each column individually.

void

ircd::db::flush(database &d,

const bool &sync)

{

const ctx::uninterruptible::nothrow ui;

const std::lock_guard<decltype(write_mutex)> lock{write_mutex};

log::debug

{

log, "'%s': @%lu FLUSH WAL",

name(d),

sequence(d)

};

throw_on_error

{

d.d->FlushWAL(sync)

};

}

/// Moves memory structures to SST files for all columns. This doesn't

/// necessarily sort anything that wasn't previously sorted, but it may create

/// new SST files and shouldn't be confused with a typical fflush().

/// Note that if blocking=true, blocking may occur for each column individually.

void

ircd::db::sort(database &d,

const bool &blocking)

{

for(const auto &c : d.columns)

{

db::column column{*c};

db::sort(column, blocking);

}

}

void

ircd::db::compact(database &d)

{

static const std::pair<string_view, string_view> range

{

{}, {}

};

for(const auto &c : d.columns)

{

db::column column{*c};

compact(column, range, -1);

compact(column, -1);

}

}

void

ircd::db::check(database &d)

{

assert(d.d);

const ctx::uninterruptible::nothrow ui;

throw_on_error

{

d.d->VerifyChecksum()

};

}

/// Writes a snapshot of this database to the directory specified. The

/// snapshot consists of hardlinks to the bulk data files of this db, but

/// copies the other stuff that usually gets corrupted. The directory can

/// then be opened as its own database either read-only or read-write.

/// Incremental backups and rollbacks can begin from this interface. Note

/// this may be an expensive blocking operation.

uint64_t

ircd::db::checkpoint(database &d)

{

if(!d.checkpointer)

throw error

{

"Checkpointing is not available for db(%p) '%s",

&d,

name(d)

};

const ctx::uninterruptible::nothrow ui;

const std::lock_guard<decltype(write_mutex)> lock{write_mutex};

const auto seqnum

{

sequence(d)

};

const std::string dir

{

db::path(name(d), seqnum)

};

throw_on_error

{

d.checkpointer->CreateCheckpoint(dir, 0)

};

log::debug

{

log, "'%s': Checkpoint at sequence %lu in `%s' complete",

name(d),

seqnum,

dir

};

return seqnum;

}

/// This wraps RocksDB's "File Deletions" which means after RocksDB

/// compresses some file it then destroys the uncompressed version;

/// setting this to false will disable that and retain both versions.

/// This is useful when a direct reference is being manually held by

/// us into the uncompressed version which must remain valid.

void

ircd::db::fdeletions(database &d,

const bool &enable,

const bool &force)

{

const std::lock_guard<decltype(write_mutex)> lock{write_mutex};

const ctx::uninterruptible::nothrow ui;

if(enable) throw_on_error

{

d.d->EnableFileDeletions(force)

};

else throw_on_error

{

d.d->DisableFileDeletions()

};

}

void

ircd::db::setopt(database &d,

const string_view &key,

const string_view &val)

{

const std::unordered_map<std::string, std::string> options

{

{ std::string{key}, std::string{val} }

};

const std::lock_guard<decltype(write_mutex)> lock{write_mutex};

const ctx::uninterruptible::nothrow ui;

throw_on_error

{

d.d->SetDBOptions(options)

};

}

uint64_t

ircd::db::ticker(const database &d,

const string_view &key)

{

return ticker(d, ticker_id(key));

}

uint64_t

ircd::db::ticker(const database &d,

const uint32_t &id)

{

return d.stats->getTickerCount(id);

}

size_t

ircd::db::bytes(const database &d)

{

return std::accumulate(begin(d.columns), end(d.columns), size_t(0), []

(auto ret, const auto &colptr)

{

db::column c{*colptr};

return ret += db::bytes(c);

});

}

size_t

ircd::db::file_count(const database &d)

{

return std::accumulate(begin(d.columns), end(d.columns), size_t(0), []

(auto ret, const auto &colptr)

{

db::column c{*colptr};

return ret += db::file_count(c);

});

}

/// Get the live file list for db; see overlord documentation.

std::vector<std::string>

ircd::db::files(const database &d)

{

uint64_t ignored;

return files(d, ignored);

}

/// Get the live file list for database relative to the database's directory.

/// One of the files is a manifest file which is over-allocated and its used

/// size is returned in the integer passed to the `msz` argument.

///

/// This list may not be completely up to date. The reliable way to get the

/// most current list is to flush all columns first and ensure no database

/// activity took place between the flushing and this query.

std::vector<std::string>

ircd::db::files(const database &cd,

uint64_t &msz)

{

std::vector<std::string> ret;

auto &d(const_cast<database &>(cd));

const ctx::uninterruptible::nothrow ui;

throw_on_error

{

d.d->GetLiveFiles(ret, &msz, false)

};

return ret;

}

uint64_t

ircd::db::sequence(const database &cd)

{

database &d(const_cast<database &>(cd));

return d.d->GetLatestSequenceNumber();

}

rocksdb::Cache *

ircd::db::cache(database &d)

{

return d.cache.get();

}

const rocksdb::Cache *

ircd::db::cache(const database &d)

{

return d.cache.get();

}

template<>

ircd::db::prop_int

ircd::db::property(const database &cd,

const string_view &name)

{

uint64_t ret;

database &d(const_cast<database &>(cd));

const ctx::uninterruptible::nothrow ui;

if(!d.d->GetAggregatedIntProperty(slice(name), &ret))

throw not_found

{

"property '%s' for all columns in '%s' not found or not an integer.",

name,

db::name(d)

};

return ret;

}

std::shared_ptr<ircd::db::database::column>

ircd::db::shared_from(database::column &column)

{

return column.shared_from_this();

}

std::shared_ptr<const ircd::db::database::column>

ircd::db::shared_from(const database::column &column)

{

return column.shared_from_this();

}

const std::string &

ircd::db::uuid(const database &d)

{

return d.uuid;

}

const std::string &

ircd::db::name(const database &d)

{

return d.name;

}

//

// database

//

namespace ircd::db

{

extern const database::description default_description;

}

// Instance list linkage

template<>

decltype(ircd::util::instance_list<ircd::db::database>::list)

ircd::util::instance_list<ircd::db::database>::list

{};

decltype(ircd::db::default_description)

ircd::db::default_description

{

/// Requirement of RocksDB going back to LevelDB. This column must

/// always exist in all descriptions and probably should be at idx[0].

{ "default" }

};

ircd::db::database &

ircd::db::database::get(column &column)

{

assert(column.d);

return *column.d;

}

const ircd::db::database &

ircd::db::database::get(const column &column)

{

assert(column.d);

return *column.d;

}

ircd::db::database &

ircd::db::database::get(const string_view &name)

{

const auto pair

{

namepoint(name)

};

return get(pair.first, pair.second);

}

ircd::db::database &

ircd::db::database::get(const string_view &name,

const uint64_t &checkpoint)

{

auto *const &d

{

get(std::nothrow, name, checkpoint)

};

if(likely(d))

return *d;

throw checkpoint == uint64_t(-1)?

std::out_of_range{"No database with that name exists"}:

std::out_of_range{"No database with that name at that checkpoint exists"};

}

ircd::db::database *

ircd::db::database::get(std::nothrow_t,

const string_view &name)

{

const auto pair

{

namepoint(name)

};

return get(std::nothrow, pair.first, pair.second);

}

ircd::db::database *

ircd::db::database::get(std::nothrow_t,

const string_view &name,

const uint64_t &checkpoint)

{

for(auto *const &d : list)

if(name == d->name)

if(checkpoint == uint64_t(-1) || checkpoint == d->checkpoint)

return d;

return nullptr;

}

//

// database::database

//

ircd::db::database::database(const string_view &name,

std::string optstr)

:database

{

name, std::move(optstr), default_description

}

{

}

ircd::db::database::database(const string_view &name,

std::string optstr,

description description)

:database

{

namepoint(name).first, namepoint(name).second, std::move(optstr), std::move(description)

}

{

}

ircd::db::database::database(const string_view &name,

const uint64_t &checkpoint,

std::string optstr,

description description)

try

:name

{

namepoint(name).first

}

,checkpoint

{

// a -1 may have been generated by the db::namepoint() util when the user

// supplied just a name without a checkpoint. In the context of database

// opening/creation -1 just defaults to 0.

checkpoint == uint64_t(-1)? 0 : checkpoint

}

,optstr

{

std::move(optstr)

}

,path

{

db::path(this->name, this->checkpoint)

}

,env

{

std::make_shared<struct env>(this)

}

,logs

{

std::make_shared<struct logs>(this)

}

,stats

{

std::make_shared<struct stats>(this)

}

,events

{

std::make_shared<struct events>(this)

}

,mergeop

{

std::make_shared<struct mergeop>(this)

}

,ssts

{

// note: the sst file manager cannot be used for now because it will spawn

// note: a pthread internally in rocksdb which does not use our callbacks

// note: we gave in the supplied env. we really don't want that.

//rocksdb::NewSstFileManager(env.get(), logs, {}, 0, true, nullptr, 0.05)

}

,cache{[this]

() -> std::shared_ptr<rocksdb::Cache>

{

//TODO: conf

const auto lru_cache_size{16_MiB};

return rocksdb::NewLRUCache(lru_cache_size);

}()}

,descriptors

{

std::move(description)

}

,column_names{[this]

{

// Existing columns at path. If any are left the descriptor set did not

// describe all of the columns found in the database at path.

const auto opts

{

make_dbopts(this->optstr)

};

const auto required

{

db::column_names(path, opts)

};

std::set<string_view> existing

{

begin(required), end(required)

};

// The names of the columns extracted from the descriptor set

std::vector<string_view> ret(descriptors.size());

std::transform(begin(descriptors), end(descriptors), begin(ret), [&existing]

(const auto &descriptor) -> string_view

{

existing.erase(descriptor.name);

return descriptor.name;

});

for(const auto &remain : existing)

throw error

{

"Failed to describe existing column '%s' (and %zd others...)",

remain,

existing.size() - 1

};

return ret;

}()}

,column_index{[this]

{

decltype(this->column_index) ret;

for(const auto &descriptor : this->descriptors)

ret.emplace(descriptor.name, -1);

return ret;

}()}

,d{[this]

{

bool fsck{false};

bool read_only{false};

auto opts

{

make_dbopts(this->optstr, &this->optstr, &read_only, &fsck)

};

// Setup sundry

opts.create_if_missing = true;

opts.create_missing_column_families = true;

opts.max_file_opening_threads = 0;

opts.stats_dump_period_sec = 0;

opts.enable_thread_tracking = true;

opts.avoid_flush_during_recovery = true;

opts.delete_obsolete_files_period_micros = 0;

opts.max_background_jobs = 2;

opts.max_background_flushes = 1;

opts.max_background_compactions = 1;

opts.max_subcompactions = 1;

opts.max_open_files = -1; //ircd::info::rlimit_nofile / 4;

opts.allow_concurrent_memtable_write = false;

opts.enable_write_thread_adaptive_yield = false;

opts.enable_pipelined_write = false;

opts.use_direct_reads = true;

opts.write_thread_max_yield_usec = 0;

opts.write_thread_slow_yield_usec = 0;

opts.use_direct_io_for_flush_and_compaction = false;

#ifdef RB_DEBUG

opts.dump_malloc_stats = true;

#endif

// Setup env

opts.env = env.get();

// Setup SST file mgmt

opts.sst_file_manager = this->ssts;

// Setup logging

logs->SetInfoLogLevel(ircd::debugmode? rocksdb::DEBUG_LEVEL : rocksdb::WARN_LEVEL);

opts.info_log_level = logs->GetInfoLogLevel();

opts.info_log = logs;

// Setup event and statistics callbacks

opts.listeners.emplace_back(this->events);

// Setup histogram collecting

//this->stats->stats_level_ = rocksdb::kAll;

this->stats->stats_level_ = rocksdb::kExceptTimeForMutex;

opts.statistics = this->stats;

// Setup performance metric options

//rocksdb::SetPerfLevel(rocksdb::PerfLevel::kDisable);

// Default corruption tolerance is zero-tolerance; db fails to open with

// error by default to inform the user. The rest of the options are

// various relaxations for how to proceed.

opts.wal_recovery_mode = rocksdb::WALRecoveryMode::kAbsoluteConsistency;

// When corrupted after crash, the DB is rolled back before the first

// corruption and erases everything after it, giving a consistent

// state up at that point, though losing some recent data.

if(ircd::pitrecdb) //TODO: no global?

opts.wal_recovery_mode = rocksdb::WALRecoveryMode::kPointInTimeRecovery;

// Skipping corrupted records will create gaps in the DB timeline where the

// application (like a matrix timeline) cannot tolerate the unexpected gap.

//opts.wal_recovery_mode = rocksdb::WALRecoveryMode::kSkipAnyCorruptedRecords;

// Tolerating corrupted records is very last-ditch for getting the database to

// open in a catastrophe. We have no use for this option but should use it for

//TODO: emergency salvage-mode.

//opts.wal_recovery_mode = rocksdb::WALRecoveryMode::kTolerateCorruptedTailRecords;

// Setup cache

opts.row_cache = this->cache;

// Setup column families

for(const auto &desc : descriptors)

{

const auto c

{

std::make_shared<column>(this, desc)

};

columns.emplace_back(c);

}

std::vector<rocksdb::ColumnFamilyHandle *> handles;

std::vector<rocksdb::ColumnFamilyDescriptor> columns(this->columns.size());

std::transform(begin(this->columns), end(this->columns), begin(columns), []

(const auto &column)

{

return static_cast<const rocksdb::ColumnFamilyDescriptor &>(*column);

});

// NOTE: rocksdb sez RepairDB is broken; can't use now

if(fsck && fs::is_dir(path))

{

const ctx::uninterruptible ui;

log::notice

{

log, "Checking database @ `%s' columns[%zu]", path, columns.size()

};

throw_on_error

{

rocksdb::RepairDB(path, opts, columns)

};

log::info

{

log, "Database @ `%s' check complete", path

};

}

// If the directory does not exist, though rocksdb will create it, we can

// avoid scaring the user with an error log message if we just do that..

if(opts.create_if_missing && !fs::is_dir(path))

fs::mkdir(path);

// Announce attempt before usual point where exceptions are thrown

const ctx::uninterruptible ui;

log::info

{

log, "Opening database \"%s\" @ `%s' with %zu columns...",

this->name,

path,

columns.size()

};

// Open DB into ptr

rocksdb::DB *ptr;

if(read_only)

throw_on_error

{

rocksdb::DB::OpenForReadOnly(opts, path, columns, &handles, &ptr)

};

else

throw_on_error

{

rocksdb::DB::Open(opts, path, columns, &handles, &ptr)

};

std::unique_ptr<rocksdb::DB> ret

{

ptr

};

for(const auto &handle : handles)

{

this->columns.at(handle->GetID())->handle.reset(handle);

this->column_index.at(handle->GetName()) = handle->GetID();

}

for(size_t i(0); i < this->columns.size(); ++i)

if(db::id(*this->columns[i]) != i)

throw error

{

"Columns misaligned: expecting id[%zd] got id[%u] '%s'",

i,

db::id(*this->columns[i]),

db::name(*this->columns[i])

};

return ret;

}()}

,uuid{[this]

{

const ctx::uninterruptible ui;

std::string ret;

throw_on_error

{

d->GetDbIdentity(ret)

};

return ret;

}()}

,checkpointer{[this]

{

const ctx::uninterruptible ui;

rocksdb::Checkpoint *checkpointer{nullptr};

throw_on_error

{

rocksdb::Checkpoint::Create(this->d.get(), &checkpointer)

};

return checkpointer;

}()}

{

if(ircd::checkdb)

{

log::notice

{

log, "'%s': Verifying database integrity. This may take several minutes...",

this->name

};

const ctx::uninterruptible ui;

check(*this);

}

log::info

{

log, "'%s': Opened database @ `%s' with %zu columns at sequence number %lu.",

this->name,

path,

columns.size(),

d->GetLatestSequenceNumber()

};

}

catch(const corruption &e)

{

throw corruption

{

"Corruption for '%s' (%s). Try restarting with the -pitrecdb command line option",

this->name,

e.what()

};

}

catch(const std::exception &e)

{

throw error

{

"Failed to open db '%s': %s",

this->name,

e.what()

};

}

ircd::db::database::~database()

noexcept try

{

const ctx::uninterruptible::nothrow ui;

log::info

{

log, "'%s': closing database @ `%s'...",

name,

path

};

rocksdb::CancelAllBackgroundWork(d.get(), true); // true = blocking

log::debug

{

log, "'%s': background_errors: %lu; flushing...",