|
|
|
// Copyright (c) 2011-present, Facebook, Inc. All rights reserved.
|
|
|
|
// This source code is licensed under both the GPLv2 (found in the
|
|
|
|
// COPYING file in the root directory) and Apache 2.0 License
|
|
|
|
// (found in the LICENSE.Apache file in the root directory).
|
|
|
|
//
|
|
|
|
// 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.
|
|
|
|
#pragma once
|
|
|
|
|
|
|
|
#include <atomic>
|
|
|
|
#include <deque>
|
|
|
|
#include <functional>
|
|
|
|
#include <limits>
|
|
|
|
#include <set>
|
|
|
|
#include <string>
|
|
|
|
#include <utility>
|
|
|
|
#include <vector>
|
|
|
|
|
|
|
|
#include "db/blob/blob_file_completion_callback.h"
|
|
|
|
#include "db/column_family.h"
|
|
|
|
#include "db/compaction/compaction_iterator.h"
|
|
|
|
#include "db/flush_scheduler.h"
|
|
|
|
#include "db/internal_stats.h"
|
|
|
|
#include "db/job_context.h"
|
|
|
|
#include "db/log_writer.h"
|
|
|
|
#include "db/memtable_list.h"
|
|
|
|
#include "db/range_del_aggregator.h"
|
|
|
|
#include "db/version_edit.h"
|
|
|
|
#include "db/write_controller.h"
|
|
|
|
#include "db/write_thread.h"
|
|
|
|
#include "logging/event_logger.h"
|
|
|
|
#include "options/cf_options.h"
|
|
|
|
#include "options/db_options.h"
|
|
|
|
#include "port/port.h"
|
|
|
|
#include "rocksdb/compaction_filter.h"
|
|
|
|
#include "rocksdb/compaction_job_stats.h"
|
|
|
|
#include "rocksdb/db.h"
|
|
|
|
#include "rocksdb/env.h"
|
|
|
|
#include "rocksdb/memtablerep.h"
|
|
|
|
#include "rocksdb/transaction_log.h"
|
|
|
|
#include "table/scoped_arena_iterator.h"
|
|
|
|
#include "util/autovector.h"
|
|
|
|
#include "util/stop_watch.h"
|
|
|
|
#include "util/thread_local.h"
|
|
|
|
|
|
|
|
namespace ROCKSDB_NAMESPACE {
|
|
|
|
|
|
|
|
class Arena;
|
|
|
|
class ErrorHandler;
|
|
|
|
class MemTable;
|
|
|
|
class SnapshotChecker;
|
|
|
|
class SystemClock;
|
|
|
|
class TableCache;
|
|
|
|
class Version;
|
|
|
|
class VersionEdit;
|
|
|
|
class VersionSet;
|
|
|
|
|
|
|
|
// CompactionJob is responsible for executing the compaction. Each (manual or
|
|
|
|
// automated) compaction corresponds to a CompactionJob object, and usually
|
|
|
|
// goes through the stages of `Prepare()`->`Run()`->`Install()`. CompactionJob
|
|
|
|
// will divide the compaction into subcompactions and execute them in parallel
|
|
|
|
// if needed.
|
|
|
|
class CompactionJob {
|
|
|
|
public:
|
|
|
|
CompactionJob(
|
|
|
|
int job_id, Compaction* compaction, const ImmutableDBOptions& db_options,
|
|
|
|
const MutableDBOptions& mutable_db_options,
|
|
|
|
const FileOptions& file_options, VersionSet* versions,
|
|
|
|
const std::atomic<bool>* shutting_down, LogBuffer* log_buffer,
|
|
|
|
FSDirectory* db_directory, FSDirectory* output_directory,
|
|
|
|
FSDirectory* blob_output_directory, Statistics* stats,
|
|
|
|
InstrumentedMutex* db_mutex, ErrorHandler* db_error_handler,
|
|
|
|
std::vector<SequenceNumber> existing_snapshots,
|
|
|
|
SequenceNumber earliest_write_conflict_snapshot,
|
CompactionIterator sees consistent view of which keys are committed (#9830)
Summary:
**This PR does not affect the functionality of `DB` and write-committed transactions.**
`CompactionIterator` uses `KeyCommitted(seq)` to determine if a key in the database is committed.
As the name 'write-committed' implies, if write-committed policy is used, a key exists in the database only if
it is committed. In fact, the implementation of `KeyCommitted()` is as follows:
```
inline bool KeyCommitted(SequenceNumber seq) {
// For non-txn-db and write-committed, snapshot_checker_ is always nullptr.
return snapshot_checker_ == nullptr ||
snapshot_checker_->CheckInSnapshot(seq, kMaxSequence) == SnapshotCheckerResult::kInSnapshot;
}
```
With that being said, we focus on write-prepared/write-unprepared transactions.
A few notes:
- A key can exist in the db even if it's uncommitted. Therefore, we rely on `snapshot_checker_` to determine data visibility. We also require that all writes go through transaction API instead of the raw `WriteBatch` + `Write`, thus at most one uncommitted version of one user key can exist in the database.
- `CompactionIterator` outputs a key as long as the key is uncommitted.
Due to the above reasons, it is possible that `CompactionIterator` decides to output an uncommitted key without
doing further checks on the key (`NextFromInput()`). By the time the key is being prepared for output, the key becomes
committed because the `snapshot_checker_(seq, kMaxSequence)` becomes true in the implementation of `KeyCommitted()`.
Then `CompactionIterator` will try to zero its sequence number and hit assertion error if the key is a tombstone.
To fix this issue, we should make the `CompactionIterator` see a consistent view of the input keys. Note that
for write-prepared/write-unprepared, the background flush/compaction jobs already take a "job snapshot" before starting
processing keys. The job snapshot is released only after the entire flush/compaction finishes. We can use this snapshot
to determine whether a key is committed or not with minor change to `KeyCommitted()`.
```
inline bool KeyCommitted(SequenceNumber sequence) {
// For non-txn-db and write-committed, snapshot_checker_ is always nullptr.
return snapshot_checker_ == nullptr ||
snapshot_checker_->CheckInSnapshot(sequence, job_snapshot_) ==
SnapshotCheckerResult::kInSnapshot;
}
```
As a result, whether a key is committed or not will remain a constant throughout compaction, causing no trouble
for `CompactionIterator`s assertions.
Pull Request resolved: https://github.com/facebook/rocksdb/pull/9830
Test Plan: make check
Reviewed By: ltamasi
Differential Revision: D35561162
Pulled By: riversand963
fbshipit-source-id: 0e00d200c195240341cfe6d34cbc86798b315b9f
3 years ago
|
|
|
const SnapshotChecker* snapshot_checker, JobContext* job_context,
|
|
|
|
std::shared_ptr<Cache> table_cache, EventLogger* event_logger,
|
|
|
|
bool paranoid_file_checks, bool measure_io_stats,
|
|
|
|
const std::string& dbname, CompactionJobStats* compaction_job_stats,
|
|
|
|
Env::Priority thread_pri, const std::shared_ptr<IOTracer>& io_tracer,
|
|
|
|
const std::atomic<bool>& manual_compaction_canceled,
|
|
|
|
const std::string& db_id = "", const std::string& db_session_id = "",
|
|
|
|
std::string full_history_ts_low = "", std::string trim_ts = "",
|
|
|
|
BlobFileCompletionCallback* blob_callback = nullptr);
|
|
|
|
|
|
|
|
virtual ~CompactionJob();
|
|
|
|
|
|
|
|
// no copy/move
|
|
|
|
CompactionJob(CompactionJob&& job) = delete;
|
|
|
|
CompactionJob(const CompactionJob& job) = delete;
|
|
|
|
CompactionJob& operator=(const CompactionJob& job) = delete;
|
|
|
|
|
|
|
|
// REQUIRED: mutex held
|
|
|
|
// Prepare for the compaction by setting up boundaries for each subcompaction
|
|
|
|
void Prepare();
|
|
|
|
// REQUIRED mutex not held
|
|
|
|
// Launch threads for each subcompaction and wait for them to finish. After
|
|
|
|
// that, verify table is usable and finally do bookkeeping to unify
|
|
|
|
// subcompaction results
|
|
|
|
Status Run();
|
Parallelize L0-L1 Compaction: Restructure Compaction Job
Summary:
As of now compactions involving files from Level 0 and Level 1 are single
threaded because the files in L0, although sorted, are not range partitioned like
the other levels. This means that during L0-L1 compaction each file from L1
needs to be merged with potentially all the files from L0.
This attempt to parallelize the L0-L1 compaction assigns a thread and a
corresponding iterator to each L1 file that then considers only the key range
found in that L1 file and only the L0 files that have those keys (and only the
specific portion of those L0 files in which those keys are found). In this way
the overlap is minimized and potentially eliminated between different iterators
focusing on the same files.
The first step is to restructure the compaction logic to break L0-L1 compactions
into multiple, smaller, sequential compactions. Eventually each of these smaller
jobs will be run simultaneously. Areas to pay extra attention to are
# Correct aggregation of compaction job statistics across multiple threads
# Proper opening/closing of output files (make sure each thread's is unique)
# Keys that span multiple L1 files
# Skewed distributions of keys within L0 files
Test Plan: Make and run db_test (newer version has separate compaction tests) and compaction_job_stats_test
Reviewers: igor, noetzli, anthony, sdong, yhchiang
Reviewed By: yhchiang
Subscribers: MarkCallaghan, dhruba, leveldb
Differential Revision: https://reviews.facebook.net/D42699
10 years ago
|
|
|
|
|
|
|
// REQUIRED: mutex held
|
|
|
|
// Add compaction input/output to the current version
|
|
|
|
Status Install(const MutableCFOptions& mutable_cf_options);
|
|
|
|
|
Pass IOStatus to write path and set retryable IO Error as hard error in BG jobs (#6487)
Summary:
In the current code base, we use Status to get and store the returned status from the call. Specifically, for IO related functions, the current Status cannot reflect the IO Error details such as error scope, error retryable attribute, and others. With the implementation of https://github.com/facebook/rocksdb/issues/5761, we have the new Wrapper for IO, which returns IOStatus instead of Status. However, the IOStatus is purged at the lower level of write path and transferred to Status.
The first job of this PR is to pass the IOStatus to the write path (flush, WAL write, and Compaction). The second job is to identify the Retryable IO Error as HardError, and set the bg_error_ as HardError. In this case, the DB Instance becomes read only. User is informed of the Status and need to take actions to deal with it (e.g., call db->Resume()).
Pull Request resolved: https://github.com/facebook/rocksdb/pull/6487
Test Plan: Added the testing case to error_handler_fs_test. Pass make asan_check
Reviewed By: anand1976
Differential Revision: D20685017
Pulled By: zhichao-cao
fbshipit-source-id: ff85f042896243abcd6ef37877834e26f36b6eb0
5 years ago
|
|
|
// Return the IO status
|
|
|
|
IOStatus io_status() const { return io_status_; }
|
|
|
|
|
|
|
|
protected:
|
|
|
|
struct SubcompactionState;
|
|
|
|
// CompactionJob state
|
|
|
|
struct CompactionState;
|
|
|
|
|
|
|
|
void AggregateStatistics();
|
|
|
|
void UpdateCompactionStats();
|
|
|
|
void LogCompaction();
|
|
|
|
virtual void RecordCompactionIOStats();
|
|
|
|
void CleanupCompaction();
|
|
|
|
|
|
|
|
// Call compaction filter. Then iterate through input and compact the
|
|
|
|
// kv-pairs
|
|
|
|
void ProcessKeyValueCompaction(SubcompactionState* sub_compact);
|
|
|
|
|
|
|
|
CompactionState* compact_;
|
|
|
|
InternalStats::CompactionStats compaction_stats_;
|
|
|
|
const ImmutableDBOptions& db_options_;
|
|
|
|
const MutableDBOptions mutable_db_options_copy_;
|
|
|
|
LogBuffer* log_buffer_;
|
|
|
|
FSDirectory* output_directory_;
|
|
|
|
Statistics* stats_;
|
|
|
|
// Is this compaction creating a file in the bottom most level?
|
|
|
|
bool bottommost_level_;
|
|
|
|
|
|
|
|
Env::WriteLifeTimeHint write_hint_;
|
|
|
|
|
|
|
|
IOStatus io_status_;
|
|
|
|
|
|
|
|
private:
|
Set Write rate limiter priority dynamically and pass it to FS (#9988)
Summary:
### Context:
Background compactions and flush generate large reads and writes, and can be long running, especially for universal compaction. In some cases, this can impact foreground reads and writes by users.
From the RocksDB perspective, there can be two kinds of rate limiters, the internal (native) one and the external one.
- The internal (native) rate limiter is introduced in [the wiki](https://github.com/facebook/rocksdb/wiki/Rate-Limiter). Currently, only IO_LOW and IO_HIGH are used and they are set statically.
- For the external rate limiter, in FSWritableFile functions, IOOptions is open for end users to set and get rate_limiter_priority for their own rate limiter. Currently, RocksDB doesn’t pass the rate_limiter_priority through IOOptions to the file system.
### Solution
During the User Read, Flush write, Compaction read/write, the WriteController is used to determine whether DB writes are stalled or slowed down. The rate limiter priority (Env::IOPriority) can be determined accordingly. We decided to always pass the priority in IOOptions. What the file system does with it should be a contract between the user and the file system. We would like to set the rate limiter priority at file level, since the Flush/Compaction job level may be too coarse with multiple files and block IO level is too granular.
**This PR is for the Write path.** The **Write:** dynamic priority for different state are listed as follows:
| State | Normal | Delayed | Stalled |
| ----- | ------ | ------- | ------- |
| Flush | IO_HIGH | IO_USER | IO_USER |
| Compaction | IO_LOW | IO_USER | IO_USER |
Flush and Compaction writes share the same call path through BlockBaseTableWriter, WritableFileWriter, and FSWritableFile. When a new FSWritableFile object is created, its io_priority_ can be set dynamically based on the state of the WriteController. In WritableFileWriter, before the call sites of FSWritableFile functions, WritableFileWriter::DecideRateLimiterPriority() determines the rate_limiter_priority. The options (IOOptions) argument of FSWritableFile functions will be updated with the rate_limiter_priority.
Pull Request resolved: https://github.com/facebook/rocksdb/pull/9988
Test Plan: Add unit tests.
Reviewed By: anand1976
Differential Revision: D36395159
Pulled By: gitbw95
fbshipit-source-id: a7c82fc29759139a1a07ec46c37dbf7e753474cf
3 years ago
|
|
|
friend class CompactionJobTestBase;
|
|
|
|
|
|
|
|
// Generates a histogram representing potential divisions of key ranges from
|
|
|
|
// the input. It adds the starting and/or ending keys of certain input files
|
|
|
|
// to the working set and then finds the approximate size of data in between
|
|
|
|
// each consecutive pair of slices. Then it divides these ranges into
|
|
|
|
// consecutive groups such that each group has a similar size.
|
|
|
|
void GenSubcompactionBoundaries();
|
Parallelize L0-L1 Compaction: Restructure Compaction Job
Summary:
As of now compactions involving files from Level 0 and Level 1 are single
threaded because the files in L0, although sorted, are not range partitioned like
the other levels. This means that during L0-L1 compaction each file from L1
needs to be merged with potentially all the files from L0.
This attempt to parallelize the L0-L1 compaction assigns a thread and a
corresponding iterator to each L1 file that then considers only the key range
found in that L1 file and only the L0 files that have those keys (and only the
specific portion of those L0 files in which those keys are found). In this way
the overlap is minimized and potentially eliminated between different iterators
focusing on the same files.
The first step is to restructure the compaction logic to break L0-L1 compactions
into multiple, smaller, sequential compactions. Eventually each of these smaller
jobs will be run simultaneously. Areas to pay extra attention to are
# Correct aggregation of compaction job statistics across multiple threads
# Proper opening/closing of output files (make sure each thread's is unique)
# Keys that span multiple L1 files
# Skewed distributions of keys within L0 files
Test Plan: Make and run db_test (newer version has separate compaction tests) and compaction_job_stats_test
Reviewers: igor, noetzli, anthony, sdong, yhchiang
Reviewed By: yhchiang
Subscribers: MarkCallaghan, dhruba, leveldb
Differential Revision: https://reviews.facebook.net/D42699
10 years ago
|
|
|
|
|
|
|
CompactionServiceJobStatus ProcessKeyValueCompactionWithCompactionService(
|
|
|
|
SubcompactionState* sub_compact);
|
|
|
|
|
|
|
|
// update the thread status for starting a compaction.
|
|
|
|
void ReportStartedCompaction(Compaction* compaction);
|
|
|
|
void AllocateCompactionOutputFileNumbers();
|
|
|
|
|
|
|
|
Status FinishCompactionOutputFile(
|
|
|
|
const Status& input_status, SubcompactionState* sub_compact,
|
|
|
|
CompactionRangeDelAggregator* range_del_agg,
|
|
|
|
CompactionIterationStats* range_del_out_stats,
|
|
|
|
const Slice* next_table_min_key = nullptr);
|
|
|
|
Status InstallCompactionResults(const MutableCFOptions& mutable_cf_options);
|
|
|
|
Status OpenCompactionOutputFile(SubcompactionState* sub_compact);
|
|
|
|
void UpdateCompactionJobStats(
|
|
|
|
const InternalStats::CompactionStats& stats) const;
|
|
|
|
void RecordDroppedKeys(const CompactionIterationStats& c_iter_stats,
|
|
|
|
CompactionJobStats* compaction_job_stats = nullptr);
|
|
|
|
|
|
|
|
void UpdateCompactionInputStatsHelper(
|
|
|
|
int* num_files, uint64_t* bytes_read, int input_level);
|
|
|
|
|
|
|
|
#ifndef ROCKSDB_LITE
|
|
|
|
void BuildSubcompactionJobInfo(
|
|
|
|
SubcompactionState* sub_compact,
|
|
|
|
SubcompactionJobInfo* subcompaction_job_info) const;
|
|
|
|
#endif // ROCKSDB_LITE
|
|
|
|
|
|
|
|
void NotifyOnSubcompactionBegin(SubcompactionState* sub_compact);
|
|
|
|
|
|
|
|
void NotifyOnSubcompactionCompleted(SubcompactionState* sub_compact);
|
|
|
|
|
|
|
|
uint32_t job_id_;
|
|
|
|
|
|
|
|
CompactionJobStats* compaction_job_stats_;
|
|
|
|
|
|
|
|
// DBImpl state
|
|
|
|
const std::string& dbname_;
|
|
|
|
const std::string db_id_;
|
|
|
|
const std::string db_session_id_;
|
Introduce a new storage specific Env API (#5761)
Summary:
The current Env API encompasses both storage/file operations, as well as OS related operations. Most of the APIs return a Status, which does not have enough metadata about an error, such as whether its retry-able or not, scope (i.e fault domain) of the error etc., that may be required in order to properly handle a storage error. The file APIs also do not provide enough control over the IO SLA, such as timeout, prioritization, hinting about placement and redundancy etc.
This PR separates out the file/storage APIs from Env into a new FileSystem class. The APIs are updated to return an IOStatus with metadata about the error, as well as to take an IOOptions structure as input in order to allow more control over the IO.
The user can set both ```options.env``` and ```options.file_system``` to specify that RocksDB should use the former for OS related operations and the latter for storage operations. Internally, a ```CompositeEnvWrapper``` has been introduced that inherits from ```Env``` and redirects individual methods to either an ```Env``` implementation or the ```FileSystem``` as appropriate. When options are sanitized during ```DB::Open```, ```options.env``` is replaced with a newly allocated ```CompositeEnvWrapper``` instance if both env and file_system have been specified. This way, the rest of the RocksDB code can continue to function as before.
This PR also ports PosixEnv to the new API by splitting it into two - PosixEnv and PosixFileSystem. PosixEnv is defined as a sub-class of CompositeEnvWrapper, and threading/time functions are overridden with Posix specific implementations in order to avoid an extra level of indirection.
The ```CompositeEnvWrapper``` translates ```IOStatus``` return code to ```Status```, and sets the severity to ```kSoftError``` if the io_status is retryable. The error handling code in RocksDB can then recover the DB automatically.
Pull Request resolved: https://github.com/facebook/rocksdb/pull/5761
Differential Revision: D18868376
Pulled By: anand1976
fbshipit-source-id: 39efe18a162ea746fabac6360ff529baba48486f
5 years ago
|
|
|
const FileOptions file_options_;
|
|
|
|
|
|
|
|
Env* env_;
|
|
|
|
std::shared_ptr<IOTracer> io_tracer_;
|
|
|
|
FileSystemPtr fs_;
|
|
|
|
// env_option optimized for compaction table reads
|
Introduce a new storage specific Env API (#5761)
Summary:
The current Env API encompasses both storage/file operations, as well as OS related operations. Most of the APIs return a Status, which does not have enough metadata about an error, such as whether its retry-able or not, scope (i.e fault domain) of the error etc., that may be required in order to properly handle a storage error. The file APIs also do not provide enough control over the IO SLA, such as timeout, prioritization, hinting about placement and redundancy etc.
This PR separates out the file/storage APIs from Env into a new FileSystem class. The APIs are updated to return an IOStatus with metadata about the error, as well as to take an IOOptions structure as input in order to allow more control over the IO.
The user can set both ```options.env``` and ```options.file_system``` to specify that RocksDB should use the former for OS related operations and the latter for storage operations. Internally, a ```CompositeEnvWrapper``` has been introduced that inherits from ```Env``` and redirects individual methods to either an ```Env``` implementation or the ```FileSystem``` as appropriate. When options are sanitized during ```DB::Open```, ```options.env``` is replaced with a newly allocated ```CompositeEnvWrapper``` instance if both env and file_system have been specified. This way, the rest of the RocksDB code can continue to function as before.
This PR also ports PosixEnv to the new API by splitting it into two - PosixEnv and PosixFileSystem. PosixEnv is defined as a sub-class of CompositeEnvWrapper, and threading/time functions are overridden with Posix specific implementations in order to avoid an extra level of indirection.
The ```CompositeEnvWrapper``` translates ```IOStatus``` return code to ```Status```, and sets the severity to ```kSoftError``` if the io_status is retryable. The error handling code in RocksDB can then recover the DB automatically.
Pull Request resolved: https://github.com/facebook/rocksdb/pull/5761
Differential Revision: D18868376
Pulled By: anand1976
fbshipit-source-id: 39efe18a162ea746fabac6360ff529baba48486f
5 years ago
|
|
|
FileOptions file_options_for_read_;
|
|
|
|
VersionSet* versions_;
|
|
|
|
const std::atomic<bool>* shutting_down_;
|
|
|
|
const std::atomic<bool>& manual_compaction_canceled_;
|
|
|
|
FSDirectory* db_directory_;
|
|
|
|
FSDirectory* blob_output_directory_;
|
|
|
|
InstrumentedMutex* db_mutex_;
|
|
|
|
ErrorHandler* db_error_handler_;
|
|
|
|
// If there were two snapshots with seq numbers s1 and
|
|
|
|
// s2 and s1 < s2, and if we find two instances of a key k1 then lies
|
|
|
|
// entirely within s1 and s2, then the earlier version of k1 can be safely
|
|
|
|
// deleted because that version is not visible in any snapshot.
|
|
|
|
std::vector<SequenceNumber> existing_snapshots_;
|
|
|
|
|
|
|
|
// This is the earliest snapshot that could be used for write-conflict
|
|
|
|
// checking by a transaction. For any user-key newer than this snapshot, we
|
|
|
|
// should make sure not to remove evidence that a write occurred.
|
|
|
|
SequenceNumber earliest_write_conflict_snapshot_;
|
|
|
|
|
|
|
|
const SnapshotChecker* const snapshot_checker_;
|
|
|
|
|
CompactionIterator sees consistent view of which keys are committed (#9830)
Summary:
**This PR does not affect the functionality of `DB` and write-committed transactions.**
`CompactionIterator` uses `KeyCommitted(seq)` to determine if a key in the database is committed.
As the name 'write-committed' implies, if write-committed policy is used, a key exists in the database only if
it is committed. In fact, the implementation of `KeyCommitted()` is as follows:
```
inline bool KeyCommitted(SequenceNumber seq) {
// For non-txn-db and write-committed, snapshot_checker_ is always nullptr.
return snapshot_checker_ == nullptr ||
snapshot_checker_->CheckInSnapshot(seq, kMaxSequence) == SnapshotCheckerResult::kInSnapshot;
}
```
With that being said, we focus on write-prepared/write-unprepared transactions.
A few notes:
- A key can exist in the db even if it's uncommitted. Therefore, we rely on `snapshot_checker_` to determine data visibility. We also require that all writes go through transaction API instead of the raw `WriteBatch` + `Write`, thus at most one uncommitted version of one user key can exist in the database.
- `CompactionIterator` outputs a key as long as the key is uncommitted.
Due to the above reasons, it is possible that `CompactionIterator` decides to output an uncommitted key without
doing further checks on the key (`NextFromInput()`). By the time the key is being prepared for output, the key becomes
committed because the `snapshot_checker_(seq, kMaxSequence)` becomes true in the implementation of `KeyCommitted()`.
Then `CompactionIterator` will try to zero its sequence number and hit assertion error if the key is a tombstone.
To fix this issue, we should make the `CompactionIterator` see a consistent view of the input keys. Note that
for write-prepared/write-unprepared, the background flush/compaction jobs already take a "job snapshot" before starting
processing keys. The job snapshot is released only after the entire flush/compaction finishes. We can use this snapshot
to determine whether a key is committed or not with minor change to `KeyCommitted()`.
```
inline bool KeyCommitted(SequenceNumber sequence) {
// For non-txn-db and write-committed, snapshot_checker_ is always nullptr.
return snapshot_checker_ == nullptr ||
snapshot_checker_->CheckInSnapshot(sequence, job_snapshot_) ==
SnapshotCheckerResult::kInSnapshot;
}
```
As a result, whether a key is committed or not will remain a constant throughout compaction, causing no trouble
for `CompactionIterator`s assertions.
Pull Request resolved: https://github.com/facebook/rocksdb/pull/9830
Test Plan: make check
Reviewed By: ltamasi
Differential Revision: D35561162
Pulled By: riversand963
fbshipit-source-id: 0e00d200c195240341cfe6d34cbc86798b315b9f
3 years ago
|
|
|
JobContext* job_context_;
|
|
|
|
|
|
|
|
std::shared_ptr<Cache> table_cache_;
|
|
|
|
|
Include bunch of more events into EventLogger
Summary:
Added these events:
* Recovery start, finish and also when recovery creates a file
* Trivial move
* Compaction start, finish and when compaction creates a file
* Flush start, finish
Also includes small fix to EventLogger
Also added option ROCKSDB_PRINT_EVENTS_TO_STDOUT which is useful when we debug things. I've spent far too much time chasing LOG files.
Still didn't get sst table properties in JSON. They are written very deeply into the stack. I'll address in separate diff.
TODO:
* Write specification. Let's first use this for a while and figure out what's good data to put here, too. After that we'll write spec
* Write tools that parse and analyze LOGs. This can be in python or go. Good intern task.
Test Plan: Ran db_bench with ROCKSDB_PRINT_EVENTS_TO_STDOUT. Here's the output: https://phabricator.fb.com/P19811976
Reviewers: sdong, yhchiang, rven, MarkCallaghan, kradhakrishnan, anthony
Reviewed By: anthony
Subscribers: dhruba, leveldb
Differential Revision: https://reviews.facebook.net/D37521
10 years ago
|
|
|
EventLogger* event_logger_;
|
|
|
|
|
|
|
|
bool paranoid_file_checks_;
|
Add options.compaction_measure_io_stats to print write I/O stats in compactions
Summary:
Add options.compaction_measure_io_stats to print out / pass to listener accumulated time spent on write calls. Example outputs in info logs:
2015/08/12-16:27:59.463944 7fd428bff700 (Original Log Time 2015/08/12-16:27:59.463922) EVENT_LOG_v1 {"time_micros": 1439422079463897, "job": 6, "event": "compaction_finished", "output_level": 1, "num_output_files": 4, "total_output_size": 6900525, "num_input_records": 111483, "num_output_records": 106877, "file_write_nanos": 15663206, "file_range_sync_nanos": 649588, "file_fsync_nanos": 349614797, "file_prepare_write_nanos": 1505812, "lsm_state": [2, 4, 0, 0, 0, 0, 0]}
Add two more counters in iostats_context.
Also add a parameter of db_bench.
Test Plan: Add a unit test. Also manually verify LOG outputs in db_bench
Subscribers: leveldb, dhruba
Differential Revision: https://reviews.facebook.net/D44115
10 years ago
|
|
|
bool measure_io_stats_;
|
|
|
|
// Stores the Slices that designate the boundaries for each subcompaction
|
|
|
|
std::vector<Slice> boundaries_;
|
|
|
|
// Stores the approx size of keys covered in the range of each subcompaction
|
|
|
|
std::vector<uint64_t> sizes_;
|
|
|
|
Env::Priority thread_pri_;
|
|
|
|
std::string full_history_ts_low_;
|
|
|
|
std::string trim_ts_;
|
|
|
|
BlobFileCompletionCallback* blob_callback_;
|
|
|
|
|
|
|
|
uint64_t GetCompactionId(SubcompactionState* sub_compact);
|
|
|
|
|
|
|
|
// Get table file name in where it's outputting to, which should also be in
|
|
|
|
// `output_directory_`.
|
|
|
|
virtual std::string GetTableFileName(uint64_t file_number);
|
Set Write rate limiter priority dynamically and pass it to FS (#9988)
Summary:
### Context:
Background compactions and flush generate large reads and writes, and can be long running, especially for universal compaction. In some cases, this can impact foreground reads and writes by users.
From the RocksDB perspective, there can be two kinds of rate limiters, the internal (native) one and the external one.
- The internal (native) rate limiter is introduced in [the wiki](https://github.com/facebook/rocksdb/wiki/Rate-Limiter). Currently, only IO_LOW and IO_HIGH are used and they are set statically.
- For the external rate limiter, in FSWritableFile functions, IOOptions is open for end users to set and get rate_limiter_priority for their own rate limiter. Currently, RocksDB doesn’t pass the rate_limiter_priority through IOOptions to the file system.
### Solution
During the User Read, Flush write, Compaction read/write, the WriteController is used to determine whether DB writes are stalled or slowed down. The rate limiter priority (Env::IOPriority) can be determined accordingly. We decided to always pass the priority in IOOptions. What the file system does with it should be a contract between the user and the file system. We would like to set the rate limiter priority at file level, since the Flush/Compaction job level may be too coarse with multiple files and block IO level is too granular.
**This PR is for the Write path.** The **Write:** dynamic priority for different state are listed as follows:
| State | Normal | Delayed | Stalled |
| ----- | ------ | ------- | ------- |
| Flush | IO_HIGH | IO_USER | IO_USER |
| Compaction | IO_LOW | IO_USER | IO_USER |
Flush and Compaction writes share the same call path through BlockBaseTableWriter, WritableFileWriter, and FSWritableFile. When a new FSWritableFile object is created, its io_priority_ can be set dynamically based on the state of the WriteController. In WritableFileWriter, before the call sites of FSWritableFile functions, WritableFileWriter::DecideRateLimiterPriority() determines the rate_limiter_priority. The options (IOOptions) argument of FSWritableFile functions will be updated with the rate_limiter_priority.
Pull Request resolved: https://github.com/facebook/rocksdb/pull/9988
Test Plan: Add unit tests.
Reviewed By: anand1976
Differential Revision: D36395159
Pulled By: gitbw95
fbshipit-source-id: a7c82fc29759139a1a07ec46c37dbf7e753474cf
3 years ago
|
|
|
// The rate limiter priority (io_priority) is determined dynamically here.
|
|
|
|
// The Compaction Read and Write priorities are the same for different
|
|
|
|
// scenarios, such as write stalled.
|
|
|
|
Env::IOPriority GetRateLimiterPriority();
|
|
|
|
};
|
|
|
|
|
|
|
|
// CompactionServiceInput is used the pass compaction information between two
|
|
|
|
// db instances. It contains the information needed to do a compaction. It
|
|
|
|
// doesn't contain the LSM tree information, which is passed though MANIFEST
|
|
|
|
// file.
|
|
|
|
struct CompactionServiceInput {
|
|
|
|
ColumnFamilyDescriptor column_family;
|
|
|
|
|
|
|
|
DBOptions db_options;
|
|
|
|
|
|
|
|
std::vector<SequenceNumber> snapshots;
|
|
|
|
|
|
|
|
// SST files for compaction, it should already be expended to include all the
|
|
|
|
// files needed for this compaction, for both input level files and output
|
|
|
|
// level files.
|
|
|
|
std::vector<std::string> input_files;
|
|
|
|
int output_level;
|
|
|
|
|
|
|
|
// db_id is used to generate unique id of sst on the remote compactor
|
|
|
|
std::string db_id;
|
|
|
|
|
|
|
|
// information for subcompaction
|
|
|
|
bool has_begin = false;
|
|
|
|
std::string begin;
|
|
|
|
bool has_end = false;
|
|
|
|
std::string end;
|
|
|
|
uint64_t approx_size = 0;
|
|
|
|
|
|
|
|
// serialization interface to read and write the object
|
|
|
|
static Status Read(const std::string& data_str, CompactionServiceInput* obj);
|
|
|
|
Status Write(std::string* output);
|
|
|
|
|
|
|
|
// Initialize a dummy ColumnFamilyDescriptor
|
|
|
|
CompactionServiceInput() : column_family("", ColumnFamilyOptions()) {}
|
|
|
|
|
|
|
|
#ifndef NDEBUG
|
|
|
|
bool TEST_Equals(CompactionServiceInput* other);
|
|
|
|
bool TEST_Equals(CompactionServiceInput* other, std::string* mismatch);
|
|
|
|
#endif // NDEBUG
|
|
|
|
};
|
|
|
|
|
|
|
|
// CompactionServiceOutputFile is the metadata for the output SST file
|
|
|
|
struct CompactionServiceOutputFile {
|
|
|
|
std::string file_name;
|
|
|
|
SequenceNumber smallest_seqno;
|
|
|
|
SequenceNumber largest_seqno;
|
|
|
|
std::string smallest_internal_key;
|
|
|
|
std::string largest_internal_key;
|
|
|
|
uint64_t oldest_ancester_time;
|
|
|
|
uint64_t file_creation_time;
|
|
|
|
uint64_t paranoid_hash;
|
|
|
|
bool marked_for_compaction;
|
|
|
|
UniqueId64x2 unique_id;
|
|
|
|
|
|
|
|
CompactionServiceOutputFile() = default;
|
|
|
|
CompactionServiceOutputFile(
|
|
|
|
const std::string& name, SequenceNumber smallest, SequenceNumber largest,
|
|
|
|
std::string _smallest_internal_key, std::string _largest_internal_key,
|
|
|
|
uint64_t _oldest_ancester_time, uint64_t _file_creation_time,
|
|
|
|
uint64_t _paranoid_hash, bool _marked_for_compaction,
|
|
|
|
UniqueId64x2 _unique_id)
|
|
|
|
: file_name(name),
|
|
|
|
smallest_seqno(smallest),
|
|
|
|
largest_seqno(largest),
|
|
|
|
smallest_internal_key(std::move(_smallest_internal_key)),
|
|
|
|
largest_internal_key(std::move(_largest_internal_key)),
|
|
|
|
oldest_ancester_time(_oldest_ancester_time),
|
|
|
|
file_creation_time(_file_creation_time),
|
|
|
|
paranoid_hash(_paranoid_hash),
|
|
|
|
marked_for_compaction(_marked_for_compaction),
|
|
|
|
unique_id(std::move(_unique_id)) {}
|
|
|
|
};
|
|
|
|
|
|
|
|
// CompactionServiceResult contains the compaction result from a different db
|
|
|
|
// instance, with these information, the primary db instance with write
|
|
|
|
// permission is able to install the result to the DB.
|
|
|
|
struct CompactionServiceResult {
|
|
|
|
Status status;
|
|
|
|
std::vector<CompactionServiceOutputFile> output_files;
|
|
|
|
int output_level;
|
|
|
|
|
|
|
|
// location of the output files
|
|
|
|
std::string output_path;
|
|
|
|
|
|
|
|
// some statistics about the compaction
|
|
|
|
uint64_t num_output_records = 0;
|
|
|
|
uint64_t total_bytes = 0;
|
|
|
|
uint64_t bytes_read = 0;
|
|
|
|
uint64_t bytes_written = 0;
|
|
|
|
CompactionJobStats stats;
|
|
|
|
|
|
|
|
// serialization interface to read and write the object
|
|
|
|
static Status Read(const std::string& data_str, CompactionServiceResult* obj);
|
|
|
|
Status Write(std::string* output);
|
|
|
|
|
|
|
|
#ifndef NDEBUG
|
|
|
|
bool TEST_Equals(CompactionServiceResult* other);
|
|
|
|
bool TEST_Equals(CompactionServiceResult* other, std::string* mismatch);
|
|
|
|
#endif // NDEBUG
|
|
|
|
};
|
|
|
|
|
|
|
|
// CompactionServiceCompactionJob is an read-only compaction job, it takes
|
|
|
|
// input information from `compaction_service_input` and put result information
|
|
|
|
// in `compaction_service_result`, the SST files are generated to `output_path`.
|
|
|
|
class CompactionServiceCompactionJob : private CompactionJob {
|
|
|
|
public:
|
|
|
|
CompactionServiceCompactionJob(
|
|
|
|
int job_id, Compaction* compaction, const ImmutableDBOptions& db_options,
|
|
|
|
const MutableDBOptions& mutable_db_options,
|
|
|
|
const FileOptions& file_options, VersionSet* versions,
|
|
|
|
const std::atomic<bool>* shutting_down, LogBuffer* log_buffer,
|
|
|
|
FSDirectory* output_directory, Statistics* stats,
|
|
|
|
InstrumentedMutex* db_mutex, ErrorHandler* db_error_handler,
|
|
|
|
std::vector<SequenceNumber> existing_snapshots,
|
|
|
|
std::shared_ptr<Cache> table_cache, EventLogger* event_logger,
|
|
|
|
const std::string& dbname, const std::shared_ptr<IOTracer>& io_tracer,
|
|
|
|
const std::atomic<bool>& manual_compaction_canceled,
|
|
|
|
const std::string& db_id, const std::string& db_session_id,
|
|
|
|
const std::string& output_path,
|
|
|
|
const CompactionServiceInput& compaction_service_input,
|
|
|
|
CompactionServiceResult* compaction_service_result);
|
|
|
|
|
|
|
|
// Run the compaction in current thread and return the result
|
|
|
|
Status Run();
|
|
|
|
|
|
|
|
void CleanupCompaction();
|
|
|
|
|
|
|
|
IOStatus io_status() const { return CompactionJob::io_status(); }
|
|
|
|
|
|
|
|
protected:
|
|
|
|
void RecordCompactionIOStats() override;
|
|
|
|
|
|
|
|
private:
|
|
|
|
// Get table file name in output_path
|
|
|
|
std::string GetTableFileName(uint64_t file_number) override;
|
|
|
|
// Specific the compaction output path, otherwise it uses default DB path
|
|
|
|
const std::string output_path_;
|
|
|
|
|
|
|
|
// Compaction job input
|
|
|
|
const CompactionServiceInput& compaction_input_;
|
|
|
|
|
|
|
|
// Compaction job result
|
|
|
|
CompactionServiceResult* compaction_result_;
|
|
|
|
};
|
|
|
|
|
|
|
|
} // namespace ROCKSDB_NAMESPACE
|