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