src/ceph/src/test/messenger/simple_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) 2013 CohortFS, LLC
   7  *
   8  * This is free software; you can redistribute it and/or
   9  * modify it under the terms of the GNU Lesser General Public
  10  * License version 2.1, as published by the Free Software
  11  * Foundation.  See file COPYING.
  12  *
  13  */
  14
  15 #include <sys/types.h>
  16
  17 #include <iostream>
  18 #include <string>
  19
  20 using namespace std;
  21
  22 #include "common/config.h"
  23 #include "msg/msg_types.h"
  24 #include "msg/Messenger.h"
  25 #include "messages/MPing.h"
  26 #include "common/Timer.h"
  27 #include "common/ceph_argparse.h"
  28 #include "global/global_init.h"
  29 #include "perfglue/heap_profiler.h"
  30 #include "common/address_helper.h"
  31 #include "message_helper.h"
  32 #include "simple_dispatcher.h"
  33
  34 #define dout_subsys ceph_subsys_simple_client
  35
  36 void usage(ostream& out)
  37 {
  38   out << "usage: simple_client [options]\n"
  39 "options:\n"
  40 "  --addr X\n"
  41 "  --port X\n"
  42 "  --msgs X\n"
  43 "  --dsize X\n"
  44     ;
  45 }
  46
  47
  48 int main(int argc, const char **argv)
  49 {
  50         vector<const char*> args;
  51         Messenger* messenger;
  52         SimpleDispatcher *dispatcher;
  53         std::vector<const char*>::iterator arg_iter;
  54         std::string val;
  55         entity_addr_t dest_addr;
  56         ConnectionRef conn;
  57         int r = 0;
  58
  59         std::string addr = "localhost";
  60         std::string port = "1234";
  61
  62         int n_msgs = 50;
  63         int n_dsize = 0;
  64
  65         struct timespec ts;
  66         ts.tv_sec = 1;
  67         ts.tv_nsec = 0;
  68
  69         argv_to_vec(argc, argv, args);
  70         env_to_vec(args);
  71
  72         auto cct = global_init(NULL, args, CEPH_ENTITY_TYPE_ANY,
  73                                CODE_ENVIRONMENT_UTILITY,
  74                                0);
  75
  76         for (arg_iter = args.begin(); arg_iter != args.end();) {
  77           if (ceph_argparse_witharg(args, arg_iter, &val, "--addr",
  78                                     (char*) NULL)) {
  79             addr = val;
  80           } else if (ceph_argparse_witharg(args, arg_iter, &val, "--port",
  81                                     (char*) NULL)) {
  82             port = val;
  83           } else if (ceph_argparse_witharg(args, arg_iter, &val, "--msgs",
  84                                     (char*) NULL)) {
  85             n_msgs = atoi(val.c_str());;
  86           } else if (ceph_argparse_witharg(args, arg_iter, &val, "--dsize",
  87                                     (char*) NULL)) {
  88             n_dsize = atoi(val.c_str());;
  89           } else {
  90             ++arg_iter;
  91           }
  92         };
  93
  94         if (!args.empty()) {
  95           cerr << "What is this? -- " << args[0] << std::endl;
  96           usage(cerr);
  97           exit(1);
  98         }
  99
 100         cout  << "simple_client starting " <<
 101           "dest addr " << addr << " " <<
 102           "dest port " << port << " " <<
 103           "initial msgs (pipe depth) " << n_msgs << " " <<
 104           "data buffer size " << n_dsize << std::endl;
 105
 106         messenger = Messenger::create(g_ceph_context, g_conf->get_val<std::string>("ms_type"),
 107                                       entity_name_t::MON(-1),
 108                                       "client",
 109                                       getpid(), 0);
 110
 111         // enable timing prints
 112         messenger->set_magic(MSG_MAGIC_TRACE_CTR);
 113         messenger->set_default_policy(Messenger::Policy::lossy_client(0));
 114
 115         string dest_str = "tcp://";
 116         dest_str += addr;
 117         dest_str += ":";
 118         dest_str += port;
 119         entity_addr_from_url(&dest_addr, dest_str.c_str());
 120         entity_inst_t dest_server(entity_name_t::MON(-1), dest_addr);
 121
 122         dispatcher = new SimpleDispatcher(messenger);
 123         messenger->add_dispatcher_head(dispatcher);
 124
 125         dispatcher->set_active(); // this side is the pinger
 126
 127         r = messenger->start();
 128         if (r < 0)
 129                 goto out;
 130
 131         conn = messenger->get_connection(dest_server);
 132
 133         // do stuff
 134         time_t t1, t2;
 135
 136         t1 = time(NULL);
 137
 138         int msg_ix;
 139         Message *m;
 140         for (msg_ix = 0; msg_ix < n_msgs; ++msg_ix) {
 141           /* add a data payload if asked */
 142           if (! n_dsize) {
 143             m = new MPing();
 144           } else {
 145             m = new_simple_ping_with_data("simple_client", n_dsize);
 146           }
 147           conn->send_message(m);
 148         }
 149
 150         // do stuff
 151         while (conn->is_connected()) {
 152           nanosleep(&ts, NULL);
 153         }
 154
 155         t2 = time(NULL);
 156         cout << "Processed " << dispatcher->get_dcount() + n_msgs
 157              << " round-trip messages in " << t2-t1 << "s"
 158              << std::endl;
 159 out:
 160         return r;
 161 }