--- /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) 2015 Haomai Wang
+ *
+ * Author: Haomai Wang <haomaiwang@gmail.com>
+ *
+ * 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.
+ *
+ */
+
+#include <stdlib.h>
+#include <stdint.h>
+#include <string>
+#include <unistd.h>
+#include <iostream>
+
+using namespace std;
+
+#include "common/ceph_argparse.h"
+#include "common/debug.h"
+#include "common/Cycles.h"
+#include "global/global_init.h"
+#include "msg/Messenger.h"
+#include "messages/MOSDOp.h"
+
+#include <atomic>
+
+class MessengerClient {
+ class ClientThread;
+ class ClientDispatcher : public Dispatcher {
+ uint64_t think_time;
+ ClientThread *thread;
+
+ public:
+ ClientDispatcher(uint64_t delay, ClientThread *t): Dispatcher(g_ceph_context), think_time(delay), thread(t) {}
+ bool ms_can_fast_dispatch_any() const override { return true; }
+ bool ms_can_fast_dispatch(const Message *m) const override {
+ switch (m->get_type()) {
+ case CEPH_MSG_OSD_OPREPLY:
+ return true;
+ default:
+ return false;
+ }
+ }
+
+ void ms_handle_fast_connect(Connection *con) override {}
+ void ms_handle_fast_accept(Connection *con) override {}
+ bool ms_dispatch(Message *m) override { return true; }
+ void ms_fast_dispatch(Message *m) override;
+ bool ms_handle_reset(Connection *con) override { return true; }
+ void ms_handle_remote_reset(Connection *con) override {}
+ bool ms_handle_refused(Connection *con) override { return false; }
+ bool ms_verify_authorizer(Connection *con, int peer_type, int protocol,
+ bufferlist& authorizer, bufferlist& authorizer_reply,
+ bool& isvalid, CryptoKey& session_key) override {
+ isvalid = true;
+ return true;
+ }
+ };
+
+ class ClientThread : public Thread {
+ Messenger *msgr;
+ int concurrent;
+ ConnectionRef conn;
+ std::atomic<unsigned> client_inc = { 0 };
+ object_t oid;
+ object_locator_t oloc;
+ pg_t pgid;
+ int msg_len;
+ bufferlist data;
+ int ops;
+ ClientDispatcher dispatcher;
+
+ public:
+ Mutex lock;
+ Cond cond;
+ uint64_t inflight;
+
+ ClientThread(Messenger *m, int c, ConnectionRef con, int len, int ops, int think_time_us):
+ msgr(m), concurrent(c), conn(con), oid("object-name"), oloc(1, 1), msg_len(len), ops(ops),
+ dispatcher(think_time_us, this), lock("MessengerBenchmark::ClientThread::lock"), inflight(0) {
+ m->add_dispatcher_head(&dispatcher);
+ bufferptr ptr(msg_len);
+ memset(ptr.c_str(), 0, msg_len);
+ data.append(ptr);
+ }
+ void *entry() override {
+ lock.Lock();
+ for (int i = 0; i < ops; ++i) {
+ if (inflight > uint64_t(concurrent)) {
+ cond.Wait(lock);
+ }
+ hobject_t hobj(oid, oloc.key, CEPH_NOSNAP, pgid.ps(), pgid.pool(),
+ oloc.nspace);
+ spg_t spgid(pgid);
+ MOSDOp *m = new MOSDOp(client_inc, 0, hobj, spgid, 0, 0, 0);
+ m->write(0, msg_len, data);
+ inflight++;
+ conn->send_message(m);
+ //cerr << __func__ << " send m=" << m << std::endl;
+ }
+ lock.Unlock();
+ msgr->shutdown();
+ return 0;
+ }
+ };
+
+ string type;
+ string serveraddr;
+ int think_time_us;
+ vector<Messenger*> msgrs;
+ vector<ClientThread*> clients;
+
+ public:
+ MessengerClient(string t, string addr, int delay):
+ type(t), serveraddr(addr), think_time_us(delay) {
+ }
+ ~MessengerClient() {
+ for (uint64_t i = 0; i < clients.size(); ++i)
+ delete clients[i];
+ for (uint64_t i = 0; i < msgrs.size(); ++i) {
+ msgrs[i]->shutdown();
+ msgrs[i]->wait();
+ }
+ }
+ void ready(int c, int jobs, int ops, int msg_len) {
+ entity_addr_t addr;
+ addr.parse(serveraddr.c_str());
+ addr.set_nonce(0);
+ for (int i = 0; i < jobs; ++i) {
+ Messenger *msgr = Messenger::create(g_ceph_context, type, entity_name_t::CLIENT(0), "client", getpid()+i, 0);
+ msgr->set_default_policy(Messenger::Policy::lossless_client(0));
+ entity_inst_t inst(entity_name_t::OSD(0), addr);
+ ConnectionRef conn = msgr->get_connection(inst);
+ ClientThread *t = new ClientThread(msgr, c, conn, msg_len, ops, think_time_us);
+ msgrs.push_back(msgr);
+ clients.push_back(t);
+ msgr->start();
+ }
+ usleep(1000*1000);
+ }
+ void start() {
+ for (uint64_t i = 0; i < clients.size(); ++i)
+ clients[i]->create("client");
+ for (uint64_t i = 0; i < msgrs.size(); ++i)
+ msgrs[i]->wait();
+ }
+};
+
+void MessengerClient::ClientDispatcher::ms_fast_dispatch(Message *m) {
+ usleep(think_time);
+ m->put();
+ Mutex::Locker l(thread->lock);
+ thread->inflight--;
+ thread->cond.Signal();
+}
+
+
+void usage(const string &name) {
+ cerr << "Usage: " << name << " [server ip:port] [numjobs] [concurrency] [ios] [thinktime us] [msg length]" << std::endl;
+ cerr << " [server ip:port]: connect to the ip:port pair" << std::endl;
+ cerr << " [numjobs]: how much client threads spawned and do benchmark" << std::endl;
+ cerr << " [concurrency]: the max inflight messages(like iodepth in fio)" << std::endl;
+ cerr << " [ios]: how much messages sent for each client" << std::endl;
+ cerr << " [thinktime]: sleep time when do fast dispatching(match client logic)" << std::endl;
+ cerr << " [msg length]: message data bytes" << std::endl;
+}
+
+int main(int argc, char **argv)
+{
+ vector<const char*> args;
+ argv_to_vec(argc, (const char **)argv, args);
+
+ auto cct = global_init(NULL, args, CEPH_ENTITY_TYPE_CLIENT,
+ CODE_ENVIRONMENT_UTILITY, 0);
+ common_init_finish(g_ceph_context);
+ g_ceph_context->_conf->apply_changes(NULL);
+
+ if (args.size() < 6) {
+ usage(argv[0]);
+ return 1;
+ }
+
+ int numjobs = atoi(args[1]);
+ int concurrent = atoi(args[2]);
+ int ios = atoi(args[3]);
+ int think_time = atoi(args[4]);
+ int len = atoi(args[5]);
+
+ std::string public_msgr_type = g_ceph_context->_conf->ms_public_type.empty() ? g_ceph_context->_conf->get_val<std::string>("ms_type") : g_ceph_context->_conf->ms_public_type;
+
+ cerr << " using ms-public-type " << public_msgr_type << std::endl;
+ cerr << " server ip:port " << args[0] << std::endl;
+ cerr << " numjobs " << numjobs << std::endl;
+ cerr << " concurrency " << concurrent << std::endl;
+ cerr << " ios " << ios << std::endl;
+ cerr << " thinktime(us) " << think_time << std::endl;
+ cerr << " message data bytes " << len << std::endl;
+
+ MessengerClient client(public_msgr_type, args[0], think_time);
+
+ client.ready(concurrent, numjobs, ios, len);
+ Cycles::init();
+ uint64_t start = Cycles::rdtsc();
+ client.start();
+ uint64_t stop = Cycles::rdtsc();
+ cerr << " Total op " << ios << " run time " << Cycles::to_microseconds(stop - start) << "us." << std::endl;
+
+ return 0;
+}
Code Review / stor4nfv.git / blobdiff