// -*- 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 #include #include #include #include #include #include #include #include #include #include #include #include #include "include/cpp-btree/btree_set.h" #include "include/ceph_assert.h" #include "include/interval_set.h" #include "include/unordered_map.h" #include "include/mempool.h" #include "include/hash.h" #include "common/bloom_filter.hpp" #include "common/Finisher.h" #include "common/ceph_mutex.h" #include "common/Throttle.h" #include "common/perf_counters.h" #include "common/PriorityCache.h" #include "compressor/Compressor.h" #include "os/ObjectStore.h" #include "bluestore_types.h" #include "BlueFS.h" #include "common/EventTrace.h" #ifdef WITH_BLKIN #include "common/zipkin_trace.h" #endif class Allocator; class FreelistManager; class BlueStoreRepairer; class SimpleBitmap; //#define DEBUG_CACHE //#define DEBUG_DEFERRED // constants for Buffer::optimize() #define MAX_BUFFER_SLOP_RATIO_DEN 8 // so actually 1/N #define CEPH_BLUESTORE_TOOL_RESTORE_ALLOCATION enum { l_bluestore_first = 732430, // space utilization stats //**************************************** l_bluestore_allocated, l_bluestore_stored, l_bluestore_fragmentation, l_bluestore_alloc_unit, //**************************************** // Update op processing state latencies //**************************************** 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_finishing_lat, l_bluestore_state_done_lat, l_bluestore_state_deferred_queued_lat, l_bluestore_state_deferred_aio_wait_lat, l_bluestore_state_deferred_cleanup_lat, l_bluestore_commit_lat, //**************************************** // Update Transaction stats //**************************************** l_bluestore_throttle_lat, l_bluestore_submit_lat, l_bluestore_txc, //**************************************** // Read op stats //**************************************** l_bluestore_read_onode_meta_lat, l_bluestore_read_wait_aio_lat, l_bluestore_csum_lat, l_bluestore_read_eio, l_bluestore_reads_with_retries, l_bluestore_read_lat, //**************************************** // kv_thread latencies //**************************************** l_bluestore_kv_flush_lat, l_bluestore_kv_commit_lat, l_bluestore_kv_sync_lat, l_bluestore_kv_final_lat, //**************************************** // write op stats //**************************************** l_bluestore_write_big, l_bluestore_write_big_bytes, l_bluestore_write_big_blobs, l_bluestore_write_big_deferred, l_bluestore_write_small, l_bluestore_write_small_bytes, l_bluestore_write_small_unused, l_bluestore_write_small_pre_read, l_bluestore_write_pad_bytes, l_bluestore_write_penalty_read_ops, l_bluestore_write_new, l_bluestore_issued_deferred_writes, l_bluestore_issued_deferred_write_bytes, l_bluestore_submitted_deferred_writes, l_bluestore_submitted_deferred_write_bytes, l_bluestore_write_big_skipped_blobs, l_bluestore_write_big_skipped_bytes, l_bluestore_write_small_skipped, l_bluestore_write_small_skipped_bytes, //**************************************** // compressions stats //**************************************** l_bluestore_compressed, l_bluestore_compressed_allocated, l_bluestore_compressed_original, l_bluestore_compress_lat, l_bluestore_decompress_lat, l_bluestore_compress_success_count, l_bluestore_compress_rejected_count, //**************************************** // onode cache stats //**************************************** l_bluestore_onodes, l_bluestore_pinned_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, //**************************************** // buffer cache stats //**************************************** l_bluestore_buffers, l_bluestore_buffer_bytes, l_bluestore_buffer_hit_bytes, l_bluestore_buffer_miss_bytes, //**************************************** // internal stats //**************************************** l_bluestore_onode_reshard, l_bluestore_blob_split, l_bluestore_extent_compress, l_bluestore_gc_merged, //**************************************** // misc //**************************************** l_bluestore_omap_iterator_count, l_bluestore_omap_rmkeys_count, l_bluestore_omap_rmkey_ranges_count, //**************************************** // other client ops latencies //**************************************** l_bluestore_omap_seek_to_first_lat, l_bluestore_omap_upper_bound_lat, l_bluestore_omap_lower_bound_lat, l_bluestore_omap_next_lat, l_bluestore_omap_get_keys_lat, l_bluestore_omap_get_values_lat, l_bluestore_omap_clear_lat, l_bluestore_clist_lat, l_bluestore_remove_lat, l_bluestore_truncate_lat, //**************************************** // allocation stats //**************************************** l_bluestore_allocate_hist, //**************************************** l_bluestore_last }; #define META_POOL_ID ((uint64_t)-1ull) using bptr_c_it_t = buffer::ptr::const_iterator; 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 ConfigProxy& conf, const std::set &changed) override; //handler for discard event void handle_discard(interval_set& to_release); void _set_csum(); void _set_compression(); void _set_throttle_params(); int _set_cache_sizes(); void _set_max_defer_interval() { max_defer_interval = cct->_conf.get_val("bluestore_max_defer_interval"); } struct TransContext; typedef std::map ready_regions_t; struct BufferSpace; struct Collection; typedef boost::intrusive_ptr 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; ceph::buffer::list data; std::shared_ptr cache_age_bin; ///< cache age bin 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, ceph::buffer::list& 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) { ceph_assert(newlen < length); if (data.length()) { ceph::buffer::list t; t.substr_of(data, 0, newlen); data = std::move(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(ceph::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 BufferCacheShard; /// map logical extent range (object) onto buffers struct BufferSpace { enum { BYPASS_CLEAN_CACHE = 0x1, // bypass clean cache }; typedef boost::intrusive::list< Buffer, boost::intrusive::member_hook< Buffer, boost::intrusive::list_member_hook<>, &Buffer::state_item> > state_list_t; mempool::bluestore_cache_meta::map> 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() { ceph_assert(buffer_map.empty()); ceph_assert(writing.empty()); } void _add_buffer(BufferCacheShard* cache, Buffer* b, int level, Buffer* near) { cache->_audit("_add_buffer start"); buffer_map[b->offset].reset(b); if (b->is_writing()) { // we might get already cached data for which resetting mempool is inppropriate // hence calling try_assign_to_mempool b->data.try_assign_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; } ceph_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(b, level, near); } cache->_audit("_add_buffer end"); } void _rm_buffer(BufferCacheShard* cache, Buffer *b) { _rm_buffer(cache, buffer_map.find(b->offset)); } void _rm_buffer(BufferCacheShard* cache, std::map>::iterator p) { ceph_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(p->second.get()); } buffer_map.erase(p); cache->_audit("_rm_buffer end"); } std::map>::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(BufferCacheShard* cache); // return value is the highest cache_private of a trimmed buffer, or 0. int discard(BufferCacheShard* cache, uint32_t offset, uint32_t length) { std::lock_guard l(cache->lock); int ret = _discard(cache, offset, length); cache->_trim(); return ret; } int _discard(BufferCacheShard* cache, uint32_t offset, uint32_t length); void write(BufferCacheShard* cache, uint64_t seq, uint32_t offset, ceph::buffer::list& bl, unsigned flags) { std::lock_guard 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); cache->_trim(); } void _finish_write(BufferCacheShard* cache, uint64_t seq); void did_read(BufferCacheShard* cache, uint32_t offset, ceph::buffer::list& bl) { std::lock_guard 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); cache->_trim(); } void read(BufferCacheShard* cache, uint32_t offset, uint32_t length, BlueStore::ready_regions_t& res, interval_set& res_intervals, int flags = 0); void truncate(BufferCacheShard* cache, uint32_t offset) { discard(cache, offset, (uint32_t)-1 - offset); } void split(BufferCacheShard* cache, size_t pos, BufferSpace &r); void dump(BufferCacheShard* cache, ceph::Formatter *f) const { std::lock_guard l(cache->lock); f->open_array_section("buffers"); for (auto& i : buffer_map) { f->open_object_section("buffer"); ceph_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(); } void dump(ceph::Formatter* f) const; friend std::ostream& operator<<(std::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, bool *unshare); void finish_write(uint64_t seq); friend bool operator==(const SharedBlob &l, const SharedBlob &r) { return l.get_sbid() == r.get_sbid(); } inline BufferCacheShard* 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 SharedBlobRef; /// a lookup table of SharedBlobs struct SharedBlobSet { /// protect lookup, insertion, removal ceph::mutex lock = ceph::make_mutex("BlueStore::SharedBlobSet::lock"); // we use a bare pointer because we don't want to affect the ref // count mempool::bluestore_cache_meta::unordered_map sb_map; SharedBlobRef lookup(uint64_t sbid) { std::lock_guard l(lock); auto p = sb_map.find(sbid); if (p == sb_map.end() || p->second->nref == 0) { return nullptr; } return p->second; } void add(Collection* coll, SharedBlob *sb) { std::lock_guard l(lock); sb_map[sb->get_sbid()] = sb; sb->coll = coll; } bool remove(SharedBlob *sb, bool verify_nref_is_zero=false) { std::lock_guard l(lock); ceph_assert(sb->get_parent() == this); if (verify_nref_is_zero && sb->nref != 0) { return false; } // only remove if it still points to us auto p = sb_map.find(sb->get_sbid()); if (p != sb_map.end() && p->second == sb) { sb_map.erase(p); } return true; } bool empty() { std::lock_guard l(lock); return sb_map.empty(); } template void dump(CephContext *cct); }; //#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 ceph::buffer::list 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(); } void dump(ceph::Formatter* f) const; friend std::ostream& operator<<(std::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 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 { ceph_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( ceph::buffer::list::contiguous_appender& p, bool include_ref_map) const { _encode(); p.append(blob_bl); if (include_ref_map) { used_in_blob.encode(p); } } void decode( ceph::buffer::ptr::const_iterator& p, bool include_ref_map, Collection */*coll*/) { 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( ceph::buffer::list::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( ceph::buffer::ptr::const_iterator& p, uint64_t struct_v, uint64_t* sbid, bool include_ref_map, Collection *coll); #endif }; typedef boost::intrusive_ptr BlobRef; typedef mempool::bluestore_cache_meta::map blob_map_t; /// a logical extent, pointing to (some portion of) a blob typedef boost::intrusive::set_base_hook > 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 dump(ceph::Formatter* f) const; void assign_blob(const BlobRef& b) { ceph_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_map_t; friend std::ostream& operator<<(std::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 typedef boost::intrusive_ptr OnodeRef; 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_meta::vector shards; ///< shards ceph::buffer::list inline_bl; ///< cached encoded map, if unsharded; empty=>dirty uint32_t needs_reshard_begin = 0; uint32_t needs_reshard_end = 0; void dup(BlueStore* b, TransContext*, CollectionRef&, OnodeRef&, OnodeRef&, uint64_t&, uint64_t&, uint64_t&); 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, size_t inline_shard_prealloc_size); ~ExtentMap() { extent_map.clear_and_dispose(DeleteDisposer()); } void clear() { extent_map.clear_and_dispose(DeleteDisposer()); shards.clear(); inline_bl.clear(); clear_needs_reshard(); } void dump(ceph::Formatter* f) const; bool encode_some(uint32_t offset, uint32_t length, ceph::buffer::list& bl, unsigned *pn); class ExtentDecoder { uint64_t pos = 0; uint64_t prev_len = 0; uint64_t extent_pos = 0; protected: virtual void consume_blobid(Extent* le, bool spanning, uint64_t blobid) = 0; virtual void consume_blob(Extent* le, uint64_t extent_no, uint64_t sbid, BlobRef b) = 0; virtual void consume_spanning_blob(uint64_t sbid, BlobRef b) = 0; virtual Extent* get_next_extent() = 0; virtual void add_extent(Extent*) = 0; void decode_extent(Extent* le, __u8 struct_v, bptr_c_it_t& p, Collection* c); public: virtual ~ExtentDecoder() { } unsigned decode_some(const ceph::buffer::list& bl, Collection* c); void decode_spanning_blobs(bptr_c_it_t& p, Collection* c); }; class ExtentDecoderFull : public ExtentDecoder { ExtentMap& extent_map; std::vector blobs; protected: void consume_blobid(Extent* le, bool spanning, uint64_t blobid) override; void consume_blob(Extent* le, uint64_t extent_no, uint64_t sbid, BlobRef b) override; void consume_spanning_blob(uint64_t sbid, BlobRef b) override; Extent* get_next_extent() override; void add_extent(Extent* ) override; public: ExtentDecoderFull (ExtentMap& _extent_map) : extent_map(_extent_map) { } }; unsigned decode_some(ceph::buffer::list& bl); void bound_encode_spanning_blobs(size_t& p); void encode_spanning_blobs(ceph::buffer::list::contiguous_appender& p); BlobRef get_spanning_blob(int id) { auto p = spanning_blob_map.find(id); ceph_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); ceph_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 std::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 -> blob1 extent2 -> blob2 extent3 -> blob1 extent4 -> blob3 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 interval_set& 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; std::map affected_blobs; ///< compressed blobs and their ref_map ///< copies that are affected by the ///< specific write ///< protrusive extents that should be collected if GC takes place interval_set extents_to_collect; boost::optional 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; ///< std::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 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; struct OnodeCacheShard; /// an in-memory object struct Onode { MEMPOOL_CLASS_HELPERS(); std::atomic_int nref = 0; ///< reference count std::atomic_int pin_nref = 0; ///< reference count replica to track pinning Collection *c; ghobject_t oid; /// key under PREFIX_OBJ where we are stored mempool::bluestore_cache_meta::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 bool cached; ///< Onode is logically in the cache /// (it can be pinned and hence physically out /// of it at the moment though) 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 flushing_count = {0}; std::atomic waiting_count = {0}; /// protect flush_txns ceph::mutex flush_lock = ceph::make_mutex("BlueStore::Onode::flush_lock"); ceph::condition_variable flush_cond; ///< wait here for uncommitted txns std::shared_ptr cache_age_bin; ///< cache age bin Onode(Collection *c, const ghobject_t& o, const mempool::bluestore_cache_meta::string& k) : c(c), oid(o), key(k), exists(false), cached(false), extent_map(this, c->store->cct->_conf-> bluestore_extent_map_inline_shard_prealloc_size) { } Onode(Collection* c, const ghobject_t& o, const std::string& k) : c(c), oid(o), key(k), exists(false), cached(false), extent_map(this, c->store->cct->_conf-> bluestore_extent_map_inline_shard_prealloc_size) { } Onode(Collection* c, const ghobject_t& o, const char* k) : c(c), oid(o), key(k), exists(false), cached(false), extent_map(this, c->store->cct->_conf-> bluestore_extent_map_inline_shard_prealloc_size) { } Onode(CephContext* cct) : c(nullptr), exists(false), cached(false), extent_map(this, cct->_conf-> bluestore_extent_map_inline_shard_prealloc_size) { } static void decode_raw( BlueStore::Onode* on, const bufferlist& v, ExtentMap::ExtentDecoder& dencoder); static Onode* create_decode( CollectionRef c, const ghobject_t& oid, const std::string& key, const ceph::buffer::list& v, bool allow_empty = false); void dump(ceph::Formatter* f) const; void flush(); void get(); void put(); inline bool is_cached() const { return cached; } inline void set_cached() { ceph_assert(!cached); cached = true; } inline void clear_cached() { ceph_assert(cached); cached = false; } static const std::string& calc_omap_prefix(uint8_t flags); static void calc_omap_header(uint8_t flags, const Onode* o, std::string* out); static void calc_omap_key(uint8_t flags, const Onode* o, const std::string& key, std::string* out); static void calc_omap_tail(uint8_t flags, const Onode* o, std::string* out); const std::string& get_omap_prefix() { return calc_omap_prefix(onode.flags); } void get_omap_header(std::string* out) { calc_omap_header(onode.flags, this, out); } void get_omap_key(const std::string& key, std::string* out) { calc_omap_key(onode.flags, this, key, out); } void get_omap_tail(std::string* out) { calc_omap_tail(onode.flags, this, out); } void rewrite_omap_key(const std::string& old, std::string *out); void decode_omap_key(const std::string& key, std::string *user_key); #ifdef HAVE_LIBZBD // Return the offset of an object on disk. This function is intended *only* // for use with zoned storage devices because in these devices, the objects // are laid out contiguously on disk, which is not the case in general. // Also, it should always be called after calling extent_map.fault_range(), // so that the extent map is loaded. int64_t zoned_get_ondisk_starting_offset() const { return extent_map.extent_map.begin()->blob-> get_blob().calc_offset(0, nullptr); } #endif private: void _decode(const ceph::buffer::list& v); }; typedef boost::intrusive_ptr OnodeRef; /// A generic Cache Shard struct CacheShard { CephContext *cct; PerfCounters *logger; /// protect lru and other structures ceph::recursive_mutex lock = { ceph::make_recursive_mutex("BlueStore::CacheShard::lock") }; std::atomic max = {0}; std::atomic num = {0}; boost::circular_buffer> age_bins; CacheShard(CephContext* cct) : cct(cct), logger(nullptr), age_bins(1) { shift_bins(); } virtual ~CacheShard() {} void set_max(uint64_t max_) { max = max_; } uint64_t _get_num() { return num; } virtual void _trim_to(uint64_t new_size) = 0; void _trim() { if (cct->_conf->objectstore_blackhole) { // do not trim if we are throwing away IOs a layer down return; } _trim_to(max); } void trim() { std::lock_guard l(lock); _trim(); } void flush() { std::lock_guard l(lock); // we should not be shutting down after the blackhole is enabled ceph_assert(!cct->_conf->objectstore_blackhole); _trim_to(0); } virtual void shift_bins() { std::lock_guard l(lock); age_bins.push_front(std::make_shared(0)); } virtual uint32_t get_bin_count() { std::lock_guard l(lock); return age_bins.capacity(); } virtual void set_bin_count(uint32_t count) { std::lock_guard l(lock); age_bins.set_capacity(count); } virtual uint64_t sum_bins(uint32_t start, uint32_t end) { std::lock_guard l(lock); auto size = age_bins.size(); if (size < start) { return 0; } uint64_t count = 0; end = (size < end) ? size : end; for (auto i = start; i < end; i++) { count += *(age_bins[i]); } return count; } #ifdef DEBUG_CACHE virtual void _audit(const char *s) = 0; #else void _audit(const char *s) { /* no-op */ } #endif }; /// A Generic onode Cache Shard struct OnodeCacheShard : public CacheShard { std::array, 64> dumped_onodes; public: OnodeCacheShard(CephContext* cct) : CacheShard(cct) {} static OnodeCacheShard *create(CephContext* cct, std::string type, PerfCounters *logger); //The following methods prefixed with '_' to be called under // Shard's lock virtual void _add(Onode* o, int level) = 0; virtual void _rm(Onode* o) = 0; virtual void _move_pinned(OnodeCacheShard *to, Onode *o) = 0; virtual void maybe_unpin(Onode* o) = 0; virtual void add_stats(uint64_t *onodes, uint64_t *pinned_onodes) = 0; bool empty() { return _get_num() == 0; } }; /// A Generic buffer Cache Shard struct BufferCacheShard : public CacheShard { std::atomic num_extents = {0}; std::atomic num_blobs = {0}; uint64_t buffer_bytes = 0; public: BufferCacheShard(CephContext* cct) : CacheShard(cct) {} static BufferCacheShard *create(CephContext* cct, std::string type, PerfCounters *logger); virtual void _add(Buffer *b, int level, Buffer *near) = 0; virtual void _rm(Buffer *b) = 0; virtual void _move(BufferCacheShard *src, Buffer *b) = 0; virtual void _touch(Buffer *b) = 0; virtual void _adjust_size(Buffer *b, int64_t delta) = 0; uint64_t _get_bytes() { return buffer_bytes; } void add_extent() { ++num_extents; } void rm_extent() { --num_extents; } void add_blob() { ++num_blobs; } void rm_blob() { --num_blobs; } virtual void add_stats(uint64_t *extents, uint64_t *blobs, uint64_t *buffers, uint64_t *bytes) = 0; bool empty() { std::lock_guard l(lock); return _get_bytes() == 0; } }; struct OnodeSpace { OnodeCacheShard *cache; private: /// forward lookups mempool::bluestore_cache_meta::unordered_map onode_map; friend struct Collection; // for split_cache() friend struct Onode; // for put() friend struct LruOnodeCacheShard; void _remove(const ghobject_t& oid); public: OnodeSpace(OnodeCacheShard *c) : cache(c) {} ~OnodeSpace() { clear(); } OnodeRef add_onode(const ghobject_t& oid, OnodeRef& o); OnodeRef lookup(const ghobject_t& o); void rename(OnodeRef& o, const ghobject_t& old_oid, const ghobject_t& new_oid, const mempool::bluestore_cache_meta::string& new_okey); void clear(); bool empty(); template void dump(CephContext *cct); /// return true if f true for any item bool map_any(std::function f); }; class OpSequencer; using OpSequencerRef = ceph::ref_t; struct Collection : public CollectionImpl { BlueStore *store; OpSequencerRef osr; BufferCacheShard *cache; ///< our cache shard bluestore_cnode_t cnode; ceph::shared_mutex lock = ceph::make_shared_mutex("BlueStore::Collection::lock", true, false); bool exists; SharedBlobSet shared_blob_set; ///< open SharedBlobs // cache onodes on a per-collection basis to avoid lock // contention. OnodeSpace onode_space; //pool options pool_opts_t pool_opts; ContextQueue *commit_queue; OnodeCacheShard* get_onode_cache() const { return onode_space.cache; } OnodeRef get_onode(const ghobject_t& oid, bool create, bool is_createop=false); // 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 std::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; } 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; } int64_t pool() const { return cid.pool(); } void split_cache(Collection *dest); bool flush_commit(Context *c) override; void flush() override; void flush_all_but_last(); Collection(BlueStore *ns, OnodeCacheShard *oc, BufferCacheShard *bc, coll_t c); }; class OmapIteratorImpl : public ObjectMap::ObjectMapIteratorImpl { PerfCounters* logger = nullptr; CollectionRef c; OnodeRef o; KeyValueDB::Iterator it; std::string head, tail; std::string _stringify() const; public: OmapIteratorImpl(PerfCounters* l, CollectionRef c, OnodeRef& o, KeyValueDB::Iterator it); virtual ~OmapIteratorImpl(); int seek_to_first() override; int upper_bound(const std::string &after) override; int lower_bound(const std::string &to) override; bool valid() override; int next() override; std::string key() override; ceph::buffer::list value() override; std::string tail_key() override { return tail; } int status() override { return 0; } }; 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(); } bool empty() const { for (size_t i = 0; i < STATFS_LAST; ++i) { if (values[i]) { return false; } } return true; } void publish(store_statfs_t* buf) const { buf->allocated = allocated(); buf->data_stored = stored(); buf->data_compressed = compressed(); buf->data_compressed_original = compressed_original(); buf->data_compressed_allocated = compressed_allocated(); } 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]; } int64_t allocated() const { return values[STATFS_ALLOCATED]; } int64_t stored() const { return values[STATFS_STORED]; } int64_t compressed_original() const { return values[STATFS_COMPRESSED_ORIGINAL]; } int64_t compressed() const { return values[STATFS_COMPRESSED]; } int64_t compressed_allocated() const { return values[STATFS_COMPRESSED_ALLOCATED]; } volatile_statfs& operator=(const store_statfs_t& st) { values[STATFS_ALLOCATED] = st.allocated; values[STATFS_STORED] = st.data_stored; values[STATFS_COMPRESSED_ORIGINAL] = st.data_compressed_original; values[STATFS_COMPRESSED] = st.data_compressed; values[STATFS_COMPRESSED_ALLOCATED] = st.data_compressed_allocated; return *this; } 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(ceph::buffer::list::const_iterator& it) { using ceph::decode; for (size_t i = 0; i < STATFS_LAST; i++) { decode(values[i], it); } } void encode(ceph::buffer::list& bl) { using ceph::encode; for (size_t i = 0; i < STATFS_LAST; i++) { encode(values[i], bl); } } }; struct TransContext final : 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; 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) 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 inline void set_state(state_t s) { state = s; #ifdef WITH_BLKIN if (trace) { trace.event(get_state_name()); } #endif } inline state_t get_state() { return state; } CollectionRef ch; OpSequencerRef osr; // this should be ch->osr boost::intrusive::list_member_hook<> sequencer_item; uint64_t bytes = 0, ios = 0, cost = 0; std::set onodes; ///< these need to be updated/written std::set modified_objects; ///< objects we modified (and need a ref) #ifdef HAVE_LIBZBD // zone refs to add/remove. each zone ref is a (zone, offset) tuple. The offset // is the first offset in the zone that the onode touched; subsequent writes // to that zone do not generate additional refs. This is a bit imprecise but // is sufficient to generate reasonably sequential reads when doing zone // cleaning with less metadata than a ref for every extent. std::map, uint64_t> new_zone_offset_refs; std::map, uint64_t> old_zone_offset_refs; #endif std::set shared_blobs; ///< these need to be updated/written std::set shared_blobs_written; ///< update these on io completion KeyValueDB::Transaction t; ///< then we will commit this std::list oncommits; ///< more commit completions std::list removed_collections; ///< colls we removed boost::intrusive::list_member_hook<> deferred_queue_item; bluestore_deferred_transaction_t *deferred_txn = nullptr; ///< if any interval_set allocated, released; volatile_statfs statfs_delta; ///< overall store statistics delta uint64_t osd_pool_id = META_POOL_ID; ///< osd pool id we're operating on IOContext ioc; bool had_ios = false; ///< true if we submitted IOs before our kv txn uint64_t seq = 0; ceph::mono_clock::time_point start; ceph::mono_clock::time_point 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 #if defined(WITH_LTTNG) bool tracing = false; #endif #ifdef WITH_BLKIN ZTracer::Trace trace; #endif explicit TransContext(CephContext* cct, Collection *c, OpSequencer *o, std::list *on_commits) : ch(c), osr(o), ioc(cct, this), start(ceph::mono_clock::now()) { last_stamp = start; if (on_commits) { oncommits.swap(*on_commits); } } ~TransContext() { #ifdef WITH_BLKIN if (trace) { trace.event("txc destruct"); } #endif 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 note_removed_object(OnodeRef& o) { modified_objects.insert(o); onodes.erase(o); } #ifdef HAVE_LIBZBD void note_write_zone_offset(OnodeRef& o, uint32_t zone, uint64_t offset) { o->onode.zone_offset_refs[zone] = offset; new_zone_offset_refs[std::make_pair(o, zone)] = offset; } void note_release_zone_offset(OnodeRef& o, uint32_t zone, uint64_t offset) { old_zone_offset_refs[std::make_pair(o, zone)] = offset; o->onode.zone_offset_refs.erase(zone); } #endif void aio_finish(BlueStore *store) override { store->txc_aio_finish(this); } private: state_t state = STATE_PREPARE; }; class BlueStoreThrottle { #if defined(WITH_LTTNG) const std::chrono::time_point time_base = ceph::mono_clock::now(); // Time of last chosen io (microseconds) std::atomic previous_emitted_tp_time_mono_mcs = {0}; std::atomic ios_started_since_last_traced = {0}; std::atomic ios_completed_since_last_traced = {0}; std::atomic_uint pending_kv_ios = {0}; std::atomic_uint pending_deferred_ios = {0}; // Min period between trace points (microseconds) std::atomic trace_period_mcs = {0}; bool should_trace( uint64_t *started, uint64_t *completed) { uint64_t min_period_mcs = trace_period_mcs.load( std::memory_order_relaxed); if (min_period_mcs == 0) { *started = 1; *completed = ios_completed_since_last_traced.exchange(0); return true; } else { ios_started_since_last_traced++; auto now_mcs = ceph::to_microseconds( ceph::mono_clock::now() - time_base); uint64_t previous_mcs = previous_emitted_tp_time_mono_mcs; uint64_t period_mcs = now_mcs - previous_mcs; if (period_mcs > min_period_mcs) { if (previous_emitted_tp_time_mono_mcs.compare_exchange_strong( previous_mcs, now_mcs)) { // This would be racy at a sufficiently extreme trace rate, but isn't // worth the overhead of doing it more carefully. *started = ios_started_since_last_traced.exchange(0); *completed = ios_completed_since_last_traced.exchange(0); return true; } } return false; } } #endif #if defined(WITH_LTTNG) void emit_initial_tracepoint( KeyValueDB &db, TransContext &txc, ceph::mono_clock::time_point); #else void emit_initial_tracepoint( KeyValueDB &db, TransContext &txc, ceph::mono_clock::time_point) {} #endif Throttle throttle_bytes; ///< submit to commit Throttle throttle_deferred_bytes; ///< submit to deferred complete public: BlueStoreThrottle(CephContext *cct) : throttle_bytes(cct, "bluestore_throttle_bytes", 0), throttle_deferred_bytes(cct, "bluestore_throttle_deferred_bytes", 0) { reset_throttle(cct->_conf); } #if defined(WITH_LTTNG) void complete_kv(TransContext &txc); void complete(TransContext &txc); #else void complete_kv(TransContext &txc) {} void complete(TransContext &txc) {} #endif ceph::mono_clock::duration log_state_latency( TransContext &txc, PerfCounters *logger, int state); bool try_start_transaction( KeyValueDB &db, TransContext &txc, ceph::mono_clock::time_point); void finish_start_transaction( KeyValueDB &db, TransContext &txc, ceph::mono_clock::time_point); void release_kv_throttle(uint64_t cost) { throttle_bytes.put(cost); } void release_deferred_throttle(uint64_t cost) { throttle_deferred_bytes.put(cost); } bool should_submit_deferred() { return throttle_deferred_bytes.past_midpoint(); } void reset_throttle(const ConfigProxy &conf) { throttle_bytes.reset_max(conf->bluestore_throttle_bytes); throttle_deferred_bytes.reset_max( conf->bluestore_throttle_bytes + conf->bluestore_throttle_deferred_bytes); #if defined(WITH_LTTNG) double rate = conf.get_val("bluestore_throttle_trace_rate"); trace_period_mcs = rate > 0 ? floor((1/rate) * 1000000.0) : 0; #endif } } throttle; typedef boost::intrusive::list< TransContext, boost::intrusive::member_hook< TransContext, boost::intrusive::list_member_hook<>, &TransContext::deferred_queue_item> > deferred_queue_t; struct DeferredBatch final : public AioContext { OpSequencer *osr; struct deferred_io { ceph::buffer::list bl; ///< data uint64_t seq; ///< deferred transaction seq }; std::map 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) std::map 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, ceph::buffer::list::const_iterator& p); void aio_finish(BlueStore *store) override { store->_deferred_aio_finish(osr); } }; class OpSequencer : public RefCountedObject { public: ceph::mutex qlock = ceph::make_mutex("BlueStore::OpSequencer::qlock"); ceph::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; ceph::mutex deferred_lock = ceph::make_mutex("BlueStore::OpSequencer::deferred_lock"); BlueStore *store; coll_t cid; 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 zombie = {false}; ///< in zombie_osr std::set (collection going away) const uint32_t sequencer_id; uint32_t get_sequencer_id() const { return sequencer_id; } void queue_new(TransContext *txc) { std::lock_guard l(qlock); txc->seq = ++last_seq; q.push_back(*txc); } void drain() { std::unique_lock l(qlock); while (!q.empty()) qcond.wait(l); } void drain_preceding(TransContext *txc) { std::unique_lock l(qlock); while (&q.front() != txc) qcond.wait(l); } bool _is_all_kv_submitted() { // caller must hold qlock & q.empty() must not empty ceph_assert(!q.empty()); TransContext *txc = &q.back(); if (txc->get_state() >= TransContext::STATE_KV_SUBMITTED) { return true; } return false; } void flush() { std::unique_lock l(qlock); while (true) { // std::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 (q.empty() || _is_all_kv_submitted()) { --kv_submitted_waiters; return; } qcond.wait(l); --kv_submitted_waiters; } } void flush_all_but_last() { std::unique_lock l(qlock); ceph_assert (q.size() >= 1); while (true) { // std::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 (q.size() <= 1) { --kv_submitted_waiters; return; } else { auto it = q.rbegin(); it++; if (it->get_state() >= TransContext::STATE_KV_SUBMITTED) { --kv_submitted_waiters; return; } } qcond.wait(l); --kv_submitted_waiters; } } bool flush_commit(Context *c) { std::lock_guard l(qlock); if (q.empty()) { return true; } TransContext *txc = &q.back(); if (txc->get_state() >= TransContext::STATE_KV_DONE) { return true; } txc->oncommits.push_back(c); return false; } private: FRIEND_MAKE_REF(OpSequencer); OpSequencer(BlueStore *store, uint32_t sequencer_id, const coll_t& c) : RefCountedObject(store->cct), store(store), cid(c), sequencer_id(sequencer_id) { } ~OpSequencer() { ceph_assert(q.empty()); } }; 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() override { store->_kv_finalize_thread(); return NULL; } }; #ifdef HAVE_LIBZBD struct ZonedCleanerThread : public Thread { BlueStore *store; explicit ZonedCleanerThread(BlueStore *s) : store(s) {} void *entry() override { store->_zoned_cleaner_thread(); return nullptr; } }; #endif struct BigDeferredWriteContext { uint64_t off = 0; // original logical offset uint32_t b_off = 0; // blob relative offset uint32_t used = 0; uint64_t head_read = 0; uint64_t tail_read = 0; BlobRef blob_ref; uint64_t blob_start = 0; PExtentVector res_extents; inline uint64_t blob_aligned_len() const { return used + head_read + tail_read; } bool can_defer(BlueStore::extent_map_t::iterator ep, uint64_t prefer_deferred_size, uint64_t block_size, uint64_t offset, uint64_t l); bool apply_defer(); }; // -------------------------------------------------------- // members private: BlueFS *bluefs = nullptr; bluefs_layout_t bluefs_layout; utime_t next_dump_on_bluefs_alloc_failure; KeyValueDB *db = nullptr; BlockDevice *bdev = nullptr; std::string freelist_type; FreelistManager *fm = nullptr; Allocator *alloc = nullptr; ///< allocator consumed by BlueStore bluefs_shared_alloc_context_t shared_alloc; ///< consumed by BlueFS (may be == alloc) uuid_d fsid; int path_fd = -1; ///< open handle to $path int fsid_fd = -1; ///< open handle (locked) to $path/fsid bool mounted = false; // store open_db options: bool db_was_opened_read_only = true; bool need_to_destage_allocation_file = false; ///< rwlock to protect coll_map/new_coll_map ceph::shared_mutex coll_lock = ceph::make_shared_mutex("BlueStore::coll_lock"); mempool::bluestore_cache_other::unordered_map coll_map; bool collections_had_errors = false; std::map new_coll_map; mempool::bluestore_cache_buffer::vector buffer_cache_shards; mempool::bluestore_cache_onode::vector onode_cache_shards; /// protect zombie_osr_set ceph::mutex zombie_osr_lock = ceph::make_mutex("BlueStore::zombie_osr_lock"); uint32_t next_sequencer_id = 0; std::map zombie_osr_set; ///< std::set of OpSequencers for deleted collections std::atomic nid_last = {0}; std::atomic nid_max = {0}; std::atomic blobid_last = {0}; std::atomic blobid_max = {0}; ceph::mutex deferred_lock = ceph::make_mutex("BlueStore::deferred_lock"); ceph::mutex atomic_alloc_and_submit_lock = ceph::make_mutex("BlueStore::atomic_alloc_and_submit_lock"); std::atomic deferred_seq = {0}; deferred_osr_queue_t deferred_queue; ///< osr's with deferred io pending std::atomic_int deferred_queue_size = {0}; ///< num txc's queued across all osrs std::atomic_int deferred_aggressive = {0}; ///< aggressive wakeup of kv thread Finisher finisher; utime_t deferred_last_submitted = utime_t(); KVSyncThread kv_sync_thread; ceph::mutex kv_lock = ceph::make_mutex("BlueStore::kv_lock"); ceph::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; std::deque kv_queue; ///< ready, already submitted std::deque kv_queue_unsubmitted; ///< ready, need submit by kv thread std::deque kv_committing; ///< currently syncing std::deque deferred_done_queue; ///< deferred ios done bool kv_sync_in_progress = false; KVFinalizeThread kv_finalize_thread; ceph::mutex kv_finalize_lock = ceph::make_mutex("BlueStore::kv_finalize_lock"); ceph::condition_variable kv_finalize_cond; std::deque kv_committing_to_finalize; ///< pending finalization std::deque deferred_stable_to_finalize; ///< pending finalization bool kv_finalize_in_progress = false; #ifdef HAVE_LIBZBD ZonedCleanerThread zoned_cleaner_thread; ceph::mutex zoned_cleaner_lock = ceph::make_mutex("BlueStore::zoned_cleaner_lock"); ceph::condition_variable zoned_cleaner_cond; bool zoned_cleaner_started = false; bool zoned_cleaner_stop = false; std::deque zoned_cleaner_queue; #endif PerfCounters *logger = nullptr; std::list removed_collections; ceph::shared_mutex debug_read_error_lock = ceph::make_shared_mutex("BlueStore::debug_read_error_lock"); std::set debug_data_error_objects; std::set debug_mdata_error_objects; std::atomic 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 optimal_io_size = 0;///< best performance io size for block device uint64_t min_alloc_size; ///< minimum allocation unit (power of 2) uint8_t min_alloc_size_order = 0;///< bits to shift to get min_alloc_size uint64_t min_alloc_size_mask;///< mask for fast checking of allocation alignment static_assert(std::numeric_limits::max() > std::numeric_limits::digits, "not enough bits for min_alloc_size"); // smr-only uint64_t zone_size = 0; ///< number of SMR zones uint64_t first_sequential_zone = 0; ///< first SMR zone that is sequential-only enum { // Please preserve the order since it's DB persistent OMAP_BULK = 0, OMAP_PER_POOL = 1, OMAP_PER_PG = 2, } per_pool_omap = OMAP_BULK; ///< maximum allocation unit (power of 2) std::atomic max_alloc_size = {0}; ///< number threshold for forced deferred writes std::atomic deferred_batch_ops = {0}; ///< size threshold for forced deferred writes std::atomic prefer_deferred_size = {0}; ///< approx cost per io, in bytes std::atomic throttle_cost_per_io = {0}; std::atomic comp_mode = {Compressor::COMP_NONE}; ///< compression mode CompressorRef compressor; std::atomic comp_min_blob_size = {0}; std::atomic comp_max_blob_size = {0}; std::atomic 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 double cache_meta_ratio = 0; ///< cache ratio dedicated to metadata double cache_kv_ratio = 0; ///< cache ratio dedicated to kv (e.g., rocksdb) double cache_kv_onode_ratio = 0; ///< cache ratio dedicated to kv onodes (e.g., rocksdb onode CF) double cache_data_ratio = 0; ///< cache ratio dedicated to object data bool cache_autotune = false; ///< cache autotune setting double cache_age_bin_interval = 0; ///< time to wait between cache age bin rotations double cache_autotune_interval = 0; ///< time to wait between cache rebalancing std::vector kv_bins; ///< kv autotune bins std::vector kv_onode_bins; ///< kv onode autotune bins std::vector meta_bins; ///< meta autotune bins std::vector data_bins; ///< data autotune bins uint64_t osd_memory_target = 0; ///< OSD memory target when autotuning cache uint64_t osd_memory_base = 0; ///< OSD base memory when autotuning cache double osd_memory_expected_fragmentation = 0; ///< expected memory fragmentation uint64_t osd_memory_cache_min = 0; ///< Min memory to assign when autotuning cache double osd_memory_cache_resize_interval = 0; ///< Time to wait between cache resizing double max_defer_interval = 0; ///< Time to wait between last deferred submit std::atomic config_changed = {0}; ///< Counter to determine if there is a configuration change. typedef std::map osd_pools_map; ceph::mutex vstatfs_lock = ceph::make_mutex("BlueStore::vstatfs_lock"); volatile_statfs vstatfs; osd_pools_map osd_pools; // protected by vstatfs_lock as well bool per_pool_stat_collection = true; struct MempoolThread : public Thread { public: BlueStore *store; ceph::condition_variable cond; ceph::mutex lock = ceph::make_mutex("BlueStore::MempoolThread::lock"); bool stop = false; std::shared_ptr binned_kv_cache = nullptr; std::shared_ptr binned_kv_onode_cache = nullptr; std::shared_ptr pcm = nullptr; struct MempoolCache : public PriorityCache::PriCache { BlueStore *store; uint64_t bins[PriorityCache::Priority::LAST+1] = {0}; int64_t cache_bytes[PriorityCache::Priority::LAST+1] = {0}; int64_t committed_bytes = 0; double cache_ratio = 0; MempoolCache(BlueStore *s) : store(s) {}; virtual uint64_t _get_used_bytes() const = 0; virtual uint64_t _sum_bins(uint32_t start, uint32_t end) const = 0; virtual int64_t request_cache_bytes( PriorityCache::Priority pri, uint64_t total_cache) const { int64_t assigned = get_cache_bytes(pri); switch (pri) { case PriorityCache::Priority::PRI0: { // BlueStore caches currently don't put anything in PRI0 break; } case PriorityCache::Priority::LAST: { uint32_t max = get_bin_count(); int64_t request = _get_used_bytes() - _sum_bins(0, max); return(request > assigned) ? request - assigned : 0; } default: { ceph_assert(pri > 0 && pri < PriorityCache::Priority::LAST); auto prev_pri = static_cast(pri - 1); uint64_t start = get_bins(prev_pri); uint64_t end = get_bins(pri); int64_t request = _sum_bins(start, end); return(request > assigned) ? request - assigned : 0; } } return -EOPNOTSUPP; } virtual int64_t get_cache_bytes(PriorityCache::Priority pri) const { return cache_bytes[pri]; } virtual int64_t get_cache_bytes() const { int64_t total = 0; for (int i = 0; i < PriorityCache::Priority::LAST + 1; i++) { PriorityCache::Priority pri = static_cast(i); total += get_cache_bytes(pri); } return total; } virtual void set_cache_bytes(PriorityCache::Priority pri, int64_t bytes) { cache_bytes[pri] = bytes; } virtual void add_cache_bytes(PriorityCache::Priority pri, int64_t bytes) { cache_bytes[pri] += bytes; } virtual int64_t commit_cache_size(uint64_t total_cache) { committed_bytes = PriorityCache::get_chunk( get_cache_bytes(), total_cache); return committed_bytes; } virtual int64_t get_committed_size() const { return committed_bytes; } virtual uint64_t get_bins(PriorityCache::Priority pri) const { if (pri > PriorityCache::Priority::PRI0 && pri < PriorityCache::Priority::LAST) { return bins[pri]; } return 0; } virtual void set_bins(PriorityCache::Priority pri, uint64_t end_bin) { if (pri <= PriorityCache::Priority::PRI0 || pri >= PriorityCache::Priority::LAST) { return; } bins[pri] = end_bin; uint64_t max = 0; for (int pri = 1; pri < PriorityCache::Priority::LAST; pri++) { if (bins[pri] > max) { max = bins[pri]; } } set_bin_count(max); } virtual void import_bins(const std::vector &bins_v) { uint64_t max = 0; for (int pri = 1; pri < PriorityCache::Priority::LAST; pri++) { unsigned i = (unsigned) pri - 1; if (i < bins_v.size()) { bins[pri] = bins_v[i]; if (bins[pri] > max) { max = bins[pri]; } } else { bins[pri] = 0; } } set_bin_count(max); } virtual double get_cache_ratio() const { return cache_ratio; } virtual void set_cache_ratio(double ratio) { cache_ratio = ratio; } virtual std::string get_cache_name() const = 0; virtual uint32_t get_bin_count() const = 0; virtual void set_bin_count(uint32_t count) = 0; }; struct MetaCache : public MempoolCache { MetaCache(BlueStore *s) : MempoolCache(s) {}; virtual uint32_t get_bin_count() const { return store->onode_cache_shards[0]->get_bin_count(); } virtual void set_bin_count(uint32_t count) { for (auto i : store->onode_cache_shards) { i->set_bin_count(count); } } virtual uint64_t _get_used_bytes() const { return mempool::bluestore_blob::allocated_bytes() + mempool::bluestore_extent::allocated_bytes() + mempool::bluestore_cache_buffer::allocated_bytes() + mempool::bluestore_cache_meta::allocated_bytes() + mempool::bluestore_cache_other::allocated_bytes() + mempool::bluestore_cache_onode::allocated_bytes() + mempool::bluestore_shared_blob::allocated_bytes() + mempool::bluestore_inline_bl::allocated_bytes(); } virtual void shift_bins() { for (auto i : store->onode_cache_shards) { i->shift_bins(); } } virtual uint64_t _sum_bins(uint32_t start, uint32_t end) const { uint64_t onodes = 0; for (auto i : store->onode_cache_shards) { onodes += i->sum_bins(start, end); } return onodes*get_bytes_per_onode(); } virtual std::string get_cache_name() const { return "BlueStore Meta Cache"; } uint64_t _get_num_onodes() const { uint64_t onode_num = mempool::bluestore_cache_onode::allocated_items(); return (2 > onode_num) ? 2 : onode_num; } double get_bytes_per_onode() const { return (double)_get_used_bytes() / (double)_get_num_onodes(); } }; std::shared_ptr meta_cache; struct DataCache : public MempoolCache { DataCache(BlueStore *s) : MempoolCache(s) {}; virtual uint32_t get_bin_count() const { return store->buffer_cache_shards[0]->get_bin_count(); } virtual void set_bin_count(uint32_t count) { for (auto i : store->buffer_cache_shards) { i->set_bin_count(count); } } virtual uint64_t _get_used_bytes() const { uint64_t bytes = 0; for (auto i : store->buffer_cache_shards) { bytes += i->_get_bytes(); } return bytes; } virtual void shift_bins() { for (auto i : store->buffer_cache_shards) { i->shift_bins(); } } virtual uint64_t _sum_bins(uint32_t start, uint32_t end) const { uint64_t bytes = 0; for (auto i : store->buffer_cache_shards) { bytes += i->sum_bins(start, end); } return bytes; } virtual std::string get_cache_name() const { return "BlueStore Data Cache"; } }; std::shared_ptr data_cache; public: explicit MempoolThread(BlueStore *s) : store(s), meta_cache(new MetaCache(s)), data_cache(new DataCache(s)) {} void *entry() override; void init() { ceph_assert(stop == false); create("bstore_mempool"); } void shutdown() { lock.lock(); stop = true; cond.notify_all(); lock.unlock(); join(); } private: void _update_cache_settings(); void _resize_shards(bool interval_stats); } mempool_thread; #ifdef WITH_BLKIN ZTracer::Endpoint trace_endpoint {"0.0.0.0", 0, "BlueStore"}; #endif // -------------------------------------------------------- // 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(); void _set_finisher_num(); void _set_per_pool_omap(); void _update_osd_memory_options(); int _open_bdev(bool create); // Verifies if disk space is enough for reserved + min bluefs // and alters the latter if needed. // Depends on min_alloc_size hence should be called after // its initialization (and outside of _open_bdev) void _validate_bdev(); void _close_bdev(); int _minimal_open_bluefs(bool create); void _minimal_close_bluefs(); int _open_bluefs(bool create, bool read_only); void _close_bluefs(); int _is_bluefs(bool create, bool* ret); /* * opens both DB and dependant super_meta, FreelistManager and allocator * in the proper order */ int _open_db_and_around(bool read_only, bool to_repair = false); void _close_db_and_around(); void _close_around_db(); int _prepare_db_environment(bool create, bool read_only, std::string* kv_dir, std::string* kv_backend); /* * @warning to_repair_db means that we open this db to repair it, will not * hold the rocksdb's file lock. */ int _open_db(bool create, bool to_repair_db=false, bool read_only = false); void _close_db(); int _open_fm(KeyValueDB::Transaction t, bool read_only, bool db_avail, bool fm_restore = false); void _close_fm(); int _write_out_fm_meta(uint64_t target_size); int _create_alloc(); int _init_alloc(std::map *zone_adjustments); void _post_init_alloc(const std::map& zone_adjustments); void _close_alloc(); int _open_collections(); void _fsck_collections(int64_t* errors); void _close_collections(); int _setup_block_symlink_or_file(std::string name, std::string path, uint64_t size, bool create); public: utime_t get_deferred_last_submitted() { std::lock_guard l(deferred_lock); return deferred_last_submitted; } static int _write_bdev_label(CephContext* cct, const std::string &path, bluestore_bdev_label_t label); static int _read_bdev_label(CephContext* cct, const std::string &path, bluestore_bdev_label_t *label); private: int _check_or_set_bdev_label(std::string path, uint64_t size, std::string desc, bool create); int _set_bdev_label_size(const std::string& path, uint64_t size); int _open_super_meta(); void _open_statfs(); void _get_statfs_overall(struct store_statfs_t *buf); void _dump_alloc_on_failure(); CollectionRef _get_collection(const coll_t& cid); CollectionRef _get_collection_by_oid(const ghobject_t& oid); void _queue_reap_collection(CollectionRef& c); void _reap_collections(); void _update_logger(); void _assign_nid(TransContext *txc, OnodeRef& o); uint64_t _assign_blobid(TransContext *txc); template friend void _dump_onode(CephContext *cct, const Onode& o); template friend void _dump_extent_map(CephContext *cct, const ExtentMap& em); template friend void _dump_transaction(CephContext *cct, Transaction *t); TransContext *_txc_create(Collection *c, OpSequencer *osr, std::list *on_commits, TrackedOpRef osd_op=TrackedOpRef()); 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(p)); } private: void _txc_finish_io(TransContext *txc); void _txc_finalize_kv(TransContext *txc, KeyValueDB::Transaction t); void _txc_apply_kv(TransContext *txc, bool sync_submit_transaction); void _txc_committed_kv(TransContext *txc); void _txc_finish(TransContext *txc); void _txc_release_alloc(TransContext *txc); void _osr_attach(Collection *c); void _osr_register_zombie(OpSequencer *osr); void _osr_drain(OpSequencer *osr); void _osr_drain_preceding(TransContext *txc); void _osr_drain_all(); void _kv_start(); void _kv_stop(); void _kv_sync_thread(); void _kv_finalize_thread(); #ifdef HAVE_LIBZBD void _zoned_cleaner_start(); void _zoned_cleaner_stop(); void _zoned_cleaner_thread(); void _zoned_clean_zone(uint64_t zone_num, class ZonedAllocator *a, class ZonedFreelistManager *f); void _clean_some(ghobject_t oid, uint32_t zone_num); #endif bluestore_deferred_op_t *_get_deferred_op(TransContext *txc, uint64_t len); 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(); bool _eliminate_outdated_deferred(bluestore_deferred_transaction_t* deferred_txn, interval_set& bluefs_extents); public: using mempool_dynamic_bitset = boost::dynamic_bitset>; using per_pool_statfs = mempool::bluestore_fsck::map; enum FSCKDepth { FSCK_REGULAR, FSCK_DEEP, FSCK_SHALLOW }; enum { MAX_FSCK_ERROR_LINES = 100, }; private: int _fsck_check_extents( std::string_view ctx_descr, const PExtentVector& extents, bool compressed, mempool_dynamic_bitset &used_blocks, uint64_t granularity, BlueStoreRepairer* repairer, store_statfs_t& expected_statfs, FSCKDepth depth); void _fsck_check_statfs( const store_statfs_t& expected_store_statfs, const per_pool_statfs& expected_pool_statfs, int64_t& errors, int64_t &warnings, BlueStoreRepairer* repairer); void _fsck_repair_shared_blobs( BlueStoreRepairer& repairer, shared_blob_2hash_tracker_t& sb_ref_counts, sb_info_space_efficient_map_t& sb_info); int _fsck(FSCKDepth depth, bool repair); int _fsck_on_open(BlueStore::FSCKDepth depth, bool repair); void _buffer_cache_write( TransContext *txc, BlobRef b, uint64_t offset, ceph::buffer::list& 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, bool legacy, std::vector *ls, ghobject_t *next); template T select_option(const std::string& opt_name, T val1, F f) { //NB: opt_name reserved for future use std::optional val2 = f(); if (val2) { return *val2; } return val1; } void _apply_padding(uint64_t head_pad, uint64_t tail_pad, ceph::buffer::list& padded); void _record_onode(OnodeRef &o, KeyValueDB::Transaction &txn); // -- ondisk version --- public: const int32_t latest_ondisk_format = 4; ///< our version const int32_t min_readable_ondisk_format = 1; ///< what we can read const int32_t min_compat_ondisk_format = 3; ///< who can read us private: int32_t ondisk_format = 0; ///< value detected on mount bool m_fast_shutdown = false; int _upgrade_super(); ///< upgrade (called during open_super) uint64_t _get_ondisk_reserved() const; void _prepare_ondisk_format_super(KeyValueDB::Transaction& t); // --- public interface --- public: BlueStore(CephContext *cct, const std::string& path); BlueStore(CephContext *cct, const std::string& path, uint64_t min_alloc_size); // Ctor for UT only ~BlueStore() override; std::string get_type() override { return "bluestore"; } bool needs_journal() override { return false; }; bool wants_journal() override { return false; }; bool allows_journal() override { return false; }; void prepare_for_fast_shutdown() override; bool has_null_manager() const override; uint64_t get_min_alloc_size() const override { return min_alloc_size; } int get_devices(std::set *ls) override; bool is_rotational() override; bool is_journal_rotational() override; bool is_db_rotational(); bool is_statfs_recoverable() const; std::string get_default_device_class() override { std::string device_class; std::map metadata; collect_metadata(&metadata); auto it = metadata.find("bluestore_bdev_type"); if (it != metadata.end()) { device_class = it->second; } return device_class; } int get_numa_node( int *numa_node, std::set *nodes, std::set *failed) override; static int get_block_device_fsid(CephContext* cct, const std::string& path, uuid_d *fsid); bool test_mount_in_use() override; private: int _mount(); public: int mount() override { return _mount(); } int umount() override; int open_db_environment(KeyValueDB **pdb, bool to_repair); int close_db_environment(); BlueFS* get_bluefs(); int write_meta(const std::string& key, const std::string& value) override; int read_meta(const std::string& key, std::string *value) override; // open in read-only and limited mode int cold_open(); int cold_close(); int fsck(bool deep) override { return _fsck(deep ? FSCK_DEEP : FSCK_REGULAR, false); } int repair(bool deep) override { return _fsck(deep ? FSCK_DEEP : FSCK_REGULAR, true); } int quick_fix() override { return _fsck(FSCK_SHALLOW, true); } void set_cache_shards(unsigned num) override; void dump_cache_stats(ceph::Formatter *f) override { int onode_count = 0, buffers_bytes = 0; for (auto i: onode_cache_shards) { onode_count += i->_get_num(); } for (auto i: buffer_cache_shards) { buffers_bytes += i->_get_bytes(); } f->dump_int("bluestore_onode", onode_count); f->dump_int("bluestore_buffers", buffers_bytes); } void dump_cache_stats(std::ostream& ss) override { int onode_count = 0, buffers_bytes = 0; for (auto i: onode_cache_shards) { onode_count += i->_get_num(); } for (auto i: buffer_cache_shards) { buffers_bytes += i->_get_bytes(); } ss << "bluestore_onode: " << onode_count; ss << "bluestore_buffers: " << buffers_bytes; } 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(ceph::Formatter *f) override; void generate_db_histogram(ceph::Formatter *f) override; void _shutdown_cache(); int flush_cache(std::ostream *os = NULL) override; void dump_perf_counters(ceph::Formatter *f) override { f->open_object_section("perf_counters"); logger->dump_formatted(f, false, false); f->close_section(); } int add_new_bluefs_device(int id, const std::string& path); int migrate_to_existing_bluefs_device(const std::set& devs_source, int id); int migrate_to_new_bluefs_device(const std::set& devs_source, int id, const std::string& path); int expand_devices(std::ostream& out); std::string get_device_path(unsigned id); int dump_bluefs_sizes(std::ostream& out); public: int statfs(struct store_statfs_t *buf, osd_alert_list_t* alerts = nullptr) override; int pool_statfs(uint64_t pool_id, struct store_statfs_t *buf, bool *per_pool_omap) override; void collect_metadata(std::map *pm) override; bool exists(CollectionHandle &c, const ghobject_t& oid) override; int set_collection_opts( CollectionHandle& c, const pool_opts_t& opts) override; int stat( CollectionHandle &c, const ghobject_t& oid, struct stat *st, bool allow_eio = false) override; int read( CollectionHandle &c, const ghobject_t& oid, uint64_t offset, size_t len, ceph::buffer::list& bl, uint32_t op_flags = 0) override; private: // -------------------------------------------------------- // intermediate data structures used while reading struct region_t { uint64_t logical_offset; uint64_t blob_xoffset; //region offset within the blob uint64_t length; // used later in read process uint64_t front = 0; region_t(uint64_t offset, uint64_t b_offs, uint64_t len, uint64_t front = 0) : logical_offset(offset), blob_xoffset(b_offs), length(len), front(front){} region_t(const region_t& from) : logical_offset(from.logical_offset), blob_xoffset(from.blob_xoffset), length(from.length), front(from.front){} friend std::ostream& operator<<(std::ostream& out, const region_t& r) { return out << "0x" << std::hex << r.logical_offset << ":" << r.blob_xoffset << "~" << r.length << std::dec; } }; // merged blob read request struct read_req_t { uint64_t r_off = 0; uint64_t r_len = 0; ceph::buffer::list bl; std::list regs; // original read regions read_req_t(uint64_t off, uint64_t len) : r_off(off), r_len(len) {} friend std::ostream& operator<<(std::ostream& out, const read_req_t& r) { out << "{<0x" << std::hex << r.r_off << ", 0x" << r.r_len << "> : ["; for (const auto& reg : r.regs) out << reg; return out << "]}" << std::dec; } }; typedef std::list regions2read_t; typedef std::map blobs2read_t; void _read_cache( OnodeRef& o, uint64_t offset, size_t length, int read_cache_policy, ready_regions_t& ready_regions, blobs2read_t& blobs2read); int _prepare_read_ioc( blobs2read_t& blobs2read, std::vector* compressed_blob_bls, IOContext* ioc); int _generate_read_result_bl( OnodeRef& o, uint64_t offset, size_t length, ready_regions_t& ready_regions, std::vector& compressed_blob_bls, blobs2read_t& blobs2read, bool buffered, bool* csum_error, ceph::buffer::list& bl); int _do_read( Collection *c, OnodeRef& o, uint64_t offset, size_t len, ceph::buffer::list& bl, uint32_t op_flags = 0, uint64_t retry_count = 0); int _do_readv( Collection *c, OnodeRef& o, const interval_set& m, ceph::buffer::list& bl, uint32_t op_flags = 0, uint64_t retry_count = 0); int _fiemap(CollectionHandle &c_, const ghobject_t& oid, uint64_t offset, size_t len, interval_set& destset); public: int fiemap(CollectionHandle &c, const ghobject_t& oid, uint64_t offset, size_t len, ceph::buffer::list& bl) override; int fiemap(CollectionHandle &c, const ghobject_t& oid, uint64_t offset, size_t len, std::map& destmap) override; int readv( CollectionHandle &c_, const ghobject_t& oid, interval_set& m, ceph::buffer::list& bl, uint32_t op_flags) override; int dump_onode(CollectionHandle &c, const ghobject_t& oid, const std::string& section_name, ceph::Formatter *f) override; int getattr(CollectionHandle &c, const ghobject_t& oid, const char *name, ceph::buffer::ptr& value) override; int getattrs(CollectionHandle &c, const ghobject_t& oid, std::map>& aset) override; int list_collections(std::vector& ls) override; CollectionHandle open_collection(const coll_t &c) override; CollectionHandle create_new_collection(const coll_t& cid) override; void set_collection_commit_queue(const coll_t& cid, ContextQueue *commit_queue) override; bool collection_exists(const coll_t& c) override; int collection_empty(CollectionHandle& c, bool *empty) override; int collection_bits(CollectionHandle& c) override; int collection_list(CollectionHandle &c, const ghobject_t& start, const ghobject_t& end, int max, std::vector *ls, ghobject_t *next) override; int collection_list_legacy(CollectionHandle &c, const ghobject_t& start, const ghobject_t& end, int max, std::vector *ls, ghobject_t *next) override; int omap_get( CollectionHandle &c, ///< [in] Collection containing oid const ghobject_t &oid, ///< [in] Object containing omap ceph::buffer::list *header, ///< [out] omap header std::map *out /// < [out] Key to value map ) override; int _omap_get( Collection *c, ///< [in] Collection containing oid const ghobject_t &oid, ///< [in] Object containing omap ceph::buffer::list *header, ///< [out] omap header std::map *out /// < [out] Key to value map ); int _onode_omap_get( const OnodeRef& o, ///< [in] Object containing omap ceph::buffer::list *header, ///< [out] omap header std::map *out /// < [out] Key to value map ); /// Get omap header int omap_get_header( CollectionHandle &c, ///< [in] Collection containing oid const ghobject_t &oid, ///< [in] Object containing omap ceph::buffer::list *header, ///< [out] omap header bool allow_eio = false ///< [in] don't assert on eio ) override; /// Get keys defined on oid int omap_get_keys( CollectionHandle &c, ///< [in] Collection containing oid const ghobject_t &oid, ///< [in] Object containing omap std::set *keys ///< [out] Keys defined on oid ) override; /// Get key values int omap_get_values( CollectionHandle &c, ///< [in] Collection containing oid const ghobject_t &oid, ///< [in] Object containing omap const std::set &keys, ///< [in] Keys to get std::map *out ///< [out] Returned keys and values ) override; #ifdef WITH_SEASTAR int omap_get_values( CollectionHandle &c, ///< [in] Collection containing oid const ghobject_t &oid, ///< [in] Object containing omap const std::optional &start_after, ///< [in] Keys to get std::map *out ///< [out] Returned keys and values ) override; #endif /// Filters keys into out which are defined on oid int omap_check_keys( CollectionHandle &c, ///< [in] Collection containing oid const ghobject_t &oid, ///< [in] Object containing omap const std::set &keys, ///< [in] Keys to check std::set *out ///< [out] Subset of keys defined on oid ) 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 os_commit_latency_ns; PerfCounters::avg_tracker os_apply_latency_ns; objectstore_perf_stat_t get_cur_stats() const { objectstore_perf_stat_t ret; ret.os_commit_latency_ns = os_commit_latency_ns.current_avg(); ret.os_apply_latency_ns = os_apply_latency_ns.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; } const PerfCounters* get_bluefs_perf_counters() const { return bluefs->get_perf_counters(); } KeyValueDB* get_kv() { return db; } int queue_transactions( CollectionHandle& ch, std::vector& tls, TrackedOpRef op = TrackedOpRef(), ThreadPool::TPHandle *handle = NULL) override; // error injection void inject_data_error(const ghobject_t& o) override { std::unique_lock l(debug_read_error_lock); debug_data_error_objects.insert(o); } void inject_mdata_error(const ghobject_t& o) override { std::unique_lock l(debug_read_error_lock); debug_mdata_error_objects.insert(o); } /// methods to inject various errors fsck can repair void inject_broken_shared_blob_key(const std::string& key, const ceph::buffer::list& bl); void inject_no_shared_blob_key(); void inject_stray_shared_blob_key(uint64_t sbid); void inject_leaked(uint64_t len); void inject_false_free(coll_t cid, ghobject_t oid); void inject_statfs(const std::string& key, const store_statfs_t& new_statfs); void inject_global_statfs(const store_statfs_t& new_statfs); void inject_misreference(coll_t cid1, ghobject_t oid1, coll_t cid2, ghobject_t oid2, uint64_t offset); void inject_zombie_spanning_blob(coll_t cid, ghobject_t oid, int16_t blob_id); // resets global per_pool_omap in DB void inject_legacy_omap(); // resets per_pool_omap | pgmeta_omap for onode void inject_legacy_omap(coll_t cid, ghobject_t oid); void inject_stray_omap(uint64_t head, const std::string& name); void inject_bluefs_file(std::string_view dir, std::string_view name, size_t new_size); void compact() override { ceph_assert(db); db->compact(); } bool has_builtin_csum() const override { return true; } inline void log_latency(const char* name, int idx, const ceph::timespan& lat, double lat_threshold, const char* info = "") const; inline void log_latency_fn(const char* name, int idx, const ceph::timespan& lat, double lat_threshold, std::function fn) const; private: bool _debug_data_eio(const ghobject_t& o) { if (!cct->_conf->bluestore_debug_inject_read_err) { return false; } std::shared_lock 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; } std::shared_lock 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) { std::unique_lock l(debug_read_error_lock); debug_data_error_objects.erase(o); debug_mdata_error_objects.erase(o); } } private: ceph::mutex qlock = ceph::make_mutex("BlueStore::Alerts::qlock"); std::string failed_cmode; std::set failed_compressors; std::string spillover_alert; std::string legacy_statfs_alert; std::string no_per_pool_omap_alert; std::string no_per_pg_omap_alert; std::string disk_size_mismatch_alert; std::string spurious_read_errors_alert; void _log_alerts(osd_alert_list_t& alerts); bool _set_compression_alert(bool cmode, const char* s) { std::lock_guard l(qlock); if (cmode) { bool ret = failed_cmode.empty(); failed_cmode = s; return ret; } return failed_compressors.emplace(s).second; } void _clear_compression_alert() { std::lock_guard l(qlock); failed_compressors.clear(); failed_cmode.clear(); } void _check_legacy_statfs_alert(); void _check_no_per_pg_or_pool_omap_alert(); void _set_disk_size_mismatch_alert(const std::string& s) { std::lock_guard l(qlock); disk_size_mismatch_alert = s; } void _set_spurious_read_errors_alert(const std::string& s) { std::lock_guard l(qlock); spurious_read_errors_alert = s; } private: // -------------------------------------------------------- // read processing internal methods int _verify_csum( OnodeRef& o, const bluestore_blob_t* blob, uint64_t blob_xoffset, const ceph::buffer::list& bl, uint64_t logical_offset) const; int _decompress(ceph::buffer::list& source, ceph::buffer::list* 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 interval_set extents_to_gc; ///< extents for garbage collection struct write_item { uint64_t logical_offset; ///< write logical offset BlobRef b; uint64_t blob_length; uint64_t b_off; ceph::buffer::list 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; ceph::buffer::list compressed_bl; size_t compressed_len = 0; write_item( uint64_t logical_offs, BlobRef b, uint64_t blob_len, uint64_t o, ceph::buffer::list& 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) {} }; std::vector 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, ceph::buffer::list& 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, ceph::buffer::list::iterator& blp, WriteContext *wctx); void _do_write_big_apply_deferred( TransContext* txc, CollectionRef& c, OnodeRef& o, BigDeferredWriteContext& dctx, bufferlist::iterator& blp, WriteContext* wctx); void _do_write_big( TransContext *txc, CollectionRef &c, OnodeRef& o, uint64_t offset, uint64_t length, ceph::buffer::list::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, std::set *maybe_unshared_blobs=0); int _write(TransContext *txc, CollectionRef& c, OnodeRef& o, uint64_t offset, size_t len, ceph::buffer::list& bl, uint32_t fadvise_flags); void _pad_zeros(ceph::buffer::list *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 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, ceph::buffer::list& bl, uint32_t fadvise_flags); void _do_write_data(TransContext *txc, CollectionRef& c, OnodeRef& o, uint64_t offset, uint64_t length, ceph::buffer::list& 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, std::set *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 std::string& name, ceph::buffer::ptr& val); int _setattrs(TransContext *txc, CollectionRef& c, OnodeRef& o, const std::map& aset); int _rmattr(TransContext *txc, CollectionRef& c, OnodeRef& o, const std::string& name); int _rmattrs(TransContext *txc, CollectionRef& c, OnodeRef& o); void _do_omap_clear(TransContext *txc, OnodeRef& o); int _omap_clear(TransContext *txc, CollectionRef& c, OnodeRef& o); int _omap_setkeys(TransContext *txc, CollectionRef& c, OnodeRef& o, ceph::buffer::list& bl); int _omap_setheader(TransContext *txc, CollectionRef& c, OnodeRef& o, ceph::buffer::list& header); int _omap_rmkeys(TransContext *txc, CollectionRef& c, OnodeRef& o, ceph::buffer::list& bl); int _omap_rmkey_range(TransContext *txc, CollectionRef& c, OnodeRef& o, const std::string& first, const std::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); void _do_remove_collection(TransContext *txc, CollectionRef *c); int _split_collection(TransContext *txc, CollectionRef& c, CollectionRef& d, unsigned bits, int rem); int _merge_collection(TransContext *txc, CollectionRef *c, CollectionRef& d, unsigned bits); void _collect_allocation_stats(uint64_t need, uint32_t alloc_size, const PExtentVector&); void _record_allocation_stats(); private: uint64_t probe_count = 0; std::atomic alloc_stats_count = {0}; std::atomic alloc_stats_fragments = { 0 }; std::atomic alloc_stats_size = { 0 }; // std::array, 5> alloc_stats_history = { std::make_tuple(0ul, 0ul, 0ul) }; inline bool _use_rotational_settings(); public: typedef btree::btree_set< uint64_t, std::less, mempool::bluestore_fsck::pool_allocator> uint64_t_btree_t; struct FSCK_ObjectCtx { int64_t& errors; int64_t& warnings; uint64_t& num_objects; uint64_t& num_extents; uint64_t& num_blobs; uint64_t& num_sharded_objects; uint64_t& num_spanning_blobs; mempool_dynamic_bitset* used_blocks; uint64_t_btree_t* used_omap_head; std::vector> *zone_refs; ceph::mutex* sb_info_lock; sb_info_space_efficient_map_t& sb_info; // approximate amount of references per shared_blob_2hash_tracker_t& sb_ref_counts; store_statfs_t& expected_store_statfs; per_pool_statfs& expected_pool_statfs; BlueStoreRepairer* repairer; FSCK_ObjectCtx(int64_t& e, int64_t& w, uint64_t& _num_objects, uint64_t& _num_extents, uint64_t& _num_blobs, uint64_t& _num_sharded_objects, uint64_t& _num_spanning_blobs, mempool_dynamic_bitset* _ub, uint64_t_btree_t* _used_omap_head, std::vector> *_zone_refs, ceph::mutex* _sb_info_lock, sb_info_space_efficient_map_t& _sb_info, shared_blob_2hash_tracker_t& _sb_ref_counts, store_statfs_t& _store_statfs, per_pool_statfs& _pool_statfs, BlueStoreRepairer* _repairer) : errors(e), warnings(w), num_objects(_num_objects), num_extents(_num_extents), num_blobs(_num_blobs), num_sharded_objects(_num_sharded_objects), num_spanning_blobs(_num_spanning_blobs), used_blocks(_ub), used_omap_head(_used_omap_head), zone_refs(_zone_refs), sb_info_lock(_sb_info_lock), sb_info(_sb_info), sb_ref_counts(_sb_ref_counts), expected_store_statfs(_store_statfs), expected_pool_statfs(_pool_statfs), repairer(_repairer) { } }; OnodeRef fsck_check_objects_shallow( FSCKDepth depth, int64_t pool_id, CollectionRef c, const ghobject_t& oid, const std::string& key, const ceph::buffer::list& value, mempool::bluestore_fsck::list* expecting_shards, std::map* referenced, const BlueStore::FSCK_ObjectCtx& ctx); #ifdef CEPH_BLUESTORE_TOOL_RESTORE_ALLOCATION int push_allocation_to_rocksdb(); int read_allocation_from_drive_for_bluestore_tool(); #endif void set_allocation_in_simple_bmap(SimpleBitmap* sbmap, uint64_t offset, uint64_t length); private: struct read_alloc_stats_t { uint32_t onode_count = 0; uint32_t shard_count = 0; uint32_t skipped_illegal_extent = 0; uint64_t shared_blob_count = 0; uint64_t compressed_blob_count = 0; uint64_t spanning_blob_count = 0; uint64_t insert_count = 0; uint64_t extent_count = 0; std::map actual_pool_vstatfs; volatile_statfs actual_store_vstatfs; }; class ExtentDecoderPartial : public ExtentMap::ExtentDecoder { BlueStore& store; read_alloc_stats_t& stats; SimpleBitmap& sbmap; sb_info_space_efficient_map_t& sb_info; uint8_t min_alloc_size_order; Extent extent; ghobject_t oid; volatile_statfs* per_pool_statfs = nullptr; blob_map_t blobs; blob_map_t spanning_blobs; void _consume_new_blob(bool spanning, uint64_t extent_no, uint64_t sbid, BlobRef b); protected: void consume_blobid(Extent*, bool spanning, uint64_t blobid) override; void consume_blob(Extent* le, uint64_t extent_no, uint64_t sbid, BlobRef b) override; void consume_spanning_blob(uint64_t sbid, BlobRef b) override; Extent* get_next_extent() override { ++stats.extent_count; extent = Extent(); return &extent; } void add_extent(Extent*) override { } public: ExtentDecoderPartial(BlueStore& _store, read_alloc_stats_t& _stats, SimpleBitmap& _sbmap, sb_info_space_efficient_map_t& _sb_info, uint8_t _min_alloc_size_order) : store(_store), stats(_stats), sbmap(_sbmap), sb_info(_sb_info), min_alloc_size_order(_min_alloc_size_order) {} const ghobject_t& get_oid() const { return oid; } void reset(const ghobject_t _oid, volatile_statfs* _per_pool_statfs); }; friend std::ostream& operator<<(std::ostream& out, const read_alloc_stats_t& stats) { out << "==========================================================" << std::endl; out << "NCB::onode_count = " ;out.width(10);out << stats.onode_count << std::endl << "NCB::shard_count = " ;out.width(10);out << stats.shard_count << std::endl << "NCB::shared_blob_count = " ;out.width(10);out << stats.shared_blob_count << std::endl << "NCB::compressed_blob_count = " ;out.width(10);out << stats.compressed_blob_count << std::endl << "NCB::spanning_blob_count = " ;out.width(10);out << stats.spanning_blob_count << std::endl << "NCB::skipped_illegal_extent = " ;out.width(10);out << stats.skipped_illegal_extent << std::endl << "NCB::extent_count = " ;out.width(10);out << stats.extent_count << std::endl << "NCB::insert_count = " ;out.width(10);out << stats.insert_count << std::endl; out << "==========================================================" << std::endl; return out; } int compare_allocators(Allocator* alloc1, Allocator* alloc2, uint64_t req_extent_count, uint64_t memory_target); Allocator* create_bitmap_allocator(uint64_t bdev_size); int add_existing_bluefs_allocation(Allocator* allocator, read_alloc_stats_t& stats); int allocator_add_restored_entries(Allocator *allocator, const void *buff, unsigned extent_count, uint64_t *p_read_alloc_size, uint64_t *p_extent_count, const void *v_header, BlueFS::FileReader *p_handle, uint64_t offset); int copy_allocator(Allocator* src_alloc, Allocator *dest_alloc, uint64_t* p_num_entries); int store_allocator(Allocator* allocator); int invalidate_allocation_file_on_bluefs(); int __restore_allocator(Allocator* allocator, uint64_t *num, uint64_t *bytes); int restore_allocator(Allocator* allocator, uint64_t *num, uint64_t *bytes); int read_allocation_from_drive_on_startup(); int reconstruct_allocations(SimpleBitmap *smbmp, read_alloc_stats_t &stats); int read_allocation_from_onodes(SimpleBitmap *smbmp, read_alloc_stats_t& stats); int commit_freelist_type(); int commit_to_null_manager(); int commit_to_real_manager(); int db_cleanup(int ret); int reset_fm_for_restore(); int verify_rocksdb_allocations(Allocator *allocator); Allocator* clone_allocator_without_bluefs(Allocator *src_allocator); Allocator* initialize_allocator_from_freelist(FreelistManager *real_fm); void copy_allocator_content_to_fm(Allocator *allocator, FreelistManager *real_fm); void _fsck_check_object_omap(FSCKDepth depth, OnodeRef& o, const BlueStore::FSCK_ObjectCtx& ctx); void _fsck_check_objects(FSCKDepth depth, FSCK_ObjectCtx& ctx); }; inline std::ostream& operator<<(std::ostream& out, const BlueStore::volatile_statfs& s) { return out << " allocated:" << s.values[BlueStore::volatile_statfs::STATFS_ALLOCATED] << " stored:" << s.values[BlueStore::volatile_statfs::STATFS_STORED] << " compressed:" << s.values[BlueStore::volatile_statfs::STATFS_COMPRESSED] << " compressed_orig:" << s.values[BlueStore::volatile_statfs::STATFS_COMPRESSED_ORIGINAL] << " compressed_alloc:" << s.values[BlueStore::volatile_statfs::STATFS_COMPRESSED_ALLOCATED]; } 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(); } class BlueStoreRepairer { ceph::mutex lock = ceph::make_mutex("BlueStore::BlueStoreRepairer::lock"); public: // to simplify future potential migration to mempools using fsck_interval = interval_set; // Structure to track what pextents are used for specific cid/oid. // Similar to Bloom filter positive and false-positive matches are // possible only. // Maintains two lists of bloom filters for both cids and oids // where each list entry is a BF for specific disk pextent // The length of the extent per filter is measured on init. // Allows to filter out 'uninteresting' pextents to speadup subsequent // 'is_used' access. struct StoreSpaceTracker { const uint64_t BLOOM_FILTER_SALT_COUNT = 2; const uint64_t BLOOM_FILTER_TABLE_SIZE = 32; // bytes per single filter const uint64_t BLOOM_FILTER_EXPECTED_COUNT = 16; // arbitrary selected static const uint64_t DEF_MEM_CAP = 128 * 1024 * 1024; typedef mempool::bluestore_fsck::vector bloom_vector; bloom_vector collections_bfs; bloom_vector objects_bfs; bool was_filtered_out = false; uint64_t granularity = 0; // extent length for a single filter StoreSpaceTracker() { } StoreSpaceTracker(const StoreSpaceTracker& from) : collections_bfs(from.collections_bfs), objects_bfs(from.objects_bfs), granularity(from.granularity) { } void init(uint64_t total, uint64_t min_alloc_size, uint64_t mem_cap = DEF_MEM_CAP) { ceph_assert(!granularity); // not initialized yet ceph_assert(std::has_single_bit(min_alloc_size)); ceph_assert(mem_cap); total = round_up_to(total, min_alloc_size); granularity = total * BLOOM_FILTER_TABLE_SIZE * 2 / mem_cap; if (!granularity) { granularity = min_alloc_size; } else { granularity = round_up_to(granularity, min_alloc_size); } uint64_t entries = round_up_to(total, granularity) / granularity; collections_bfs.resize(entries, bloom_filter(BLOOM_FILTER_SALT_COUNT, BLOOM_FILTER_TABLE_SIZE, 0, BLOOM_FILTER_EXPECTED_COUNT)); objects_bfs.resize(entries, bloom_filter(BLOOM_FILTER_SALT_COUNT, BLOOM_FILTER_TABLE_SIZE, 0, BLOOM_FILTER_EXPECTED_COUNT)); } inline uint32_t get_hash(const coll_t& cid) const { return cid.hash_to_shard(1); } inline void set_used(uint64_t offset, uint64_t len, const coll_t& cid, const ghobject_t& oid) { ceph_assert(granularity); // initialized // can't call this func after filter_out has been applied ceph_assert(!was_filtered_out); if (!len) { return; } auto pos = offset / granularity; auto end_pos = (offset + len - 1) / granularity; while (pos <= end_pos) { collections_bfs[pos].insert(get_hash(cid)); objects_bfs[pos].insert(oid.hobj.get_hash()); ++pos; } } // filter-out entries unrelated to the specified(broken) extents. // 'is_used' calls are permitted after that only size_t filter_out(const fsck_interval& extents); // determines if collection's present after filtering-out inline bool is_used(const coll_t& cid) const { ceph_assert(was_filtered_out); for(auto& bf : collections_bfs) { if (bf.contains(get_hash(cid))) { return true; } } return false; } // determines if object's present after filtering-out inline bool is_used(const ghobject_t& oid) const { ceph_assert(was_filtered_out); for(auto& bf : objects_bfs) { if (bf.contains(oid.hobj.get_hash())) { return true; } } return false; } // determines if collection's present before filtering-out inline bool is_used(const coll_t& cid, uint64_t offs) const { ceph_assert(granularity); // initialized ceph_assert(!was_filtered_out); auto &bf = collections_bfs[offs / granularity]; if (bf.contains(get_hash(cid))) { return true; } return false; } // determines if object's present before filtering-out inline bool is_used(const ghobject_t& oid, uint64_t offs) const { ceph_assert(granularity); // initialized ceph_assert(!was_filtered_out); auto &bf = objects_bfs[offs / granularity]; if (bf.contains(oid.hobj.get_hash())) { return true; } return false; } }; public: void fix_per_pool_omap(KeyValueDB *db, int); bool remove_key(KeyValueDB *db, const std::string& prefix, const std::string& key); bool fix_shared_blob(KeyValueDB::Transaction txn, uint64_t sbid, bluestore_extent_ref_map_t* ref_map, size_t repaired = 1); bool fix_statfs(KeyValueDB *db, const std::string& key, const store_statfs_t& new_statfs); bool fix_leaked(KeyValueDB *db, FreelistManager* fm, uint64_t offset, uint64_t len); bool fix_false_free(KeyValueDB *db, FreelistManager* fm, uint64_t offset, uint64_t len); bool fix_spanning_blobs( KeyValueDB* db, std::function f); bool preprocess_misreference(KeyValueDB *db); unsigned apply(KeyValueDB* db); void note_misreference(uint64_t offs, uint64_t len, bool inc_error) { std::lock_guard l(lock); misreferenced_extents.union_insert(offs, len); if (inc_error) { ++to_repair_cnt; } } ////////////////////// //In fact two methods below are the only ones in this class which are thread-safe!! void inc_repaired(size_t n = 1) { to_repair_cnt += n; } void request_compaction() { need_compact = true; } ////////////////////// void init_space_usage_tracker( uint64_t total_space, uint64_t lres_tracking_unit_size) { //NB: not for use in multithreading mode!!! space_usage_tracker.init(total_space, lres_tracking_unit_size); } void set_space_used(uint64_t offset, uint64_t len, const coll_t& cid, const ghobject_t& oid) { std::lock_guard l(lock); space_usage_tracker.set_used(offset, len, cid, oid); } inline bool is_used(const coll_t& cid) const { //NB: not for use in multithreading mode!!! return space_usage_tracker.is_used(cid); } inline bool is_used(const ghobject_t& oid) const { //NB: not for use in multithreading mode!!! return space_usage_tracker.is_used(oid); } const fsck_interval& get_misreferences() const { //NB: not for use in multithreading mode!!! return misreferenced_extents; } KeyValueDB::Transaction get_fix_misreferences_txn() { //NB: not for use in multithreading mode!!! return fix_misreferences_txn; } private: std::atomic to_repair_cnt = { 0 }; std::atomic need_compact = { false }; KeyValueDB::Transaction fix_per_pool_omap_txn; KeyValueDB::Transaction fix_fm_leaked_txn; KeyValueDB::Transaction fix_fm_false_free_txn; KeyValueDB::Transaction remove_key_txn; KeyValueDB::Transaction fix_statfs_txn; KeyValueDB::Transaction fix_shared_blob_txn; KeyValueDB::Transaction fix_misreferences_txn; KeyValueDB::Transaction fix_onode_txn; StoreSpaceTracker space_usage_tracker; // non-shared extents with multiple references fsck_interval misreferenced_extents; }; class RocksDBBlueFSVolumeSelector : public BlueFSVolumeSelector { template class matrix_2d { T values[MaxX][MaxY]; public: matrix_2d() { clear(); } T& at(size_t x, size_t y) { ceph_assert(x < MaxX); ceph_assert(y < MaxY); return values[x][y]; } size_t get_max_x() const { return MaxX; } size_t get_max_y() const { return MaxY; } void clear() { memset(values, 0, sizeof(values)); } }; enum { // use 0/nullptr as unset indication LEVEL_FIRST = 1, LEVEL_LOG = LEVEL_FIRST, // BlueFS log LEVEL_WAL, LEVEL_DB, LEVEL_SLOW, LEVEL_MAX }; // add +1 row for corresponding per-device totals // add +1 column for per-level actual (taken from file size) total typedef matrix_2d, BlueFS::MAX_BDEV + 1, LEVEL_MAX - LEVEL_FIRST + 1> per_level_per_dev_usage_t; per_level_per_dev_usage_t per_level_per_dev_usage; // file count per level, add +1 to keep total file count std::atomic per_level_files[LEVEL_MAX - LEVEL_FIRST + 1] = { 0 }; // Note: maximum per-device totals below might be smaller than corresponding // perf counters by up to a single alloc unit (1M) due to superblock extent. // The later is not accounted here. per_level_per_dev_usage_t per_level_per_dev_max; uint64_t l_totals[LEVEL_MAX - LEVEL_FIRST]; uint64_t db_avail4slow = 0; enum { OLD_POLICY, USE_SOME_EXTRA }; public: RocksDBBlueFSVolumeSelector( uint64_t _wal_total, uint64_t _db_total, uint64_t _slow_total, uint64_t _level0_size, uint64_t _level_base, uint64_t _level_multiplier, double reserved_factor, uint64_t reserved, bool new_pol) { l_totals[LEVEL_LOG - LEVEL_FIRST] = 0; // not used at the moment l_totals[LEVEL_WAL - LEVEL_FIRST] = _wal_total; l_totals[LEVEL_DB - LEVEL_FIRST] = _db_total; l_totals[LEVEL_SLOW - LEVEL_FIRST] = _slow_total; if (!new_pol) { return; } // Calculating how much extra space is available at DB volume. // Depending on the presence of explicit reserved size specification it might be either // * DB volume size - reserved // or // * DB volume size - sum_max_level_size(0, L-1) - max_level_size(L) * reserved_factor if (!reserved) { uint64_t prev_levels = _level0_size; uint64_t cur_level = _level_base; uint64_t cur_threshold = 0; do { uint64_t next_level = cur_level * _level_multiplier; uint64_t next_threshold = prev_levels + cur_level + next_level * reserved_factor; if (_db_total <= next_threshold) { db_avail4slow = cur_threshold ? _db_total - cur_threshold : 0; break; } else { prev_levels += cur_level; cur_level = next_level; cur_threshold = next_threshold; } } while (true); } else { db_avail4slow = _db_total - reserved; } } void* get_hint_for_log() const override { return reinterpret_cast(LEVEL_LOG); } void* get_hint_by_dir(std::string_view dirname) const override; void add_usage(void* hint, const bluefs_fnode_t& fnode) override { if (hint == nullptr) return; size_t pos = (size_t)hint - LEVEL_FIRST; for (auto& p : fnode.extents) { auto& cur = per_level_per_dev_usage.at(p.bdev, pos); auto& max = per_level_per_dev_max.at(p.bdev, pos); uint64_t v = cur.fetch_add(p.length) + p.length; while (v > max) { max.exchange(v); } { //update per-device totals auto& cur = per_level_per_dev_usage.at(p.bdev, LEVEL_MAX - LEVEL_FIRST); auto& max = per_level_per_dev_max.at(p.bdev, LEVEL_MAX - LEVEL_FIRST); uint64_t v = cur.fetch_add(p.length) + p.length; while (v > max) { max.exchange(v); } } } { //update per-level actual totals auto& cur = per_level_per_dev_usage.at(BlueFS::MAX_BDEV, pos); auto& max = per_level_per_dev_max.at(BlueFS::MAX_BDEV, pos); uint64_t v = cur.fetch_add(fnode.size) + fnode.size; while (v > max) { max.exchange(v); } } ++per_level_files[pos]; ++per_level_files[LEVEL_MAX - LEVEL_FIRST]; } void sub_usage(void* hint, const bluefs_fnode_t& fnode) override { if (hint == nullptr) return; size_t pos = (size_t)hint - LEVEL_FIRST; for (auto& p : fnode.extents) { auto& cur = per_level_per_dev_usage.at(p.bdev, pos); ceph_assert(cur >= p.length); cur -= p.length; //update per-device totals auto& cur2 = per_level_per_dev_usage.at(p.bdev, LEVEL_MAX - LEVEL_FIRST); ceph_assert(cur2 >= p.length); cur2 -= p.length; } //update per-level actual totals auto& cur = per_level_per_dev_usage.at(BlueFS::MAX_BDEV, pos); ceph_assert(cur >= fnode.size); cur -= fnode.size; ceph_assert(per_level_files[pos] > 0); --per_level_files[pos]; ceph_assert(per_level_files[LEVEL_MAX - LEVEL_FIRST] > 0); --per_level_files[LEVEL_MAX - LEVEL_FIRST]; } void add_usage(void* hint, uint64_t size_more) override { if (hint == nullptr) return; size_t pos = (size_t)hint - LEVEL_FIRST; //update per-level actual totals auto& cur = per_level_per_dev_usage.at(BlueFS::MAX_BDEV, pos); auto& max = per_level_per_dev_max.at(BlueFS::MAX_BDEV, pos); uint64_t v = cur.fetch_add(size_more) + size_more; while (v > max) { max.exchange(v); } } void sub_usage(void* hint, uint64_t size_less) override { if (hint == nullptr) return; size_t pos = (size_t)hint - LEVEL_FIRST; //update per-level actual totals auto& cur = per_level_per_dev_usage.at(BlueFS::MAX_BDEV, pos); ceph_assert(cur >= size_less); cur -= size_less; } uint8_t select_prefer_bdev(void* h) override; void get_paths( const std::string& base, BlueFSVolumeSelector::paths& res) const override; void dump(std::ostream& sout) override; BlueFSVolumeSelector* clone_empty() const override; bool compare(BlueFSVolumeSelector* other) override; }; #endif