+++ /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