--- /dev/null
+// -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
+
+#include <boost/scoped_ptr.hpp>
+#include <boost/lexical_cast.hpp>
+#include <boost/program_options/option.hpp>
+#include <boost/program_options/options_description.hpp>
+#include <boost/program_options/variables_map.hpp>
+#include <boost/program_options/cmdline.hpp>
+#include <boost/program_options/parsers.hpp>
+#include <iostream>
+#include <set>
+#include <sstream>
+#include <stdlib.h>
+#include <fstream>
+
+#include "common/Formatter.h"
+
+#include "bencher.h"
+#include "rados_backend.h"
+#include "detailed_stat_collector.h"
+#include "distribution.h"
+#include "global/global_init.h"
+#include "common/WorkQueue.h"
+#include "common/Semaphore.h"
+#include "common/Finisher.h"
+
+namespace po = boost::program_options;
+using namespace std;
+class Queueable {
+public:
+ virtual void queue(unsigned *) = 0;
+ virtual void start() = 0;
+ virtual void stop() = 0;
+ virtual ~Queueable() {};
+};
+class Base : public Queueable {
+ DetailedStatCollector *col;
+ Semaphore *sem;
+public:
+ Base(DetailedStatCollector *col,
+ Semaphore *sem) : col(col), sem(sem) {}
+ void queue(unsigned *item) override {
+ col->read_complete(*item);
+ sem->Put();
+ delete item;
+ }
+ void start() override {}
+ void stop() override {}
+};
+class WQWrapper : public Queueable {
+ boost::scoped_ptr<ThreadPool::WorkQueue<unsigned> > wq;
+ boost::scoped_ptr<ThreadPool> tp;
+public:
+ WQWrapper(ThreadPool::WorkQueue<unsigned> *wq, ThreadPool *tp):
+ wq(wq), tp(tp) {}
+ void queue(unsigned *item) override { wq->queue(item); }
+ void start() override { tp->start(); }
+ void stop() override { tp->stop(); }
+};
+class FinisherWrapper : public Queueable {
+ class CB : public Context {
+ Queueable *next;
+ unsigned *item;
+ public:
+ CB(Queueable *next, unsigned *item) : next(next), item(item) {}
+ void finish(int) override {
+ next->queue(item);
+ }
+ };
+ Finisher f;
+ Queueable *next;
+public:
+ FinisherWrapper(CephContext *cct, Queueable *next) :
+ f(cct), next(next) {}
+ void queue(unsigned *item) override {
+ f.queue(new CB(next, item));
+ }
+ void start() override { f.start(); }
+ void stop() override { f.stop(); }
+};
+class PassAlong : public ThreadPool::WorkQueue<unsigned> {
+ Queueable *next;
+ list<unsigned*> q;
+ bool _enqueue(unsigned *item) override {
+ q.push_back(item);
+ return true;
+ }
+ void _dequeue(unsigned *item) override { ceph_abort(); }
+ unsigned *_dequeue() override {
+ if (q.empty())
+ return 0;
+ unsigned *val = q.front();
+ q.pop_front();
+ return val;
+ }
+ void _process(unsigned *item, ThreadPool::TPHandle &) override {
+ next->queue(item);
+ }
+ void _clear() override { q.clear(); }
+ bool _empty() override { return q.empty(); }
+public:
+ PassAlong(ThreadPool *tp, Queueable *_next) :
+ ThreadPool::WorkQueue<unsigned>("TestQueue", 100, 100, tp), next(_next) {}
+};
+
+int main(int argc, char **argv)
+{
+ po::options_description desc("Allowed options");
+ desc.add_options()
+ ("help", "produce help message")
+ ("num-threads", po::value<unsigned>()->default_value(10),
+ "set number of threads")
+ ("queue-size", po::value<unsigned>()->default_value(30),
+ "queue size")
+ ("num-items", po::value<unsigned>()->default_value(3000000),
+ "num items")
+ ("layers", po::value<string>()->default_value(""),
+ "layer desc")
+ ;
+
+ vector<string> ceph_option_strings;
+ po::variables_map vm;
+ try {
+ po::parsed_options parsed =
+ po::command_line_parser(argc, argv).options(desc).allow_unregistered().run();
+ po::store(
+ parsed,
+ vm);
+ po::notify(vm);
+
+ ceph_option_strings = po::collect_unrecognized(parsed.options,
+ po::include_positional);
+ } catch(po::error &e) {
+ std::cerr << e.what() << std::endl;
+ return 1;
+ }
+ vector<const char *> ceph_options, def_args;
+ ceph_options.reserve(ceph_option_strings.size());
+ for (vector<string>::iterator i = ceph_option_strings.begin();
+ i != ceph_option_strings.end();
+ ++i) {
+ ceph_options.push_back(i->c_str());
+ }
+
+ auto cct = global_init(
+ &def_args, ceph_options, CEPH_ENTITY_TYPE_CLIENT,
+ CODE_ENVIRONMENT_UTILITY,
+ CINIT_FLAG_NO_DEFAULT_CONFIG_FILE);
+ common_init_finish(g_ceph_context);
+ g_ceph_context->_conf->apply_changes(NULL);
+
+ if (vm.count("help")) {
+ cout << desc << std::endl;
+ return 1;
+ }
+
+ DetailedStatCollector col(1, new JSONFormatter, 0, &cout);
+ Semaphore sem;
+ for (unsigned i = 0; i < vm["queue-size"].as<unsigned>(); ++i)
+ sem.Put();
+
+ typedef list<Queueable*> QQ;
+ QQ wqs;
+ wqs.push_back(
+ new Base(&col, &sem));
+ string layers(vm["layers"].as<string>());
+ unsigned num = 0;
+ for (string::reverse_iterator i = layers.rbegin();
+ i != layers.rend(); ++i) {
+ stringstream ss;
+ ss << "Test " << num;
+ if (*i == 'q') {
+ ThreadPool *tp =
+ new ThreadPool(
+ g_ceph_context, ss.str(), "tp_test", vm["num-threads"].as<unsigned>(), 0);
+ wqs.push_back(
+ new WQWrapper(
+ new PassAlong(tp, wqs.back()),
+ tp
+ ));
+ } else if (*i == 'f') {
+ wqs.push_back(
+ new FinisherWrapper(
+ g_ceph_context, wqs.back()));
+ }
+ ++num;
+ }
+
+ for (QQ::iterator i = wqs.begin();
+ i != wqs.end();
+ ++i) {
+ (*i)->start();
+ }
+
+ for (uint64_t i = 0; i < vm["num-items"].as<unsigned>(); ++i) {
+ sem.Get();
+ unsigned *item = new unsigned(col.next_seq());
+ col.start_read(*item, 1);
+ wqs.back()->queue(item);
+ }
+
+ for (QQ::iterator i = wqs.begin();
+ i != wqs.end();
+ ++i) {
+ (*i)->stop();
+ }
+ for (QQ::iterator i = wqs.begin(); i != wqs.end(); wqs.erase(i++)) {
+ delete *i;
+ }
+ return 0;
+}
Code Review / stor4nfv.git / blobdiff