Caching compressed binlog events for reuse between dump threads

Summary:
With packet compression each dump thread compresses the same event
redundantly. This feature introduces a new way of network compression between
master and slave where events are compressed once and cached. This compression
can be enabled by using the variable `slave_compressed_event_protocol`. Slave
needs to be restarted (stop slave; start slave;) for the variable to take effect

Reviewed By: anirbanr-fb

Differential Revision: D5107304

fbshipit-source-id: 06b212d

client/mysqlbinlog.cc

@@ -104,6 +104,7 @@ static const char* default_dbug_option = "d:t:o,/tmp/mysqlbinlog.trace";
 static const char *load_default_groups[]= { "mysqlbinlog","client",0 };
 
 static my_bool opt_compress=0;
+static my_bool opt_compress_event=0;
 static my_bool one_database=0, disable_log_bin= 0;
 static my_bool opt_hexdump= 0;
 const char *base64_output_mode_names[]=
@@ -1866,6 +1867,9 @@ static struct my_option my_long_options[] =
    &charsets_dir, 0, GET_STR, REQUIRED_ARG, 0, 0, 0, 0, 0, 0},
   {"compress", 'C', "Use compression in server/client protocol.",
    &opt_compress, &opt_compress, 0, GET_BOOL, NO_ARG, 0, 0, 0, 0, 0, 0},
+  {"compress_event", 'E', "Use event compression in server/client protocol.",
+   &opt_compress_event, &opt_compress_event, 0, GET_BOOL, NO_ARG,
+   0, 0, 0, 0, 0, 0},
   {"database", 'd', "List entries for just this database (local log only).",
    &database, &database, 0, GET_STR_ALLOC, REQUIRED_ARG,
    0, 0, 0, 0, 0, 0},
@@ -2416,6 +2420,11 @@ static Exit_status safe_connect()
 
   if (opt_compress)
     mysql_options(mysql, MYSQL_OPT_COMPRESS, NullS);
+  if (opt_compress_event)
+    mysql_options(mysql, MYSQL_OPT_COMP_EVENT, NullS);
+  if (opt_compress && opt_compress_event)
+    warning("Both packet and event compression were enabled. Disabling packet "
+            "compression to avoid double compression.");
   if (opt_protocol)
     mysql_options(mysql, MYSQL_OPT_PROTOCOL, (char*) &opt_protocol);
   if (opt_bind_addr)
@@ -2927,8 +2936,18 @@ static Exit_status dump_remote_log_entries(PRINT_EVENT_INFO *print_event_info,
     const char *error_msg= NULL;
     Log_event *ev= NULL;
     Log_event_type type= UNKNOWN_EVENT;
+    len= 0;
+
+    if (mysql->net.vio != 0)
+      len= my_net_read(&mysql->net);
+
+#ifdef HAVE_COMPRESS
+  // case: event was compressed before sending, so we have to uncompress
+  if (mysql->net.compress_event && len != 0 && len != packet_error)
+    len= uncompress_event(&mysql->net, len);
+#endif
 
-    len= cli_safe_read(mysql, NULL);
+    len= cli_safe_read_complete(mysql, len, 0, NULL);
     if (len == packet_error)
     {
       uint mysql_error_number= mysql_errno(mysql);

include/mysql.h

@@ -189,7 +189,8 @@ enum mysql_option
   MYSQL_OPT_WRITE_TIMEOUT_MS,
   MYSQL_OPT_SSL_SESSION,
   MYSQL_OPT_SSL_CONTEXT,
-  MYSQL_OPT_COMP_LIB
+  MYSQL_OPT_COMP_LIB,
+  MYSQL_OPT_COMP_EVENT
 };
 
 /**
@@ -213,7 +214,7 @@ struct st_mysql_options {
   char *shared_memory_base_name;
   unsigned long max_allowed_packet;
   my_bool use_ssl;				/* if to use SSL or not */
-  my_bool compress,named_pipe;
+  my_bool compress,compress_event,named_pipe;
   my_bool unused1;
   my_bool unused2;
   my_bool unused3;

include/mysql.h.pp

@@ -96,6 +96,7 @@
   my_bool unused1;
   my_bool unused2;
   my_bool compress;
+  my_bool compress_event;
   my_bool unused3;
   enum mysql_compression_lib comp_lib;
   ZSTD_CCtx *cctx;
@@ -393,7 +394,8 @@
   MYSQL_OPT_WRITE_TIMEOUT_MS,
   MYSQL_OPT_SSL_SESSION,
   MYSQL_OPT_SSL_CONTEXT,
-  MYSQL_OPT_COMP_LIB
+  MYSQL_OPT_COMP_LIB,
+  MYSQL_OPT_COMP_EVENT
 };
 struct st_mysql_options_extention;
 struct st_mysql_options {
@@ -411,7 +413,7 @@
   char *shared_memory_base_name;
   unsigned long max_allowed_packet;
   my_bool use_ssl;
-  my_bool compress,named_pipe;
+  my_bool compress,compress_event,named_pipe;
   my_bool unused1;
   my_bool unused2;
   my_bool unused3;

include/mysql_com.h

@@ -213,11 +213,14 @@ enum enum_server_command
 /* Client no longer needs EOF packet */
 #define CLIENT_DEPRECATE_EOF (1UL << 24)
 
+/* Event compression */
+#define CLIENT_COMPRESS_EVENT (1UL << 25)
+
 #define CLIENT_SSL_VERIFY_SERVER_CERT (1UL << 30)
 #define CLIENT_REMEMBER_OPTIONS (1UL << 31)
 
 #ifdef HAVE_COMPRESS
-#define CAN_CLIENT_COMPRESS CLIENT_COMPRESS
+#define CAN_CLIENT_COMPRESS (CLIENT_COMPRESS | CLIENT_COMPRESS_EVENT)
 #else
 #define CAN_CLIENT_COMPRESS 0
 #endif
@@ -250,15 +253,17 @@ enum enum_server_command
                            | CLIENT_CAN_HANDLE_EXPIRED_PASSWORDS \
                            | CLIENT_SESSION_TRACK \
                            | CLIENT_DEPRECATE_EOF \
+                           | CLIENT_COMPRESS_EVENT \
 )
 
 /*
   Switch off the flags that are optional and depending on build flags
   If any of the optional flags is supported by the build it will be switched
   on before sending to the client during the connection handshake.
 */
-#define CLIENT_BASIC_FLAGS (((CLIENT_ALL_FLAGS & ~CLIENT_SSL) \
+#define CLIENT_BASIC_FLAGS ((((CLIENT_ALL_FLAGS & ~CLIENT_SSL) \
                                                & ~CLIENT_COMPRESS) \
+                                               & ~CLIENT_COMPRESS_EVENT) \
                                                & ~CLIENT_SSL_VERIFY_SERVER_CERT)
 
 /**
@@ -470,6 +475,7 @@ typedef struct st_net {
   my_bool unused1; /* Please remove with the next incompatible ABI change */
   my_bool unused2; /* Please remove with the next incompatible ABI change */
   my_bool compress;
+  my_bool compress_event;
   my_bool unused3; /* Please remove with the next incompatible ABI change. */
   enum mysql_compression_lib comp_lib;
   ZSTD_CCtx *cctx;

include/sql_common.h

@@ -125,6 +125,11 @@ cli_advanced_command(MYSQL *mysql, enum enum_server_command command,
 unsigned long cli_safe_read(MYSQL *mysql, my_bool *is_data_packet);
 unsigned long cli_safe_read_with_ok(MYSQL *mysql, my_bool parse_ok,
                                     my_bool *is_data_packet);
+unsigned long cli_safe_read_complete(MYSQL *mysql, ulong len,
+                                     my_bool parse_ok, my_bool *is_data_packet);
+#ifdef HAVE_COMPRESS
+unsigned long uncompress_event(NET* net, ulong len);
+#endif
 void net_clear_error(NET *net);
 void set_stmt_errmsg(MYSQL_STMT *stmt, NET *net);
 void set_stmt_error(MYSQL_STMT *stmt, int errcode, const char *sqlstate,

libmysqld/libmysqld.c

@@ -165,7 +165,7 @@ mysql_real_connect(MYSQL *mysql,const char *host, const char *user,
     no compression in embedded as we don't send any data,
     and no pluggable auth, as we cannot do a client-server dialog
   */
-  client_flag&= ~(CLIENT_COMPRESS | CLIENT_PLUGIN_AUTH);
+  client_flag&= ~(CLIENT_COMPRESS | CLIENT_COMPRESS_EVENT | CLIENT_PLUGIN_AUTH);
   if (db)
     client_flag|=CLIENT_CONNECT_WITH_DB;
 

mysql-test/r/mysqld--help-notwin-profiling.result

@@ -532,6 +532,10 @@ The following options may be given as the first argument:
  max_relay_log_size is 0
  --max-binlog-stmt-cache-size=# 
  Sets the total size of the statement cache
+ --max-compressed-event-cache-size[=#] 
+ Max size of the compressed event cache in MB, this makes
+ sense only when some connected slaves are using
+ compressed event protocol.
  --max-connect-errors=# 
  If there is more than this number of interrupted
  connections from a host this host will be blocked from
@@ -1440,6 +1444,8 @@ The following options may be given as the first argument:
  Gather workers' activities to Update progress status of
  Multi-threaded slave and flush the relay log info to disk
  after every #th milli-seconds.
+ --slave-compressed-event-protocol 
+ Use event compression on master/slave protocol
  --slave-compressed-protocol 
  Use compression on master/slave protocol
  --slave-compression-lib[=name] 
@@ -1824,6 +1830,7 @@ max-binlog-cache-size 18446744073709547520
 max-binlog-dump-events 0
 max-binlog-size 1073741824
 max-binlog-stmt-cache-size 18446744073709547520
+max-compressed-event-cache-size 1
 max-connect-errors 100
 max-connections 151
 max-delayed-threads 20
@@ -2116,6 +2123,7 @@ skip-slave-start FALSE
 slave-allow-batching FALSE
 slave-checkpoint-group 512
 slave-checkpoint-period 300
+slave-compressed-event-protocol FALSE
 slave-compressed-protocol FALSE
 slave-compression-lib zlib
 slave-exec-mode STRICT

mysql-test/r/mysqld--help-notwin.result

@@ -532,6 +532,10 @@ The following options may be given as the first argument:
  max_relay_log_size is 0
  --max-binlog-stmt-cache-size=# 
  Sets the total size of the statement cache
+ --max-compressed-event-cache-size[=#] 
+ Max size of the compressed event cache in MB, this makes
+ sense only when some connected slaves are using
+ compressed event protocol.
  --max-connect-errors=# 
  If there is more than this number of interrupted
  connections from a host this host will be blocked from
@@ -1438,6 +1442,8 @@ The following options may be given as the first argument:
  Gather workers' activities to Update progress status of
  Multi-threaded slave and flush the relay log info to disk
  after every #th milli-seconds.
+ --slave-compressed-event-protocol 
+ Use event compression on master/slave protocol
  --slave-compressed-protocol 
  Use compression on master/slave protocol
  --slave-compression-lib[=name] 
@@ -1822,6 +1828,7 @@ max-binlog-cache-size 18446744073709547520
 max-binlog-dump-events 0
 max-binlog-size 1073741824
 max-binlog-stmt-cache-size 18446744073709547520
+max-compressed-event-cache-size 1
 max-connect-errors 100
 max-connections 151
 max-delayed-threads 20
@@ -2113,6 +2120,7 @@ skip-slave-start FALSE
 slave-allow-batching FALSE
 slave-checkpoint-group 512
 slave-checkpoint-period 300
+slave-compressed-event-protocol FALSE
 slave-compressed-protocol FALSE
 slave-compression-lib zlib
 slave-exec-mode STRICT