Adding upstream version 14.2.21.upstream/14.2.21upstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 352 insertions, 0 deletions

diff --git a/src/osd/OSDMapMapping.h b/src/osd/OSDMapMapping.h
new file mode 100644
index 00000000..b0965fc1
--- /dev/null
+++ b/src/osd/OSDMapMapping.h
@@ -0,0 +1,352 @@
+// -*- 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"
+#include "common/Cond.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() {
+      ceph_assert(shards == 0);
+    }
+
+    // child must implement either form of process
+    virtual void process(const vector<pg_t>& pgs) = 0;
+    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() {
+      std::lock_guard l(lock);
+      return shards == 0;
+    }
+    utime_t get_duration() {
+      return finish - start;
+    }
+    void wait() {
+      std::lock_guard l(lock);
+      while (shards > 0) {
+	cond.Wait(lock);
+      }
+    }
+    bool wait_for(double duration) {
+      utime_t until = start;
+      until += duration;
+      std::lock_guard l(lock);
+      while (shards > 0) {
+	if (ceph_clock_now() >= until) {
+	  return false;
+	}
+	cond.Wait(lock);
+      }
+      return true;
+    }
+    void abort() {
+      Context *fin = nullptr;
+      {
+	std::lock_guard l(lock);
+	aborted = true;
+	fin = onfinish;
+	onfinish = nullptr;
+	while (shards > 0) {
+	  cond.Wait(lock);
+	}
+      }
+      if (fin) {
+	fin->complete(-ECANCELED);
+      }
+    }
+
+    void start_one() {
+      std::lock_guard l(lock);
+      ++shards;
+    }
+    void finish_one();
+  };
+
+protected:
+  CephContext *cct;
+
+  struct Item {
+    Job *job;
+    int64_t pool;
+    unsigned begin, end;
+    vector<pg_t> pgs;
+
+    Item(Job *j, vector<pg_t> pgs) : job(j), pgs(pgs) {}
+    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 {
+      ceph_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,
+    const vector<pg_t>& input_pgs);
+
+  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;
+    bool erasure = false;
+    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, bool e)
+      : size(s),
+	pg_num(p),
+	erasure(e),
+	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;
+      // these should always be <= the pool size, but just in case, avoid
+      // blowing out the array.  Note that our mapping is not completely
+      // accurate in this case--this is just to avoid crashing.
+      row[2] = std::min<int32_t>(acting.size(), size);
+      row[3] = std::min<int32_t>(up.size(), 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(const vector<pg_t>& pgs) override {}
+    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());
+    ceph_assert(p != pools.end());
+    ceph_assert(pgid.ps() < p->second.pg_num);
+    p->second.get(pgid.ps(), up, up_primary, acting, acting_primary);
+  }
+
+  bool get_primary_and_shard(pg_t pgid,
+			     int *acting_primary,
+			     spg_t *spgid) {
+    auto p = pools.find(pgid.pool());
+    ceph_assert(p != pools.end());
+    ceph_assert(pgid.ps() < p->second.pg_num);
+    vector<int> acting;
+    p->second.get(pgid.ps(), nullptr, nullptr, &acting, acting_primary);
+    if (p->second.erasure) {
+      for (uint8_t i = 0; i < acting.size(); ++i) {
+	if (acting[i] == *acting_primary) {
+	  *spgid = spg_t(pgid, shard_id_t(i));
+	  return true;
+	}
+      }
+      return false;
+    } else {
+      *spgid = spg_t(pgid);
+      return true;
+    }
+  }
+
+  const mempool::osdmap_mapping::vector<pg_t>& get_osd_acting_pgs(unsigned osd) {
+    ceph_assert(osd < acting_rmap.size());
+    return acting_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