Merge branch 10.5 into 10.6

4 files changed, 60 insertions, 27 deletions

diff --git a/tpool/aio_linux.cc b/tpool/aio_linux.cc
index 4abc2139881..fc6e5b53e1a 100644
--- a/tpool/aio_linux.cc
+++ b/tpool/aio_linux.cc
@@ -128,6 +128,8 @@ class aio_linux final : public aio
           {
             iocb->m_ret_len= event.res;
             iocb->m_err= 0;
+            if (iocb->m_ret_len != iocb->m_len)
+              finish_synchronous(iocb);
           }
           iocb->m_internal_task.m_func= iocb->m_callback;
           iocb->m_internal_task.m_arg= iocb;
diff --git a/tpool/aio_simulated.cc b/tpool/aio_simulated.cc
index 6b6fe71c8ab..4bc58c2930c 100644
--- a/tpool/aio_simulated.cc
+++ b/tpool/aio_simulated.cc
@@ -136,32 +136,7 @@ public:
   static void simulated_aio_callback(void *param)
   {
     aiocb *cb= (aiocb *) param;
-#ifdef _WIN32
-    size_t ret_len;
-#else
-    ssize_t ret_len;
-#endif
-    int err= 0;
-    switch (cb->m_opcode)
-    {
-    case aio_opcode::AIO_PREAD:
-      ret_len= pread(cb->m_fh, cb->m_buffer, cb->m_len, cb->m_offset);
-      break;
-    case aio_opcode::AIO_PWRITE:
-      ret_len= pwrite(cb->m_fh, cb->m_buffer, cb->m_len, cb->m_offset);
-      break;
-    default:
-      abort();
-    }
-#ifdef _WIN32
-    if (static_cast<int>(ret_len) < 0)
-      err= GetLastError();
-#else
-    if (ret_len < 0)
-      err= errno;
-#endif
-    cb->m_ret_len = ret_len;
-    cb->m_err = err;
+    synchronous(cb);
     cb->m_internal_task.m_func= cb->m_callback;
     thread_pool *pool= (thread_pool *)cb->m_internal;
     pool->submit_task(&cb->m_internal_task);
diff --git a/tpool/tpool.h b/tpool/tpool.h
index f857dddd57a..2c61c2d62b2 100644
--- a/tpool/tpool.h
+++ b/tpool/tpool.h
@@ -161,7 +161,7 @@ class aio
 {
 public:
   /**
-    Submit asyncronous IO.
+    Submit asynchronous IO.
     On completion, cb->m_callback is executed.
   */
   virtual int submit_io(aiocb *cb)= 0;
@@ -170,6 +170,10 @@ public:
   /** "Unind" file to AIO handler (used on Windows only) */
   virtual int unbind(const native_file_handle &fd)= 0;
   virtual ~aio(){};
+protected:
+  static void synchronous(aiocb *cb);
+  /** finish a partial read/write callback synchronously */
+  static void finish_synchronous(aiocb *cb);
 };
 
 class timer
diff --git a/tpool/tpool_generic.cc b/tpool/tpool_generic.cc
index ecc489c3357..a1b9a3ce945 100644
--- a/tpool/tpool_generic.cc
+++ b/tpool/tpool_generic.cc
@@ -52,6 +52,58 @@ namespace tpool
   static const int  OVERSUBSCRIBE_FACTOR = 2;
 
 /**
+  Process the cb synchronously
+*/
+void aio::synchronous(aiocb *cb)
+{
+#ifdef _WIN32
+  size_t ret_len;
+#else
+  ssize_t ret_len;
+#endif
+  int err= 0;
+  switch (cb->m_opcode)
+  {
+  case aio_opcode::AIO_PREAD:
+    ret_len= pread(cb->m_fh, cb->m_buffer, cb->m_len, cb->m_offset);
+    break;
+  case aio_opcode::AIO_PWRITE:
+    ret_len= pwrite(cb->m_fh, cb->m_buffer, cb->m_len, cb->m_offset);
+    break;
+  default:
+    abort();
+  }
+#ifdef _WIN32
+  if (static_cast<int>(ret_len) < 0)
+    err= GetLastError();
+#else
+  if (ret_len < 0)
+  {
+    err= errno;
+    ret_len= 0;
+  }
+#endif
+  cb->m_ret_len = ret_len;
+  cb->m_err = err;
+  if (!err && cb->m_ret_len != cb->m_len)
+    finish_synchronous(cb);
+}
+
+
+/**
+  A partial read/write has occured, continue synchronously.
+*/
+void aio::finish_synchronous(aiocb *cb)
+{
+  assert(cb->m_ret_len != (unsigned int) cb->m_len && !cb->m_err);
+  /* partial read/write */
+  cb->m_buffer= (char *) cb->m_buffer + cb->m_ret_len;
+  cb->m_len-= (unsigned int) cb->m_ret_len;
+  cb->m_offset+= cb->m_ret_len;
+  synchronous(cb);
+}
+
+/**
   Implementation of generic threadpool.
   This threadpool consists of the following components