Add single-buffer optimisation for send and receive.

When we can determine at compile time that the user has supplied a
single buffer, use send/recv rather than sendmsg/recvmsg, as the former
system calls can be faster.

Author: chriskohlhoff

include/boost/asio/detail/buffer_sequence_adapter.hpp

@@ -105,6 +105,8 @@ class buffer_sequence_adapter
   : buffer_sequence_adapter_base
 {
 public:
+  enum { is_single_buffer = false };
+
   explicit buffer_sequence_adapter(const Buffers& buffer_sequence)
     : count_(0), total_buffer_size_(0)
   {
@@ -246,6 +248,8 @@ class buffer_sequence_adapter<Buffer, boost::asio::mutable_buffer>
   : buffer_sequence_adapter_base
 {
 public:
+  enum { is_single_buffer = true };
+
   explicit buffer_sequence_adapter(
       const boost::asio::mutable_buffer& buffer_sequence)
   {
@@ -306,6 +310,8 @@ class buffer_sequence_adapter<Buffer, boost::asio::const_buffer>
   : buffer_sequence_adapter_base
 {
 public:
+  enum { is_single_buffer = true };
+
   explicit buffer_sequence_adapter(
       const boost::asio::const_buffer& buffer_sequence)
   {
@@ -368,6 +374,8 @@ class buffer_sequence_adapter<Buffer, boost::asio::mutable_buffers_1>
   : buffer_sequence_adapter_base
 {
 public:
+  enum { is_single_buffer = true };
+
   explicit buffer_sequence_adapter(
       const boost::asio::mutable_buffers_1& buffer_sequence)
   {
@@ -428,6 +436,8 @@ class buffer_sequence_adapter<Buffer, boost::asio::const_buffers_1>
   : buffer_sequence_adapter_base
 {
 public:
+  enum { is_single_buffer = true };
+
   explicit buffer_sequence_adapter(
       const boost::asio::const_buffers_1& buffer_sequence)
   {
@@ -490,6 +500,8 @@ class buffer_sequence_adapter<Buffer, boost::array<Elem, 2> >
   : buffer_sequence_adapter_base
 {
 public:
+  enum { is_single_buffer = false };
+
   explicit buffer_sequence_adapter(
       const boost::array<Elem, 2>& buffer_sequence)
   {
@@ -560,6 +572,8 @@ class buffer_sequence_adapter<Buffer, std::array<Elem, 2> >
   : buffer_sequence_adapter_base
 {
 public:
+  enum { is_single_buffer = false };
+
   explicit buffer_sequence_adapter(
       const std::array<Elem, 2>& buffer_sequence)
   {

include/boost/asio/detail/impl/socket_ops.ipp

@@ -788,6 +788,38 @@ signed_size_type recv(socket_type s, buf* bufs, size_t count,
 #endif // defined(BOOST_ASIO_WINDOWS) || defined(__CYGWIN__)
 }
 
+signed_size_type recv1(socket_type s, void* data, size_t size,
+    int flags, boost::system::error_code& ec)
+{
+  clear_last_error();
+#if defined(BOOST_ASIO_WINDOWS) || defined(__CYGWIN__)
+  // Receive some data.
+  WSABUF buf;
+  buf.buf = const_cast<char*>(static_cast<const char*>(data));
+  buf.len = static_cast<ULONG>(size);
+  DWORD bytes_transferred = 0;
+  DWORD recv_flags = flags;
+  int result = error_wrapper(::WSARecv(s, &buf, 1,
+        &bytes_transferred, &recv_flags, 0, 0), ec);
+  if (ec.value() == ERROR_NETNAME_DELETED)
+    ec = boost::asio::error::connection_reset;
+  else if (ec.value() == ERROR_PORT_UNREACHABLE)
+    ec = boost::asio::error::connection_refused;
+  else if (ec.value() == WSAEMSGSIZE || ec.value() == ERROR_MORE_DATA)
+    result = 0;
+  if (result != 0)
+    return socket_error_retval;
+  ec = boost::system::error_code();
+  return bytes_transferred;
+#else // defined(BOOST_ASIO_WINDOWS) || defined(__CYGWIN__)
+  signed_size_type result = error_wrapper(::recv(s,
+        static_cast<char*>(data), size, flags), ec);
+  if (result >= 0)
+    ec = boost::system::error_code();
+  return result;
+#endif // defined(BOOST_ASIO_WINDOWS) || defined(__CYGWIN__)
+}
+
 size_t sync_recv(socket_type s, state_type state, buf* bufs,
     size_t count, int flags, bool all_empty, boost::system::error_code& ec)
 {
@@ -833,6 +865,51 @@ size_t sync_recv(socket_type s, state_type state, buf* bufs,
   }
 }
 
+size_t sync_recv1(socket_type s, state_type state, void* data,
+    size_t size, int flags, boost::system::error_code& ec)
+{
+  if (s == invalid_socket)
+  {
+    ec = boost::asio::error::bad_descriptor;
+    return 0;
+  }
+
+  // A request to read 0 bytes on a stream is a no-op.
+  if (size == 0 && (state & stream_oriented))
+  {
+    ec = boost::system::error_code();
+    return 0;
+  }
+
+  // Read some data.
+  for (;;)
+  {
+    // Try to complete the operation without blocking.
+    signed_size_type bytes = socket_ops::recv1(s, data, size, flags, ec);
+
+    // Check if operation succeeded.
+    if (bytes > 0)
+      return bytes;
+
+    // Check for EOF.
+    if ((state & stream_oriented) && bytes == 0)
+    {
+      ec = boost::asio::error::eof;
+      return 0;
+    }
+
+    // Operation failed.
+    if ((state & user_set_non_blocking)
+        || (ec != boost::asio::error::would_block
+          && ec != boost::asio::error::try_again))
+      return 0;
+
+    // Wait for socket to become ready.
+    if (socket_ops::poll_read(s, 0, -1, ec) < 0)
+      return 0;
+  }
+}
+
 #if defined(BOOST_ASIO_HAS_IOCP)
 
 void complete_iocp_recv(state_type state,
@@ -905,6 +982,44 @@ bool non_blocking_recv(socket_type s,
   }
 }
 
+bool non_blocking_recv1(socket_type s,
+    void* data, size_t size, int flags, bool is_stream,
+    boost::system::error_code& ec, size_t& bytes_transferred)
+{
+  for (;;)
+  {
+    // Read some data.
+    signed_size_type bytes = socket_ops::recv1(s, data, size, flags, ec);
+
+    // Check for end of stream.
+    if (is_stream && bytes == 0)
+    {
+      ec = boost::asio::error::eof;
+      return true;
+    }
+
+    // Retry operation if interrupted by signal.
+    if (ec == boost::asio::error::interrupted)
+      continue;
+
+    // Check if we need to run the operation again.
+    if (ec == boost::asio::error::would_block
+        || ec == boost::asio::error::try_again)
+      return false;
+
+    // Operation is complete.
+    if (bytes >= 0)
+    {
+      ec = boost::system::error_code();
+      bytes_transferred = bytes;
+    }
+    else
+      bytes_transferred = 0;
+
+    return true;
+  }
+}
+
 #endif // defined(BOOST_ASIO_HAS_IOCP)
 
 signed_size_type recvfrom(socket_type s, buf* bufs, size_t count,
@@ -1188,6 +1303,39 @@ signed_size_type send(socket_type s, const buf* bufs, size_t count,
 #endif // defined(BOOST_ASIO_WINDOWS) || defined(__CYGWIN__)
 }
 
+signed_size_type send1(socket_type s, const void* data, size_t size,
+    int flags, boost::system::error_code& ec)
+{
+  clear_last_error();
+#if defined(BOOST_ASIO_WINDOWS) || defined(__CYGWIN__)
+  // Send the data.
+  WSABUF buf;
+  buf.buf = const_cast<char*>(static_cast<const char*>(data));
+  buf.len = static_cast<ULONG>(size);
+  DWORD bytes_transferred = 0;
+  DWORD send_flags = flags;
+  int result = error_wrapper(::WSASend(s, &buf, 1,
+        &bytes_transferred, send_flags, 0, 0), ec);
+  if (ec.value() == ERROR_NETNAME_DELETED)
+    ec = boost::asio::error::connection_reset;
+  else if (ec.value() == ERROR_PORT_UNREACHABLE)
+    ec = boost::asio::error::connection_refused;
+  if (result != 0)
+    return socket_error_retval;
+  ec = boost::system::error_code();
+  return bytes_transferred;
+#else // defined(BOOST_ASIO_WINDOWS) || defined(__CYGWIN__)
+#if defined(__linux__)
+  flags |= MSG_NOSIGNAL;
+#endif // defined(__linux__)
+  signed_size_type result = error_wrapper(::send(s,
+        static_cast<const char*>(data), size, flags), ec);
+  if (result >= 0)
+    ec = boost::system::error_code();
+  return result;
+#endif // defined(BOOST_ASIO_WINDOWS) || defined(__CYGWIN__)
+}
+
 size_t sync_send(socket_type s, state_type state, const buf* bufs,
     size_t count, int flags, bool all_empty, boost::system::error_code& ec)
 {
@@ -1226,6 +1374,44 @@ size_t sync_send(socket_type s, state_type state, const buf* bufs,
   }
 }
 
+size_t sync_send1(socket_type s, state_type state, const void* data,
+    size_t size, int flags, boost::system::error_code& ec)
+{
+  if (s == invalid_socket)
+  {
+    ec = boost::asio::error::bad_descriptor;
+    return 0;
+  }
+
+  // A request to write 0 bytes to a stream is a no-op.
+  if (size == 0 && (state & stream_oriented))
+  {
+    ec = boost::system::error_code();
+    return 0;
+  }
+
+  // Read some data.
+  for (;;)
+  {
+    // Try to complete the operation without blocking.
+    signed_size_type bytes = socket_ops::send1(s, data, size, flags, ec);
+
+    // Check if operation succeeded.
+    if (bytes >= 0)
+      return bytes;
+
+    // Operation failed.
+    if ((state & user_set_non_blocking)
+        || (ec != boost::asio::error::would_block
+          && ec != boost::asio::error::try_again))
+      return 0;
+
+    // Wait for socket to become ready.
+    if (socket_ops::poll_write(s, 0, -1, ec) < 0)
+      return 0;
+  }
+}
+
 #if defined(BOOST_ASIO_HAS_IOCP)
 
 void complete_iocp_send(
@@ -1279,6 +1465,37 @@ bool non_blocking_send(socket_type s,
   }
 }
 
+bool non_blocking_send1(socket_type s,
+    const void* data, size_t size, int flags,
+    boost::system::error_code& ec, size_t& bytes_transferred)
+{
+  for (;;)
+  {
+    // Write some data.
+    signed_size_type bytes = socket_ops::send1(s, data, size, flags, ec);
+
+    // Retry operation if interrupted by signal.
+    if (ec == boost::asio::error::interrupted)
+      continue;
+
+    // Check if we need to run the operation again.
+    if (ec == boost::asio::error::would_block
+        || ec == boost::asio::error::try_again)
+      return false;
+
+    // Operation is complete.
+    if (bytes >= 0)
+    {
+      ec = boost::system::error_code();
+      bytes_transferred = bytes;
+    }
+    else
+      bytes_transferred = 0;
+
+    return true;
+  }
+}
+
 #endif // defined(BOOST_ASIO_HAS_IOCP)
 
 signed_size_type sendto(socket_type s, const buf* bufs, size_t count,

include/boost/asio/detail/reactive_socket_recv_op.hpp

@@ -49,13 +49,26 @@ class reactive_socket_recv_op_base : public reactor_op
     reactive_socket_recv_op_base* o(
         static_cast<reactive_socket_recv_op_base*>(base));
 
-    buffer_sequence_adapter<boost::asio::mutable_buffer,
-        MutableBufferSequence> bufs(o->buffers_);
+    typedef buffer_sequence_adapter<boost::asio::mutable_buffer,
+        MutableBufferSequence> bufs_type;
 
-    status result = socket_ops::non_blocking_recv(o->socket_,
-        bufs.buffers(), bufs.count(), o->flags_,
-        (o->state_ & socket_ops::stream_oriented) != 0,
-        o->ec_, o->bytes_transferred_) ? done : not_done;
+    status result;
+    if (bufs_type::is_single_buffer)
+    {
+      result = socket_ops::non_blocking_recv1(o->socket_,
+          bufs_type::first(o->buffers_).data(),
+          bufs_type::first(o->buffers_).size(), o->flags_,
+          (o->state_ & socket_ops::stream_oriented) != 0,
+          o->ec_, o->bytes_transferred_) ? done : not_done;
+    }
+    else
+    {
+      bufs_type bufs(o->buffers_);
+      result = socket_ops::non_blocking_recv(o->socket_,
+          bufs.buffers(), bufs.count(), o->flags_,
+          (o->state_ & socket_ops::stream_oriented) != 0,
+          o->ec_, o->bytes_transferred_) ? done : not_done;
+    }
 
     if (result == done)
       if ((o->state_ & socket_ops::stream_oriented) != 0)

include/boost/asio/detail/reactive_socket_send_op.hpp

@@ -49,17 +49,34 @@ class reactive_socket_send_op_base : public reactor_op
     reactive_socket_send_op_base* o(
         static_cast<reactive_socket_send_op_base*>(base));
 
-    buffer_sequence_adapter<boost::asio::const_buffer,
-        ConstBufferSequence> bufs(o->buffers_);
+    typedef buffer_sequence_adapter<boost::asio::const_buffer,
+        ConstBufferSequence> bufs_type;
 
-    status result = socket_ops::non_blocking_send(o->socket_,
-          bufs.buffers(), bufs.count(), o->flags_,
+    status result;
+    if (bufs_type::is_single_buffer)
+    {
+      result = socket_ops::non_blocking_send1(o->socket_,
+          bufs_type::first(o->buffers_).data(),
+          bufs_type::first(o->buffers_).size(), o->flags_,
           o->ec_, o->bytes_transferred_) ? done : not_done;
 
-    if (result == done)
-      if ((o->state_ & socket_ops::stream_oriented) != 0)
-        if (o->bytes_transferred_ < bufs.total_size())
-          result = done_and_exhausted;
+      if (result == done)
+        if ((o->state_ & socket_ops::stream_oriented) != 0)
+          if (o->bytes_transferred_ < bufs_type::first(o->buffers_).size())
+            result = done_and_exhausted;
+    }
+    else
+    {
+      bufs_type bufs(o->buffers_);
+      result = socket_ops::non_blocking_send(o->socket_,
+            bufs.buffers(), bufs.count(), o->flags_,
+            o->ec_, o->bytes_transferred_) ? done : not_done;
+
+      if (result == done)
+        if ((o->state_ & socket_ops::stream_oriented) != 0)
+          if (o->bytes_transferred_ < bufs.total_size())
+            result = done_and_exhausted;
+    }
 
     BOOST_ASIO_HANDLER_REACTOR_OPERATION((*o, "non_blocking_send",
           o->ec_, o->bytes_transferred_));