rest_bench: some more implementation

Signed-off-by: Yehuda Sadeh <yehuda@hq.newdream.net>

diff --git a/src/tools/rest_bench.cc b/src/tools/rest_bench.cc
index 78532704e422b2cb3a205d3fcf6036483da937f8..f902d9e84508e8800e0135a26abd6820171267e8 100644 (file)
--- a/src/tools/rest_bench.cc
+++ b/src/tools/rest_bench.cc
@@ -13,6 +13,7 @@
  */
 
 #include "include/types.h"
+#include "include/atomic.h"
 
 #include "common/obj_bencher.h"
 #include "common/config.h"
@@ -40,10 +41,58 @@ static void usage_exit()
   exit(1);
 }
 
+struct req_context {
+  atomic_t complete;
+  S3Status status;
+  S3RequestContext *ctx;
+  void (*cb)(void *, void *);
+  void *arg;
+
+  req_context() : status(S3StatusOK), ctx(NULL), cb(NULL), arg(NULL) {}
+  ~req_context() {
+    if (ctx) {
+      S3_destroy_request_context(ctx);
+    }
+  }
+
+  int init_ctx() {
+    S3Status status = S3_create_request_context(&ctx);
+    if (status != S3StatusOK) {
+      cerr << "failed to create context: " << S3_get_status_name(status) << std::endl;
+      return -EINVAL;
+    }
+
+    return 0;
+  }
+};
+
+static S3Status properties_callback(const S3ResponseProperties *properties, void *cb_data)
+{
+  return S3StatusOK;
+}
+
+static void complete_callback(S3Status status, const S3ErrorDetails *details, void *cb_data)
+{
+  if (!cb_data)
+    return;
+
+  struct req_context *ctx = (struct req_context *)cb_data;
+  ctx->complete.set(1);
+  ctx->status = status;
+
+  if (ctx->cb) {
+    ctx->cb((void *)ctx->cb, ctx->arg);
+  }
+}
+
+static S3ResponseHandler response_cb;
+
 class RESTBencher : public ObjBencher {
-  void **completions;
+  struct req_context **completions;
+  string bucket;
 protected:
   int completions_init(int concurrentios) {
+    completions = new req_context *[concurrentios];
     return 0;
   }
   void completions_done() {
@@ -51,40 +100,81 @@ protected:
     completions = NULL;
   }
   int create_completion(int slot, void (*cb)(void *, void*), void *arg) {
-    if (!completions[slot])
-      return -EINVAL;
+    struct req_context *ctx = new req_context;
+    int ret = ctx->init_ctx();
+    if (ret < 0) {
+      return ret;
+    }
+
+    ctx->cb = cb;
+    ctx->arg = arg;
+
+    completions[slot] = ctx;
 
     return 0;
   }
   void release_completion(int slot) {
+    struct req_context *ctx = completions[slot];
 
+    delete ctx;
     completions[slot] = 0;
   }
 
   int aio_read(const std::string& oid, int slot, bufferlist *pbl, size_t len) {
+    S3_get_object(bucket_ctx, oid.c_str, NULL, 0, len, completions[slot]->ctx, 
+                  &response_cb, completions[slot]);
+    return 0;
   }
 
   int aio_write(const std::string& oid, int slot, const bufferlist& bl, size_t len) {
+    /* void S3_put_object(const S3BucketContext *bucketContext, const char *key,
+                   uint64_t contentLength,
+                   const S3PutProperties *putProperties,
+                   S3RequestContext *requestContext,
+                   const S3PutObjectHandler *handler, void *callbackData); */
   }
 
   int sync_read(const std::string& oid, bufferlist& bl, size_t len) {
+    S3_get_object(&bucket_ctx, oid.c_str, NULL, 0, len, completions[slot]->ctx, 
+                  &response_cb, completions[slot]);
   }
   int sync_write(const std::string& oid, bufferlist& bl, size_t len) {
   }
 
   bool completion_is_done(int slot) {
+    int val = completions[slot]->complete.read();
+    return (val != 0);
   }
 
   int completion_wait(int slot) {
   }
+
   int completion_ret(int slot) {
+    S3Status status = completions[slot]->status;
+    if (status >= 200 && status < 300)
+      return 0;
+    return -EIO;
   }
 
 public:
-  RESTBencher() : completions(NULL) {}
+  RESTBencher(string _bucket) : completions(NULL), bucket(_bucket) {}
   ~RESTBencher() { }
 };
 
+static int rest_init(string& user_agent, string& host)
+{
+  S3Status status = S3_initialize(user_agent.c_str(), S3_INIT_ALL, host.c_str());
+  if (status != S3StatusOK) {
+    cerr << "failed to init: " << S3_get_status_name(status) << std::endl;
+    return -EINVAL;
+  }
+
+  response_cb.propertiesCallback = properties_callback;
+  response_cb.completeCallback = complete_callback;
+
+  return 0;
+}
+
 int main(int argc, const char **argv)
 {
   vector<const char*> args;
@@ -101,7 +191,10 @@ int main(int argc, const char **argv)
   std::string user_agent;
   std::string access_key;
   std::string secret;
-  std::string bucket;
+  std::string bucket = DEFAULT_BUCKET;
+  S3Protocol protocol = S3ProtocolHTTP;
+  std::string proto_str;
+
   for (i = args.begin(); i != args.end(); ) {
     if (ceph_argparse_double_dash(args, i)) {
       break;
@@ -110,11 +203,21 @@ int main(int argc, const char **argv)
       exit(0);
     } else if (ceph_argparse_witharg(args, i, &user_agent, "--agent", (char*)NULL)) {
       /* nothing */
-    } else if (ceph_argparse_witharg(args, i, &user_agent, "--access-key", (char*)NULL)) {
+    } else if (ceph_argparse_witharg(args, i, &access_key, "--access-key", (char*)NULL)) {
+      /* nothing */
+    } else if (ceph_argparse_witharg(args, i, &secret, "--secret", (char*)NULL)) {
       /* nothing */
-    } else if (ceph_argparse_witharg(args, i, &user_agent, "--secret", (char*)NULL)) {
+    } else if (ceph_argparse_witharg(args, i, &bucket, "--bucket", (char*)NULL)) {
       /* nothing */
-    } else if (ceph_argparse_witharg(args, i, &user_agent, "--bucket", (char*)NULL)) {
+    } else if (ceph_argparse_witharg(args, i, &proto_str, "--protocol", (char*)NULL)) {
+      if (strcasecmp(proto_str.c_str(), "http") == 0) {
+        protocol = S3ProtocolHTTP;
+      } else if (strcasecmp(proto_str.c_str(), "http") == 0) {
+        protocol = S3ProtocolHTTPS;
+      } else {
+        cerr << "bad protocol" << std::endl;
+        usage_exit();
+      }
       /* nothing */
     } else {
       if (val[0] == '-')
@@ -142,11 +245,25 @@ int main(int argc, const char **argv)
   if (user_agent.empty())
     user_agent = DEFAULT_USER_AGENT;
 
-  S3Status status = S3_initialize(user_agent.c_str(), S3_INIT_ALL, host.c_str());
-  if (status != S3StatusOK) {
-    cerr << "failed to init: " << S3_get_status_name(status) << std::endl;
+  int ret = rest_init(user_agent, host);
+  if (ret < 0) {
+    exit(1);
+  }
+
+  RESTBencher bencher(bucket);
+
+  struct req_context ctx;
+
+  ret = ctx.init_ctx();
+  if (ret < 0) {
+    exit(1);
   }
 
-  RESTBencher bencher();
+  S3_create_bucket(protocol, access_key.c_str(), secret.c_str(), NULL,
+                   bucket.c_str(), S3CannedAclPrivate,
+                   NULL, /* locationConstraint */
+                   NULL, /* requestContext */
+                   &response_cb, /* handler */
+                   (void *)&ctx  /* callbackData */);
 }