From bb68e0226700c48da7a3c4392b46b1bd298e89ed Mon Sep 17 00:00:00 2001 From: Jay Edgar Date: Wed, 3 Jun 2020 12:39:06 -0700 Subject: [PATCH] Coallesce small buffers to send them as one Summary: The new async client function to write a vector of buffers can cause lots of small buffers to go through SSL encoding which can be expensive. This diff attempts to combine the first portions (which are most likely to be small) into a single buffer. The original code already had a small buffer to build the header into. This expands that buffer and then continues copying data into it until the next buffer won't fit. This diff also avoids adding 0 length buffers which could occur with the original implementation. Note that this buffer coalescing is disabled when the overall packet is very large and gets turned off part way through if network compression is enabled. Differential Revision: D21867050 --- include/mysql_async.h | 12 ++++- sql-common/net_serv.cc | 109 ++++++++++++++++++++++++----------------- 2 files changed, 75 insertions(+), 46 deletions(-) diff --git a/include/mysql_async.h b/include/mysql_async.h index cf78bfe4c831..a8e76336d03f 100644 --- a/include/mysql_async.h +++ b/include/mysql_async.h @@ -91,6 +91,15 @@ struct io_vec { size_t iov_len; /**< Number of bytes to transfer */ }; +#define ASYNC_COALESCE_BUFFER_SIZE 1024 + +// Make sure the coalesce buffer size is larger than we used to have +#define ASYNC_INLINE_ORIGINAL_SIZE \ + NET_HEADER_SIZE + COMP_HEADER_SIZE + NET_HEADER_SIZE + 1 +#if ASYNC_COALESCE_BUFFER_SIZE < ASYNC_INLINE_ORIGINAL_SIZE +#error The coalesce buffer size is too small +#endif + typedef struct NET_ASYNC { /** The position in buff we continue reads from when data is next @@ -162,8 +171,7 @@ typedef struct NET_ASYNC { B12 : COM_COMMAND ~~~~~~~~~~~~~~~~~~~ */ - unsigned char inline_async_write_header[NET_HEADER_SIZE + COMP_HEADER_SIZE + - NET_HEADER_SIZE + 1]; + unsigned char inline_async_write_header[ASYNC_COALESCE_BUFFER_SIZE]; struct io_vec inline_async_write_vector[3]; /** Keep track of compressed buffers */ diff --git a/sql-common/net_serv.cc b/sql-common/net_serv.cc index 88e3a2ae08a2..224db16a0a72 100644 --- a/sql-common/net_serv.cc +++ b/sql-common/net_serv.cc @@ -551,6 +551,8 @@ static int begin_packet_write_state(NET *net, uchar command, struct io_vec *vec; uchar *headers; uchar **compressed_buffers = nullptr; + bool try_coalesce = true; + if (total_len < MAX_PACKET_LENGTH) { /* Most writes hit this case, ie, less than MAX_PACKET_LENGTH of @@ -574,6 +576,11 @@ static int begin_packet_write_state(NET *net, uchar command, my_free(vec); return 0; } + + // Don't try any coalescing on messages that require multiple network + // packets - the efficiency gains would be minimal and the code would + // be more complex. + try_coalesce = false; } /* Regardless of where vec and headers come from, these are what we @@ -631,10 +638,7 @@ static int begin_packet_write_state(NET *net, uchar command, } int3store(buf, comp_packet_len); buf[3] = (uchar)net->compress_pkt_nr++; - /* - The bytes in COMP_HEADER_SIZE are implicitly zero because they were - zero filled. A zero length means that the contents are uncompressed. - */ + memset(buf + NET_HEADER_SIZE, 0, COMP_HEADER_SIZE); buf += NET_HEADER_SIZE + COMP_HEADER_SIZE; } @@ -653,65 +657,82 @@ static int begin_packet_write_state(NET *net, uchar command, ++bytes_queued; } - ++vec; - /* Second iovec (if any), our optional prefix. */ if (packet_num == 0 && optional_prefix != nullptr) { - (*vec).iov_base = const_cast(optional_prefix); - (*vec).iov_len = prefix_len; + if (try_coalesce && + vec->iov_len + prefix_len <= ASYNC_COALESCE_BUFFER_SIZE) { + memcpy((uchar *)vec->iov_base + vec->iov_len, optional_prefix, + prefix_len); + vec->iov_len += prefix_len; + } else { + ++vec; + (*vec).iov_base = const_cast(optional_prefix); + (*vec).iov_len = prefix_len; + try_coalesce = false; + } bytes_queued += prefix_len; - ++vec; } /* Final iovec, the payload itself. Send however many bytes from packet we have left, and advance our packet pointer. */ const size_t remaining_bytes = packet_size - bytes_queued; - (*vec).iov_base = const_cast(packet); - (*vec).iov_len = remaining_bytes; - - bytes_queued += remaining_bytes; + if (remaining_bytes != 0) { + // Turn off coalescing if compressing + try_coalesce = try_coalesce && !net->compress; + + if (try_coalesce && + vec->iov_len + remaining_bytes <= ASYNC_COALESCE_BUFFER_SIZE) { + memcpy((uchar *)vec->iov_base + vec->iov_len, packet, remaining_bytes); + vec->iov_len += remaining_bytes; + } else { + ++vec; + (*vec).iov_base = const_cast(packet); + (*vec).iov_len = remaining_bytes; + } - packet += remaining_bytes; - total_len -= bytes_queued; - /* clang-format off */ - /* - If we have a payload to compress, then compress_packet will add - compression headers for us. This is what we have at this point where - each line is an iovec. - | len |cpn|uncompress len| len | pn | command | - | prefix + command | 0 |0 | total_len = command + prefix + payload | 0 |COM_* | + bytes_queued += remaining_bytes; - | prefix | - |... | + packet += remaining_bytes; + /* clang-format off */ + /* + If we have a payload to compress, then compress_packet will + add compression headers for us. This is what we have at this point where + each line is an iovec. + | len |cpn|uncompress len| len | pn | command | + | prefix + command | 0| 0| total_len = command + prefix + payload | 0 | COM_* | - |payload | - |... | + | prefix | + | ... | - We want to transform into this: + | payload | + | ... | - | len |cpn|uncompress len| len | pn | command | - | prefix +command | 0 |0 | total_len = command + prefix + payload | 0 |COM_* | + We want to transform into this: + | len |cpn|uncompress len| len | pn | command | + | prefix + command | 0| 0| total_len = command + prefix + payload | 0 | COM_* | - | prefix | - |... | + | prefix | + | ... | - | len |cpn|uncompress len| compressed payload | - | len(compressed payload) | 1| len(payload)| compress(payload) | - */ - /* clang-format on */ - - if (net->compress && remaining_bytes) { - (*vec).iov_base = - compress_packet(net, (uchar *)(*vec).iov_base, &(*vec).iov_len); - if (!(*vec).iov_base) { - reset_packet_write_state(net); - return 0; + | len |cpn|uncompress len| compressed payload | + | len(compressed payload) | 1| len(payload)| compress(payload) | + */ + /* clang-format on */ + if (net->compress) { + (*vec).iov_base = compress_packet( + net, static_cast((*vec).iov_base), &(*vec).iov_len); + if (!(*vec).iov_base) { + reset_packet_write_state(net); + return 0; + } + compressed_buffers[net_async->compressed_buffers_size++] = + static_cast((*vec).iov_base); } - compressed_buffers[net_async->compressed_buffers_size++] = - (uchar *)(*vec).iov_base; } ++vec; + total_len -= bytes_queued; + try_coalesce = false; /* Make sure we sent entire packets. */ if (total_len > 0) {