From 8f1beb1b86bda14c54c5a193cf29b6971ff7bf21 Mon Sep 17 00:00:00 2001
From: Yehuda Sadeh <yehuda@hq.newdream.net>
Date: Mon, 20 Jun 2011 17:09:27 -0700
Subject: [PATCH] rgw: put data using a window

---
 src/rgw/rgw_common.h |  3 ++-
 src/rgw/rgw_op.cc    | 39 +++++++++++++++++++++++++++++++++++----
 src/rgw/rgw_rados.cc |  1 +
 3 files changed, 38 insertions(+), 5 deletions(-)

diff --git a/src/rgw/rgw_common.h b/src/rgw/rgw_common.h
index f30d4908113..c498395509e 100644
--- a/src/rgw/rgw_common.h
+++ b/src/rgw/rgw_common.h
@@ -46,7 +46,8 @@ extern string rgw_root_bucket;
 
 #define USER_INFO_VER 7
 
-#define RGW_MAX_CHUNK_SIZE	(4*1024*1024)
+#define RGW_MAX_CHUNK_SIZE	(512*1024)
+#define RGW_MAX_PENDING_CHUNKS  16
 
 #define RGW_LOG_BEGIN "RADOS S3 Gateway:"
 #define RGW_LOG(x) pdout(x, g_conf->rgw_log)
diff --git a/src/rgw/rgw_op.cc b/src/rgw/rgw_op.cc
index e2ba0adf4c1..508fc528058 100644
--- a/src/rgw/rgw_op.cc
+++ b/src/rgw/rgw_op.cc
@@ -432,11 +432,32 @@ void RGWDeleteBucket::execute()
   send_response();
 }
 
+struct put_obj_aio_info {
+  void *data;
+  void *handle;
+};
+
+static int drain_pending(std::list<struct put_obj_aio_info>& pending)
+{
+  int ret = 0;
+  while (!pending.empty()) {
+   struct put_obj_aio_info info = pending.front();
+   int r = rgwstore->aio_wait(info.handle);
+   free(info.data);
+   if (r < 0)
+     ret = r;
+
+   pending.pop_front();
+  }
+  return ret;
+}
+
 void RGWPutObj::execute()
 {
   bool multipart;
   string multipart_meta_obj;
   string part_num;
+  list<struct put_obj_aio_info> pending;
 
   ret = -EINVAL;
   if (!s->object) {
@@ -503,6 +524,7 @@ void RGWPutObj::execute()
     do {
       get_data();
       if (len > 0) {
+        struct put_obj_aio_info info;
 	// For the first call to put_obj_data, pass -1 as the offset to
 	// do a write_full.
         void *handle;
@@ -513,13 +535,21 @@ void RGWPutObj::execute()
           goto done;
 
         hash.Update((unsigned char *)data, len);
-        ret = rgwstore->aio_wait(handle);
-        free(data);
-        if (ret < 0)
-          goto done;
+        info.handle = handle;
+        info.data = data;
+        pending.push_back(info);
+        if (pending.size() > RGW_MAX_PENDING_CHUNKS) {
+          info = pending.front();
+          pending.pop_front();
+          ret = rgwstore->aio_wait(info.handle);
+          free(info.data);
+          if (ret < 0)
+            goto done;
+        }
         ofs += len;
       }
     } while ( len > 0);
+    drain_pending(pending);
 
     s->obj_size = ofs;
 
@@ -573,6 +603,7 @@ void RGWPutObj::execute()
     }
   }
 done:
+  drain_pending(pending);
   send_response();
 }
 
diff --git a/src/rgw/rgw_rados.cc b/src/rgw/rgw_rados.cc
index eb928d892b7..1627b4baeef 100644
--- a/src/rgw/rgw_rados.cc
+++ b/src/rgw/rgw_rados.cc
@@ -371,6 +371,7 @@ int RGWRados::aio_put_obj_data(std::string& id, rgw_obj& obj,
   bl.append(data, len);
 
   AioCompletion *c = librados::Rados::aio_create_completion(NULL, NULL, NULL);
+  *handle = c;
   
 
   if (ofs == -1) {