os/FileJournal: limit size of aio submission

Limit size of each aio submission to IOV_MAX-1 (to be safe).  Take care to
only mark the last aio with the seq to signal completion.

Signed-off-by: Sage Weil <sage@inktank.com>

1 files changed, 40 insertions, 29 deletions

diff --git a/src/os/FileJournal.cc b/src/os/FileJournal.cc
index ee8f7029eb5..20324f37a3d 100644
--- a/src/os/FileJournal.cc
+++ b/src/os/FileJournal.cc
@@ -1251,40 +1251,51 @@ int FileJournal::write_aio_bl(off64_t& pos, bufferlist& bl, uint64_t seq)
 
   dout(20) << "write_aio_bl " << pos << "~" << bl.length() << " seq " << seq << dendl;
   
-  aio_queue.push_back(aio_info(bl, pos, seq));
-  aio_info& aio = aio_queue.back();
+  while (bl.length() > 0) {
+    int max = MIN(bl.buffers().size(), IOV_MAX-1);
+    iovec *iov = new iovec[max];
+    int n = 0;
+    unsigned len = 0;
+    for (std::list<buffer::ptr>::const_iterator p = bl.buffers().begin();
+	 n < max;
+	 ++p, ++n) {
+      assert(p != bl.buffers().end());
+      iov[n].iov_base = (void *)p->c_str();
+      iov[n].iov_len = p->length();
+      len += p->length();
+    }
 
-  aio.iov = new iovec[aio.bl.buffers().size()];
-  int n = 0;
-  for (std::list<buffer::ptr>::const_iterator p = aio.bl.buffers().begin(); 
-       p != aio.bl.buffers().end();
-       ++p, ++n) {
-    aio.iov[n].iov_base = (void *)p->c_str();
-    aio.iov[n].iov_len = p->length();
-  }
-  io_prep_pwritev(&aio.iocb, fd, aio.iov, n, pos);
+    bufferlist tbl;
+    bl.splice(0, len, &tbl);  // move bytes from bl -> tbl
 
-  dout(20) << "write_aio_bl .. " << aio.off << "~" << aio.len
-	   << " in " << n << dendl;
+    aio_queue.push_back(aio_info(tbl, pos, bl.length() > 0 ? 0 : seq));
+    aio_info& aio = aio_queue.back();
+    aio.iov = iov;
 
-  aio_num++;
-  aio_bytes += aio.len;
+    io_prep_pwritev(&aio.iocb, fd, aio.iov, n, pos);
 
-  iocb *piocb = &aio.iocb;
-  int attempts = 10;
-  do {
-    int r = io_submit(aio_ctx, 1, &piocb);
-    if (r < 0) {
-      derr << "io_submit to " << aio.off << "~" << aio.len
-	   << " got " << cpp_strerror(r) << dendl;
-      if (r == -EAGAIN && attempts-- > 0) {
-	usleep(500);
-	continue;
+    dout(20) << "write_aio_bl .. " << aio.off << "~" << aio.len
+	     << " in " << n << dendl;
+
+    aio_num++;
+    aio_bytes += aio.len;
+
+    iocb *piocb = &aio.iocb;
+    int attempts = 10;
+    do {
+      int r = io_submit(aio_ctx, 1, &piocb);
+      if (r < 0) {
+	derr << "io_submit to " << aio.off << "~" << aio.len
+	     << " got " << cpp_strerror(r) << dendl;
+	if (r == -EAGAIN && attempts-- > 0) {
+	  usleep(500);
+	  continue;
+	}
+	assert(0 == "io_submit got unexpected error");
       }
-      assert(0 == "io_submit got unexpected error");
-    }
-  } while (false);
-  pos += aio.len;
+    } while (false);
+    pos += aio.len;
+  }
   write_finish_cond.Signal();
   return 0;
 }