CDRIVER-532 - Coalesce small vectorized TLS writes

Rather than calling BIO_write multiple times for each iovec_t in a call
to mongoc_stream_tls_writev(), buffer them and send them all at once.
Attempt to "do the right thing"tm, when buffering provides no
performance benefit.

Author: hanumantmk

src/mongoc/mongoc-stream-tls.c

@@ -42,6 +42,8 @@
 #undef MONGOC_LOG_DOMAIN
 #define MONGOC_LOG_DOMAIN "stream-tls"
 
+#define MONGOC_STREAM_TLS_BUFFER_SIZE 4096
+
 
 /**
  * mongoc_stream_tls_t:
@@ -448,6 +450,53 @@ _mongoc_stream_tls_flush (mongoc_stream_t *stream)
 }
 
 
+static ssize_t
+_mongoc_stream_tls_write (mongoc_stream_tls_t *tls,
+                          char                *buf,
+                          size_t               buf_len)
+{
+   ssize_t ret;
+
+   int64_t now;
+   int64_t expire = 0;
+
+   BSON_ASSERT (tls);
+   BSON_ASSERT (buf);
+   BSON_ASSERT (buf_len);
+
+   if (tls->timeout_msec >= 0) {
+      expire = bson_get_monotonic_time () + (tls->timeout_msec * 1000UL);
+   }
+
+   ret = BIO_write (tls->bio, buf, buf_len);
+
+   if (ret < 0) {
+      return ret;
+   }
+
+   if (expire) {
+      now = bson_get_monotonic_time ();
+
+      if ((expire - now) < 0) {
+         if (ret < buf_len) {
+            mongoc_counter_streams_timeout_inc();
+#ifdef _WIN32
+            errno = WSAETIMEDOUT;
+#else
+            errno = ETIMEDOUT;
+#endif
+         }
+
+         tls->timeout_msec = 0;
+      } else {
+         tls->timeout_msec = (expire - now) / 1000L;
+      }
+   }
+
+   return ret;
+}
+
+
 /*
  *--------------------------------------------------------------------------
  *
@@ -473,62 +522,112 @@ _mongoc_stream_tls_writev (mongoc_stream_t *stream,
                            int32_t          timeout_msec)
 {
    mongoc_stream_tls_t *tls = (mongoc_stream_tls_t *)stream;
+   char buf[MONGOC_STREAM_TLS_BUFFER_SIZE];
+
    ssize_t ret = 0;
+   ssize_t child_ret;
    size_t i;
    size_t iov_pos = 0;
-   int write_ret;
 
-   int64_t now;
-   int64_t expire = 0;
+   /* There's a bit of a dance to coalesce vectorized writes into
+    * MONGOC_STREAM_TLS_BUFFER_SIZE'd writes to avoid lots of small tls
+    * packets.
+    *
+    * The basic idea is that we want to combine writes in the buffer if they're
+    * smaller than the buffer, flushing as it gets full.  For larger writes, or
+    * the last write in the iovec array, we want to ignore the buffer and just
+    * write immediately.  We take care of doing buffer writes by re-invoking
+    * ourself with a single iovec_t, pointing at our stack buffer.
+    */
+   char *buf_head = buf;
+   char *buf_tail = buf;
+   char *buf_end = buf + MONGOC_STREAM_TLS_BUFFER_SIZE;
+   size_t bytes;
+
+   char *to_write = NULL;
+   size_t to_write_len;
 
    BSON_ASSERT (tls);
    BSON_ASSERT (iov);
    BSON_ASSERT (iovcnt);
 
    tls->timeout_msec = timeout_msec;
 
-   if (timeout_msec >= 0) {
-      expire = bson_get_monotonic_time () + (timeout_msec * 1000UL);
-   }
-
    for (i = 0; i < iovcnt; i++) {
       iov_pos = 0;
 
       while (iov_pos < iov[i].iov_len) {
-         write_ret = BIO_write (tls->bio, (char *)iov[i].iov_base + iov_pos,
-                              (int)(iov[i].iov_len - iov_pos));
+         if (buf_head != buf_tail ||
+             ((i + 1 < iovcnt) &&
+              ((buf_end - buf_tail) > (iov[i].iov_len - iov_pos)))) {
+            /* If we have either of:
+             *   - buffered bytes already
+             *   - another iovec to send after this one and we don't have more
+             *     bytes to send than the size of the buffer.
+             *
+             * copy into the buffer */
+
+            bytes = BSON_MIN (iov[i].iov_len - iov_pos, buf_end - buf_tail);
+
+            memcpy (buf_tail, iov[i].iov_base + iov_pos, bytes);
+            buf_tail += bytes;
+            iov_pos += bytes;
+
+            if (buf_tail == buf_end) {
+               /* If we're full, request send */
+
+               to_write = buf_head;
+               to_write_len = buf_tail - buf_head;
+
+               buf_tail = buf_head = buf;
+            }
+         } else {
+            /* Didn't buffer, so just write it through */
+
+            to_write = (char *)iov[i].iov_base + iov_pos;
+            to_write_len = iov[i].iov_len - iov_pos;
 
-         if (write_ret < 0) {
-            return write_ret;
+            iov_pos += to_write_len;
          }
 
-         if (expire) {
-            now = bson_get_monotonic_time ();
+         if (to_write) {
+            /* We get here if we buffered some bytes and filled the buffer, or
+             * if we didn't buffer and have to send out of the iovec */
 
-            if ((expire - now) < 0) {
-               if (write_ret == 0) {
-                  mongoc_counter_streams_timeout_inc();
-#ifdef _WIN32
-                  errno = WSAETIMEDOUT;
-#else
-                  errno = ETIMEDOUT;
-#endif
-                  return -1;
-               }
+            child_ret = _mongoc_stream_tls_write (tls, to_write, to_write_len);
 
-               tls->timeout_msec = 0;
-            } else {
-               tls->timeout_msec = (expire - now) / 1000L;
+            if (child_ret < 0) {
+               /* Buffer write failed, just return the error */
+               return child_ret;
+            }
+
+            ret += child_ret;
+
+            if (child_ret < to_write_len) {
+               /* we timed out, so send back what we could send */
+
+               return ret;
             }
+
+            to_write = NULL;
          }
+      }
+   }
+
+   if (buf_head != buf_tail) {
+      /* If we have any bytes buffered, send */
 
-         ret += write_ret;
-         iov_pos += write_ret;
+      child_ret = _mongoc_stream_tls_write (tls, buf_head, buf_tail - buf_head);
+
+      if (child_ret < 0) {
+         return child_ret;
       }
+
+      ret += child_ret;
    }
 
    if (ret >= 0) {
-      mongoc_counter_streams_egress_add(ret);
+      mongoc_counter_streams_egress_add (ret);
    }
 
    return ret;

tests/ssl-test.c

@@ -9,6 +9,8 @@
 
 #define TIMEOUT 1000
 
+#define NUM_IOVECS 2000
+
 #define LOCALHOST "127.0.0.1"
 
 typedef struct ssl_test_data
@@ -45,7 +47,7 @@ ssl_test_server (void * ptr)
    mongoc_socket_t *listen_sock;
    mongoc_socket_t *conn_sock;
    socklen_t sock_len;
-   char buf[1024];
+   char buf[4 * NUM_IOVECS];
    ssize_t r;
    mongoc_iovec_t iov;
    struct sockaddr_in server_addr = { 0 };
@@ -166,11 +168,13 @@ ssl_test_client (void * ptr)
    mongoc_stream_t *sock_stream;
    mongoc_stream_t *ssl_stream;
    mongoc_socket_t *conn_sock;
+   int i;
    int errno_captured;
    char buf[1024];
    ssize_t r;
    mongoc_iovec_t riov;
    mongoc_iovec_t wiov;
+   mongoc_iovec_t wiov_many[NUM_IOVECS];
    struct sockaddr_in server_addr = { 0 };
    int len;
 
@@ -237,18 +241,21 @@ ssl_test_client (void * ptr)
       return NULL;
    }
 
-   len = 4;
+   len = 4 * NUM_IOVECS;
 
    wiov.iov_base = (void *)&len;
    wiov.iov_len = 4;
    r = mongoc_stream_writev(ssl_stream, &wiov, 1, TIMEOUT);
 
    assert(r == wiov.iov_len);
 
-   wiov.iov_base = "foo";
-   wiov.iov_len = 4;
-   r = mongoc_stream_writev(ssl_stream, &wiov, 1, TIMEOUT);
-   assert(r == wiov.iov_len);
+   for (i = 0; i < NUM_IOVECS; i++) {
+      wiov_many[i].iov_base = "foo";
+      wiov_many[i].iov_len = 4;
+   }
+
+   r = mongoc_stream_writev(ssl_stream, wiov_many, NUM_IOVECS, TIMEOUT);
+   assert(r == wiov_many[0].iov_len * NUM_IOVECS);
 
    riov.iov_len = 1;