WL#13510: COMPRESSION PROTOCOL FOR ASYNC CLIENT

We implemented asynchronous clients support through WL#11381 and added
compression protocol support for usual clients through WL#12475.
This worklog will enable compression protocol support for asynchronous clients.

It will reduce the network traffic across data-centers.
This worklog is based on Facebook's contribution.

FR-1: Asynchronous client must be able to send/receive compressed packets.

FR-2: Asynchronous client must use compression algorithms supported by the
      server. For instance : 'zstd' and 'zlib'.

FR-3: Asynchronous client must be able to execute single or multiple statements.

FR-4: Wire format of the compressed packet must be same as it is sent by the
      usual blocking client.

Background:
----------
Wire format of the compressed packet is not changed. Compressed packet consists
7 byte packet header followed by the payload.

For instance :

The first meta packet consists information about the subsequent packet. Both of
them have packet header of 7 bytes followed by the payload.

     | len  |cpn|uncompress len| payload size | packet num | command (COM_*)|

e.g. |  5   | 0 |  000         | 560          |      0     |   3            |
        ∆
        |                      |   3 byte     |   1 byte   |     1 byte     |
        └──────────────────────┴──────────────┴────────────┴────────────────┘


     | len  |cpn|uncompress len| compressed payload |

e.g. |  150 |  1|  560         |  compress(payload) |
         ∆                               V
     |   |      |              |         |
     |   └──────┼──────────────┼─────────┘
     |          |              |
     | 4 byte   |  3 byte      |
     └──────────┴──────────────┘
Refer the manual for the more details :
https://dev.mysql.com/doc/dev/mysql-
server/latest/page_protocol_basic_compression_packet.html

(H1) Send the compressed packet through asynchronous client
     ------------------------------------------------------

    (1) The first async state machine is to prepare the packet for writing on
        VIO. This state is named NET_ASYNC_OP_IDLE. Async client will prepare
        the compressed packet during this state.

    (2) Following two variables will be added to the NET_ASYNC structure to
        keep track of compressed data packets.

        NET_ASYNC {
          + unsigned char **compressed_write_buffers;
          + size_t compressed_buffers_size;
       }
    (3) The compressed packets are represented through io vector structure.
        This IO Vector is written to the VIO in the next async state known as
        NET_ASYNC_OP_WRITING.

    (4) Control may go back to the client during NET_ASYNC_OP_WRITING state
        while packet is being written to the VIO. In the NET_ASYNC_OP_IDLE
        state, while async state machine is compressing the data control
        does not return to the client.

(H2) Read the compressed packet through asynchronous client
     ------------------------------------------------------
    (1) Like usual nonblocking read, async client will read the header of
        compressed packet in the NET_ASYNC_PACKET_READ_HEADER state. It will
        retrieve the length and the packet sequence number.

    (2) Async state machine will fall through in NET_ASYNC_PACKET_READ_BODY
        state. While reading the packet from the VIO, it may return the control
        to the client. If the async state machine is able to read the
        compressed data then it uncompresses the data before moving itself to
        the NET_ASYNC_COMPLETE state.

    (3) The uncompressed data may comprises multiple packets. Hence
        the data is parsed until all packets are read in the uncompressed
        payload.

    (4) In cases where the payload length is more than MAX_PACKET_LENGTH, the
        packet is divided into the multiple packets. The subsequent packets
        comprises data in continuation. In multibyte packets we could directly
        read the data in the subsequent packets. There shall be handling of such
        packets.

  (H3) Enable packets compression for asynchronous clients
       ---------------------------------------------------

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

       For instance :
       -------------
          // Set the compression algorithm name
          if (mysql_options(mysql_local, MYSQL_OPT_COMPRESSION_ALGORITHMS,
                            "zstd"))
            exit(1);
          // Set the compression level. This is optional.
          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));
          }

(L1) Sending the compressed packet asynchronously
     --------------------------------------------
     (1) net_write_command_nonblocking() API writes to the  wire in various
         steps. These steps are called net_async_operation states. In the
         first state, NET_ASYNC_OP_IDLE, packet is prepared to be written.
         In the second state packet is sent to the wire.
         The packet to be written is represented by a structure known as io_vec.

         net_write_command_nonblocking () {
           switch (net_async->async_operation) {
               case NET_ASYNC_OP_IDLE:
                 if (!begin_packet_write_state()) {
                   return NET_ASYNC_COMPLETE;
                 }
                 net_async->async_operation = NET_ASYNC_OP_WRITING;
                 /* fallthrough */
               case NET_ASYNC_OP_WRITING:
                 status = net_write_vector_nonblocking();
                 if (status == NET_ASYNC_COMPLETE) {
                  return NET_ASYNC_COMPLETE;
                 } else
                  return NET_ASYNC_NOT_READY;
          }

     (2) The compress packet is prepared during the first state
         NET_ASYNC_OP_IDLE. The payload is compressed and added to the io_vec.
         The compressed and uncompressed payload length are added to the packet
         header.
         begin_packet_write_state {
          struct io_vec *vec = ;
          for (size_t packet_num = 0; packet_num < packet_count; ++packet_num) {
            /*
              For each packet the iovec points to the packet header and payload
              buffer as following.
            */
            vec[packet_num][0] =   | len  |cpn|uncompress len| payload size |
                                   | packet num | command (COM_*)|
                                   
            vec[packet_num[1]  =   | prefix |
            
            vec[packet_num][2] =   | len    |cpn|uncompress len| 
                                   |  compressed payload |
        }
       }

(L2) Reading the compressed packet asynchronously :
     --------------------------------------------
  (1) my_net_read_nonblocking() API reads the packets asynchronously.
      This API will call the respective read method depending upon the
      compress or uncompress packet to be read.

      if (net->compress) {
        if (my_net_read_compressed_nonblocking(net, len_ptr, complen_ptr) ==
            NET_ASYNC_NOT_READY) {
              return NET_ASYNC_NOT_READY;
        }
      } else if (net_read_packet_nonblocking(net, len_ptr, complen_ptr) ==
             NET_ASYNC_NOT_READY) {
        return NET_ASYNC_NOT_READY;
      }

   (2) Since the wire format is not changed, the pattern to read the
       uncompressed packet is same for synchronous and asynchronous clients
       both. We could read the compressed packet as following.

       read_compressed_packets {
         // initialize offsets in the NET structure
         initialize_offsets(net);
         for (;;)
         {
            if(process_buffer(net)) break;
            if(asynchronous_client)
               read_packet_nonblocking(packet);    // packet = net->buff
               uncompress_packet(packet)
            else
              read_packet_uncompress_it(packet);
         }
         // Update the offsets in the NET structure after the packet is read.
         update_offsets(net);
      }

   (3) The nonblocking packet read happens in different steps. These steps are
       called net_read_packet_nonblocking states. Control may reach to the
       client while the packet is being read. During uncompress operation
       control does not reach to the client.

       read_packet_nonblocking() {
          switch (net_read_packet_nonblocking) {
            case NET_ASYNC_PACKET_READ_HEADER:
              if (net_read_packet_header_nonblocking(net, &err) ==
                NET_ASYNC_NOT_READY) {
                  return NET_ASYNC_NOT_READY;
              }
            case NET_ASYNC_PACKET_READ_BODY:
              if (net_read_data_nonblocking(net, net_async->async_packet_length,
                                    &err) == NET_ASYNC_NOT_READY) {
                return NET_ASYNC_NOT_READY;
              }
          }
          if(is_compressed) {
            my_uncompress(compress_context, net->buff, uncomp_length,
                          original_len);
          }
     }