WL#13510 COMPRESSION PROTOCOL FOR ASYNC CLIENT

Description
-----------

This worklog adds support of compression protocol for asynchronous clients.
it does not change the wire format used in compressing packets for the usual
clients.

User need to set the compression algorithm name and if supported compression
level in the asynchronous client to able the packets compression.

For instance :
-------------

   if (mysql_options(mysql_local, MYSQL_OPT_COMPRESSION_ALGORITHMS,
                     "zstd"))
     exit(1);

   if (mysql_options(mysql_local, MYSQL_OPT_ZSTD_COMPRESSION_LEVEL, 4))
     exit(1);

   if(mysql_real_connect(mysql_local, opt_host, opt_user, opt_password,
                         current_db, opt_port, opt_unix_socket,
                         client_flag))
     exit(1);

   stmt_text = "SELECT col2 FROM test_table";

   /* run query in asynchronous way */
   status = mysql_real_query_nonblocking(mysql_local, stmt_text,
                                            (ulong)strlen(stmt_text));
   /* do some other task */
   perform_arithmatic();
   while (status == NET_ASYNC_NOT_READY) {
     status = mysql_real_query_nonblocking(mysql_local, stmt_text,
                                          (ulong)strlen(stmt_text));
   }

Testing
-------
  - Added following two tests in the testclients/mysql_client_test.cc.
     1. test_wl13510()
     2. test_wl13510_multi_statements()
    Above two tests are verified through main.mysql_client_test file.

  - Added MTR test file main.async_compress_client.test
  - Improved the existing big packet tests and added scenarios for
    testing with Nonblocking as well as Nonblocking compress clients.

Worklog PB2 page - mysql-trunk-wl13510

Review
------
RB#23665

Author: RahulSisondia

client/mysqltest.cc

@@ -6452,10 +6452,6 @@ static void do_connect(struct st_command *command) {
     opt_protocol = MYSQL_PROTOCOL_PIPE;
   }
 
-  if (opt_compress || con_compress) {
-    enable_async_client = false;
-  }
-
   if (opt_protocol) {
     mysql_options(&con_slot->mysql, MYSQL_OPT_PROTOCOL, (char *)&opt_protocol);
     /*
@@ -9173,9 +9169,6 @@ int main(int argc, char **argv) {
       opt_ssl_mode = SSL_MODE_VERIFY_CA;
   }
 
-  if (opt_compress) {
-    enable_async_client = false;
-  }
   if (SSL_SET_OPTIONS(&con->mysql)) die("%s", SSL_SET_OPTIONS_ERROR);
 #if defined(_WIN32)
   if (shared_memory_base_name)

include/errmsg.h

@@ -123,7 +123,7 @@ extern const char *client_errors[]; /* Error messages */
 #define CR_INSECURE_API_ERR 2062
 #define CR_FILE_NAME_TOO_LONG 2063
 #define CR_SSL_FIPS_MODE_ERR 2064
-#define CR_COMPRESSION_NOT_SUPPORTED 2065
+#define CR_DEPRECATED_COMPRESSION_NOT_SUPPORTED 2065
 #define CR_COMPRESSION_WRONGLY_CONFIGURED 2066
 #define CR_KERBEROS_USER_NOT_FOUND 2067
 #define CR_LOAD_DATA_LOCAL_INFILE_REJECTED 2068

include/mysql_async.h

@@ -1,4 +1,4 @@
-/* Copyright (c) 2019, Oracle and/or its affiliates. All rights reserved.
+/* Copyright (c) 2020, Oracle and/or its affiliates. All rights reserved.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License, version 2.0,
@@ -23,54 +23,55 @@
 #ifndef MYSQL_ASYNC_INCLUDED
 #define MYSQL_ASYNC_INCLUDED
 
-#define MYSQL_ASYNC_INCLUDED
-
 #include <mysql.h>
 
-/*
-  NOTE:this file should not be included as part of packaging.
+/**
+  @file mysql_async.h
+
+  Declarations for asynchronous client communication.
+
+  @note this file should not be included as part of packaging.
 */
-/* Async MySQL extension fields here. */
 
-/*
+/**
   This enum is to represent different asynchronous operations like reading the
   network, writing to network, idle state, or complete state.
 */
 enum net_async_operation {
-  NET_ASYNC_OP_IDLE = 0, /* default state */
-  NET_ASYNC_OP_READING,  /* used by my_net_read calls */
-  NET_ASYNC_OP_WRITING,  /* used by my_net_write calls */
-  NET_ASYNC_OP_COMPLETE  /* network read or write is complete */
+  NET_ASYNC_OP_IDLE = 0, /**< default state */
+  NET_ASYNC_OP_READING,  /**< used by my_net_read calls */
+  NET_ASYNC_OP_WRITING,  /**< used by my_net_write calls */
+  NET_ASYNC_OP_COMPLETE  /**< network read or write is complete */
 };
 
-/* Reading a packet is a multi-step process, so we have a state machine. */
+/** Reading a packet is a multi-step process, so we have a state machine. */
 enum net_async_read_packet_state {
-  NET_ASYNC_PACKET_READ_IDLE = 0, /* default packet read state */
-  NET_ASYNC_PACKET_READ_HEADER,   /* read packet header */
-  NET_ASYNC_PACKET_READ_BODY,     /* read packet contents */
-  NET_ASYNC_PACKET_READ_COMPLETE  /* state to define if packet is
-                                     completely read */
+  NET_ASYNC_PACKET_READ_IDLE = 0, /**< default packet read state */
+  NET_ASYNC_PACKET_READ_HEADER,   /**< read packet header */
+  NET_ASYNC_PACKET_READ_BODY,     /**< read packet contents */
+  NET_ASYNC_PACKET_READ_COMPLETE  /**< state to define if packet is
+                                       completely read */
 };
 
-/* Different states when reading a query result. */
+/** Different states when reading a query result. */
 enum net_read_query_result_status {
-  NET_ASYNC_READ_QUERY_RESULT_IDLE = 0,    /* default state */
-  NET_ASYNC_READ_QUERY_RESULT_FIELD_COUNT, /* read Ok or read field
-                                              count sent as part of
-                                              COM_QUERY */
-  NET_ASYNC_READ_QUERY_RESULT_FIELD_INFO   /* read result of above
-                                              COM_* command */
+  NET_ASYNC_READ_QUERY_RESULT_IDLE = 0,    /**< default state */
+  NET_ASYNC_READ_QUERY_RESULT_FIELD_COUNT, /**< read Ok or read field
+                                                count sent as part of
+                                                COM_QUERY */
+  NET_ASYNC_READ_QUERY_RESULT_FIELD_INFO   /**< read result of above
+                                                COM_* command */
 };
 
-/* Sending a command involves the write as well as reading the status. */
+/** Sending a command involves the write as well as reading the status. */
 enum net_send_command_status {
-  NET_ASYNC_SEND_COMMAND_IDLE = 0,      /* default send command state */
-  NET_ASYNC_SEND_COMMAND_WRITE_COMMAND, /* send COM_* command */
-  NET_ASYNC_SEND_COMMAND_READ_STATUS    /* read result of above COM_*
-                                           command */
+  NET_ASYNC_SEND_COMMAND_IDLE = 0,      /**< default send command state */
+  NET_ASYNC_SEND_COMMAND_WRITE_COMMAND, /**< send COM_* command */
+  NET_ASYNC_SEND_COMMAND_READ_STATUS    /**< read result of above COM_*
+                                             command */
 };
 
-/*
+/**
   Async operations are broadly classified into 3 phases:
   Connection phase, phase of sending data to server (which is writing phase)
   and reading data from server (which is reading phase). Below enum describes
@@ -82,49 +83,93 @@ enum net_async_block_state {
   NET_NONBLOCKING_WRITE
 };
 
+/**
+  Represents the packet to be sent on wire asynchronously.
+*/
 struct io_vec {
-  void *iov_base; /* Starting address */
-  size_t iov_len; /* Number of bytes to transfer */
+  void *iov_base; /**< Starting address */
+  size_t iov_len; /**< Number of bytes to transfer */
 };
 
 typedef struct NET_ASYNC {
-  /* The position in buff we continue reads from when data is next available */
+  /**
+    The position in buff we continue reads from when data is next
+    available
+  */
   unsigned char *cur_pos;
   /** Blocking state */
   enum net_async_block_state async_blocking_state;
   /** Our current operation */
   enum net_async_operation async_operation;
   /** How many bytes we want to read */
   size_t async_bytes_wanted;
-  /*
+  /**
     Simple state to know if we're reading the first row, and
     command/query statuses.
   */
   bool read_rows_is_first_read;
   enum net_send_command_status async_send_command_status;
   enum net_read_query_result_status async_read_query_result_status;
 
-  /* State when waiting on an async read */
+  /** State when waiting on an async read */
   enum net_async_read_packet_state async_packet_read_state;
-  /* Size of the packet we're currently reading */
+  /** Size of the packet we're currently reading */
   size_t async_packet_length;
 
-  /*
+  /**
     Headers and vector for our async writes; see net_serv.c for
     detailed description.
   */
   unsigned char *async_write_headers;
   struct io_vec *async_write_vector;
   size_t async_write_vector_size;
   size_t async_write_vector_current;
-  unsigned char
-      inline_async_write_header[NET_HEADER_SIZE + COMP_HEADER_SIZE + 1 + 1];
+
+  /**
+    If the packet length is less than MAX_PACKET_LENGTH then use a static array
+    to hold the meta packet header. The array either holds the usual packet
+    header or a compressed meta packet header as following. The compressed
+    meta packet header is followwed by usual compresses packet heder that is
+    7 bytes in length.
+
+
+    Packet
+
+        Header
+    ~~~~~~~~~~~~~~~~~~~
+    B1 B2 B3  : Packet length
+    B4        : Packet number
+    ~~~~~~~~~~~~~~~~~~~
+
+        Payload
+    ~~~~~~~~~~~~~~~~~~~
+    B5        : COM_COMMAND
+    ~~~~~~~~~~~~~~~~~~~
+
+    Compressed Packet
+
+        Header
+    ~~~~~~~~~~~~~~~~~~~
+    B1 B2 B3  : Compress packet length
+    B4        : Compress Packet Nunmber
+    00 00 00  : Indicates following payload is uncompressed
+    ~~~~~~~~~~~~~~~~~~~
+
+        Payload
+    ~~~~~~~~~~~~~~~~~~~
+    B8 B9 B10 : Packet size
+    B11       : Packet number
+    B12       : COM_COMMAND
+    ~~~~~~~~~~~~~~~~~~~
+  */
+  unsigned char inline_async_write_header[NET_HEADER_SIZE + COMP_HEADER_SIZE +
+                                          NET_HEADER_SIZE + 1];
   struct io_vec inline_async_write_vector[3];
 
-  /* State for reading responses that are larger than MAX_PACKET_LENGTH */
-  unsigned long async_multipacket_read_saved_whereb;
-  unsigned long async_multipacket_read_total_len;
-  bool async_multipacket_read_started;
+  /** Keep track of compressed buffers */
+  unsigned char **compressed_write_buffers;
+  /** Size of the compressed buffer */
+  size_t compressed_buffers_size;
 } NET_ASYNC;
 
 struct NET_EXTENSION {
@@ -141,7 +186,7 @@ void net_extension_free(NET *);
 #define NET_ASYNC_DATA(M) \
   ((NET_EXTENSION_PTR(M)) ? (NET_EXTENSION_PTR(M)->net_async_context) : NULL)
 
-/*
+/**
   Asynchronous operations are broadly classified into 2 categories.
   1. Connection
   2. Query execution
@@ -153,7 +198,7 @@ enum mysql_async_operation_status {
   ASYNC_OP_QUERY
 };
 
-/*
+/**
   Query execution in an asynchronous fashion is broadly divided into 3 states
   which is described in below enum
 */
@@ -164,24 +209,24 @@ enum mysql_async_query_state_enum {
 };
 
 typedef struct MYSQL_ASYNC {
-  /* Buffer storing the rows result for cli_read_rows_nonblocking */
+  /** Buffer storing the rows result for cli_read_rows_nonblocking */
   MYSQL_DATA *rows_result_buffer;
-  /* a pointer to keep track of the previous row of the current result row */
+  /** a pointer to keep track of the previous row of the current result row */
   MYSQL_ROWS **prev_row_ptr;
-  /* Context needed to track the state of a connection being established */
+  /** Context needed to track the state of a connection being established */
   struct mysql_async_connect *connect_context;
-  /* Status of the current async op */
+  /** Status of the current async op */
   enum mysql_async_operation_status async_op_status;
-  /* Size of the current running async query */
+  /** Size of the current running async query */
   size_t async_query_length;
-  /* If a query is running, this is its state */
+  /** If a query is running, this is its state */
   enum mysql_async_query_state_enum async_query_state;
-  /* context needed to support metadata read operation */
+  /** context needed to support metadata read operation */
   unsigned long *async_read_metadata_field_len;
   MYSQL_FIELD *async_read_metadata_fields;
   MYSQL_ROWS async_read_metadata_data;
   unsigned int async_read_metadata_cur_field;
-  /* a pointer to keep track of the result sets */
+  /** a pointer to keep track of the result sets */
   struct MYSQL_RES *async_store_result_result;
 } MYSQL_ASYNC;
 
@@ -192,8 +237,7 @@ enum net_async_status net_write_command_nonblocking(
     NET *net, unsigned char command, const unsigned char *prefix,
     size_t prefix_len, const unsigned char *packet, size_t packet_len,
     bool *res);
-enum net_async_status my_net_read_nonblocking(NET *net, unsigned long *len_ptr,
-                                              unsigned long *complen_ptr);
+enum net_async_status my_net_read_nonblocking(NET *net, unsigned long *len_ptr);
 
 int mysql_get_socket_descriptor(MYSQL *mysql);
 

mysql-test/r/async_client.result

@@ -14,15 +14,15 @@ CREATE TABLE t1(i INT, j VARCHAR(2048));
 INSERT INTO t1 VALUES(1,repeat('a',1000)),(2,repeat('def',600));
 SELECT * FROM t1;
 i	j
-1	aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa
-2	defdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdefdef
+count	big_value
+count	big_value
 # case2: request a large packet
 SET GLOBAL max_allowed_packet=4*1024;
 Warnings:
 Warning	1708	The value of 'max_allowed_packet' should be no less than the value of 'net_buffer_length'
 SELECT SPACE(@@global.max_allowed_packet);
 SPACE(@@global.max_allowed_packet)
-                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                
+lot_of_spaces
 SET GLOBAL max_allowed_packet=default;
 # connect with wrong password
 connect(localhost,caching_sha2,caching1,wl11381,MASTER_PORT,MASTER_SOCKET);
@@ -64,7 +64,7 @@ caching_sha2@localhost
 # case3: authenticate user with sha256_password
 CREATE USER sha256@localhost IDENTIFIED WITH 'sha256_password' BY 'auth_string';
 # restart: --default-authentication-plugin=sha256_password
-# connect as sha256 
+# connect as sha256
 SELECT USER();
 USER()
 sha256@localhost