--- /dev/null
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+/*
+ * Ceph - scalable distributed file system
+ *
+ * Copyright (C) 2014 Red Hat
+ *
+ * This is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Lesser General Public
+ * License version 2.1, as published by the Free Software
+ * Foundation. See file COPYING.
+ *
+ */
+
+#ifndef CEPH_OSD_BLUESTORE_H
+#define CEPH_OSD_BLUESTORE_H
+
+#include "acconfig.h"
+
+#include <unistd.h>
+
+#include <atomic>
+#include <mutex>
+#include <condition_variable>
+
+#include <boost/intrusive/list.hpp>
+#include <boost/intrusive/unordered_set.hpp>
+#include <boost/intrusive/set.hpp>
+#include <boost/functional/hash.hpp>
+#include <boost/dynamic_bitset.hpp>
+
+#include "include/assert.h"
+#include "include/unordered_map.h"
+#include "include/memory.h"
+#include "include/mempool.h"
+#include "common/Finisher.h"
+#include "common/perf_counters.h"
+#include "compressor/Compressor.h"
+#include "os/ObjectStore.h"
+
+#include "bluestore_types.h"
+#include "BlockDevice.h"
+#include "common/EventTrace.h"
+
+class Allocator;
+class FreelistManager;
+class BlueFS;
+
+//#define DEBUG_CACHE
+//#define DEBUG_DEFERRED
+
+
+
+// constants for Buffer::optimize()
+#define MAX_BUFFER_SLOP_RATIO_DEN 8 // so actually 1/N
+
+
+enum {
+ l_bluestore_first = 732430,
+ l_bluestore_kv_flush_lat,
+ l_bluestore_kv_commit_lat,
+ l_bluestore_kv_lat,
+ l_bluestore_state_prepare_lat,
+ l_bluestore_state_aio_wait_lat,
+ l_bluestore_state_io_done_lat,
+ l_bluestore_state_kv_queued_lat,
+ l_bluestore_state_kv_committing_lat,
+ l_bluestore_state_kv_done_lat,
+ l_bluestore_state_deferred_queued_lat,
+ l_bluestore_state_deferred_aio_wait_lat,
+ l_bluestore_state_deferred_cleanup_lat,
+ l_bluestore_state_finishing_lat,
+ l_bluestore_state_done_lat,
+ l_bluestore_throttle_lat,
+ l_bluestore_submit_lat,
+ l_bluestore_commit_lat,
+ l_bluestore_read_lat,
+ l_bluestore_read_onode_meta_lat,
+ l_bluestore_read_wait_aio_lat,
+ l_bluestore_compress_lat,
+ l_bluestore_decompress_lat,
+ l_bluestore_csum_lat,
+ l_bluestore_compress_success_count,
+ l_bluestore_compress_rejected_count,
+ l_bluestore_write_pad_bytes,
+ l_bluestore_deferred_write_ops,
+ l_bluestore_deferred_write_bytes,
+ l_bluestore_write_penalty_read_ops,
+ l_bluestore_allocated,
+ l_bluestore_stored,
+ l_bluestore_compressed,
+ l_bluestore_compressed_allocated,
+ l_bluestore_compressed_original,
+ l_bluestore_onodes,
+ l_bluestore_onode_hits,
+ l_bluestore_onode_misses,
+ l_bluestore_onode_shard_hits,
+ l_bluestore_onode_shard_misses,
+ l_bluestore_extents,
+ l_bluestore_blobs,
+ l_bluestore_buffers,
+ l_bluestore_buffer_bytes,
+ l_bluestore_buffer_hit_bytes,
+ l_bluestore_buffer_miss_bytes,
+ l_bluestore_write_big,
+ l_bluestore_write_big_bytes,
+ l_bluestore_write_big_blobs,
+ l_bluestore_write_small,
+ l_bluestore_write_small_bytes,
+ l_bluestore_write_small_unused,
+ l_bluestore_write_small_deferred,
+ l_bluestore_write_small_pre_read,
+ l_bluestore_write_small_new,
+ l_bluestore_txc,
+ l_bluestore_onode_reshard,
+ l_bluestore_blob_split,
+ l_bluestore_extent_compress,
+ l_bluestore_gc_merged,
+ l_bluestore_last
+};
+
+class BlueStore : public ObjectStore,
+ public md_config_obs_t {
+ // -----------------------------------------------------
+ // types
+public:
+ // config observer
+ const char** get_tracked_conf_keys() const override;
+ void handle_conf_change(const struct md_config_t *conf,
+ const std::set<std::string> &changed) override;
+
+ void _set_csum();
+ void _set_compression();
+ void _set_throttle_params();
+ int _set_cache_sizes();
+
+ class TransContext;
+
+ typedef map<uint64_t, bufferlist> ready_regions_t;
+
+ struct BufferSpace;
+ struct Collection;
+ typedef boost::intrusive_ptr<Collection> CollectionRef;
+
+ struct AioContext {
+ virtual void aio_finish(BlueStore *store) = 0;
+ virtual ~AioContext() {}
+ };
+
+ /// cached buffer
+ struct Buffer {
+ MEMPOOL_CLASS_HELPERS();
+
+ enum {
+ STATE_EMPTY, ///< empty buffer -- used for cache history
+ STATE_CLEAN, ///< clean data that is up to date
+ STATE_WRITING, ///< data that is being written (io not yet complete)
+ };
+ static const char *get_state_name(int s) {
+ switch (s) {
+ case STATE_EMPTY: return "empty";
+ case STATE_CLEAN: return "clean";
+ case STATE_WRITING: return "writing";
+ default: return "???";
+ }
+ }
+ enum {
+ FLAG_NOCACHE = 1, ///< trim when done WRITING (do not become CLEAN)
+ // NOTE: fix operator<< when you define a second flag
+ };
+ static const char *get_flag_name(int s) {
+ switch (s) {
+ case FLAG_NOCACHE: return "nocache";
+ default: return "???";
+ }
+ }
+
+ BufferSpace *space;
+ uint16_t state; ///< STATE_*
+ uint16_t cache_private = 0; ///< opaque (to us) value used by Cache impl
+ uint32_t flags; ///< FLAG_*
+ uint64_t seq;
+ uint32_t offset, length;
+ bufferlist data;
+
+ boost::intrusive::list_member_hook<> lru_item;
+ boost::intrusive::list_member_hook<> state_item;
+
+ Buffer(BufferSpace *space, unsigned s, uint64_t q, uint32_t o, uint32_t l,
+ unsigned f = 0)
+ : space(space), state(s), flags(f), seq(q), offset(o), length(l) {}
+ Buffer(BufferSpace *space, unsigned s, uint64_t q, uint32_t o, bufferlist& b,
+ unsigned f = 0)
+ : space(space), state(s), flags(f), seq(q), offset(o),
+ length(b.length()), data(b) {}
+
+ bool is_empty() const {
+ return state == STATE_EMPTY;
+ }
+ bool is_clean() const {
+ return state == STATE_CLEAN;
+ }
+ bool is_writing() const {
+ return state == STATE_WRITING;
+ }
+
+ uint32_t end() const {
+ return offset + length;
+ }
+
+ void truncate(uint32_t newlen) {
+ assert(newlen < length);
+ if (data.length()) {
+ bufferlist t;
+ t.substr_of(data, 0, newlen);
+ data.claim(t);
+ }
+ length = newlen;
+ }
+ void maybe_rebuild() {
+ if (data.length() &&
+ (data.get_num_buffers() > 1 ||
+ data.front().wasted() > data.length() / MAX_BUFFER_SLOP_RATIO_DEN)) {
+ data.rebuild();
+ }
+ }
+
+ void dump(Formatter *f) const {
+ f->dump_string("state", get_state_name(state));
+ f->dump_unsigned("seq", seq);
+ f->dump_unsigned("offset", offset);
+ f->dump_unsigned("length", length);
+ f->dump_unsigned("data_length", data.length());
+ }
+ };
+
+ struct Cache;
+
+ /// map logical extent range (object) onto buffers
+ struct BufferSpace {
+ typedef boost::intrusive::list<
+ Buffer,
+ boost::intrusive::member_hook<
+ Buffer,
+ boost::intrusive::list_member_hook<>,
+ &Buffer::state_item> > state_list_t;
+
+ mempool::bluestore_cache_other::map<uint32_t, std::unique_ptr<Buffer>>
+ buffer_map;
+
+ // we use a bare intrusive list here instead of std::map because
+ // it uses less memory and we expect this to be very small (very
+ // few IOs in flight to the same Blob at the same time).
+ state_list_t writing; ///< writing buffers, sorted by seq, ascending
+
+ ~BufferSpace() {
+ assert(buffer_map.empty());
+ assert(writing.empty());
+ }
+
+ void _add_buffer(Cache* cache, Buffer *b, int level, Buffer *near) {
+ cache->_audit("_add_buffer start");
+ buffer_map[b->offset].reset(b);
+ if (b->is_writing()) {
+ b->data.reassign_to_mempool(mempool::mempool_bluestore_writing);
+ if (writing.empty() || writing.rbegin()->seq <= b->seq) {
+ writing.push_back(*b);
+ } else {
+ auto it = writing.begin();
+ while (it->seq < b->seq) {
+ ++it;
+ }
+
+ assert(it->seq >= b->seq);
+ // note that this will insert b before it
+ // hence the order is maintained
+ writing.insert(it, *b);
+ }
+ } else {
+ b->data.reassign_to_mempool(mempool::mempool_bluestore_cache_data);
+ cache->_add_buffer(b, level, near);
+ }
+ cache->_audit("_add_buffer end");
+ }
+ void _rm_buffer(Cache* cache, Buffer *b) {
+ _rm_buffer(cache, buffer_map.find(b->offset));
+ }
+ void _rm_buffer(Cache* cache,
+ map<uint32_t, std::unique_ptr<Buffer>>::iterator p) {
+ assert(p != buffer_map.end());
+ cache->_audit("_rm_buffer start");
+ if (p->second->is_writing()) {
+ writing.erase(writing.iterator_to(*p->second));
+ } else {
+ cache->_rm_buffer(p->second.get());
+ }
+ buffer_map.erase(p);
+ cache->_audit("_rm_buffer end");
+ }
+
+ map<uint32_t,std::unique_ptr<Buffer>>::iterator _data_lower_bound(
+ uint32_t offset) {
+ auto i = buffer_map.lower_bound(offset);
+ if (i != buffer_map.begin()) {
+ --i;
+ if (i->first + i->second->length <= offset)
+ ++i;
+ }
+ return i;
+ }
+
+ // must be called under protection of the Cache lock
+ void _clear(Cache* cache);
+
+ // return value is the highest cache_private of a trimmed buffer, or 0.
+ int discard(Cache* cache, uint32_t offset, uint32_t length) {
+ std::lock_guard<std::recursive_mutex> l(cache->lock);
+ return _discard(cache, offset, length);
+ }
+ int _discard(Cache* cache, uint32_t offset, uint32_t length);
+
+ void write(Cache* cache, uint64_t seq, uint32_t offset, bufferlist& bl,
+ unsigned flags) {
+ std::lock_guard<std::recursive_mutex> l(cache->lock);
+ Buffer *b = new Buffer(this, Buffer::STATE_WRITING, seq, offset, bl,
+ flags);
+ b->cache_private = _discard(cache, offset, bl.length());
+ _add_buffer(cache, b, (flags & Buffer::FLAG_NOCACHE) ? 0 : 1, nullptr);
+ }
+ void finish_write(Cache* cache, uint64_t seq);
+ void did_read(Cache* cache, uint32_t offset, bufferlist& bl) {
+ std::lock_guard<std::recursive_mutex> l(cache->lock);
+ Buffer *b = new Buffer(this, Buffer::STATE_CLEAN, 0, offset, bl);
+ b->cache_private = _discard(cache, offset, bl.length());
+ _add_buffer(cache, b, 1, nullptr);
+ }
+
+ void read(Cache* cache, uint32_t offset, uint32_t length,
+ BlueStore::ready_regions_t& res,
+ interval_set<uint32_t>& res_intervals);
+
+ void truncate(Cache* cache, uint32_t offset) {
+ discard(cache, offset, (uint32_t)-1 - offset);
+ }
+
+ void split(Cache* cache, size_t pos, BufferSpace &r);
+
+ void dump(Cache* cache, Formatter *f) const {
+ std::lock_guard<std::recursive_mutex> l(cache->lock);
+ f->open_array_section("buffers");
+ for (auto& i : buffer_map) {
+ f->open_object_section("buffer");
+ assert(i.first == i.second->offset);
+ i.second->dump(f);
+ f->close_section();
+ }
+ f->close_section();
+ }
+ };
+
+ struct SharedBlobSet;
+
+ /// in-memory shared blob state (incl cached buffers)
+ struct SharedBlob {
+ MEMPOOL_CLASS_HELPERS();
+
+ std::atomic_int nref = {0}; ///< reference count
+ bool loaded = false;
+
+ CollectionRef coll;
+ union {
+ uint64_t sbid_unloaded; ///< sbid if persistent isn't loaded
+ bluestore_shared_blob_t *persistent; ///< persistent part of the shared blob if any
+ };
+ BufferSpace bc; ///< buffer cache
+
+ SharedBlob(Collection *_coll) : coll(_coll), sbid_unloaded(0) {
+ if (get_cache()) {
+ get_cache()->add_blob();
+ }
+ }
+ SharedBlob(uint64_t i, Collection *_coll);
+ ~SharedBlob();
+
+ uint64_t get_sbid() const {
+ return loaded ? persistent->sbid : sbid_unloaded;
+ }
+
+ friend void intrusive_ptr_add_ref(SharedBlob *b) { b->get(); }
+ friend void intrusive_ptr_release(SharedBlob *b) { b->put(); }
+
+ friend ostream& operator<<(ostream& out, const SharedBlob& sb);
+
+ void get() {
+ ++nref;
+ }
+ void put();
+
+ /// get logical references
+ void get_ref(uint64_t offset, uint32_t length);
+
+ /// put logical references, and get back any released extents
+ void put_ref(uint64_t offset, uint32_t length,
+ PExtentVector *r, set<SharedBlob*> *maybe_unshared_blobs);
+
+ friend bool operator==(const SharedBlob &l, const SharedBlob &r) {
+ return l.get_sbid() == r.get_sbid();
+ }
+ inline Cache* get_cache() {
+ return coll ? coll->cache : nullptr;
+ }
+ inline SharedBlobSet* get_parent() {
+ return coll ? &(coll->shared_blob_set) : nullptr;
+ }
+ inline bool is_loaded() const {
+ return loaded;
+ }
+
+ };
+ typedef boost::intrusive_ptr<SharedBlob> SharedBlobRef;
+
+ /// a lookup table of SharedBlobs
+ struct SharedBlobSet {
+ std::mutex lock; ///< protect lookup, insertion, removal
+
+ // we use a bare pointer because we don't want to affect the ref
+ // count
+ mempool::bluestore_cache_other::unordered_map<uint64_t,SharedBlob*> sb_map;
+
+ SharedBlobRef lookup(uint64_t sbid) {
+ std::lock_guard<std::mutex> l(lock);
+ auto p = sb_map.find(sbid);
+ if (p == sb_map.end()) {
+ return nullptr;
+ }
+ return p->second;
+ }
+
+ void add(Collection* coll, SharedBlob *sb) {
+ std::lock_guard<std::mutex> l(lock);
+ sb_map[sb->get_sbid()] = sb;
+ sb->coll = coll;
+ }
+
+ bool try_remove(SharedBlob *sb) {
+ std::lock_guard<std::mutex> l(lock);
+ if (sb->nref == 0) {
+ assert(sb->get_parent() == this);
+ sb_map.erase(sb->get_sbid());
+ return true;
+ }
+ return false;
+ }
+
+ void remove(SharedBlob *sb) {
+ std::lock_guard<std::mutex> l(lock);
+ assert(sb->get_parent() == this);
+ sb_map.erase(sb->get_sbid());
+ }
+
+ bool empty() {
+ std::lock_guard<std::mutex> l(lock);
+ return sb_map.empty();
+ }
+
+ void dump(CephContext *cct, int lvl);
+ };
+
+//#define CACHE_BLOB_BL // not sure if this is a win yet or not... :/
+
+ /// in-memory blob metadata and associated cached buffers (if any)
+ struct Blob {
+ MEMPOOL_CLASS_HELPERS();
+
+ std::atomic_int nref = {0}; ///< reference count
+ int16_t id = -1; ///< id, for spanning blobs only, >= 0
+ int16_t last_encoded_id = -1; ///< (ephemeral) used during encoding only
+ SharedBlobRef shared_blob; ///< shared blob state (if any)
+
+ private:
+ mutable bluestore_blob_t blob; ///< decoded blob metadata
+#ifdef CACHE_BLOB_BL
+ mutable bufferlist blob_bl; ///< cached encoded blob, blob is dirty if empty
+#endif
+ /// refs from this shard. ephemeral if id<0, persisted if spanning.
+ bluestore_blob_use_tracker_t used_in_blob;
+
+ public:
+
+ friend void intrusive_ptr_add_ref(Blob *b) { b->get(); }
+ friend void intrusive_ptr_release(Blob *b) { b->put(); }
+
+ friend ostream& operator<<(ostream& out, const Blob &b);
+
+ const bluestore_blob_use_tracker_t& get_blob_use_tracker() const {
+ return used_in_blob;
+ }
+ bool is_referenced() const {
+ return used_in_blob.is_not_empty();
+ }
+ uint32_t get_referenced_bytes() const {
+ return used_in_blob.get_referenced_bytes();
+ }
+
+ bool is_spanning() const {
+ return id >= 0;
+ }
+
+ bool can_split() const {
+ std::lock_guard<std::recursive_mutex> l(shared_blob->get_cache()->lock);
+ // splitting a BufferSpace writing list is too hard; don't try.
+ return shared_blob->bc.writing.empty() &&
+ used_in_blob.can_split() &&
+ get_blob().can_split();
+ }
+
+ bool can_split_at(uint32_t blob_offset) const {
+ return used_in_blob.can_split_at(blob_offset) &&
+ get_blob().can_split_at(blob_offset);
+ }
+
+ bool can_reuse_blob(uint32_t min_alloc_size,
+ uint32_t target_blob_size,
+ uint32_t b_offset,
+ uint32_t *length0);
+
+ void dup(Blob& o) {
+ o.shared_blob = shared_blob;
+ o.blob = blob;
+#ifdef CACHE_BLOB_BL
+ o.blob_bl = blob_bl;
+#endif
+ }
+
+ inline const bluestore_blob_t& get_blob() const {
+ return blob;
+ }
+ inline bluestore_blob_t& dirty_blob() {
+#ifdef CACHE_BLOB_BL
+ blob_bl.clear();
+#endif
+ return blob;
+ }
+
+ /// discard buffers for unallocated regions
+ void discard_unallocated(Collection *coll);
+
+ /// get logical references
+ void get_ref(Collection *coll, uint32_t offset, uint32_t length);
+ /// put logical references, and get back any released extents
+ bool put_ref(Collection *coll, uint32_t offset, uint32_t length,
+ PExtentVector *r);
+
+ /// split the blob
+ void split(Collection *coll, uint32_t blob_offset, Blob *o);
+
+ void get() {
+ ++nref;
+ }
+ void put() {
+ if (--nref == 0)
+ delete this;
+ }
+
+
+#ifdef CACHE_BLOB_BL
+ void _encode() const {
+ if (blob_bl.length() == 0 ) {
+ ::encode(blob, blob_bl);
+ } else {
+ assert(blob_bl.length());
+ }
+ }
+ void bound_encode(
+ size_t& p,
+ bool include_ref_map) const {
+ _encode();
+ p += blob_bl.length();
+ if (include_ref_map) {
+ used_in_blob.bound_encode(p);
+ }
+ }
+ void encode(
+ bufferlist::contiguous_appender& p,
+ bool include_ref_map) const {
+ _encode();
+ p.append(blob_bl);
+ if (include_ref_map) {
+ used_in_blob.encode(p);
+ }
+ }
+ void decode(
+ Collection */*coll*/,
+ bufferptr::iterator& p,
+ bool include_ref_map) {
+ const char *start = p.get_pos();
+ denc(blob, p);
+ const char *end = p.get_pos();
+ blob_bl.clear();
+ blob_bl.append(start, end - start);
+ if (include_ref_map) {
+ used_in_blob.decode(p);
+ }
+ }
+#else
+ void bound_encode(
+ size_t& p,
+ uint64_t struct_v,
+ uint64_t sbid,
+ bool include_ref_map) const {
+ denc(blob, p, struct_v);
+ if (blob.is_shared()) {
+ denc(sbid, p);
+ }
+ if (include_ref_map) {
+ used_in_blob.bound_encode(p);
+ }
+ }
+ void encode(
+ bufferlist::contiguous_appender& p,
+ uint64_t struct_v,
+ uint64_t sbid,
+ bool include_ref_map) const {
+ denc(blob, p, struct_v);
+ if (blob.is_shared()) {
+ denc(sbid, p);
+ }
+ if (include_ref_map) {
+ used_in_blob.encode(p);
+ }
+ }
+ void decode(
+ Collection *coll,
+ bufferptr::iterator& p,
+ uint64_t struct_v,
+ uint64_t* sbid,
+ bool include_ref_map);
+#endif
+ };
+ typedef boost::intrusive_ptr<Blob> BlobRef;
+ typedef mempool::bluestore_cache_other::map<int,BlobRef> blob_map_t;
+
+ /// a logical extent, pointing to (some portion of) a blob
+ typedef boost::intrusive::set_base_hook<boost::intrusive::optimize_size<true> > ExtentBase; //making an alias to avoid build warnings
+ struct Extent : public ExtentBase {
+ MEMPOOL_CLASS_HELPERS();
+
+ uint32_t logical_offset = 0; ///< logical offset
+ uint32_t blob_offset = 0; ///< blob offset
+ uint32_t length = 0; ///< length
+ BlobRef blob; ///< the blob with our data
+
+ /// ctor for lookup only
+ explicit Extent(uint32_t lo) : ExtentBase(), logical_offset(lo) { }
+ /// ctor for delayed initialization (see decode_some())
+ explicit Extent() : ExtentBase() {
+ }
+ /// ctor for general usage
+ Extent(uint32_t lo, uint32_t o, uint32_t l, BlobRef& b)
+ : ExtentBase(),
+ logical_offset(lo), blob_offset(o), length(l) {
+ assign_blob(b);
+ }
+ ~Extent() {
+ if (blob) {
+ blob->shared_blob->get_cache()->rm_extent();
+ }
+ }
+
+ void assign_blob(const BlobRef& b) {
+ assert(!blob);
+ blob = b;
+ blob->shared_blob->get_cache()->add_extent();
+ }
+
+ // comparators for intrusive_set
+ friend bool operator<(const Extent &a, const Extent &b) {
+ return a.logical_offset < b.logical_offset;
+ }
+ friend bool operator>(const Extent &a, const Extent &b) {
+ return a.logical_offset > b.logical_offset;
+ }
+ friend bool operator==(const Extent &a, const Extent &b) {
+ return a.logical_offset == b.logical_offset;
+ }
+
+ uint32_t blob_start() const {
+ return logical_offset - blob_offset;
+ }
+
+ uint32_t blob_end() const {
+ return blob_start() + blob->get_blob().get_logical_length();
+ }
+
+ uint32_t logical_end() const {
+ return logical_offset + length;
+ }
+
+ // return true if any piece of the blob is out of
+ // the given range [o, o + l].
+ bool blob_escapes_range(uint32_t o, uint32_t l) const {
+ return blob_start() < o || blob_end() > o + l;
+ }
+ };
+ typedef boost::intrusive::set<Extent> extent_map_t;
+
+
+ friend ostream& operator<<(ostream& out, const Extent& e);
+
+ struct OldExtent {
+ boost::intrusive::list_member_hook<> old_extent_item;
+ Extent e;
+ PExtentVector r;
+ bool blob_empty; // flag to track the last removed extent that makes blob
+ // empty - required to update compression stat properly
+ OldExtent(uint32_t lo, uint32_t o, uint32_t l, BlobRef& b)
+ : e(lo, o, l, b), blob_empty(false) {
+ }
+ static OldExtent* create(CollectionRef c,
+ uint32_t lo,
+ uint32_t o,
+ uint32_t l,
+ BlobRef& b);
+ };
+ typedef boost::intrusive::list<
+ OldExtent,
+ boost::intrusive::member_hook<
+ OldExtent,
+ boost::intrusive::list_member_hook<>,
+ &OldExtent::old_extent_item> > old_extent_map_t;
+
+ struct Onode;
+
+ /// a sharded extent map, mapping offsets to lextents to blobs
+ struct ExtentMap {
+ Onode *onode;
+ extent_map_t extent_map; ///< map of Extents to Blobs
+ blob_map_t spanning_blob_map; ///< blobs that span shards
+
+ struct Shard {
+ bluestore_onode_t::shard_info *shard_info = nullptr;
+ unsigned extents = 0; ///< count extents in this shard
+ bool loaded = false; ///< true if shard is loaded
+ bool dirty = false; ///< true if shard is dirty and needs reencoding
+ };
+ mempool::bluestore_cache_other::vector<Shard> shards; ///< shards
+
+ bufferlist inline_bl; ///< cached encoded map, if unsharded; empty=>dirty
+
+ uint32_t needs_reshard_begin = 0;
+ uint32_t needs_reshard_end = 0;
+
+ bool needs_reshard() const {
+ return needs_reshard_end > needs_reshard_begin;
+ }
+ void clear_needs_reshard() {
+ needs_reshard_begin = needs_reshard_end = 0;
+ }
+ void request_reshard(uint32_t begin, uint32_t end) {
+ if (begin < needs_reshard_begin) {
+ needs_reshard_begin = begin;
+ }
+ if (end > needs_reshard_end) {
+ needs_reshard_end = end;
+ }
+ }
+
+ struct DeleteDisposer {
+ void operator()(Extent *e) { delete e; }
+ };
+
+ ExtentMap(Onode *o);
+ ~ExtentMap() {
+ extent_map.clear_and_dispose(DeleteDisposer());
+ }
+
+ void clear() {
+ extent_map.clear_and_dispose(DeleteDisposer());
+ shards.clear();
+ inline_bl.clear();
+ clear_needs_reshard();
+ }
+
+ bool encode_some(uint32_t offset, uint32_t length, bufferlist& bl,
+ unsigned *pn);
+ unsigned decode_some(bufferlist& bl);
+
+ void bound_encode_spanning_blobs(size_t& p);
+ void encode_spanning_blobs(bufferlist::contiguous_appender& p);
+ void decode_spanning_blobs(bufferptr::iterator& p);
+
+ BlobRef get_spanning_blob(int id) {
+ auto p = spanning_blob_map.find(id);
+ assert(p != spanning_blob_map.end());
+ return p->second;
+ }
+
+ void update(KeyValueDB::Transaction t, bool force);
+ decltype(BlueStore::Blob::id) allocate_spanning_blob_id();
+ void reshard(
+ KeyValueDB *db,
+ KeyValueDB::Transaction t);
+
+ /// initialize Shards from the onode
+ void init_shards(bool loaded, bool dirty);
+
+ /// return index of shard containing offset
+ /// or -1 if not found
+ int seek_shard(uint32_t offset) {
+ size_t end = shards.size();
+ size_t mid, left = 0;
+ size_t right = end; // one passed the right end
+
+ while (left < right) {
+ mid = left + (right - left) / 2;
+ if (offset >= shards[mid].shard_info->offset) {
+ size_t next = mid + 1;
+ if (next >= end || offset < shards[next].shard_info->offset)
+ return mid;
+ //continue to search forwards
+ left = next;
+ } else {
+ //continue to search backwards
+ right = mid;
+ }
+ }
+
+ return -1; // not found
+ }
+
+ /// check if a range spans a shard
+ bool spans_shard(uint32_t offset, uint32_t length) {
+ if (shards.empty()) {
+ return false;
+ }
+ int s = seek_shard(offset);
+ assert(s >= 0);
+ if (s == (int)shards.size() - 1) {
+ return false; // last shard
+ }
+ if (offset + length <= shards[s+1].shard_info->offset) {
+ return false;
+ }
+ return true;
+ }
+
+ /// ensure that a range of the map is loaded
+ void fault_range(KeyValueDB *db,
+ uint32_t offset, uint32_t length);
+
+ /// ensure a range of the map is marked dirty
+ void dirty_range(uint32_t offset, uint32_t length);
+
+ /// for seek_lextent test
+ extent_map_t::iterator find(uint64_t offset);
+
+ /// seek to the first lextent including or after offset
+ extent_map_t::iterator seek_lextent(uint64_t offset);
+ extent_map_t::const_iterator seek_lextent(uint64_t offset) const;
+
+ /// add a new Extent
+ void add(uint32_t lo, uint32_t o, uint32_t l, BlobRef& b) {
+ extent_map.insert(*new Extent(lo, o, l, b));
+ }
+
+ /// remove (and delete) an Extent
+ void rm(extent_map_t::iterator p) {
+ extent_map.erase_and_dispose(p, DeleteDisposer());
+ }
+
+ bool has_any_lextents(uint64_t offset, uint64_t length);
+
+ /// consolidate adjacent lextents in extent_map
+ int compress_extent_map(uint64_t offset, uint64_t length);
+
+ /// punch a logical hole. add lextents to deref to target list.
+ void punch_hole(CollectionRef &c,
+ uint64_t offset, uint64_t length,
+ old_extent_map_t *old_extents);
+
+ /// put new lextent into lextent_map overwriting existing ones if
+ /// any and update references accordingly
+ Extent *set_lextent(CollectionRef &c,
+ uint64_t logical_offset,
+ uint64_t offset, uint64_t length,
+ BlobRef b,
+ old_extent_map_t *old_extents);
+
+ /// split a blob (and referring extents)
+ BlobRef split_blob(BlobRef lb, uint32_t blob_offset, uint32_t pos);
+ };
+
+ /// Compressed Blob Garbage collector
+ /*
+ The primary idea of the collector is to estimate a difference between
+ allocation units(AU) currently present for compressed blobs and new AUs
+ required to store that data uncompressed.
+ Estimation is performed for protrusive extents within a logical range
+ determined by a concatenation of old_extents collection and specific(current)
+ write request.
+ The root cause for old_extents use is the need to handle blob ref counts
+ properly. Old extents still hold blob refs and hence we need to traverse
+ the collection to determine if blob to be released.
+ Protrusive extents are extents that fit into the blob set in action
+ (ones that are below the logical range from above) but not removed totally
+ due to the current write.
+ E.g. for
+ extent1 <loffs = 100, boffs = 100, len = 100> ->
+ blob1<compressed, len_on_disk=4096, logical_len=8192>
+ extent2 <loffs = 200, boffs = 200, len = 100> ->
+ blob2<raw, len_on_disk=4096, llen=4096>
+ extent3 <loffs = 300, boffs = 300, len = 100> ->
+ blob1<compressed, len_on_disk=4096, llen=8192>
+ extent4 <loffs = 4096, boffs = 0, len = 100> ->
+ blob3<raw, len_on_disk=4096, llen=4096>
+ write(300~100)
+ protrusive extents are within the following ranges <0~300, 400~8192-400>
+ In this case existing AUs that might be removed due to GC (i.e. blob1)
+ use 2x4K bytes.
+ And new AUs expected after GC = 0 since extent1 to be merged into blob2.
+ Hence we should do a collect.
+ */
+ class GarbageCollector
+ {
+ public:
+ /// return amount of allocation units that might be saved due to GC
+ int64_t estimate(
+ uint64_t offset,
+ uint64_t length,
+ const ExtentMap& extent_map,
+ const old_extent_map_t& old_extents,
+ uint64_t min_alloc_size);
+
+ /// return a collection of extents to perform GC on
+ const vector<AllocExtent>& get_extents_to_collect() const {
+ return extents_to_collect;
+ }
+ GarbageCollector(CephContext* _cct) : cct(_cct) {}
+
+ private:
+ struct BlobInfo {
+ uint64_t referenced_bytes = 0; ///< amount of bytes referenced in blob
+ int64_t expected_allocations = 0; ///< new alloc units required
+ ///< in case of gc fulfilled
+ bool collect_candidate = false; ///< indicate if blob has any extents
+ ///< eligible for GC.
+ extent_map_t::const_iterator first_lextent; ///< points to the first
+ ///< lextent referring to
+ ///< the blob if any.
+ ///< collect_candidate flag
+ ///< determines the validity
+ extent_map_t::const_iterator last_lextent; ///< points to the last
+ ///< lextent referring to
+ ///< the blob if any.
+
+ BlobInfo(uint64_t ref_bytes) :
+ referenced_bytes(ref_bytes) {
+ }
+ };
+ CephContext* cct;
+ map<Blob*, BlobInfo> affected_blobs; ///< compressed blobs and their ref_map
+ ///< copies that are affected by the
+ ///< specific write
+
+ vector<AllocExtent> extents_to_collect; ///< protrusive extents that should
+ ///< be collected if GC takes place
+
+ boost::optional<uint64_t > used_alloc_unit; ///< last processed allocation
+ ///< unit when traversing
+ ///< protrusive extents.
+ ///< Other extents mapped to
+ ///< this AU to be ignored
+ ///< (except the case where
+ ///< uncompressed extent follows
+ ///< compressed one - see below).
+ BlobInfo* blob_info_counted = nullptr; ///< set if previous allocation unit
+ ///< caused expected_allocations
+ ///< counter increment at this blob.
+ ///< if uncompressed extent follows
+ ///< a decrement for the
+ ///< expected_allocations counter
+ ///< is needed
+ int64_t expected_allocations = 0; ///< new alloc units required in case
+ ///< of gc fulfilled
+ int64_t expected_for_release = 0; ///< alloc units currently used by
+ ///< compressed blobs that might
+ ///< gone after GC
+ uint64_t gc_start_offset; ///starting offset for GC
+ uint64_t gc_end_offset; ///ending offset for GC
+
+ protected:
+ void process_protrusive_extents(const BlueStore::ExtentMap& extent_map,
+ uint64_t start_offset,
+ uint64_t end_offset,
+ uint64_t start_touch_offset,
+ uint64_t end_touch_offset,
+ uint64_t min_alloc_size);
+ };
+
+ struct OnodeSpace;
+
+ /// an in-memory object
+ struct Onode {
+ MEMPOOL_CLASS_HELPERS();
+
+ std::atomic_int nref; ///< reference count
+ Collection *c;
+
+ ghobject_t oid;
+
+ /// key under PREFIX_OBJ where we are stored
+ mempool::bluestore_cache_other::string key;
+
+ boost::intrusive::list_member_hook<> lru_item;
+
+ bluestore_onode_t onode; ///< metadata stored as value in kv store
+ bool exists; ///< true if object logically exists
+
+ ExtentMap extent_map;
+
+ // track txc's that have not been committed to kv store (and whose
+ // effects cannot be read via the kvdb read methods)
+ std::atomic<int> flushing_count = {0};
+ std::mutex flush_lock; ///< protect flush_txns
+ std::condition_variable flush_cond; ///< wait here for uncommitted txns
+
+ Onode(Collection *c, const ghobject_t& o,
+ const mempool::bluestore_cache_other::string& k)
+ : nref(0),
+ c(c),
+ oid(o),
+ key(k),
+ exists(false),
+ extent_map(this) {
+ }
+
+ void flush();
+ void get() {
+ ++nref;
+ }
+ void put() {
+ if (--nref == 0)
+ delete this;
+ }
+ };
+ typedef boost::intrusive_ptr<Onode> OnodeRef;
+
+
+ /// a cache (shard) of onodes and buffers
+ struct Cache {
+ CephContext* cct;
+ PerfCounters *logger;
+ std::recursive_mutex lock; ///< protect lru and other structures
+
+ std::atomic<uint64_t> num_extents = {0};
+ std::atomic<uint64_t> num_blobs = {0};
+
+ static Cache *create(CephContext* cct, string type, PerfCounters *logger);
+
+ Cache(CephContext* cct) : cct(cct), logger(nullptr) {}
+ virtual ~Cache() {}
+
+ virtual void _add_onode(OnodeRef& o, int level) = 0;
+ virtual void _rm_onode(OnodeRef& o) = 0;
+ virtual void _touch_onode(OnodeRef& o) = 0;
+
+ virtual void _add_buffer(Buffer *b, int level, Buffer *near) = 0;
+ virtual void _rm_buffer(Buffer *b) = 0;
+ virtual void _move_buffer(Cache *src, Buffer *b) = 0;
+ virtual void _adjust_buffer_size(Buffer *b, int64_t delta) = 0;
+ virtual void _touch_buffer(Buffer *b) = 0;
+
+ virtual uint64_t _get_num_onodes() = 0;
+ virtual uint64_t _get_buffer_bytes() = 0;
+
+ void add_extent() {
+ ++num_extents;
+ }
+ void rm_extent() {
+ --num_extents;
+ }
+
+ void add_blob() {
+ ++num_blobs;
+ }
+ void rm_blob() {
+ --num_blobs;
+ }
+
+ void trim(uint64_t target_bytes,
+ float target_meta_ratio,
+ float target_data_ratio,
+ float bytes_per_onode);
+
+ void trim_all();
+
+ virtual void _trim(uint64_t onode_max, uint64_t buffer_max) = 0;
+
+ virtual void add_stats(uint64_t *onodes, uint64_t *extents,
+ uint64_t *blobs,
+ uint64_t *buffers,
+ uint64_t *bytes) = 0;
+
+ bool empty() {
+ std::lock_guard<std::recursive_mutex> l(lock);
+ return _get_num_onodes() == 0 && _get_buffer_bytes() == 0;
+ }
+
+#ifdef DEBUG_CACHE
+ virtual void _audit(const char *s) = 0;
+#else
+ void _audit(const char *s) { /* no-op */ }
+#endif
+ };
+
+ /// simple LRU cache for onodes and buffers
+ struct LRUCache : public Cache {
+ private:
+ typedef boost::intrusive::list<
+ Onode,
+ boost::intrusive::member_hook<
+ Onode,
+ boost::intrusive::list_member_hook<>,
+ &Onode::lru_item> > onode_lru_list_t;
+ typedef boost::intrusive::list<
+ Buffer,
+ boost::intrusive::member_hook<
+ Buffer,
+ boost::intrusive::list_member_hook<>,
+ &Buffer::lru_item> > buffer_lru_list_t;
+
+ onode_lru_list_t onode_lru;
+
+ buffer_lru_list_t buffer_lru;
+ uint64_t buffer_size = 0;
+
+ public:
+ LRUCache(CephContext* cct) : Cache(cct) {}
+ uint64_t _get_num_onodes() override {
+ return onode_lru.size();
+ }
+ void _add_onode(OnodeRef& o, int level) override {
+ if (level > 0)
+ onode_lru.push_front(*o);
+ else
+ onode_lru.push_back(*o);
+ }
+ void _rm_onode(OnodeRef& o) override {
+ auto q = onode_lru.iterator_to(*o);
+ onode_lru.erase(q);
+ }
+ void _touch_onode(OnodeRef& o) override;
+
+ uint64_t _get_buffer_bytes() override {
+ return buffer_size;
+ }
+ void _add_buffer(Buffer *b, int level, Buffer *near) override {
+ if (near) {
+ auto q = buffer_lru.iterator_to(*near);
+ buffer_lru.insert(q, *b);
+ } else if (level > 0) {
+ buffer_lru.push_front(*b);
+ } else {
+ buffer_lru.push_back(*b);
+ }
+ buffer_size += b->length;
+ }
+ void _rm_buffer(Buffer *b) override {
+ assert(buffer_size >= b->length);
+ buffer_size -= b->length;
+ auto q = buffer_lru.iterator_to(*b);
+ buffer_lru.erase(q);
+ }
+ void _move_buffer(Cache *src, Buffer *b) override {
+ src->_rm_buffer(b);
+ _add_buffer(b, 0, nullptr);
+ }
+ void _adjust_buffer_size(Buffer *b, int64_t delta) override {
+ assert((int64_t)buffer_size + delta >= 0);
+ buffer_size += delta;
+ }
+ void _touch_buffer(Buffer *b) override {
+ auto p = buffer_lru.iterator_to(*b);
+ buffer_lru.erase(p);
+ buffer_lru.push_front(*b);
+ _audit("_touch_buffer end");
+ }
+
+ void _trim(uint64_t onode_max, uint64_t buffer_max) override;
+
+ void add_stats(uint64_t *onodes, uint64_t *extents,
+ uint64_t *blobs,
+ uint64_t *buffers,
+ uint64_t *bytes) override {
+ std::lock_guard<std::recursive_mutex> l(lock);
+ *onodes += onode_lru.size();
+ *extents += num_extents;
+ *blobs += num_blobs;
+ *buffers += buffer_lru.size();
+ *bytes += buffer_size;
+ }
+
+#ifdef DEBUG_CACHE
+ void _audit(const char *s) override;
+#endif
+ };
+
+ // 2Q cache for buffers, LRU for onodes
+ struct TwoQCache : public Cache {
+ private:
+ // stick with LRU for onodes for now (fixme?)
+ typedef boost::intrusive::list<
+ Onode,
+ boost::intrusive::member_hook<
+ Onode,
+ boost::intrusive::list_member_hook<>,
+ &Onode::lru_item> > onode_lru_list_t;
+ typedef boost::intrusive::list<
+ Buffer,
+ boost::intrusive::member_hook<
+ Buffer,
+ boost::intrusive::list_member_hook<>,
+ &Buffer::lru_item> > buffer_list_t;
+
+ onode_lru_list_t onode_lru;
+
+ buffer_list_t buffer_hot; ///< "Am" hot buffers
+ buffer_list_t buffer_warm_in; ///< "A1in" newly warm buffers
+ buffer_list_t buffer_warm_out; ///< "A1out" empty buffers we've evicted
+ uint64_t buffer_bytes = 0; ///< bytes
+
+ enum {
+ BUFFER_NEW = 0,
+ BUFFER_WARM_IN, ///< in buffer_warm_in
+ BUFFER_WARM_OUT, ///< in buffer_warm_out
+ BUFFER_HOT, ///< in buffer_hot
+ BUFFER_TYPE_MAX
+ };
+
+ uint64_t buffer_list_bytes[BUFFER_TYPE_MAX] = {0}; ///< bytes per type
+
+ public:
+ TwoQCache(CephContext* cct) : Cache(cct) {}
+ uint64_t _get_num_onodes() override {
+ return onode_lru.size();
+ }
+ void _add_onode(OnodeRef& o, int level) override {
+ if (level > 0)
+ onode_lru.push_front(*o);
+ else
+ onode_lru.push_back(*o);
+ }
+ void _rm_onode(OnodeRef& o) override {
+ auto q = onode_lru.iterator_to(*o);
+ onode_lru.erase(q);
+ }
+ void _touch_onode(OnodeRef& o) override;
+
+ uint64_t _get_buffer_bytes() override {
+ return buffer_bytes;
+ }
+ void _add_buffer(Buffer *b, int level, Buffer *near) override;
+ void _rm_buffer(Buffer *b) override;
+ void _move_buffer(Cache *src, Buffer *b) override;
+ void _adjust_buffer_size(Buffer *b, int64_t delta) override;
+ void _touch_buffer(Buffer *b) override {
+ switch (b->cache_private) {
+ case BUFFER_WARM_IN:
+ // do nothing (somewhat counter-intuitively!)
+ break;
+ case BUFFER_WARM_OUT:
+ // move from warm_out to hot LRU
+ assert(0 == "this happens via discard hint");
+ break;
+ case BUFFER_HOT:
+ // move to front of hot LRU
+ buffer_hot.erase(buffer_hot.iterator_to(*b));
+ buffer_hot.push_front(*b);
+ break;
+ }
+ _audit("_touch_buffer end");
+ }
+
+ void _trim(uint64_t onode_max, uint64_t buffer_max) override;
+
+ void add_stats(uint64_t *onodes, uint64_t *extents,
+ uint64_t *blobs,
+ uint64_t *buffers,
+ uint64_t *bytes) override {
+ std::lock_guard<std::recursive_mutex> l(lock);
+ *onodes += onode_lru.size();
+ *extents += num_extents;
+ *blobs += num_blobs;
+ *buffers += buffer_hot.size() + buffer_warm_in.size();
+ *bytes += buffer_bytes;
+ }
+
+#ifdef DEBUG_CACHE
+ void _audit(const char *s) override;
+#endif
+ };
+
+ struct OnodeSpace {
+ private:
+ Cache *cache;
+
+ /// forward lookups
+ mempool::bluestore_cache_other::unordered_map<ghobject_t,OnodeRef> onode_map;
+
+ friend class Collection; // for split_cache()
+
+ public:
+ OnodeSpace(Cache *c) : cache(c) {}
+ ~OnodeSpace() {
+ clear();
+ }
+
+ OnodeRef add(const ghobject_t& oid, OnodeRef o);
+ OnodeRef lookup(const ghobject_t& o);
+ void remove(const ghobject_t& oid) {
+ onode_map.erase(oid);
+ }
+ void rename(OnodeRef& o, const ghobject_t& old_oid,
+ const ghobject_t& new_oid,
+ const mempool::bluestore_cache_other::string& new_okey);
+ void clear();
+ bool empty();
+
+ void dump(CephContext *cct, int lvl);
+
+ /// return true if f true for any item
+ bool map_any(std::function<bool(OnodeRef)> f);
+ };
+
+ struct Collection : public CollectionImpl {
+ BlueStore *store;
+ Cache *cache; ///< our cache shard
+ coll_t cid;
+ bluestore_cnode_t cnode;
+ RWLock lock;
+
+ bool exists;
+
+ SharedBlobSet shared_blob_set; ///< open SharedBlobs
+
+ // cache onodes on a per-collection basis to avoid lock
+ // contention.
+ OnodeSpace onode_map;
+
+ //pool options
+ pool_opts_t pool_opts;
+
+ OnodeRef get_onode(const ghobject_t& oid, bool create);
+
+ // the terminology is confusing here, sorry!
+ //
+ // blob_t shared_blob_t
+ // !shared unused -> open
+ // shared !loaded -> open + shared
+ // shared loaded -> open + shared + loaded
+ //
+ // i.e.,
+ // open = SharedBlob is instantiated
+ // shared = blob_t shared flag is set; SharedBlob is hashed.
+ // loaded = SharedBlob::shared_blob_t is loaded from kv store
+ void open_shared_blob(uint64_t sbid, BlobRef b);
+ void load_shared_blob(SharedBlobRef sb);
+ void make_blob_shared(uint64_t sbid, BlobRef b);
+ uint64_t make_blob_unshared(SharedBlob *sb);
+
+ BlobRef new_blob() {
+ BlobRef b = new Blob();
+ b->shared_blob = new SharedBlob(this);
+ return b;
+ }
+
+ const coll_t &get_cid() override {
+ return cid;
+ }
+
+ bool contains(const ghobject_t& oid) {
+ if (cid.is_meta())
+ return oid.hobj.pool == -1;
+ spg_t spgid;
+ if (cid.is_pg(&spgid))
+ return
+ spgid.pgid.contains(cnode.bits, oid) &&
+ oid.shard_id == spgid.shard;
+ return false;
+ }
+
+ void split_cache(Collection *dest);
+
+ Collection(BlueStore *ns, Cache *ca, coll_t c);
+ };
+
+ class OmapIteratorImpl : public ObjectMap::ObjectMapIteratorImpl {
+ CollectionRef c;
+ OnodeRef o;
+ KeyValueDB::Iterator it;
+ string head, tail;
+ public:
+ OmapIteratorImpl(CollectionRef c, OnodeRef o, KeyValueDB::Iterator it);
+ int seek_to_first() override;
+ int upper_bound(const string &after) override;
+ int lower_bound(const string &to) override;
+ bool valid() override;
+ int next(bool validate=true) override;
+ string key() override;
+ bufferlist value() override;
+ int status() override {
+ return 0;
+ }
+ };
+
+ class OpSequencer;
+ typedef boost::intrusive_ptr<OpSequencer> OpSequencerRef;
+
+ struct volatile_statfs{
+ enum {
+ STATFS_ALLOCATED = 0,
+ STATFS_STORED,
+ STATFS_COMPRESSED_ORIGINAL,
+ STATFS_COMPRESSED,
+ STATFS_COMPRESSED_ALLOCATED,
+ STATFS_LAST
+ };
+ int64_t values[STATFS_LAST];
+ volatile_statfs() {
+ memset(this, 0, sizeof(volatile_statfs));
+ }
+ void reset() {
+ *this = volatile_statfs();
+ }
+ volatile_statfs& operator+=(const volatile_statfs& other) {
+ for (size_t i = 0; i < STATFS_LAST; ++i) {
+ values[i] += other.values[i];
+ }
+ return *this;
+ }
+ int64_t& allocated() {
+ return values[STATFS_ALLOCATED];
+ }
+ int64_t& stored() {
+ return values[STATFS_STORED];
+ }
+ int64_t& compressed_original() {
+ return values[STATFS_COMPRESSED_ORIGINAL];
+ }
+ int64_t& compressed() {
+ return values[STATFS_COMPRESSED];
+ }
+ int64_t& compressed_allocated() {
+ return values[STATFS_COMPRESSED_ALLOCATED];
+ }
+ bool is_empty() {
+ return values[STATFS_ALLOCATED] == 0 &&
+ values[STATFS_STORED] == 0 &&
+ values[STATFS_COMPRESSED] == 0 &&
+ values[STATFS_COMPRESSED_ORIGINAL] == 0 &&
+ values[STATFS_COMPRESSED_ALLOCATED] == 0;
+ }
+ void decode(bufferlist::iterator& it) {
+ for (size_t i = 0; i < STATFS_LAST; i++) {
+ ::decode(values[i], it);
+ }
+ }
+
+ void encode(bufferlist& bl) {
+ for (size_t i = 0; i < STATFS_LAST; i++) {
+ ::encode(values[i], bl);
+ }
+ }
+ };
+
+ struct TransContext : public AioContext {
+ MEMPOOL_CLASS_HELPERS();
+
+ typedef enum {
+ STATE_PREPARE,
+ STATE_AIO_WAIT,
+ STATE_IO_DONE,
+ STATE_KV_QUEUED, // queued for kv_sync_thread submission
+ STATE_KV_SUBMITTED, // submitted to kv; not yet synced
+ STATE_KV_DONE,
+ STATE_DEFERRED_QUEUED, // in deferred_queue (pending or running)
+ STATE_DEFERRED_CLEANUP, // remove deferred kv record
+ STATE_DEFERRED_DONE,
+ STATE_FINISHING,
+ STATE_DONE,
+ } state_t;
+
+ state_t state = STATE_PREPARE;
+
+ const char *get_state_name() {
+ switch (state) {
+ case STATE_PREPARE: return "prepare";
+ case STATE_AIO_WAIT: return "aio_wait";
+ case STATE_IO_DONE: return "io_done";
+ case STATE_KV_QUEUED: return "kv_queued";
+ case STATE_KV_SUBMITTED: return "kv_submitted";
+ case STATE_KV_DONE: return "kv_done";
+ case STATE_DEFERRED_QUEUED: return "deferred_queued";
+ case STATE_DEFERRED_CLEANUP: return "deferred_cleanup";
+ case STATE_DEFERRED_DONE: return "deferred_done";
+ case STATE_FINISHING: return "finishing";
+ case STATE_DONE: return "done";
+ }
+ return "???";
+ }
+
+#if defined(WITH_LTTNG) && defined(WITH_EVENTTRACE)
+ const char *get_state_latency_name(int state) {
+ switch (state) {
+ case l_bluestore_state_prepare_lat: return "prepare";
+ case l_bluestore_state_aio_wait_lat: return "aio_wait";
+ case l_bluestore_state_io_done_lat: return "io_done";
+ case l_bluestore_state_kv_queued_lat: return "kv_queued";
+ case l_bluestore_state_kv_committing_lat: return "kv_committing";
+ case l_bluestore_state_kv_done_lat: return "kv_done";
+ case l_bluestore_state_deferred_queued_lat: return "deferred_queued";
+ case l_bluestore_state_deferred_cleanup_lat: return "deferred_cleanup";
+ case l_bluestore_state_finishing_lat: return "finishing";
+ case l_bluestore_state_done_lat: return "done";
+ }
+ return "???";
+ }
+#endif
+
+ void log_state_latency(PerfCounters *logger, int state) {
+ utime_t lat, now = ceph_clock_now();
+ lat = now - last_stamp;
+ logger->tinc(state, lat);
+#if defined(WITH_LTTNG) && defined(WITH_EVENTTRACE)
+ if (state >= l_bluestore_state_prepare_lat && state <= l_bluestore_state_done_lat) {
+ double usecs = (now.to_nsec()-last_stamp.to_nsec())/1000;
+ OID_ELAPSED("", usecs, get_state_latency_name(state));
+ }
+#endif
+ last_stamp = now;
+ }
+
+ OpSequencerRef osr;
+ boost::intrusive::list_member_hook<> sequencer_item;
+
+ uint64_t bytes = 0, cost = 0;
+
+ set<OnodeRef> onodes; ///< these need to be updated/written
+ set<OnodeRef> modified_objects; ///< objects we modified (and need a ref)
+ set<SharedBlobRef> shared_blobs; ///< these need to be updated/written
+ set<SharedBlobRef> shared_blobs_written; ///< update these on io completion
+
+ KeyValueDB::Transaction t; ///< then we will commit this
+ Context *oncommit = nullptr; ///< signal on commit
+ Context *onreadable = nullptr; ///< signal on readable
+ Context *onreadable_sync = nullptr; ///< signal on readable
+ list<Context*> oncommits; ///< more commit completions
+ list<CollectionRef> removed_collections; ///< colls we removed
+
+ boost::intrusive::list_member_hook<> deferred_queue_item;
+ bluestore_deferred_transaction_t *deferred_txn = nullptr; ///< if any
+
+ interval_set<uint64_t> allocated, released;
+ volatile_statfs statfs_delta;
+
+ IOContext ioc;
+ bool had_ios = false; ///< true if we submitted IOs before our kv txn
+
+ uint64_t seq = 0;
+ utime_t start;
+ utime_t last_stamp;
+
+ uint64_t last_nid = 0; ///< if non-zero, highest new nid we allocated
+ uint64_t last_blobid = 0; ///< if non-zero, highest new blobid we allocated
+
+ explicit TransContext(CephContext* cct, OpSequencer *o)
+ : osr(o),
+ ioc(cct, this),
+ start(ceph_clock_now()) {
+ last_stamp = start;
+ }
+ ~TransContext() {
+ delete deferred_txn;
+ }
+
+ void write_onode(OnodeRef &o) {
+ onodes.insert(o);
+ }
+ void write_shared_blob(SharedBlobRef &sb) {
+ shared_blobs.insert(sb);
+ }
+ void unshare_blob(SharedBlob *sb) {
+ shared_blobs.erase(sb);
+ }
+
+ /// note we logically modified object (when onode itself is unmodified)
+ void note_modified_object(OnodeRef &o) {
+ // onode itself isn't written, though
+ modified_objects.insert(o);
+ }
+ void removed(OnodeRef& o) {
+ onodes.erase(o);
+ modified_objects.erase(o);
+ }
+
+ void aio_finish(BlueStore *store) override {
+ store->txc_aio_finish(this);
+ }
+ };
+
+ typedef boost::intrusive::list<
+ TransContext,
+ boost::intrusive::member_hook<
+ TransContext,
+ boost::intrusive::list_member_hook<>,
+ &TransContext::deferred_queue_item> > deferred_queue_t;
+
+ struct DeferredBatch : public AioContext {
+ OpSequencer *osr;
+ struct deferred_io {
+ bufferlist bl; ///< data
+ uint64_t seq; ///< deferred transaction seq
+ };
+ map<uint64_t,deferred_io> iomap; ///< map of ios in this batch
+ deferred_queue_t txcs; ///< txcs in this batch
+ IOContext ioc; ///< our aios
+ /// bytes of pending io for each deferred seq (may be 0)
+ map<uint64_t,int> seq_bytes;
+
+ void _discard(CephContext *cct, uint64_t offset, uint64_t length);
+ void _audit(CephContext *cct);
+
+ DeferredBatch(CephContext *cct, OpSequencer *osr)
+ : osr(osr), ioc(cct, this) {}
+
+ /// prepare a write
+ void prepare_write(CephContext *cct,
+ uint64_t seq, uint64_t offset, uint64_t length,
+ bufferlist::const_iterator& p);
+
+ void aio_finish(BlueStore *store) override {
+ store->_deferred_aio_finish(osr);
+ }
+ };
+
+ class OpSequencer : public Sequencer_impl {
+ public:
+ std::mutex qlock;
+ std::condition_variable qcond;
+ typedef boost::intrusive::list<
+ TransContext,
+ boost::intrusive::member_hook<
+ TransContext,
+ boost::intrusive::list_member_hook<>,
+ &TransContext::sequencer_item> > q_list_t;
+ q_list_t q; ///< transactions
+
+ boost::intrusive::list_member_hook<> deferred_osr_queue_item;
+
+ DeferredBatch *deferred_running = nullptr;
+ DeferredBatch *deferred_pending = nullptr;
+
+ Sequencer *parent;
+ BlueStore *store;
+
+ uint64_t last_seq = 0;
+
+ std::atomic_int txc_with_unstable_io = {0}; ///< num txcs with unstable io
+
+ std::atomic_int kv_committing_serially = {0};
+
+ std::atomic_int kv_submitted_waiters = {0};
+
+ std::atomic_bool registered = {true}; ///< registered in BlueStore's osr_set
+ std::atomic_bool zombie = {false}; ///< owning Sequencer has gone away
+
+ OpSequencer(CephContext* cct, BlueStore *store)
+ : Sequencer_impl(cct),
+ parent(NULL), store(store) {
+ store->register_osr(this);
+ }
+ ~OpSequencer() override {
+ assert(q.empty());
+ _unregister();
+ }
+
+ void discard() override {
+ // Note that we may have txc's in flight when the parent Sequencer
+ // goes away. Reflect this with zombie==registered==true and let
+ // _osr_drain_all clean up later.
+ assert(!zombie);
+ zombie = true;
+ parent = nullptr;
+ bool empty;
+ {
+ std::lock_guard<std::mutex> l(qlock);
+ empty = q.empty();
+ }
+ if (empty) {
+ _unregister();
+ }
+ }
+
+ void _unregister() {
+ if (registered) {
+ store->unregister_osr(this);
+ registered = false;
+ }
+ }
+
+ void queue_new(TransContext *txc) {
+ std::lock_guard<std::mutex> l(qlock);
+ txc->seq = ++last_seq;
+ q.push_back(*txc);
+ }
+
+ void drain() {
+ std::unique_lock<std::mutex> l(qlock);
+ while (!q.empty())
+ qcond.wait(l);
+ }
+
+ void drain_preceding(TransContext *txc) {
+ std::unique_lock<std::mutex> l(qlock);
+ while (!q.empty() && &q.front() != txc)
+ qcond.wait(l);
+ }
+
+ bool _is_all_kv_submitted() {
+ // caller must hold qlock
+ if (q.empty()) {
+ return true;
+ }
+ TransContext *txc = &q.back();
+ if (txc->state >= TransContext::STATE_KV_SUBMITTED) {
+ return true;
+ }
+ return false;
+ }
+
+ void flush() override {
+ std::unique_lock<std::mutex> l(qlock);
+ while (true) {
+ // set flag before the check because the condition
+ // may become true outside qlock, and we need to make
+ // sure those threads see waiters and signal qcond.
+ ++kv_submitted_waiters;
+ if (_is_all_kv_submitted()) {
+ return;
+ }
+ qcond.wait(l);
+ --kv_submitted_waiters;
+ }
+ }
+
+ bool flush_commit(Context *c) override {
+ std::lock_guard<std::mutex> l(qlock);
+ if (q.empty()) {
+ return true;
+ }
+ TransContext *txc = &q.back();
+ if (txc->state >= TransContext::STATE_KV_DONE) {
+ return true;
+ }
+ txc->oncommits.push_back(c);
+ return false;
+ }
+ };
+
+ typedef boost::intrusive::list<
+ OpSequencer,
+ boost::intrusive::member_hook<
+ OpSequencer,
+ boost::intrusive::list_member_hook<>,
+ &OpSequencer::deferred_osr_queue_item> > deferred_osr_queue_t;
+
+ struct KVSyncThread : public Thread {
+ BlueStore *store;
+ explicit KVSyncThread(BlueStore *s) : store(s) {}
+ void *entry() override {
+ store->_kv_sync_thread();
+ return NULL;
+ }
+ };
+ struct KVFinalizeThread : public Thread {
+ BlueStore *store;
+ explicit KVFinalizeThread(BlueStore *s) : store(s) {}
+ void *entry() {
+ store->_kv_finalize_thread();
+ return NULL;
+ }
+ };
+
+ struct DBHistogram {
+ struct value_dist {
+ uint64_t count;
+ uint32_t max_len;
+ };
+
+ struct key_dist {
+ uint64_t count;
+ uint32_t max_len;
+ map<int, struct value_dist> val_map; ///< slab id to count, max length of value and key
+ };
+
+ map<string, map<int, struct key_dist> > key_hist;
+ map<int, uint64_t> value_hist;
+ int get_key_slab(size_t sz);
+ string get_key_slab_to_range(int slab);
+ int get_value_slab(size_t sz);
+ string get_value_slab_to_range(int slab);
+ void update_hist_entry(map<string, map<int, struct key_dist> > &key_hist,
+ const string &prefix, size_t key_size, size_t value_size);
+ void dump(Formatter *f);
+ };
+
+ // --------------------------------------------------------
+ // members
+private:
+ BlueFS *bluefs = nullptr;
+ unsigned bluefs_shared_bdev = 0; ///< which bluefs bdev we are sharing
+ bool bluefs_single_shared_device = true;
+ utime_t bluefs_last_balance;
+
+ KeyValueDB *db = nullptr;
+ BlockDevice *bdev = nullptr;
+ std::string freelist_type;
+ FreelistManager *fm = nullptr;
+ Allocator *alloc = nullptr;
+ uuid_d fsid;
+ int path_fd = -1; ///< open handle to $path
+ int fsid_fd = -1; ///< open handle (locked) to $path/fsid
+ bool mounted = false;
+
+ RWLock coll_lock = {"BlueStore::coll_lock"}; ///< rwlock to protect coll_map
+ mempool::bluestore_cache_other::unordered_map<coll_t, CollectionRef> coll_map;
+
+ vector<Cache*> cache_shards;
+
+ std::mutex osr_lock; ///< protect osd_set
+ std::set<OpSequencerRef> osr_set; ///< set of all OpSequencers
+
+ std::atomic<uint64_t> nid_last = {0};
+ std::atomic<uint64_t> nid_max = {0};
+ std::atomic<uint64_t> blobid_last = {0};
+ std::atomic<uint64_t> blobid_max = {0};
+
+ Throttle throttle_bytes; ///< submit to commit
+ Throttle throttle_deferred_bytes; ///< submit to deferred complete
+
+ interval_set<uint64_t> bluefs_extents; ///< block extents owned by bluefs
+ interval_set<uint64_t> bluefs_extents_reclaiming; ///< currently reclaiming
+
+ std::mutex deferred_lock;
+ std::atomic<uint64_t> deferred_seq = {0};
+ deferred_osr_queue_t deferred_queue; ///< osr's with deferred io pending
+ int deferred_queue_size = 0; ///< num txc's queued across all osrs
+ atomic_int deferred_aggressive = {0}; ///< aggressive wakeup of kv thread
+ Finisher deferred_finisher;
+
+ int m_finisher_num = 1;
+ vector<Finisher*> finishers;
+
+ KVSyncThread kv_sync_thread;
+ std::mutex kv_lock;
+ std::condition_variable kv_cond;
+ bool _kv_only = false;
+ bool kv_sync_started = false;
+ bool kv_stop = false;
+ bool kv_finalize_started = false;
+ bool kv_finalize_stop = false;
+ deque<TransContext*> kv_queue; ///< ready, already submitted
+ deque<TransContext*> kv_queue_unsubmitted; ///< ready, need submit by kv thread
+ deque<TransContext*> kv_committing; ///< currently syncing
+ deque<DeferredBatch*> deferred_done_queue; ///< deferred ios done
+ deque<DeferredBatch*> deferred_stable_queue; ///< deferred ios done + stable
+
+ KVFinalizeThread kv_finalize_thread;
+ std::mutex kv_finalize_lock;
+ std::condition_variable kv_finalize_cond;
+ deque<TransContext*> kv_committing_to_finalize; ///< pending finalization
+ deque<DeferredBatch*> deferred_stable_to_finalize; ///< pending finalization
+
+ PerfCounters *logger = nullptr;
+
+ std::mutex reap_lock;
+ list<CollectionRef> removed_collections;
+
+ RWLock debug_read_error_lock = {"BlueStore::debug_read_error_lock"};
+ set<ghobject_t> debug_data_error_objects;
+ set<ghobject_t> debug_mdata_error_objects;
+
+ std::atomic<int> csum_type = {Checksummer::CSUM_CRC32C};
+
+ uint64_t block_size = 0; ///< block size of block device (power of 2)
+ uint64_t block_mask = 0; ///< mask to get just the block offset
+ size_t block_size_order = 0; ///< bits to shift to get block size
+
+ uint64_t min_alloc_size = 0; ///< minimum allocation unit (power of 2)
+ ///< bits for min_alloc_size
+ uint8_t min_alloc_size_order = 0;
+ static_assert(std::numeric_limits<uint8_t>::max() >
+ std::numeric_limits<decltype(min_alloc_size)>::digits,
+ "not enough bits for min_alloc_size");
+
+ ///< maximum allocation unit (power of 2)
+ std::atomic<uint64_t> max_alloc_size = {0};
+
+ ///< number threshold for forced deferred writes
+ std::atomic<int> deferred_batch_ops = {0};
+
+ ///< size threshold for forced deferred writes
+ std::atomic<uint64_t> prefer_deferred_size = {0};
+
+ ///< approx cost per io, in bytes
+ std::atomic<uint64_t> throttle_cost_per_io = {0};
+
+ std::atomic<Compressor::CompressionMode> comp_mode =
+ {Compressor::COMP_NONE}; ///< compression mode
+ CompressorRef compressor;
+ std::atomic<uint64_t> comp_min_blob_size = {0};
+ std::atomic<uint64_t> comp_max_blob_size = {0};
+
+ std::atomic<uint64_t> max_blob_size = {0}; ///< maximum blob size
+
+ uint64_t kv_ios = 0;
+ uint64_t kv_throttle_costs = 0;
+
+ // cache trim control
+ uint64_t cache_size = 0; ///< total cache size
+ float cache_meta_ratio = 0; ///< cache ratio dedicated to metadata
+ float cache_kv_ratio = 0; ///< cache ratio dedicated to kv (e.g., rocksdb)
+ float cache_data_ratio = 0; ///< cache ratio dedicated to object data
+
+ std::mutex vstatfs_lock;
+ volatile_statfs vstatfs;
+
+ struct MempoolThread : public Thread {
+ BlueStore *store;
+ Cond cond;
+ Mutex lock;
+ bool stop = false;
+ public:
+ explicit MempoolThread(BlueStore *s)
+ : store(s),
+ lock("BlueStore::MempoolThread::lock") {}
+ void *entry() override;
+ void init() {
+ assert(stop == false);
+ create("bstore_mempool");
+ }
+ void shutdown() {
+ lock.Lock();
+ stop = true;
+ cond.Signal();
+ lock.Unlock();
+ join();
+ }
+ } mempool_thread;
+
+ // --------------------------------------------------------
+ // private methods
+
+ void _init_logger();
+ void _shutdown_logger();
+ int _reload_logger();
+
+ int _open_path();
+ void _close_path();
+ int _open_fsid(bool create);
+ int _lock_fsid();
+ int _read_fsid(uuid_d *f);
+ int _write_fsid();
+ void _close_fsid();
+ void _set_alloc_sizes();
+ void _set_blob_size();
+
+ int _open_bdev(bool create);
+ void _close_bdev();
+ int _open_db(bool create);
+ void _close_db();
+ int _open_fm(bool create);
+ void _close_fm();
+ int _open_alloc();
+ void _close_alloc();
+ int _open_collections(int *errors=0);
+ void _close_collections();
+
+ int _setup_block_symlink_or_file(string name, string path, uint64_t size,
+ bool create);
+
+public:
+ static int _write_bdev_label(CephContext* cct,
+ string path, bluestore_bdev_label_t label);
+ static int _read_bdev_label(CephContext* cct, string path,
+ bluestore_bdev_label_t *label);
+private:
+ int _check_or_set_bdev_label(string path, uint64_t size, string desc,
+ bool create);
+
+ int _open_super_meta();
+
+ void _open_statfs();
+
+ int _reconcile_bluefs_freespace();
+ int _balance_bluefs_freespace(PExtentVector *extents);
+ void _commit_bluefs_freespace(const PExtentVector& extents);
+
+ CollectionRef _get_collection(const coll_t& cid);
+ void _queue_reap_collection(CollectionRef& c);
+ void _reap_collections();
+ void _update_cache_logger();
+
+ void _assign_nid(TransContext *txc, OnodeRef o);
+ uint64_t _assign_blobid(TransContext *txc);
+
+ void _dump_onode(OnodeRef o, int log_level=30);
+ void _dump_extent_map(ExtentMap& em, int log_level=30);
+ void _dump_transaction(Transaction *t, int log_level = 30);
+
+ TransContext *_txc_create(OpSequencer *osr);
+ void _txc_update_store_statfs(TransContext *txc);
+ void _txc_add_transaction(TransContext *txc, Transaction *t);
+ void _txc_calc_cost(TransContext *txc);
+ void _txc_write_nodes(TransContext *txc, KeyValueDB::Transaction t);
+ void _txc_state_proc(TransContext *txc);
+ void _txc_aio_submit(TransContext *txc);
+public:
+ void txc_aio_finish(void *p) {
+ _txc_state_proc(static_cast<TransContext*>(p));
+ }
+private:
+ void _txc_finish_io(TransContext *txc);
+ void _txc_finalize_kv(TransContext *txc, KeyValueDB::Transaction t);
+ void _txc_applied_kv(TransContext *txc);
+ void _txc_committed_kv(TransContext *txc);
+ void _txc_finish(TransContext *txc);
+ void _txc_release_alloc(TransContext *txc);
+
+ void _osr_drain_preceding(TransContext *txc);
+ void _osr_drain_all();
+ void _osr_unregister_all();
+
+ void _kv_start();
+ void _kv_stop();
+ void _kv_sync_thread();
+ void _kv_finalize_thread();
+
+ bluestore_deferred_op_t *_get_deferred_op(TransContext *txc, OnodeRef o);
+ void _deferred_queue(TransContext *txc);
+public:
+ void deferred_try_submit();
+private:
+ void _deferred_submit_unlock(OpSequencer *osr);
+ void _deferred_aio_finish(OpSequencer *osr);
+ int _deferred_replay();
+
+public:
+ using mempool_dynamic_bitset =
+ boost::dynamic_bitset<uint64_t,
+ mempool::bluestore_fsck::pool_allocator<uint64_t>>;
+
+private:
+ int _fsck_check_extents(
+ const ghobject_t& oid,
+ const PExtentVector& extents,
+ bool compressed,
+ mempool_dynamic_bitset &used_blocks,
+ store_statfs_t& expected_statfs);
+
+ void _buffer_cache_write(
+ TransContext *txc,
+ BlobRef b,
+ uint64_t offset,
+ bufferlist& bl,
+ unsigned flags) {
+ b->shared_blob->bc.write(b->shared_blob->get_cache(), txc->seq, offset, bl,
+ flags);
+ txc->shared_blobs_written.insert(b->shared_blob);
+ }
+
+ int _collection_list(
+ Collection *c, const ghobject_t& start, const ghobject_t& end,
+ int max, vector<ghobject_t> *ls, ghobject_t *next);
+
+ template <typename T, typename F>
+ T select_option(const std::string& opt_name, T val1, F f) {
+ //NB: opt_name reserved for future use
+ boost::optional<T> val2 = f();
+ if (val2) {
+ return *val2;
+ }
+ return val1;
+ }
+
+ void _apply_padding(uint64_t head_pad,
+ uint64_t tail_pad,
+ bufferlist& padded);
+
+ // -- ondisk version ---
+public:
+ const int32_t latest_ondisk_format = 2; ///< our version
+ const int32_t min_readable_ondisk_format = 1; ///< what we can read
+ const int32_t min_compat_ondisk_format = 2; ///< who can read us
+
+private:
+ int32_t ondisk_format = 0; ///< value detected on mount
+
+ int _upgrade_super(); ///< upgrade (called during open_super)
+ void _prepare_ondisk_format_super(KeyValueDB::Transaction& t);
+
+ // --- public interface ---
+public:
+ BlueStore(CephContext *cct, const string& path);
+ BlueStore(CephContext *cct, const string& path, uint64_t min_alloc_size); // Ctor for UT only
+ ~BlueStore() override;
+
+ string get_type() override {
+ return "bluestore";
+ }
+
+ bool needs_journal() override { return false; };
+ bool wants_journal() override { return false; };
+ bool allows_journal() override { return false; };
+
+ bool is_rotational() override;
+ bool is_journal_rotational() override;
+
+ string get_default_device_class() override {
+ string device_class;
+ map<string, string> metadata;
+ collect_metadata(&metadata);
+ auto it = metadata.find("bluestore_bdev_type");
+ if (it != metadata.end()) {
+ device_class = it->second;
+ }
+ return device_class;
+ }
+
+ static int get_block_device_fsid(CephContext* cct, const string& path,
+ uuid_d *fsid);
+
+ bool test_mount_in_use() override;
+
+private:
+ int _mount(bool kv_only);
+public:
+ int mount() override {
+ return _mount(false);
+ }
+ int umount() override;
+
+ int start_kv_only(KeyValueDB **pdb) {
+ int r = _mount(true);
+ if (r < 0)
+ return r;
+ *pdb = db;
+ return 0;
+ }
+
+ int write_meta(const std::string& key, const std::string& value) override;
+ int read_meta(const std::string& key, std::string *value) override;
+
+
+ int fsck(bool deep) override {
+ return _fsck(deep, false);
+ }
+ int repair(bool deep) override {
+ return _fsck(deep, true);
+ }
+ int _fsck(bool deep, bool repair);
+
+ void set_cache_shards(unsigned num) override;
+
+ int validate_hobject_key(const hobject_t &obj) const override {
+ return 0;
+ }
+ unsigned get_max_attr_name_length() override {
+ return 256; // arbitrary; there is no real limit internally
+ }
+
+ int mkfs() override;
+ int mkjournal() override {
+ return 0;
+ }
+
+ void get_db_statistics(Formatter *f) override;
+ void generate_db_histogram(Formatter *f) override;
+ void _flush_cache();
+ void flush_cache() override;
+ void dump_perf_counters(Formatter *f) override {
+ f->open_object_section("perf_counters");
+ logger->dump_formatted(f, false);
+ f->close_section();
+ }
+
+ void register_osr(OpSequencer *osr) {
+ std::lock_guard<std::mutex> l(osr_lock);
+ osr_set.insert(osr);
+ }
+ void unregister_osr(OpSequencer *osr) {
+ std::lock_guard<std::mutex> l(osr_lock);
+ osr_set.erase(osr);
+ }
+
+public:
+ int statfs(struct store_statfs_t *buf) override;
+
+ void collect_metadata(map<string,string> *pm) override;
+
+ bool exists(const coll_t& cid, const ghobject_t& oid) override;
+ bool exists(CollectionHandle &c, const ghobject_t& oid) override;
+ int set_collection_opts(
+ const coll_t& cid,
+ const pool_opts_t& opts) override;
+ int stat(
+ const coll_t& cid,
+ const ghobject_t& oid,
+ struct stat *st,
+ bool allow_eio = false) override;
+ int stat(
+ CollectionHandle &c,
+ const ghobject_t& oid,
+ struct stat *st,
+ bool allow_eio = false) override;
+ int read(
+ const coll_t& cid,
+ const ghobject_t& oid,
+ uint64_t offset,
+ size_t len,
+ bufferlist& bl,
+ uint32_t op_flags = 0) override;
+ int read(
+ CollectionHandle &c,
+ const ghobject_t& oid,
+ uint64_t offset,
+ size_t len,
+ bufferlist& bl,
+ uint32_t op_flags = 0) override;
+ int _do_read(
+ Collection *c,
+ OnodeRef o,
+ uint64_t offset,
+ size_t len,
+ bufferlist& bl,
+ uint32_t op_flags = 0);
+
+private:
+ int _fiemap(CollectionHandle &c_, const ghobject_t& oid,
+ uint64_t offset, size_t len, interval_set<uint64_t>& destset);
+public:
+ int fiemap(const coll_t& cid, const ghobject_t& oid,
+ uint64_t offset, size_t len, bufferlist& bl) override;
+ int fiemap(CollectionHandle &c, const ghobject_t& oid,
+ uint64_t offset, size_t len, bufferlist& bl) override;
+ int fiemap(const coll_t& cid, const ghobject_t& oid,
+ uint64_t offset, size_t len, map<uint64_t, uint64_t>& destmap) override;
+ int fiemap(CollectionHandle &c, const ghobject_t& oid,
+ uint64_t offset, size_t len, map<uint64_t, uint64_t>& destmap) override;
+
+
+ int getattr(const coll_t& cid, const ghobject_t& oid, const char *name,
+ bufferptr& value) override;
+ int getattr(CollectionHandle &c, const ghobject_t& oid, const char *name,
+ bufferptr& value) override;
+
+ int getattrs(const coll_t& cid, const ghobject_t& oid,
+ map<string,bufferptr>& aset) override;
+ int getattrs(CollectionHandle &c, const ghobject_t& oid,
+ map<string,bufferptr>& aset) override;
+
+ int list_collections(vector<coll_t>& ls) override;
+
+ CollectionHandle open_collection(const coll_t &c) override;
+
+ bool collection_exists(const coll_t& c) override;
+ int collection_empty(const coll_t& c, bool *empty) override;
+ int collection_bits(const coll_t& c) override;
+
+ int collection_list(const coll_t& cid,
+ const ghobject_t& start,
+ const ghobject_t& end,
+ int max,
+ vector<ghobject_t> *ls, ghobject_t *next) override;
+ int collection_list(CollectionHandle &c,
+ const ghobject_t& start,
+ const ghobject_t& end,
+ int max,
+ vector<ghobject_t> *ls, ghobject_t *next) override;
+
+ int omap_get(
+ const coll_t& cid, ///< [in] Collection containing oid
+ const ghobject_t &oid, ///< [in] Object containing omap
+ bufferlist *header, ///< [out] omap header
+ map<string, bufferlist> *out /// < [out] Key to value map
+ ) override;
+ int omap_get(
+ CollectionHandle &c, ///< [in] Collection containing oid
+ const ghobject_t &oid, ///< [in] Object containing omap
+ bufferlist *header, ///< [out] omap header
+ map<string, bufferlist> *out /// < [out] Key to value map
+ ) override;
+
+ /// Get omap header
+ int omap_get_header(
+ const coll_t& cid, ///< [in] Collection containing oid
+ const ghobject_t &oid, ///< [in] Object containing omap
+ bufferlist *header, ///< [out] omap header
+ bool allow_eio = false ///< [in] don't assert on eio
+ ) override;
+ int omap_get_header(
+ CollectionHandle &c, ///< [in] Collection containing oid
+ const ghobject_t &oid, ///< [in] Object containing omap
+ bufferlist *header, ///< [out] omap header
+ bool allow_eio = false ///< [in] don't assert on eio
+ ) override;
+
+ /// Get keys defined on oid
+ int omap_get_keys(
+ const coll_t& cid, ///< [in] Collection containing oid
+ const ghobject_t &oid, ///< [in] Object containing omap
+ set<string> *keys ///< [out] Keys defined on oid
+ ) override;
+ int omap_get_keys(
+ CollectionHandle &c, ///< [in] Collection containing oid
+ const ghobject_t &oid, ///< [in] Object containing omap
+ set<string> *keys ///< [out] Keys defined on oid
+ ) override;
+
+ /// Get key values
+ int omap_get_values(
+ const coll_t& cid, ///< [in] Collection containing oid
+ const ghobject_t &oid, ///< [in] Object containing omap
+ const set<string> &keys, ///< [in] Keys to get
+ map<string, bufferlist> *out ///< [out] Returned keys and values
+ ) override;
+ int omap_get_values(
+ CollectionHandle &c, ///< [in] Collection containing oid
+ const ghobject_t &oid, ///< [in] Object containing omap
+ const set<string> &keys, ///< [in] Keys to get
+ map<string, bufferlist> *out ///< [out] Returned keys and values
+ ) override;
+
+ /// Filters keys into out which are defined on oid
+ int omap_check_keys(
+ const coll_t& cid, ///< [in] Collection containing oid
+ const ghobject_t &oid, ///< [in] Object containing omap
+ const set<string> &keys, ///< [in] Keys to check
+ set<string> *out ///< [out] Subset of keys defined on oid
+ ) override;
+ int omap_check_keys(
+ CollectionHandle &c, ///< [in] Collection containing oid
+ const ghobject_t &oid, ///< [in] Object containing omap
+ const set<string> &keys, ///< [in] Keys to check
+ set<string> *out ///< [out] Subset of keys defined on oid
+ ) override;
+
+ ObjectMap::ObjectMapIterator get_omap_iterator(
+ const coll_t& cid, ///< [in] collection
+ const ghobject_t &oid ///< [in] object
+ ) override;
+ ObjectMap::ObjectMapIterator get_omap_iterator(
+ CollectionHandle &c, ///< [in] collection
+ const ghobject_t &oid ///< [in] object
+ ) override;
+
+ void set_fsid(uuid_d u) override {
+ fsid = u;
+ }
+ uuid_d get_fsid() override {
+ return fsid;
+ }
+
+ uint64_t estimate_objects_overhead(uint64_t num_objects) override {
+ return num_objects * 300; //assuming per-object overhead is 300 bytes
+ }
+
+ struct BSPerfTracker {
+ PerfCounters::avg_tracker<uint64_t> os_commit_latency;
+ PerfCounters::avg_tracker<uint64_t> os_apply_latency;
+
+ objectstore_perf_stat_t get_cur_stats() const {
+ objectstore_perf_stat_t ret;
+ ret.os_commit_latency = os_commit_latency.current_avg();
+ ret.os_apply_latency = os_apply_latency.current_avg();
+ return ret;
+ }
+
+ void update_from_perfcounters(PerfCounters &logger);
+ } perf_tracker;
+
+ objectstore_perf_stat_t get_cur_stats() override {
+ perf_tracker.update_from_perfcounters(*logger);
+ return perf_tracker.get_cur_stats();
+ }
+ const PerfCounters* get_perf_counters() const override {
+ return logger;
+ }
+
+ int queue_transactions(
+ Sequencer *osr,
+ vector<Transaction>& tls,
+ TrackedOpRef op = TrackedOpRef(),
+ ThreadPool::TPHandle *handle = NULL) override;
+
+ // error injection
+ void inject_data_error(const ghobject_t& o) override {
+ RWLock::WLocker l(debug_read_error_lock);
+ debug_data_error_objects.insert(o);
+ }
+ void inject_mdata_error(const ghobject_t& o) override {
+ RWLock::WLocker l(debug_read_error_lock);
+ debug_mdata_error_objects.insert(o);
+ }
+ void compact() override {
+ assert(db);
+ db->compact();
+ }
+
+private:
+ bool _debug_data_eio(const ghobject_t& o) {
+ if (!cct->_conf->bluestore_debug_inject_read_err) {
+ return false;
+ }
+ RWLock::RLocker l(debug_read_error_lock);
+ return debug_data_error_objects.count(o);
+ }
+ bool _debug_mdata_eio(const ghobject_t& o) {
+ if (!cct->_conf->bluestore_debug_inject_read_err) {
+ return false;
+ }
+ RWLock::RLocker l(debug_read_error_lock);
+ return debug_mdata_error_objects.count(o);
+ }
+ void _debug_obj_on_delete(const ghobject_t& o) {
+ if (cct->_conf->bluestore_debug_inject_read_err) {
+ RWLock::WLocker l(debug_read_error_lock);
+ debug_data_error_objects.erase(o);
+ debug_mdata_error_objects.erase(o);
+ }
+ }
+
+private:
+
+ // --------------------------------------------------------
+ // read processing internal methods
+ int _verify_csum(
+ OnodeRef& o,
+ const bluestore_blob_t* blob,
+ uint64_t blob_xoffset,
+ const bufferlist& bl,
+ uint64_t logical_offset) const;
+ int _decompress(bufferlist& source, bufferlist* result);
+
+
+ // --------------------------------------------------------
+ // write ops
+
+ struct WriteContext {
+ bool buffered = false; ///< buffered write
+ bool compress = false; ///< compressed write
+ uint64_t target_blob_size = 0; ///< target (max) blob size
+ unsigned csum_order = 0; ///< target checksum chunk order
+
+ old_extent_map_t old_extents; ///< must deref these blobs
+
+ struct write_item {
+ uint64_t logical_offset; ///< write logical offset
+ BlobRef b;
+ uint64_t blob_length;
+ uint64_t b_off;
+ bufferlist bl;
+ uint64_t b_off0; ///< original offset in a blob prior to padding
+ uint64_t length0; ///< original data length prior to padding
+
+ bool mark_unused;
+ bool new_blob; ///< whether new blob was created
+
+ bool compressed = false;
+ bufferlist compressed_bl;
+ size_t compressed_len = 0;
+
+ write_item(
+ uint64_t logical_offs,
+ BlobRef b,
+ uint64_t blob_len,
+ uint64_t o,
+ bufferlist& bl,
+ uint64_t o0,
+ uint64_t l0,
+ bool _mark_unused,
+ bool _new_blob)
+ :
+ logical_offset(logical_offs),
+ b(b),
+ blob_length(blob_len),
+ b_off(o),
+ bl(bl),
+ b_off0(o0),
+ length0(l0),
+ mark_unused(_mark_unused),
+ new_blob(_new_blob) {}
+ };
+ vector<write_item> writes; ///< blobs we're writing
+
+ /// partial clone of the context
+ void fork(const WriteContext& other) {
+ buffered = other.buffered;
+ compress = other.compress;
+ target_blob_size = other.target_blob_size;
+ csum_order = other.csum_order;
+ }
+ void write(
+ uint64_t loffs,
+ BlobRef b,
+ uint64_t blob_len,
+ uint64_t o,
+ bufferlist& bl,
+ uint64_t o0,
+ uint64_t len0,
+ bool _mark_unused,
+ bool _new_blob) {
+ writes.emplace_back(loffs,
+ b,
+ blob_len,
+ o,
+ bl,
+ o0,
+ len0,
+ _mark_unused,
+ _new_blob);
+ }
+ /// Checks for writes to the same pextent within a blob
+ bool has_conflict(
+ BlobRef b,
+ uint64_t loffs,
+ uint64_t loffs_end,
+ uint64_t min_alloc_size);
+ };
+
+ void _do_write_small(
+ TransContext *txc,
+ CollectionRef &c,
+ OnodeRef o,
+ uint64_t offset, uint64_t length,
+ bufferlist::iterator& blp,
+ WriteContext *wctx);
+ void _do_write_big(
+ TransContext *txc,
+ CollectionRef &c,
+ OnodeRef o,
+ uint64_t offset, uint64_t length,
+ bufferlist::iterator& blp,
+ WriteContext *wctx);
+ int _do_alloc_write(
+ TransContext *txc,
+ CollectionRef c,
+ OnodeRef o,
+ WriteContext *wctx);
+ void _wctx_finish(
+ TransContext *txc,
+ CollectionRef& c,
+ OnodeRef o,
+ WriteContext *wctx,
+ set<SharedBlob*> *maybe_unshared_blobs=0);
+
+ int _do_transaction(Transaction *t,
+ TransContext *txc,
+ ThreadPool::TPHandle *handle);
+
+ int _write(TransContext *txc,
+ CollectionRef& c,
+ OnodeRef& o,
+ uint64_t offset, size_t len,
+ bufferlist& bl,
+ uint32_t fadvise_flags);
+ void _pad_zeros(bufferlist *bl, uint64_t *offset,
+ uint64_t chunk_size);
+
+ void _choose_write_options(CollectionRef& c,
+ OnodeRef o,
+ uint32_t fadvise_flags,
+ WriteContext *wctx);
+
+ int _do_gc(TransContext *txc,
+ CollectionRef& c,
+ OnodeRef o,
+ const GarbageCollector& gc,
+ const WriteContext& wctx,
+ uint64_t *dirty_start,
+ uint64_t *dirty_end);
+
+ int _do_write(TransContext *txc,
+ CollectionRef &c,
+ OnodeRef o,
+ uint64_t offset, uint64_t length,
+ bufferlist& bl,
+ uint32_t fadvise_flags);
+ void _do_write_data(TransContext *txc,
+ CollectionRef& c,
+ OnodeRef o,
+ uint64_t offset,
+ uint64_t length,
+ bufferlist& bl,
+ WriteContext *wctx);
+
+ int _touch(TransContext *txc,
+ CollectionRef& c,
+ OnodeRef& o);
+ int _do_zero(TransContext *txc,
+ CollectionRef& c,
+ OnodeRef& o,
+ uint64_t offset, size_t len);
+ int _zero(TransContext *txc,
+ CollectionRef& c,
+ OnodeRef& o,
+ uint64_t offset, size_t len);
+ void _do_truncate(TransContext *txc,
+ CollectionRef& c,
+ OnodeRef o,
+ uint64_t offset,
+ set<SharedBlob*> *maybe_unshared_blobs=0);
+ int _truncate(TransContext *txc,
+ CollectionRef& c,
+ OnodeRef& o,
+ uint64_t offset);
+ int _remove(TransContext *txc,
+ CollectionRef& c,
+ OnodeRef& o);
+ int _do_remove(TransContext *txc,
+ CollectionRef& c,
+ OnodeRef o);
+ int _setattr(TransContext *txc,
+ CollectionRef& c,
+ OnodeRef& o,
+ const string& name,
+ bufferptr& val);
+ int _setattrs(TransContext *txc,
+ CollectionRef& c,
+ OnodeRef& o,
+ const map<string,bufferptr>& aset);
+ int _rmattr(TransContext *txc,
+ CollectionRef& c,
+ OnodeRef& o,
+ const string& name);
+ int _rmattrs(TransContext *txc,
+ CollectionRef& c,
+ OnodeRef& o);
+ void _do_omap_clear(TransContext *txc, uint64_t id);
+ int _omap_clear(TransContext *txc,
+ CollectionRef& c,
+ OnodeRef& o);
+ int _omap_setkeys(TransContext *txc,
+ CollectionRef& c,
+ OnodeRef& o,
+ bufferlist& bl);
+ int _omap_setheader(TransContext *txc,
+ CollectionRef& c,
+ OnodeRef& o,
+ bufferlist& header);
+ int _omap_rmkeys(TransContext *txc,
+ CollectionRef& c,
+ OnodeRef& o,
+ bufferlist& bl);
+ int _omap_rmkey_range(TransContext *txc,
+ CollectionRef& c,
+ OnodeRef& o,
+ const string& first, const string& last);
+ int _set_alloc_hint(
+ TransContext *txc,
+ CollectionRef& c,
+ OnodeRef& o,
+ uint64_t expected_object_size,
+ uint64_t expected_write_size,
+ uint32_t flags);
+ int _do_clone_range(TransContext *txc,
+ CollectionRef& c,
+ OnodeRef& oldo,
+ OnodeRef& newo,
+ uint64_t srcoff, uint64_t length, uint64_t dstoff);
+ int _clone(TransContext *txc,
+ CollectionRef& c,
+ OnodeRef& oldo,
+ OnodeRef& newo);
+ int _clone_range(TransContext *txc,
+ CollectionRef& c,
+ OnodeRef& oldo,
+ OnodeRef& newo,
+ uint64_t srcoff, uint64_t length, uint64_t dstoff);
+ int _rename(TransContext *txc,
+ CollectionRef& c,
+ OnodeRef& oldo,
+ OnodeRef& newo,
+ const ghobject_t& new_oid);
+ int _create_collection(TransContext *txc, const coll_t &cid,
+ unsigned bits, CollectionRef *c);
+ int _remove_collection(TransContext *txc, const coll_t &cid,
+ CollectionRef *c);
+ int _split_collection(TransContext *txc,
+ CollectionRef& c,
+ CollectionRef& d,
+ unsigned bits, int rem);
+};
+
+inline ostream& operator<<(ostream& out, const BlueStore::OpSequencer& s) {
+ return out << *s.parent;
+}
+
+static inline void intrusive_ptr_add_ref(BlueStore::Onode *o) {
+ o->get();
+}
+static inline void intrusive_ptr_release(BlueStore::Onode *o) {
+ o->put();
+}
+
+static inline void intrusive_ptr_add_ref(BlueStore::OpSequencer *o) {
+ o->get();
+}
+static inline void intrusive_ptr_release(BlueStore::OpSequencer *o) {
+ o->put();
+}
+
+#endif
Code Review / stor4nfv.git / blobdiff