ceph/src/rgw/rgw_loadgen_process.cc

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab ft=cpp
   3
   4 #include "common/errno.h"
   5 #include "common/Throttle.h"
   6 #include "common/WorkQueue.h"
   7
   8 #include "rgw_rest.h"
   9 #include "rgw_frontend.h"
  10 #include "rgw_request.h"
  11 #include "rgw_process.h"
  12 #include "rgw_loadgen.h"
  13 #include "rgw_client_io.h"
  14
  15 #include <atomic>
  16
  17 #define dout_subsys ceph_subsys_rgw
  18
  19 extern void signal_shutdown();
  20
  21 void RGWLoadGenProcess::checkpoint()
  22 {
  23   m_tp.drain(&req_wq);
  24 }
  25
  26 void RGWLoadGenProcess::run()
  27 {
  28   m_tp.start(); /* start thread pool */
  29
  30   int i;
  31
  32   int num_objs;
  33
  34   conf->get_val("num_objs", 1000, &num_objs);
  35
  36   int num_buckets;
  37   conf->get_val("num_buckets", 1, &num_buckets);
  38
  39   vector<string> buckets(num_buckets);
  40
  41   std::atomic<bool> failed = { false };
  42
  43   for (i = 0; i < num_buckets; i++) {
  44     buckets[i] = "/loadgen";
  45     string& bucket = buckets[i];
  46     append_rand_alpha(cct, bucket, bucket, 16);
  47
  48     /* first create a bucket */
  49     gen_request("PUT", bucket, 0, &failed);
  50     checkpoint();
  51   }
  52
  53   string *objs = new string[num_objs];
  54
  55   if (failed) {
  56     derr << "ERROR: bucket creation failed" << dendl;
  57     goto done;
  58   }
  59
  60   for (i = 0; i < num_objs; i++) {
  61     char buf[16 + 1];
  62     gen_rand_alphanumeric(cct, buf, sizeof(buf));
  63     buf[16] = '\0';
  64     objs[i] = buckets[i % num_buckets] + "/" + buf;
  65   }
  66
  67   for (i = 0; i < num_objs; i++) {
  68     gen_request("PUT", objs[i], 4096, &failed);
  69   }
  70
  71   checkpoint();
  72
  73   if (failed) {
  74     derr << "ERROR: bucket creation failed" << dendl;
  75     goto done;
  76   }
  77
  78   for (i = 0; i < num_objs; i++) {
  79     gen_request("GET", objs[i], 4096, NULL);
  80   }
  81
  82   checkpoint();
  83
  84   for (i = 0; i < num_objs; i++) {
  85     gen_request("DELETE", objs[i], 0, NULL);
  86   }
  87
  88   checkpoint();
  89
  90   for (i = 0; i < num_buckets; i++) {
  91     gen_request("DELETE", buckets[i], 0, NULL);
  92   }
  93
  94 done:
  95   checkpoint();
  96
  97   m_tp.stop();
  98
  99   delete[] objs;
 100
 101   signal_shutdown();
 102 } /* RGWLoadGenProcess::run() */
 103
 104 void RGWLoadGenProcess::gen_request(const string& method,
 105                                     const string& resource,
 106                                     int content_length, std::atomic<bool>* fail_flag)
 107 {
 108   RGWLoadGenRequest* req =
 109     new RGWLoadGenRequest(store->getRados()->get_new_req_id(), method, resource,
 110                           content_length, fail_flag);
 111   dout(10) << "allocated request req=" << hex << req << dec << dendl;
 112   req_throttle.get(1);
 113   req_wq.queue(req);
 114 } /* RGWLoadGenProcess::gen_request */
 115
 116 void RGWLoadGenProcess::handle_request(RGWRequest* r)
 117 {
 118   RGWLoadGenRequest* req = static_cast<RGWLoadGenRequest*>(r);
 119
 120   RGWLoadGenRequestEnv env;
 121
 122   utime_t tm = ceph_clock_now();
 123
 124   env.port = 80;
 125   env.content_length = req->content_length;
 126   env.content_type = "binary/octet-stream";
 127   env.request_method = req->method;
 128   env.uri = req->resource;
 129   env.set_date(tm);
 130   env.sign(access_key);
 131
 132   RGWLoadGenIO real_client_io(&env);
 133   RGWRestfulIO client_io(cct, &real_client_io);
 134
 135   int ret = process_request(store, rest, req, uri_prefix,
 136                             *auth_registry, &client_io, olog,
 137                             null_yield, nullptr, nullptr, nullptr);
 138   if (ret < 0) {
 139     /* we don't really care about return code */
 140     dout(20) << "process_request() returned " << ret << dendl;
 141
 142     if (req->fail_flag) {
 143       req->fail_flag++;
 144     }
 145   }
 146
 147   delete req;
 148 } /* RGWLoadGenProcess::handle_request */