rgw: loadgen shutdown, error out on failures

Signed-off-by: Yehuda Sadeh <yehuda@inktank.com>

diff --git a/src/rgw/rgw_main.cc b/src/rgw/rgw_main.cc
index 5df04d5abe35ae3325f56afe662824377e22bfe6..1a890f5e38ce46f4b76257af6332c8d0cdbad832 100644 (file)
--- a/src/rgw/rgw_main.cc
+++ b/src/rgw/rgw_main.cc
@@ -77,6 +77,8 @@ class RGWProcess;
 static int signal_fd[2] = {0, 0};
 static atomic_t disable_signal_fd;
 
+static void signal_shutdown();
+
 
 #define SOCKET_BACKLOG 1024
 
@@ -168,6 +170,7 @@ protected:
   Throttle req_throttle;
   RGWREST *rest;
   RGWFrontendConfig *conf;
+  int sock_fd;
 
   struct RGWWQ : public ThreadPool::WorkQueue<RGWRequest> {
     RGWProcess *process;
@@ -227,34 +230,33 @@ public:
       req_throttle(cct, "rgw_ops", num_threads * 2),
       rest(pe->rest),
       conf(_conf),
+      sock_fd(-1),
       req_wq(this, g_conf->rgw_op_thread_timeout,
             g_conf->rgw_op_thread_suicide_timeout, &m_tp),
       max_req_id(0) {}
   virtual ~RGWProcess() {}
   virtual void run() = 0;
   virtual void handle_request(RGWRequest *req) = 0;
-  virtual void close_fd() {}
+
+  void close_fd() {
+    if (sock_fd >= 0) {
+      ::close(sock_fd);
+      sock_fd = -1;
+    }
+  }
 };
 
 
 class RGWFCGXProcess : public RGWProcess {
-  int sock_fd;
 public:
   RGWFCGXProcess(CephContext *cct, RGWProcessEnv *pe, int num_threads, RGWFrontendConfig *_conf) :
-    RGWProcess(cct, pe, num_threads, _conf), sock_fd(-1) {}
+    RGWProcess(cct, pe, num_threads, _conf) {}
   void run();
   void handle_request(RGWRequest *req);
-
-  void close_fd() {
-    if (sock_fd >= 0)
-      close(sock_fd);
-  }
 };
 
 void RGWFCGXProcess::run()
 {
-  sock_fd = 0;
-
   string socket_path;
   string socket_port;
   string socket_host;
@@ -325,9 +327,11 @@ struct RGWLoadGenRequest : public RGWRequest {
   string method;
   string resource;
   int content_length;
+  atomic_t *fail_flag;
 
 
-  RGWLoadGenRequest(const string& _m, const  string& _r, int _cl) : method(_m), resource(_r), content_length(_cl) {}
+  RGWLoadGenRequest(const string& _m, const  string& _r, int _cl,
+                    atomic_t *ff) : method(_m), resource(_r), content_length(_cl), fail_flag(ff) {}
 };
 
 class RGWLoadGenProcess : public RGWProcess {
@@ -338,9 +342,7 @@ public:
   void run();
   void checkpoint();
   void handle_request(RGWRequest *req);
-  void gen_request(const string& method, const string& resource, int content_length);
-
-  void close_fd() { }
+  void gen_request(const string& method, const string& resource, int content_length, atomic_t *fail_flag);
 
   void set_access_key(RGWAccessKey& key) { access_key = key; }
 };
@@ -362,12 +364,18 @@ void RGWLoadGenProcess::run()
 
   string bucket_resource = string("/") + bucket;
 
+  atomic_t failed;
+  string *objs = new string[num_objs];
+
   /* first create a bucket */
-  gen_request("PUT", bucket_resource, 0);
+  gen_request("PUT", bucket_resource, 0, &failed);
 
   checkpoint();
 
-  string *objs = new string[num_objs];
+  if (failed.read()) {
+    derr << "ERROR: bucket creation failed" << dendl;
+    goto done;
+  }
 
   for (i = 0; i < num_objs; i++) {
     char buf[32];
@@ -376,26 +384,41 @@ void RGWLoadGenProcess::run()
   }
 
   for (i = 0; i < num_objs; i++) {
-    gen_request("PUT", objs[i], 4096);
+    gen_request("PUT", objs[i], 4096, &failed);
+  }
+
+  checkpoint();
+
+  if (failed.read()) {
+    derr << "ERROR: bucket creation failed" << dendl;
+    goto done;
   }
-#if 0
-  for (i = 0; i < 10; i++) {
-    gen_request("GET", 4096);
+
+  for (i = 0; i < num_objs; i++) {
+    gen_request("GET", objs[i], 4096, NULL);
   }
-  for (i = 0; i < 10; i++) {
-    gen_request("DELETE", 4096);
+
+  checkpoint();
+
+  for (i = 0; i < num_objs; i++) {
+    gen_request("DELETE", objs[i], 0, NULL);
   }
-#endif
-  m_tp.drain();
+
+  gen_request("DELETE", bucket_resource, 0, NULL);
+
+  checkpoint();
 
   m_tp.stop();
 
+done:
   delete[] objs;
+
+  signal_shutdown();
 }
 
-void RGWLoadGenProcess::gen_request(const string& method, const string& resource, int content_length)
+void RGWLoadGenProcess::gen_request(const string& method, const string& resource, int content_length, atomic_t *fail_flag)
 {
-  RGWLoadGenRequest *req = new RGWLoadGenRequest(method, resource, content_length);
+  RGWLoadGenRequest *req = new RGWLoadGenRequest(method, resource, content_length, fail_flag);
   req->id = ++max_req_id;
   dout(10) << "allocated request req=" << hex << req << dec << dendl;
   req_throttle.get(1);
@@ -585,7 +608,7 @@ done:
 
   dout(1) << "====== req done req=" << hex << req << dec << " http_status=" << http_ret << " ======" << dendl;
 
-  return ret;
+  return (ret < 0 ? ret : s->err.ret);
 }
 
 void RGWFCGXProcess::handle_request(RGWRequest *r)
@@ -628,6 +651,10 @@ void RGWLoadGenProcess::handle_request(RGWRequest *r)
   if (ret < 0) {
     /* we don't really care about return code */
     dout(20) << "process_request() returned " << ret << dendl;
+
+    if (req->fail_flag) {
+      req->fail_flag->inc();
+    }
   }
 
   delete req;

diff --git a/src/rgw/rgw_rest.cc b/src/rgw/rgw_rest.cc
index a18f28ecde457160f891eb91ef856a2c0d8c2620..3402a28ce2584a23862e0719a7a85fafb82c44d0 100644 (file)
--- a/src/rgw/rgw_rest.cc
+++ b/src/rgw/rgw_rest.cc
@@ -199,7 +199,7 @@ void set_req_state_err(struct req_state *s, int err_no)
 
   if (err_no < 0)
     err_no = -err_no;
-  s->err.ret = err_no;
+  s->err.ret = -err_no;
   if (s->prot_flags & RGW_REST_SWIFT) {
     r = search_err(err_no, RGW_HTTP_SWIFT_ERRORS, ARRAY_LEN(RGW_HTTP_SWIFT_ERRORS));
     if (r) {