ceph/src/test/msgr/perf_msgr_client.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3 /*
   4  * Ceph - scalable distributed file system
   5  *
   6  * Copyright (C) 2015 Haomai Wang
   7  *
   8  * Author: Haomai Wang <haomaiwang@gmail.com>
   9  *
  10  * This is free software; you can redistribute it and/or
  11  * modify it under the terms of the GNU Lesser General Public
  12  * License version 2.1, as published by the Free Software
  13  * Foundation.  See file COPYING.
  14  *
  15  */
  16
  17 #include <stdlib.h>
  18 #include <stdint.h>
  19 #include <string>
  20 #include <unistd.h>
  21 #include <iostream>
  22
  23 using namespace std;
  24
  25 #include "common/ceph_argparse.h"
  26 #include "common/debug.h"
  27 #include "common/Cycles.h"
  28 #include "global/global_init.h"
  29 #include "msg/Messenger.h"
  30 #include "messages/MOSDOp.h"
  31
  32 #include <atomic>
  33
  34 class MessengerClient {
  35   class ClientThread;
  36   class ClientDispatcher : public Dispatcher {
  37     uint64_t think_time;
  38     ClientThread *thread;
  39
  40    public:
  41     ClientDispatcher(uint64_t delay, ClientThread *t): Dispatcher(g_ceph_context), think_time(delay), thread(t) {}
  42     bool ms_can_fast_dispatch_any() const override { return true; }
  43     bool ms_can_fast_dispatch(const Message *m) const override {
  44       switch (m->get_type()) {
  45       case CEPH_MSG_OSD_OPREPLY:
  46         return true;
  47       default:
  48         return false;
  49       }
  50     }
  51
  52     void ms_handle_fast_connect(Connection *con) override {}
  53     void ms_handle_fast_accept(Connection *con) override {}
  54     bool ms_dispatch(Message *m) override { return true; }
  55     void ms_fast_dispatch(Message *m) override;
  56     bool ms_handle_reset(Connection *con) override { return true; }
  57     void ms_handle_remote_reset(Connection *con) override {}
  58     bool ms_handle_refused(Connection *con) override { return false; }
  59     int ms_handle_authentication(Connection *con) override {
  60       return 1;
  61     }
  62   };
  63
  64   class ClientThread : public Thread {
  65     Messenger *msgr;
  66     int concurrent;
  67     ConnectionRef conn;
  68     std::atomic<unsigned> client_inc = { 0 };
  69     object_t oid;
  70     object_locator_t oloc;
  71     pg_t pgid;
  72     int msg_len;
  73     bufferlist data;
  74     int ops;
  75     ClientDispatcher dispatcher;
  76
  77    public:
  78     Mutex lock;
  79     Cond cond;
  80     uint64_t inflight;
  81
  82     ClientThread(Messenger *m, int c, ConnectionRef con, int len, int ops, int think_time_us):
  83         msgr(m), concurrent(c), conn(con), oid("object-name"), oloc(1, 1), msg_len(len), ops(ops),
  84         dispatcher(think_time_us, this), lock("MessengerBenchmark::ClientThread::lock"), inflight(0) {
  85       m->add_dispatcher_head(&dispatcher);
  86       bufferptr ptr(msg_len);
  87       memset(ptr.c_str(), 0, msg_len);
  88       data.append(ptr);
  89     }
  90     void *entry() override {
  91       lock.Lock();
  92       for (int i = 0; i < ops; ++i) {
  93         if (inflight > uint64_t(concurrent)) {
  94           cond.Wait(lock);
  95         }
  96         hobject_t hobj(oid, oloc.key, CEPH_NOSNAP, pgid.ps(), pgid.pool(),
  97                        oloc.nspace);
  98         spg_t spgid(pgid);
  99         MOSDOp *m = new MOSDOp(client_inc, 0, hobj, spgid, 0, 0, 0);
 100         bufferlist msg_data(data);
 101         m->write(0, msg_len, msg_data);
 102         inflight++;
 103         conn->send_message(m);
 104         //cerr << __func__ << " send m=" << m << std::endl;
 105       }
 106       lock.Unlock();
 107       msgr->shutdown();
 108       return 0;
 109     }
 110   };
 111
 112   string type;
 113   string serveraddr;
 114   int think_time_us;
 115   vector<Messenger*> msgrs;
 116   vector<ClientThread*> clients;
 117
 118  public:
 119   MessengerClient(const string &t, const string &addr, int delay):
 120       type(t), serveraddr(addr), think_time_us(delay) {
 121   }
 122   ~MessengerClient() {
 123     for (uint64_t i = 0; i < clients.size(); ++i)
 124       delete clients[i];
 125     for (uint64_t i = 0; i < msgrs.size(); ++i) {
 126       msgrs[i]->shutdown();
 127       msgrs[i]->wait();
 128     }
 129   }
 130   void ready(int c, int jobs, int ops, int msg_len) {
 131     entity_addr_t addr;
 132     addr.parse(serveraddr.c_str());
 133     addr.set_nonce(0);
 134     for (int i = 0; i < jobs; ++i) {
 135       Messenger *msgr = Messenger::create(g_ceph_context, type, entity_name_t::CLIENT(0), "client", getpid()+i, 0);
 136       msgr->set_default_policy(Messenger::Policy::lossless_client(0));
 137       msgr->start();
 138       entity_addrvec_t addrs(addr);
 139       ConnectionRef conn = msgr->connect_to_osd(addrs);
 140       ClientThread *t = new ClientThread(msgr, c, conn, msg_len, ops, think_time_us);
 141       msgrs.push_back(msgr);
 142       clients.push_back(t);
 143     }
 144     usleep(1000*1000);
 145   }
 146   void start() {
 147     for (uint64_t i = 0; i < clients.size(); ++i)
 148       clients[i]->create("client");
 149     for (uint64_t i = 0; i < msgrs.size(); ++i)
 150       msgrs[i]->wait();
 151   }
 152 };
 153
 154 void MessengerClient::ClientDispatcher::ms_fast_dispatch(Message *m) {
 155   usleep(think_time);
 156   m->put();
 157   Mutex::Locker l(thread->lock);
 158   thread->inflight--;
 159   thread->cond.Signal();
 160 }
 161
 162
 163 void usage(const string &name) {
 164   cerr << "Usage: " << name << " [server ip:port] [numjobs] [concurrency] [ios] [thinktime us] [msg length]" << std::endl;
 165   cerr << "       [server ip:port]: connect to the ip:port pair" << std::endl;
 166   cerr << "       [numjobs]: how much client threads spawned and do benchmark" << std::endl;
 167   cerr << "       [concurrency]: the max inflight messages(like iodepth in fio)" << std::endl;
 168   cerr << "       [ios]: how much messages sent for each client" << std::endl;
 169   cerr << "       [thinktime]: sleep time when do fast dispatching(match client logic)" << std::endl;
 170   cerr << "       [msg length]: message data bytes" << std::endl;
 171 }
 172
 173 int main(int argc, char **argv)
 174 {
 175   vector<const char*> args;
 176   argv_to_vec(argc, (const char **)argv, args);
 177
 178   auto cct = global_init(NULL, args, CEPH_ENTITY_TYPE_CLIENT,
 179                          CODE_ENVIRONMENT_UTILITY,
 180                          CINIT_FLAG_NO_DEFAULT_CONFIG_FILE);
 181   common_init_finish(g_ceph_context);
 182   g_ceph_context->_conf.apply_changes(nullptr);
 183
 184   if (args.size() < 6) {
 185     usage(argv[0]);
 186     return 1;
 187   }
 188
 189   int numjobs = atoi(args[1]);
 190   int concurrent = atoi(args[2]);
 191   int ios = atoi(args[3]);
 192   int think_time = atoi(args[4]);
 193   int len = atoi(args[5]);
 194
 195   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;
 196
 197   cerr << " using ms-public-type " << public_msgr_type << std::endl;
 198   cerr << "       server ip:port " << args[0] << std::endl;
 199   cerr << "       numjobs " << numjobs << std::endl;
 200   cerr << "       concurrency " << concurrent << std::endl;
 201   cerr << "       ios " << ios << std::endl;
 202   cerr << "       thinktime(us) " << think_time << std::endl;
 203   cerr << "       message data bytes " << len << std::endl;
 204
 205   MessengerClient client(public_msgr_type, args[0], think_time);
 206
 207   client.ready(concurrent, numjobs, ios, len);
 208   Cycles::init();
 209   uint64_t start = Cycles::rdtsc();
 210   client.start();
 211   uint64_t stop = Cycles::rdtsc();
 212   cerr << " Total op " << ios << " run time " << Cycles::to_microseconds(stop - start) << "us." << std::endl;
 213
 214   return 0;
 215 }