--- /dev/null
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+// vim: ts=8 sw=2 smarttab
+
+
+#ifndef CEPH_OSDMAPMAPPING_H
+#define CEPH_OSDMAPMAPPING_H
+
+#include <vector>
+#include <map>
+
+#include "osd/osd_types.h"
+#include "common/WorkQueue.h"
+
+class OSDMap;
+
+/// work queue to perform work on batches of pgids on multiple CPUs
+class ParallelPGMapper {
+public:
+ struct Job {
+ utime_t start, finish;
+ unsigned shards = 0;
+ const OSDMap *osdmap;
+ bool aborted = false;
+ Context *onfinish = nullptr;
+
+ Mutex lock = {"ParallelPGMapper::Job::lock"};
+ Cond cond;
+
+ Job(const OSDMap *om) : start(ceph_clock_now()), osdmap(om) {}
+ virtual ~Job() {
+ assert(shards == 0);
+ }
+
+ // child must implement this
+ virtual void process(int64_t poolid, unsigned ps_begin, unsigned ps_end) = 0;
+ virtual void complete() = 0;
+
+ void set_finish_event(Context *fin) {
+ lock.Lock();
+ if (shards == 0) {
+ // already done.
+ lock.Unlock();
+ fin->complete(0);
+ } else {
+ // set finisher
+ onfinish = fin;
+ lock.Unlock();
+ }
+ }
+ bool is_done() {
+ Mutex::Locker l(lock);
+ return shards == 0;
+ }
+ utime_t get_duration() {
+ return finish - start;
+ }
+ void wait() {
+ Mutex::Locker l(lock);
+ while (shards > 0) {
+ cond.Wait(lock);
+ }
+ }
+ bool wait_for(double duration) {
+ utime_t until = start;
+ until += duration;
+ Mutex::Locker l(lock);
+ while (shards > 0) {
+ if (ceph_clock_now() >= until) {
+ return false;
+ }
+ cond.Wait(lock);
+ }
+ return true;
+ }
+ void abort() {
+ Context *fin = nullptr;
+ {
+ Mutex::Locker l(lock);
+ aborted = true;
+ fin = onfinish;
+ onfinish = nullptr;
+ while (shards > 0) {
+ cond.Wait(lock);
+ }
+ }
+ if (fin) {
+ fin->complete(-ECANCELED);
+ }
+ }
+
+ void start_one() {
+ Mutex::Locker l(lock);
+ ++shards;
+ }
+ void finish_one();
+ };
+
+protected:
+ CephContext *cct;
+
+ struct Item {
+ Job *job;
+ int64_t pool;
+ unsigned begin, end;
+
+ Item(Job *j, int64_t p, unsigned b, unsigned e)
+ : job(j),
+ pool(p),
+ begin(b),
+ end(e) {}
+ };
+ std::deque<Item*> q;
+
+ struct WQ : public ThreadPool::WorkQueue<Item> {
+ ParallelPGMapper *m;
+
+ WQ(ParallelPGMapper *m_, ThreadPool *tp)
+ : ThreadPool::WorkQueue<Item>("ParallelPGMapper::WQ", 0, 0, tp),
+ m(m_) {}
+
+ bool _enqueue(Item *i) override {
+ m->q.push_back(i);
+ return true;
+ }
+ void _dequeue(Item *i) override {
+ ceph_abort();
+ }
+ Item *_dequeue() override {
+ while (!m->q.empty()) {
+ Item *i = m->q.front();
+ m->q.pop_front();
+ if (i->job->aborted) {
+ i->job->finish_one();
+ delete i;
+ } else {
+ return i;
+ }
+ }
+ return nullptr;
+ }
+
+ void _process(Item *i, ThreadPool::TPHandle &h) override;
+
+ void _clear() override {
+ assert(_empty());
+ }
+
+ bool _empty() override {
+ return m->q.empty();
+ }
+ } wq;
+
+public:
+ ParallelPGMapper(CephContext *cct, ThreadPool *tp)
+ : cct(cct),
+ wq(this, tp) {}
+
+ void queue(
+ Job *job,
+ unsigned pgs_per_item);
+
+ void drain() {
+ wq.drain();
+ }
+};
+
+
+/// a precalculated mapping of every PG for a given OSDMap
+class OSDMapMapping {
+public:
+ MEMPOOL_CLASS_HELPERS();
+private:
+
+ struct PoolMapping {
+ MEMPOOL_CLASS_HELPERS();
+
+ unsigned size = 0;
+ unsigned pg_num = 0;
+ mempool::osdmap_mapping::vector<int32_t> table;
+
+ size_t row_size() const {
+ return
+ 1 + // acting_primary
+ 1 + // up_primary
+ 1 + // num acting
+ 1 + // num up
+ size + // acting
+ size; // up
+ }
+
+ PoolMapping(int s, int p)
+ : size(s),
+ pg_num(p),
+ table(pg_num * row_size()) {
+ }
+
+ void get(size_t ps,
+ std::vector<int> *up,
+ int *up_primary,
+ std::vector<int> *acting,
+ int *acting_primary) const {
+ const int32_t *row = &table[row_size() * ps];
+ if (acting_primary) {
+ *acting_primary = row[0];
+ }
+ if (up_primary) {
+ *up_primary = row[1];
+ }
+ if (acting) {
+ acting->resize(row[2]);
+ for (int i = 0; i < row[2]; ++i) {
+ (*acting)[i] = row[4 + i];
+ }
+ }
+ if (up) {
+ up->resize(row[3]);
+ for (int i = 0; i < row[3]; ++i) {
+ (*up)[i] = row[4 + size + i];
+ }
+ }
+ }
+
+ void set(size_t ps,
+ const std::vector<int>& up,
+ int up_primary,
+ const std::vector<int>& acting,
+ int acting_primary) {
+ int32_t *row = &table[row_size() * ps];
+ row[0] = acting_primary;
+ row[1] = up_primary;
+ row[2] = acting.size();
+ row[3] = up.size();
+ for (int i = 0; i < row[2]; ++i) {
+ row[4 + i] = acting[i];
+ }
+ for (int i = 0; i < row[3]; ++i) {
+ row[4 + size + i] = up[i];
+ }
+ }
+ };
+
+ mempool::osdmap_mapping::map<int64_t,PoolMapping> pools;
+ mempool::osdmap_mapping::vector<
+ mempool::osdmap_mapping::vector<pg_t>> acting_rmap; // osd -> pg
+ //unused: mempool::osdmap_mapping::vector<std::vector<pg_t>> up_rmap; // osd -> pg
+ epoch_t epoch = 0;
+ uint64_t num_pgs = 0;
+
+ void _init_mappings(const OSDMap& osdmap);
+ void _update_range(
+ const OSDMap& map,
+ int64_t pool,
+ unsigned pg_begin, unsigned pg_end);
+
+ void _build_rmap(const OSDMap& osdmap);
+
+ void _start(const OSDMap& osdmap) {
+ _init_mappings(osdmap);
+ }
+ void _finish(const OSDMap& osdmap);
+
+ void _dump();
+
+ friend class ParallelPGMapper;
+
+ struct MappingJob : public ParallelPGMapper::Job {
+ OSDMapMapping *mapping;
+ MappingJob(const OSDMap *osdmap, OSDMapMapping *m)
+ : Job(osdmap), mapping(m) {
+ mapping->_start(*osdmap);
+ }
+ void process(int64_t pool, unsigned ps_begin, unsigned ps_end) override {
+ mapping->_update_range(*osdmap, pool, ps_begin, ps_end);
+ }
+ void complete() override {
+ mapping->_finish(*osdmap);
+ }
+ };
+
+public:
+ void get(pg_t pgid,
+ std::vector<int> *up,
+ int *up_primary,
+ std::vector<int> *acting,
+ int *acting_primary) const {
+ auto p = pools.find(pgid.pool());
+ assert(p != pools.end());
+ assert(pgid.ps() < p->second.pg_num);
+ p->second.get(pgid.ps(), up, up_primary, acting, acting_primary);
+ }
+
+ const mempool::osdmap_mapping::vector<pg_t>& get_osd_acting_pgs(unsigned osd) {
+ assert(osd < acting_rmap.size());
+ return acting_rmap[osd];
+ }
+ /* unsued
+ const std::vector<pg_t>& get_osd_up_pgs(unsigned osd) {
+ assert(osd < up_rmap.size());
+ return up_rmap[osd];
+ }
+ */
+
+ void update(const OSDMap& map);
+ void update(const OSDMap& map, pg_t pgid);
+
+ std::unique_ptr<MappingJob> start_update(
+ const OSDMap& map,
+ ParallelPGMapper& mapper,
+ unsigned pgs_per_item) {
+ std::unique_ptr<MappingJob> job(new MappingJob(&map, this));
+ mapper.queue(job.get(), pgs_per_item);
+ return job;
+ }
+
+ epoch_t get_epoch() const {
+ return epoch;
+ }
+
+ uint64_t get_num_pgs() const {
+ return num_pgs;
+ }
+};
+
+
+#endif
Code Review / stor4nfv.git / blobdiff