diff --git a/mysql-test/include/wait_for_engine_binlog.inc b/mysql-test/include/wait_for_engine_binlog.inc
new file mode 100644
index 00000000000..040d1de6279
--- /dev/null
+++ b/mysql-test/include/wait_for_engine_binlog.inc
@@ -0,0 +1,77 @@
+# include/wait_for_engine_binlog.inc
+#
+# SUMMARY
+#
+#    Waits until a specific (engine-implemented) binlog file is seen with
+#    the specified size in SHOW BINARY LOGS.
+#    Used to avoid sporadic failures due to races with the binlog
+#    pre-allocation thread.
+#
+# USAGE
+#
+#    --let $binlog_name= binlog-000002.ibb
+#    --let $binlog_size= 262144
+#    --source include/wait_for_engine_binlog.inc
+#
+#   OPTIONALLY:
+#
+#    let $wait_timeout= 60; # Override default 30 seconds with 60.
+#    let $wait_notfound= 1; # Wait until specified binlog _not_ found.
+#
+# EXAMPLE
+#    binlog_in_engine.binlog_flush_purge.test
+#
+
+--disable_query_log
+
+let $_wait_counter= 300;
+if ($wait_timeout)
+{
+  let $_wait_counter= `SELECT $wait_timeout * 10`;
+}
+# Reset $wait_timeout so that its value won't be used on subsequent
+# calls, and default will be used instead.
+let $wait_timeout= 0;
+
+--let $_expect= 1
+--let $_message= exist
+if ($wait_notfound) {
+  --let $_expect= 0
+  --let $_message= no longer exist
+}
+let $wait_notfound= 0;
+
+--let $_done= 0
+--let $_i= 0
+while (!$_done) {
+  --let $_j= 1
+  --let $_end= 0
+  --let $_found= 0
+  while (!$_end) {
+    --let $_x= query_get_value(SHOW BINARY LOGS, Log_name, $_j)
+    if ($_x == No such row) {
+      --let $_end= 1
+    }
+    if ($_x == $binlog_name) {
+      --let $_y= query_get_value(SHOW BINARY LOGS, File_size, $_j)
+      if ($_y == $binlog_size) {
+        --let $_found= 1
+	--let $_end= 1
+      }
+    }
+    inc $_j;
+  }
+  if ($_found == $_expect) {
+    --let $_done= 1
+  }
+  if (!$_done) {
+    sleep 0.1;
+    inc $_i;
+    if ($_i >= $_wait_counter) {
+      SHOW BINARY LOGS;
+      --die Timeout waiting for binlog '$binlog_name' to $_message
+    }
+  }
+}
+
+--enable_query_log
diff --git a/mysql-test/suite/binlog_in_engine/binlog_flush_purge.result b/mysql-test/suite/binlog_in_engine/binlog_flush_purge.result
new file mode 100644
index 00000000000..c5aeb26924d
--- /dev/null
+++ b/mysql-test/suite/binlog_in_engine/binlog_flush_purge.result
@@ -0,0 +1,132 @@
+RESET MASTER;
+CREATE TABLE t1 (a INT PRIMARY KEY) ENGINE=InnoDB;
+CREATE TABLE t2 (a INT PRIMARY KEY, b VARCHAR(2048)) ENGINE=InnoDB;
+INSERT INTO t1 VALUES (1);
+BEGIN;
+INSERT INTO t1 VALUES (2);
+INSERT INTO t1 VALUES (3);
+COMMIT;
+INSERT INTO t2 VALUES (0, REPEAT("x", 2048));
+INSERT INTO t2 SELECT a+1, b FROM t2;
+INSERT INTO t2 SELECT a+2, b FROM t2;
+INSERT INTO t2 SELECT a+4, b FROM t2;
+INSERT INTO t2 SELECT a+8, b FROM t2;
+SHOW BINARY LOGS;
+Log_name	File_size
+binlog-000000.ibb	262144
+binlog-000001.ibb	262144
+FLUSH BINARY LOGS;
+SHOW BINARY LOGS;
+Log_name	File_size
+binlog-000000.ibb	49152
+binlog-000001.ibb	262144
+binlog-000002.ibb	262144
+RESET MASTER;
+SHOW BINARY LOGS;
+Log_name	File_size
+binlog-000000.ibb	262144
+binlog-000001.ibb	262144
+INSERT INTO t1 VALUES (100);
+INSERT INTO t2 VALUES (100, 'xyzzy');
+DROP TABLE t1, t2;
+CREATE TABLE t1 (a INT PRIMARY KEY, b VARCHAR(2048)) ENGINE=InnoDB;
+SET @old_min_slaves= @@GLOBAL.slave_connections_needed_for_purge;
+SET GLOBAL slave_connections_needed_for_purge= 1;
+PURGE BINARY LOGS TO 'binlog-000001.ibb';
+ERROR HY000: A purgeable log is in use, will not purge
+SHOW WARNINGS;
+Level	Code	Message
+Note	1375	Binary log 'binlog-000000.ibb' is not purged because less than 'slave_connections_needed_for_purge' slaves have processed it
+Error	1378	A purgeable log is in use, will not purge
+SET GLOBAL slave_connections_needed_for_purge= 0;
+PURGE BINARY LOGS TO 'binlog-000001.ibb';
+ERROR HY000: A purgeable log is in use, will not purge
+SHOW WARNINGS;
+Level	Code	Message
+Note	1375	Binary log 'binlog-000000.ibb' is not purged because the binlog file is in active use
+Error	1378	A purgeable log is in use, will not purge
+SET @old_max_total= @@GLOBAL.max_binlog_total_size;
+SET GLOBAL max_binlog_total_size= 4*@@GLOBAL.max_binlog_size;
+SET SESSION binlog_format= ROW;
+*** Do 1500 transactions ...
+SHOW BINARY LOGS;
+Log_name	File_size
+binlog-000011.ibb	262144
+binlog-000012.ibb	262144
+binlog-000013.ibb	262144
+binlog-000014.ibb	262144
+*** Test purge by date.
+SET GLOBAL max_binlog_total_size= 0;
+SET @old_expire= @@GLOBAL.binlog_expire_logs_seconds;
+SET GLOBAL binlog_expire_logs_seconds= 1;
+*** Do 187 inserts ...
+SET GLOBAL binlog_expire_logs_seconds= 0;
+*** Do 1000 transactions ...
+SHOW BINARY LOGS;
+Log_name	File_size
+binlog-000013.ibb	262144
+binlog-000014.ibb	262144
+binlog-000015.ibb	262144
+binlog-000016.ibb	262144
+binlog-000017.ibb	262144
+binlog-000018.ibb	262144
+binlog-000019.ibb	262144
+binlog-000020.ibb	262144
+binlog-000021.ibb	262144
+binlog-000022.ibb	262144
+binlog-000023.ibb	262144
+binlog-000024.ibb	262144
+SET @now= NOW();
+*** Do 187 inserts ...
+PURGE BINARY LOGS BEFORE @now;
+SHOW BINARY LOGS;
+Log_name	File_size
+binlog-000023.ibb	262144
+binlog-000024.ibb	262144
+binlog-000025.ibb	262144
+*** Test PURGE BINARY LOGS TO
+PURGE BINARY LOGS TO 'binlog-000025.ibb';
+ERROR HY000: A purgeable log is in use, will not purge
+SHOW WARNINGS;
+Level	Code	Message
+Note	1375	Binary log 'binlog-000024.ibb' is not purged because the binlog file is in active use
+Error	1378	A purgeable log is in use, will not purge
+SHOW BINARY LOGS;
+Log_name	File_size
+binlog-000024.ibb	262144
+binlog-000025.ibb	262144
+*** Do 436 inserts ...
+SHOW BINARY LOGS;
+Log_name	File_size
+binlog-000024.ibb	262144
+binlog-000025.ibb	262144
+binlog-000026.ibb	262144
+binlog-000027.ibb	262144
+binlog-000028.ibb	262144
+binlog-000029.ibb	262144
+PURGE BINARY LOGS TO 'binlog-000025.ibb';
+SHOW BINARY LOGS;
+Log_name	File_size
+binlog-000025.ibb	262144
+binlog-000026.ibb	262144
+binlog-000027.ibb	262144
+binlog-000028.ibb	262144
+binlog-000029.ibb	262144
+PURGE BINARY LOGS TO 'binlog-999999.ibb';
+ERROR HY000: Target log not found in binlog index
+SHOW WARNINGS;
+Level	Code	Message
+Error	1373	Target log not found in binlog index
+*** Test purging logs when setting the maximum size.
+SET GLOBAL max_binlog_total_size= ceil(1.5*@@GLOBAL.max_binlog_size);
+Warnings:
+Note	1375	Binary log 'binlog-000028.ibb' is not purged because the binlog file is in active use
+SHOW BINARY LOGS;
+Log_name	File_size
+binlog-000028.ibb	262144
+binlog-000029.ibb	262144
+SET SESSION binlog_format= MIXED;
+DROP TABLE t1;
+SET GLOBAL max_binlog_total_size= @old_max_total;
+SET GLOBAL slave_connections_needed_for_purge= @old_min_slaves;
+SET GLOBAL binlog_expire_logs_seconds= @old_expire;
diff --git a/mysql-test/suite/binlog_in_engine/binlog_flush_purge.test b/mysql-test/suite/binlog_in_engine/binlog_flush_purge.test
index f8b74f56992..f6c0a77fba9 100644
--- a/mysql-test/suite/binlog_in_engine/binlog_flush_purge.test
+++ b/mysql-test/suite/binlog_in_engine/binlog_flush_purge.test
@@ -16,15 +16,20 @@ INSERT INTO t2 SELECT a+2, b FROM t2;
 INSERT INTO t2 SELECT a+4, b FROM t2;
 INSERT INTO t2 SELECT a+8, b FROM t2;
 
+--let $binlog_name= binlog-000001.ibb
+--let $binlog_size= 262144
+--source include/wait_for_engine_binlog.inc
 SHOW BINARY LOGS;
 FLUSH BINARY LOGS;
+--let $binlog_name= binlog-000002.ibb
+--let $binlog_size= 262144
+--source include/wait_for_engine_binlog.inc
 SHOW BINARY LOGS;
 
 RESET MASTER;
-# ToDo: This will be racy, the second binlog file binlog-000001.ibb is
-# pre-allocated in the background, so will be visible or not depending on exact
-# timing. So just omit this SHOW BINARY LOGS or wait for both to be created
-# or something.
+--let $binlog_name= binlog-000001.ibb
+--let $binlog_size= 262144
+--source include/wait_for_engine_binlog.inc
 SHOW BINARY LOGS;
 
 INSERT INTO t1 VALUES (100);
@@ -32,4 +37,153 @@ INSERT INTO t2 VALUES (100, 'xyzzy');
 
 DROP TABLE t1, t2;
 
---exec $MYSQL_BINLOG --read-from-remote-server --user=root --host=127.0.0.1 --port=$MASTER_MYPORT master-bin.000001 --start-position=0-1-1 > $MYSQLTEST_VARDIR/tmp/mysqlbinlog.txt
+# Test purge by size
+CREATE TABLE t1 (a INT PRIMARY KEY, b VARCHAR(2048)) ENGINE=InnoDB;
+
+SET @old_min_slaves= @@GLOBAL.slave_connections_needed_for_purge;
+SET GLOBAL slave_connections_needed_for_purge= 1;
+--error ER_LOG_IN_USE
+PURGE BINARY LOGS TO 'binlog-000001.ibb';
+SHOW WARNINGS;
+SET GLOBAL slave_connections_needed_for_purge= 0;
+--error ER_LOG_IN_USE
+PURGE BINARY LOGS TO 'binlog-000001.ibb';
+SHOW WARNINGS;
+SET @old_max_total= @@GLOBAL.max_binlog_total_size;
+SET GLOBAL max_binlog_total_size= 4*@@GLOBAL.max_binlog_size;
+SET SESSION binlog_format= ROW;
+--let $num_trx= 1500
+--echo *** Do $num_trx transactions ...
+--disable_query_log
+--let $i= 0
+while ($i < $num_trx) {
+  eval INSERT INTO t1 VALUES ($i+100000, REPEAT("x", 2048));
+  inc $i;
+}
+--enable_query_log
+# The precise point at which we move to the next binlog file depends on the
+# exact size of binlogged transactions, which might change as server code is
+# developed, and then this test will fail with a different set of binlog files
+# appearing in SHOW BINARY LOGS.
+#
+# In this case, just check that the general structure of the present binlogs
+# is similar, and then update the $binlog_name waited for and the .result file.
+--let $binlog_name= binlog-000010.ibb
+--let $binlog_size= 262144
+--let $wait_notfound= 1
+--source include/wait_for_engine_binlog.inc
+--let $binlog_name= binlog-000014.ibb
+--let $binlog_size= 262144
+--source include/wait_for_engine_binlog.inc
+SHOW BINARY LOGS;
+
+--echo *** Test purge by date.
+SET GLOBAL max_binlog_total_size= 0;
+SET @old_expire= @@GLOBAL.binlog_expire_logs_seconds;
+SET GLOBAL binlog_expire_logs_seconds= 1;
+--sleep 2
+--let $num_insert= `SELECT floor(256*1.5*1024/2100)`
+--echo *** Do $num_insert inserts ...
+--disable_query_log
+BEGIN;
+--let $i= 0
+while ($i < $num_insert) {
+  eval INSERT INTO t1 VALUES ($i+200000, REPEAT("x", 2048));
+  inc $i;
+}
+COMMIT;
+--enable_query_log
+--let $binlog_name= binlog-000012.ibb
+--let $binlog_size= 262144
+--let $wait_notfound= 1
+--source include/wait_for_engine_binlog.inc
+# SHOW BINARY LOGS here will not be stable.
+# We can wait for the log before the --sleep 2 to expire.
+# But the next log might also expire, if there is a random delay sufficiently
+# long before the automatic purge runs.
+#SHOW BINARY LOGS;
+
+SET GLOBAL binlog_expire_logs_seconds= 0;
+
+--let $num_trx= 1000
+--echo *** Do $num_trx transactions ...
+--disable_query_log
+--let $i= 0
+while ($i < $num_trx) {
+  eval INSERT INTO t1 VALUES ($i+300000, REPEAT("x", 2048));
+  inc $i;
+}
+--enable_query_log
+
+--let $binlog_name= binlog-000024.ibb
+--let $binlog_size= 262144
+--source include/wait_for_engine_binlog.inc
+SHOW BINARY LOGS;
+--sleep 1
+SET @now= NOW();
+--sleep 1
+--let $num_insert= `SELECT floor(256*1.5*1024/2100)`
+--echo *** Do $num_insert inserts ...
+--disable_query_log
+BEGIN;
+--let $i= 0
+while ($i < $num_insert) {
+  eval INSERT INTO t1 VALUES ($i+400000, REPEAT("x", 2048));
+  inc $i;
+}
+COMMIT;
+--enable_query_log
+PURGE BINARY LOGS BEFORE @now;
+--let $binlog_name= binlog-000022.ibb
+--let $binlog_size= 262144
+--let $wait_notfound= 1
+--source include/wait_for_engine_binlog.inc
+--let $binlog_name= binlog-000025.ibb
+--let $binlog_size= 262144
+--source include/wait_for_engine_binlog.inc
+SHOW BINARY LOGS;
+
+--echo *** Test PURGE BINARY LOGS TO
+--let $current= query_get_value(SHOW BINARY LOGS, Log_name, 3)
+--error ER_LOG_IN_USE
+eval PURGE BINARY LOGS TO '$current';
+SHOW WARNINGS;
+SHOW BINARY LOGS;
+--let $num_insert= `SELECT floor(256*3.5*1024/2100)`
+--echo *** Do $num_insert inserts ...
+--disable_query_log
+BEGIN;
+--let $i= 0
+while ($i < $num_insert) {
+  eval INSERT INTO t1 VALUES ($i+500000, REPEAT("x", 2048));
+  inc $i;
+}
+COMMIT;
+--enable_query_log
+--source include/wait_for_engine_binlog.inc
+--let $binlog_name= binlog-000029.ibb
+--let $binlog_size= 262144
+--source include/wait_for_engine_binlog.inc
+SHOW BINARY LOGS;
+eval PURGE BINARY LOGS TO '$current';
+--let $binlog_name= binlog-000024.ibb
+--let $binlog_size= 262144
+--let $wait_notfound= 1
+--source include/wait_for_engine_binlog.inc
+SHOW BINARY LOGS;
+
+
+--error ER_UNKNOWN_TARGET_BINLOG
+PURGE BINARY LOGS TO 'binlog-999999.ibb';
+SHOW WARNINGS;
+
+--echo *** Test purging logs when setting the maximum size.
+SET GLOBAL max_binlog_total_size= ceil(1.5*@@GLOBAL.max_binlog_size);
+SHOW BINARY LOGS;
+
+SET SESSION binlog_format= MIXED;
+DROP TABLE t1;
+
+SET GLOBAL max_binlog_total_size= @old_max_total;
+SET GLOBAL slave_connections_needed_for_purge= @old_min_slaves;
+SET GLOBAL binlog_expire_logs_seconds= @old_expire;
diff --git a/sql/handler.h b/sql/handler.h
index 54696df4455..ab3e0a3ea7f 100644
--- a/sql/handler.h
+++ b/sql/handler.h
@@ -64,6 +64,7 @@ struct rpl_gtid;
 struct slave_connection_state;
 struct rpl_binlog_state_base;
 struct handler_binlog_event_group_info;
+struct handler_binlog_purge_info;
 
 // the following is for checking tables
 
@@ -1572,6 +1573,13 @@ struct handlerton
   bool (*binlog_flush)();
   /* Engine implementation of RESET MASTER. */
   bool (*reset_binlogs)();
+  /*
+    Engine implementation of PURGE BINARY LOGS.
+    Return 0 for ok or one of LOG_INFO_* errors.
+
+    See also ha_binlog_purge_info() for auto-purge.
+  */
+  int (*binlog_purge)(handler_binlog_purge_info *purge_info);
 
    /*
      Optional clauses in the CREATE/ALTER TABLE
@@ -5865,6 +5873,16 @@ struct handler_binlog_event_group_info {
   Class for reading a binlog implemented in an engine.
 */
 class handler_binlog_reader {
+public:
+  /*
+    Approximate current position (from which next call to read_binlog_data()
+    will need to read). Updated by the engine. Used to know which binlog files
+    the active dump threads are currently reading from, to avoid purging
+    actively used binlogs.
+  */
+  uint64_t cur_file_no;
+  uint64_t cur_file_pos;
+
 private:
   /* Position and length of any remaining data in buf[]. */
   uint32_t buf_data_pos;
@@ -5874,15 +5892,59 @@ private:
 
 public:
   handler_binlog_reader()
-    : buf_data_pos(0), buf_data_remain(0)
+    : cur_file_no(~(uint64_t)0), cur_file_pos(~(uint64_t)0),
+      buf_data_pos(0), buf_data_remain(0)
   { }
   virtual ~handler_binlog_reader() { };
   virtual int read_binlog_data(uchar *buf, uint32_t len) = 0;
   virtual bool data_available()= 0;
+  /*
+    This initializes the current read position to the point of the slave GTID
+    position passed in as POS. It is permissible to start at a position a bit
+    earlier in the binlog, only cost is the extra read cost of reading not
+    needed event data.
+
+    If position is found, must return the corresponding binlog state in the
+    STATE output parameter and initialize cur_file_no and cur_file_pos members.
+
+    Returns:
+     -1  Error
+      0  The requested GTID position not found, needed binlogs have been purged
+      1  Ok, position found and returned.
+  */
   virtual int init_gtid_pos(slave_connection_state *pos,
                             rpl_binlog_state_base *state) = 0;
 
   int read_log_event(String *packet, uint32_t ev_offset, size_t max_allowed);
 };
 
+
+/* Structure returned by ha_binlog_purge_info(). */
+struct handler_binlog_purge_info {
+  /* The earliest binlog file that is in use by a dump thread. */
+  uint64_t limit_file_no;
+  /*
+    Set by engine to give a reason why a requested purge could not be done.
+    If set, then nonpurge_filename should be set to the filename.
+
+    Also set by ha_binlog_purge_info() when it returns false, to the reason
+    why no purge is possible. In this case, the nonpurge_filename is set
+    to the empty string.
+  */
+  const char *nonpurge_reason;
+  /* The user-configured maximum total size of the binlog. */
+  ulonglong limit_size;
+  /* Binlog name, for PURGE BINARY LOGS TO. */
+  const char *limit_name;
+  /* The earliest file date (unix timestamp) that should not be purged. */
+  time_t limit_date;
+  /* Whether purge by date and/or by size and/or name is requested. */
+  bool purge_by_date, purge_by_size, purge_by_name;
+  /*
+    The name of the file that could not be purged, when nonpurge_reason
+    is given.
+  */
+  char nonpurge_filename[FN_REFLEN];
+};
+
 #endif /* HANDLER_INCLUDED */
diff --git a/sql/log.cc b/sql/log.cc
index adae5ecd1bf..77f3cc47997 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -5326,8 +5326,8 @@ int MYSQL_BIN_LOG::purge_index_entry(THD *thd, ulonglong *reclaimed_space,
     }
     else
     {
-      if (unlikely((error= find_log_pos(&check_log_info,
-                                        log_info.log_file_name, need_mutex))))
+      if (likely((error= find_log_pos(&check_log_info,
+                                      log_info.log_file_name, need_mutex))))
       {
         if (error != LOG_INFO_EOF)
         {
@@ -5600,6 +5600,40 @@ err:
   DBUG_RETURN(error);
 }
 
+
+void
+MYSQL_BIN_LOG::engine_purge_logs_by_size(ulonglong max_total_size)
+{
+  DBUG_ASSERT(opt_binlog_engine_hton);
+  if (!is_open())
+    return;
+
+  handler_binlog_purge_info purge_info;
+  auto p= engine_binlog_in_use();
+  purge_info.limit_file_no= p.first;
+  uint num_dump_threads= p.second;
+  if (num_dump_threads < slave_connections_needed_for_purge)
+  {
+    purge_info.limit_file_no= 0;
+    purge_info.nonpurge_reason= "less than "
+      "'slave_connections_needed_for_purge' slaves have processed it";
+  }
+  else
+    purge_info.nonpurge_reason= nullptr;
+  purge_info.nonpurge_filename[0]= '\0';
+  purge_info.purge_by_date= false;
+  purge_info.limit_date= my_time(0);
+  purge_info.purge_by_size= true;
+  purge_info.limit_size= max_total_size;
+  purge_info.purge_by_name= false;
+  purge_info.limit_name= nullptr;
+  int res= (*opt_binlog_engine_hton->binlog_purge)(&purge_info);
+  if (res && purge_info.nonpurge_reason)
+    give_purge_note(purge_info.nonpurge_reason,
+                    purge_info.nonpurge_filename, true);
+}
+
+
 /*
   @param  log_file_name_arg  Name of log file to check
   @param  interactive        True if called by a PURGE BINLOG command.
@@ -5630,6 +5664,7 @@ MYSQL_BIN_LOG::can_purge_log(const char *log_file_name_arg,
   int res;
   const char *reason;
 
+  DBUG_ASSERT(is_relay_log || !opt_binlog_engine_hton);
   if (is_active(log_file_name_arg) ||
       (!is_relay_log && waiting_for_slave_to_change_binlog &&
        purge_sending_new_binlog_file == sending_new_binlog_file &&
@@ -5693,23 +5728,39 @@ error:
 
     /* purge_warning_given is reset after next sucessful purge */
     purge_warning_given= 1;
-    if (interactive)
-    {
-      my_printf_error(ER_BINLOG_PURGE_PROHIBITED,
-                      "Binary log '%s' is not purged because %s",
-                      MYF(ME_NOTE), log_file_name_arg, reason);
-    }
-    else
-    {
-      sql_print_information("Binary log '%s' is not purged because %s",
-                            log_file_name_arg, reason);
-    }
+    give_purge_note(reason, log_file_name_arg, interactive);
   }
   return 0;
 }
 #endif /* HAVE_REPLICATION */
 
 
+void
+give_purge_note(const char *reason, const char *file_name, bool interactive)
+{
+  if (interactive)
+  {
+    if (file_name && file_name[0])
+      my_printf_error(ER_BINLOG_PURGE_PROHIBITED,
+                      "Binary log '%s' is not purged because %s",
+                      MYF(ME_NOTE), file_name, reason);
+    else
+      my_printf_error(ER_BINLOG_PURGE_PROHIBITED,
+                      "Binary log purge is prevented because %s",
+                      MYF(ME_NOTE), reason);
+  }
+  else
+  {
+    if (file_name && file_name[0])
+      sql_print_information("Binary log '%s' is not purged because %s",
+                            file_name, reason);
+    else
+      sql_print_information("Binary log purge is prevented because %s",
+                            reason);
+  }
+}
+
+
 /**
   Count a total size of binary logs (except the active one) to the variable
   binlog_space_total.
@@ -5727,6 +5778,7 @@ int MYSQL_BIN_LOG::count_binlog_space()
   LOG_INFO log_info;
   DBUG_ENTER("count_binlog_space");
 
+  DBUG_ASSERT(!opt_binlog_engine_hton);
   binlog_space_total = 0;
   if ((error= find_log_pos(&log_info, NullS, false /*need_lock_index=false*/)))
     goto done;
diff --git a/sql/log.h b/sql/log.h
index dc461ea9173..d1a59ba8154 100644
--- a/sql/log.h
+++ b/sql/log.h
@@ -28,6 +28,8 @@ class Gtid_log_event;
 
 bool reopen_fstreams(const char *filename, FILE *outstream, FILE *errstream);
 void setup_log_handling();
+void give_purge_note(const char *reason, const char *file_name,
+                     bool interactive);
 bool trans_has_updated_trans_table(const THD* thd);
 bool stmt_has_updated_trans_table(const THD *thd);
 bool use_trans_cache(const THD* thd, bool is_transactional);
@@ -266,12 +268,14 @@ class Relay_log_info;
  */
 typedef struct st_log_info
 {
+  /* file_no only used when --binlog-storage-engine set. */
+  std::atomic<uint64_t> file_no;
+  /* log_file_name and *_offset only used when --binlog-storage-engine unset. */
   char log_file_name[FN_REFLEN];
   my_off_t index_file_offset, index_file_start_offset;
   my_off_t pos;
-  bool fatal; // if the purge happens to give us a negative offset
-  st_log_info() : index_file_offset(0), index_file_start_offset(0),
-      pos(0), fatal(0)
+  st_log_info() : file_no(~(uint64_t)0), index_file_offset(0),
+                  index_file_start_offset(0), pos(0)
   {
     DBUG_ENTER("LOG_INFO");
     log_file_name[0] = '\0';
@@ -1084,6 +1088,7 @@ public:
       return 0;
     return real_purge_logs_by_size(binlog_pos);
   }
+  void engine_purge_logs_by_size(ulonglong max_total_size);
   int set_purge_index_file_name(const char *base_file_name);
   int open_purge_index_file(bool destroy);
   bool truncate_and_remove_binlogs(const char *truncate_file,
diff --git a/sql/sql_repl.cc b/sql/sql_repl.cc
index 304bc67350f..2da28dc553b 100644
--- a/sql/sql_repl.cc
+++ b/sql/sql_repl.cc
@@ -612,9 +612,7 @@ static my_bool adjust_callback(THD *thd, my_off_t *purge_offset)
       we just started reading the index file. In that case
       we have nothing to adjust
     */
-    if (linfo->index_file_offset < *purge_offset)
-      linfo->fatal= (linfo->index_file_offset != 0);
-    else
+    if (linfo->index_file_offset >= *purge_offset)
       linfo->index_file_offset-= *purge_offset;
   }
   mysql_mutex_unlock(&thd->LOCK_thd_data);
@@ -654,7 +652,7 @@ static my_bool log_in_use_callback(THD *thd, st_log_in_use *arg)
 
 
 /*
-  Check if a log is in use.
+  Check if a log is in use (legacy binlog).
 
   @return 0  Not used
   @return 1  A slave is reading from the log
@@ -676,6 +674,94 @@ int log_in_use(const char* log_name, uint min_connected)
 }
 
 
+struct st_engine_binlog_in_use {
+  uint64_t min_file_no;
+  uint count;
+};
+
+
+my_bool
+engine_binlog_in_use_callback(THD *thd, st_engine_binlog_in_use *arg)
+{
+  if (thd->current_linfo)
+  {
+    mysql_mutex_lock(&thd->LOCK_thd_data);
+    if (LOG_INFO *linfo= thd->current_linfo)
+    {
+      uint64_t file_no= linfo->file_no.load(std::memory_order_relaxed);
+      if (file_no < arg->min_file_no)
+        arg->min_file_no= file_no;
+      if (file_no != ~(uint64_t)0)
+        ++arg->count;
+    }
+    mysql_mutex_unlock(&thd->LOCK_thd_data);
+  }
+  return FALSE;
+}
+
+
+/*
+  Find earliest binlog file in use (--binlog-storage-engine).
+
+  Returns a pair of the earliest file_no binlog in use by a dump thread,
+  and the number of actively running dump threads.
+*/
+std::pair<uint64_t, uint>
+engine_binlog_in_use()
+{
+  DBUG_ASSERT(opt_binlog_engine_hton);
+  st_engine_binlog_in_use arg{~(uint64_t)0, 0};
+  server_threads.iterate(engine_binlog_in_use_callback, &arg);
+  return {arg.min_file_no, arg.count};
+}
+
+
+/*
+  Inform engine about server state relevant for automatic binlog purge.
+  Used by engines that implement --binlog-storage-engine.
+
+  Returns true if automatic purge should proceed with supplied information,
+  false if automatic purge is disabled due to
+  --slave-connections-needed-for-purge.
+*/
+bool
+ha_binlog_purge_info(handler_binlog_purge_info *out_info)
+{
+  auto p= engine_binlog_in_use();
+  out_info->limit_file_no= p.first;
+  uint num_dump_threads= p.second;
+  out_info->purge_by_name= false;
+  out_info->limit_name= nullptr;
+  if (binlog_expire_logs_seconds)
+  {
+    out_info->purge_by_date= true;
+    out_info->limit_date= my_time(0) - binlog_expire_logs_seconds;
+  }
+  else
+    out_info->purge_by_date= false;
+  if (binlog_space_limit)
+  {
+    out_info->purge_by_size= true;
+    out_info->limit_size= binlog_space_limit;
+  }
+  else
+    out_info->purge_by_size= false;
+
+  out_info->nonpurge_filename[0]= '\0';
+  if (num_dump_threads >= slave_connections_needed_for_purge)
+  {
+    out_info->nonpurge_reason= nullptr;
+    return true;
+  }
+  else
+  {
+    out_info->nonpurge_reason= "less than 'slave_connections_needed_for_purge' "
+      "slaves have processed it";
+    return false;
+  }
+}
+
+
 bool purge_error_message(THD* thd, int res)
 {
   uint errcode;
@@ -703,17 +789,53 @@ bool purge_error_message(THD* thd, int res)
 */
 bool purge_master_logs(THD* thd, const char* to_log)
 {
-  char search_file_name[FN_REFLEN];
   if (!mysql_bin_log.is_open())
   {
     my_ok(thd);
     return FALSE;
   }
 
-  mysql_bin_log.make_log_name(search_file_name, to_log);
-  return purge_error_message(thd,
-			     mysql_bin_log.purge_logs(thd, search_file_name,
-                                                      0, 1, 1, 1, NULL));
+  int res;
+  if (!opt_binlog_engine_hton)
+  {
+    char search_file_name[FN_REFLEN];
+    mysql_bin_log.make_log_name(search_file_name, to_log);
+    res= mysql_bin_log.purge_logs(thd, search_file_name, 0, 1, 1, 1, NULL);
+  }
+  else
+  {
+    handler_binlog_purge_info purge_info;
+    auto p= engine_binlog_in_use();
+    purge_info.limit_file_no= p.first;
+    uint num_dump_threads= p.second;
+    if (num_dump_threads < slave_connections_needed_for_purge)
+    {
+      /*
+        Prevent purging any file.
+        We need to do it this way, since we have to call into the engine to let
+        it check if there are any files to potentially purge. If there are, we
+        want to give an error that purge was not possible. But if there were no
+        files to purge in any case, we do not want to give any error.
+      */
+      purge_info.limit_file_no= 0;
+      purge_info.nonpurge_reason= "less than "
+        "'slave_connections_needed_for_purge' slaves have processed it";
+    }
+    else
+      purge_info.nonpurge_reason= nullptr;
+    purge_info.nonpurge_filename[0]= '\0';
+    purge_info.purge_by_date= false;
+    purge_info.limit_date= (time_t)0;
+    purge_info.purge_by_size= false;
+    purge_info.limit_size= 0;
+    purge_info.purge_by_name= true;
+    purge_info.limit_name= to_log;
+    res= (*opt_binlog_engine_hton->binlog_purge)(&purge_info);
+    if (res && purge_info.nonpurge_reason)
+      give_purge_note(purge_info.nonpurge_reason,
+                      purge_info.nonpurge_filename, true);
+  }
+  return purge_error_message(thd, res);
 }
 
 
@@ -735,10 +857,36 @@ bool purge_master_logs_before_date(THD* thd, time_t purge_time)
     my_ok(thd);
     return 0;
   }
-  return purge_error_message(thd,
-                             mysql_bin_log.purge_logs_before_date(thd,
-                                                                  purge_time,
-                                                                  1));
+  int res;
+  if (!opt_binlog_engine_hton)
+    res= mysql_bin_log.purge_logs_before_date(thd, purge_time, 1);
+  else
+  {
+    handler_binlog_purge_info purge_info;
+    auto p= engine_binlog_in_use();
+    purge_info.limit_file_no= p.first;
+    uint num_dump_threads= p.second;
+    if (num_dump_threads < slave_connections_needed_for_purge)
+    {
+      purge_info.limit_file_no= 0;
+      purge_info.nonpurge_reason= "less than "
+        "'slave_connections_needed_for_purge' slaves have processed it";
+    }
+    else
+      purge_info.nonpurge_reason= nullptr;
+    purge_info.nonpurge_filename[0]= '\0';
+    purge_info.purge_by_date= true;
+    purge_info.limit_date= purge_time;
+    purge_info.purge_by_size= false;
+    purge_info.limit_size= 0;
+    purge_info.purge_by_name= false;
+    purge_info.limit_name= nullptr;
+    res= (*opt_binlog_engine_hton->binlog_purge)(&purge_info);
+    if (res && purge_info.nonpurge_reason)
+      give_purge_note(purge_info.nonpurge_reason,
+                      purge_info.nonpurge_filename, true);
+  }
+  return purge_error_message(thd, res);
 }
 
 void set_read_error(binlog_send_info *info, int error)
@@ -3169,7 +3317,7 @@ static int send_events(binlog_send_info *info, IO_CACHE* log, LOG_INFO* linfo,
  * return 0 - OK
  *        else NOK
  */
-static int send_engine_events(binlog_send_info *info)
+static int send_engine_events(binlog_send_info *info, LOG_INFO* linfo)
 {
   int error;
   ulong ev_offset;
@@ -3191,6 +3339,10 @@ static int send_engine_events(binlog_send_info *info)
       set_read_error(info, error);
       return 1;
     }
+
+    linfo->file_no.store(reader->cur_file_no, std::memory_order_relaxed);
+    linfo->pos= (my_off_t) reader->cur_file_pos;
+
     if (error == LOG_READ_EOF)
     {
       PSI_stage_info old_stage;
@@ -3280,7 +3432,7 @@ static int send_one_binlog_file(binlog_send_info *info,
     if (opt_binlog_engine_hton)
     {
       info->dirlen= 0;
-      if (send_engine_events(info))
+      if (send_engine_events(info, linfo))
         return 1;
     }
     else
@@ -5105,7 +5257,8 @@ retry:
     cur_link->name.str+=    dir_len;
     cur_link->name.length-= dir_len;
 
-    if (mysql_bin_log.get_reset_master_count() > expected_reset_masters)
+    if (!opt_binlog_engine_hton &&
+        mysql_bin_log.get_reset_master_count() > expected_reset_masters)
     {
       /*
         Reset master was called after we cached filenames.
@@ -5115,7 +5268,8 @@ retry:
       goto retry;
     }
 
-    if (!(strncmp(fname+dir_len, cur.log_file_name+cur_dir_len, length)))
+    if (!opt_binlog_engine_hton &&
+        !(strncmp(fname+dir_len, cur.log_file_name+cur_dir_len, length)))
       cur_link->size= cur.pos;  /* The active log, use the active position */
     else
     {
diff --git a/sql/sql_repl.h b/sql/sql_repl.h
index c03384aa5a3..0e5f0f27d0a 100644
--- a/sql/sql_repl.h
+++ b/sql/sql_repl.h
@@ -39,6 +39,7 @@ int reset_master(THD* thd, rpl_gtid *init_state, uint32 init_state_len,
 bool purge_master_logs(THD* thd, const char* to_log);
 bool purge_master_logs_before_date(THD* thd, time_t purge_time);
 int log_in_use(const char* log_name, uint min_connections);
+std::pair<uint64_t, uint> engine_binlog_in_use();
 void adjust_linfo_offsets(my_off_t purge_offset);
 void show_binlogs_get_fields(THD *thd, List<Item> *field_list);
 bool show_binlogs(THD* thd);
diff --git a/sql/sys_vars.cc b/sql/sys_vars.cc
index 43cfcf90acc..8ee45c5b750 100644
--- a/sql/sys_vars.cc
+++ b/sql/sys_vars.cc
@@ -1266,14 +1266,22 @@ static bool update_binlog_space_limit(sys_var *, THD *,
 
   if (opt_bin_log)
   {
-    if (binlog_space_limit)
-      mysql_bin_log.count_binlog_space();
-    /* Inform can_purge_log() that it should do a recheck of log_in_use() */
-    sending_new_binlog_file++;
-    mysql_bin_log.unlock_index();
-    mysql_bin_log.purge(1);
-    mysql_mutex_lock(&LOCK_global_system_variables);
-    return 0;
+    if (opt_binlog_engine_hton)
+    {
+      if (loc_binlog_space_limit)
+        mysql_bin_log.engine_purge_logs_by_size(loc_binlog_space_limit);
+    }
+    else
+    {
+      if (loc_binlog_space_limit)
+        mysql_bin_log.count_binlog_space();
+      /* Inform can_purge_log() that it should do a recheck of log_in_use() */
+      sending_new_binlog_file++;
+      mysql_bin_log.unlock_index();
+      mysql_bin_log.purge(1);
+      mysql_mutex_lock(&LOCK_global_system_variables);
+      return 0;
+    }
   }
   mysql_bin_log.unlock_index();
   mysql_mutex_lock(&LOCK_global_system_variables);
diff --git a/storage/innobase/fsp/fsp_binlog.cc b/storage/innobase/fsp/fsp_binlog.cc
index 1e52396fb7b..51bee2f08cb 100644
--- a/storage/innobase/fsp/fsp_binlog.cc
+++ b/storage/innobase/fsp/fsp_binlog.cc
@@ -265,13 +265,13 @@ fsp_binlog_open(const char *file_name, pfs_os_file_t fh,
 @param[in]  file_no	 Index of the binlog tablespace
 @param[out] new_space	 The newly created tablespace
 @return DB_SUCCESS or error code */
-dberr_t fsp_binlog_tablespace_create(uint64_t file_no, fil_space_t **new_space)
+dberr_t fsp_binlog_tablespace_create(uint64_t file_no, uint32_t size_in_pages,
+                                     fil_space_t **new_space)
 {
 	pfs_os_file_t	fh;
 	bool		ret;
 
         *new_space= nullptr;
-	uint32_t size= innodb_binlog_size_in_pages;
 	if(srv_read_only_mode)
 		return DB_ERROR;
 
@@ -296,7 +296,8 @@ dberr_t fsp_binlog_tablespace_create(uint64_t file_no, fil_space_t **new_space)
 
 	/* We created the binlog file and now write it full of zeros */
 	if (!os_file_set_size(name, fh,
-			      os_offset_t{size} << srv_page_size_shift)) {
+			      os_offset_t{size_in_pages} << srv_page_size_shift)
+            ) {
 		sql_print_error("Unable to allocate file %s", name);
 		os_file_close(fh);
 		os_file_delete(innodb_data_file_key, name);
@@ -317,7 +318,8 @@ dberr_t fsp_binlog_tablespace_create(uint64_t file_no, fil_space_t **new_space)
 		return DB_ERROR;
 	}
 
-	fil_node_t* node = (*new_space)->add(name, fh, size, false, true);
+	fil_node_t* node = (*new_space)->add(name, fh, size_in_pages,
+                                             false, true);
 	node->find_metadata();
 	mysql_mutex_unlock(&fil_system.mutex);
 
@@ -575,6 +577,8 @@ fsp_binlog_flush()
   chunk_data_flush dummy_data;
   mtr_t mtr;
 
+  mysql_mutex_lock(&purge_binlog_mutex);
+
   mtr.start();
   mtr.x_lock_space(space);
   /*
@@ -588,6 +592,12 @@ fsp_binlog_flush()
   {
     mtr.trim_pages(page_id_t(space_id, page_no + 1));
     mtr.commit_shrink(*space, page_no + 1);
+
+    size_t reclaimed= (space->size - (page_no + 1)) << srv_page_size_shift;
+    if (UNIV_LIKELY(total_binlog_used_size >= reclaimed))
+      total_binlog_used_size-= reclaimed;
+    else
+      ut_ad(0);
   }
   else
     mtr.commit();
@@ -596,6 +606,8 @@ fsp_binlog_flush()
   while (buf_flush_list_space(space))
     ;
 
+  mysql_mutex_unlock(&purge_binlog_mutex);
+
   /*
     Now get a new GTID state record written to the next binlog tablespace.
     This ensures that the new state (in case of DELETE_DOMAIN_ID) will be
diff --git a/storage/innobase/handler/ha_innodb.cc b/storage/innobase/handler/ha_innodb.cc
index 2bddf3b5893..948f6405293 100644
--- a/storage/innobase/handler/ha_innodb.cc
+++ b/storage/innobase/handler/ha_innodb.cc
@@ -549,6 +549,7 @@ mysql_pfs_key_t	trx_sys_mutex_key;
 mysql_pfs_key_t	srv_threads_mutex_key;
 mysql_pfs_key_t	tpool_cache_mutex_key;
 mysql_pfs_key_t fsp_active_binlog_mutex_key;
+mysql_pfs_key_t fsp_purge_binlog_mutex_key;
 
 /* all_innodb_mutexes array contains mutexes that are
 performance schema instrumented if "UNIV_PFS_MUTEX"
@@ -4135,6 +4136,7 @@ static int innodb_init(void* p)
         innobase_hton->get_binlog_file_list= innodb_get_binlog_file_list;
         innobase_hton->binlog_flush= innodb_binlog_flush;
         innobase_hton->reset_binlogs= innodb_reset_binlogs;
+        innobase_hton->binlog_purge= innodb_binlog_purge;
 
 	innodb_remember_check_sysvar_funcs();
 
diff --git a/storage/innobase/handler/innodb_binlog.cc b/storage/innobase/handler/innodb_binlog.cc
index c387da3c71d..e163a3d253c 100644
--- a/storage/innobase/handler/innodb_binlog.cc
+++ b/storage/innobase/handler/innodb_binlog.cc
@@ -57,6 +57,32 @@ rpl_binlog_state_base binlog_diff_state;
 static std::thread binlog_prealloc_thr_obj;
 static bool prealloc_thread_end= false;
 
+/*
+  Mutex around purge operations, including earliest_binlog_file_no and
+  total_binlog_used_size.
+*/
+mysql_mutex_t purge_binlog_mutex;
+
+/* The earliest binlog tablespace file. Used in binlog purge. */
+static uint64_t earliest_binlog_file_no;
+
+/*
+  The total space in use by binlog tablespace files. Maintained in-memory to
+  not have to stat(2) every file for every new binlog tablespace allocated in
+  case of --max-binlog-total-size.
+
+  Initialized at server startup (and in RESET MASTER), and updated as binlog
+  files are pre-allocated and purged.
+*/
+size_t total_binlog_used_size;
+
+static bool purge_warning_given= false;
+
+
+#ifdef UNIV_PFS_THREAD
+mysql_pfs_key_t binlog_prealloc_thread_key;
+#endif
+
 
 /* Structure holding context for out-of-band chunks of binlogged event group. */
 struct binlog_oob_context {
@@ -375,8 +401,8 @@ private:
 
 
 struct found_binlogs {
-  uint64_t last_file_no, prev_file_no;
-  size_t last_size, prev_size;
+  uint64_t last_file_no, prev_file_no, earliest_file_no;
+  size_t last_size, prev_size, total_size;
   int found_binlogs;
 };
 
@@ -384,6 +410,7 @@ struct found_binlogs {
 static void innodb_binlog_prealloc_thread();
 static int innodb_binlog_discover();
 static bool binlog_state_recover();
+static void innodb_binlog_autopurge(uint64_t first_open_file_no);
 
 
 /*
@@ -420,6 +447,7 @@ void
 innodb_binlog_startup_init()
 {
   fsp_binlog_init();
+  mysql_mutex_init(fsp_purge_binlog_mutex_key, &purge_binlog_mutex, nullptr);
   binlog_diff_state.init();
   innodb_binlog_inited= 1;
 }
@@ -432,6 +460,8 @@ innodb_binlog_init_state()
   binlog_cur_end_offset[0].store(~(uint64_t)0, std::memory_order_relaxed);
   binlog_cur_end_offset[1].store(~(uint64_t)0, std::memory_order_relaxed);
   last_created_binlog_file_no= ~(uint64_t)0;
+  earliest_binlog_file_no= ~(uint64_t)0;
+  total_binlog_used_size= 0;
   active_binlog_file_no.store(~(uint64_t)0, std::memory_order_release);
   active_binlog_space= nullptr;
   binlog_cur_page_no= 0;
@@ -531,6 +561,18 @@ process_binlog_name(found_binlogs *bls, uint64_t idx, size_t size)
     bls->prev_file_no= idx;
     bls->prev_size= size;
   }
+
+  if (bls->found_binlogs == 0)
+  {
+    bls->earliest_file_no= idx;
+    bls->total_size= size;
+  }
+  else
+  {
+    if (idx < bls->earliest_file_no)
+      bls->earliest_file_no= idx;
+    bls->total_size+= size;
+  }
 }
 
 
@@ -702,6 +744,9 @@ innodb_binlog_discover()
   if (!page_buf)
     return -1;
   if (binlog_files.found_binlogs >= 1) {
+    earliest_binlog_file_no= binlog_files.earliest_file_no;
+    total_binlog_used_size= binlog_files.total_size;
+
     int res= find_pos_in_binlog(binlog_files.last_file_no,
                                 binlog_files.last_size,
                                 page_buf.get(),
@@ -771,6 +816,8 @@ innodb_binlog_discover()
 
   /* No binlog files found, start from scratch. */
   file_no= 0;
+  earliest_binlog_file_no= 0;
+  total_binlog_used_size= 0;
   ib::info() << "Starting a new binlog from file number " << file_no << ".";
   return 0;
 }
@@ -802,6 +849,7 @@ void innodb_binlog_close(bool shutdown)
   if (shutdown && innodb_binlog_inited >= 1)
   {
     binlog_diff_state.free();
+    mysql_mutex_destroy(&purge_binlog_mutex);
     fsp_binlog_shutdown();
   }
 }
@@ -813,6 +861,10 @@ void innodb_binlog_close(bool shutdown)
 static void
 innodb_binlog_prealloc_thread()
 {
+  my_thread_init();
+#ifdef UNIV_PFS_THREAD
+  pfs_register_thread(binlog_prealloc_thread_key);
+#endif
 
   mysql_mutex_lock(&active_binlog_mutex);
   while (1)
@@ -832,7 +884,18 @@ innodb_binlog_prealloc_thread()
       */
       ++last_created;
       mysql_mutex_unlock(&active_binlog_mutex);
-      dberr_t res2= fsp_binlog_tablespace_create(last_created, &new_space);
+
+      mysql_mutex_lock(&purge_binlog_mutex);
+      uint32_t size_in_pages=  innodb_binlog_size_in_pages;
+      dberr_t res2= fsp_binlog_tablespace_create(last_created, size_in_pages,
+                                                 &new_space);
+      if (earliest_binlog_file_no == ~(uint64_t)0)
+        earliest_binlog_file_no= last_created;
+      total_binlog_used_size+= (size_in_pages << srv_page_size_shift);
+
+      innodb_binlog_autopurge(first_open);
+      mysql_mutex_unlock(&purge_binlog_mutex);
+
       mysql_mutex_lock(&active_binlog_mutex);
       ut_a(res2 == DB_SUCCESS /* ToDo: Error handling. */);
       ut_a(new_space);
@@ -874,6 +937,12 @@ innodb_binlog_prealloc_thread()
 
   }
   mysql_mutex_unlock(&active_binlog_mutex);
+
+  my_thread_end();
+
+#ifdef UNIV_PFS_THREAD
+  pfs_delete_thread();
+#endif
 }
 
 
@@ -1650,6 +1719,8 @@ int ha_innodb_binlog_reader::read_binlog_data(uchar *buf, uint32_t len)
 {
   int res= read_data(buf, len);
   chunk_rd.release(res == 0);
+  cur_file_no= chunk_rd.current_file_no();
+  cur_file_pos= chunk_rd.current_pos();
   return res;
 }
 
@@ -2074,6 +2145,8 @@ ha_innodb_binlog_reader::init_gtid_pos(slave_connection_state *pos,
   {
     chunk_rd.seek(file_no, offset);
     chunk_rd.skip_partial(true);
+    cur_file_no= chunk_rd.current_file_no();
+    cur_file_pos= chunk_rd.current_pos();
   }
   return res;
 }
@@ -2121,43 +2194,17 @@ innobase_binlog_write_direct(IO_CACHE *cache,
 bool
 innodb_find_binlogs(uint64_t *out_first, uint64_t *out_last)
 {
-  MY_DIR *dir= my_dir(innodb_binlog_directory, MYF(0));
-  if (!dir)
+  mysql_mutex_lock(&active_binlog_mutex);
+  *out_last= last_created_binlog_file_no;
+  mysql_mutex_unlock(&active_binlog_mutex);
+  mysql_mutex_lock(&purge_binlog_mutex);
+  *out_first= earliest_binlog_file_no;
+  mysql_mutex_unlock(&purge_binlog_mutex);
+  if (*out_first == ~(uint64_t)0 || *out_last == ~(uint64_t)0)
   {
-    sql_print_error("Could not read the binlog directory '%s', error code %d",
-                    innodb_binlog_directory, my_errno);
+    ut_ad(0 /* Impossible, we wait at startup for binlog to be created. */);
     return true;
   }
-
-  size_t num_entries= dir->number_of_files;
-  fileinfo *entries= dir->dir_entry;
-  uint64_t first_file_no, last_file_no;
-  uint64_t num_file_no= 0;
-  for (size_t i= 0; i < num_entries; ++i) {
-    const char *name= entries[i].name;
-    uint64_t file_no;
-    if (!is_binlog_name(name, &file_no))
-      continue;
-    if (num_file_no == 0 || file_no < first_file_no)
-      first_file_no= file_no;
-    if (num_file_no == 0 || file_no > last_file_no)
-      last_file_no= file_no;
-    ++num_file_no;
-  }
-  my_dirend(dir);
-
-  if (num_file_no == 0)
-  {
-    sql_print_error("No binlog files found (deleted externally?)");
-    return true;
-  }
-  if (num_file_no != last_file_no - first_file_no + 1)
-  {
-    sql_print_error("Missing binlog files (deleted externally?)");
-    return true;
-  }
-  *out_first= first_file_no;
-  *out_last= last_file_no;
   return false;
 }
 
@@ -2207,3 +2254,232 @@ innodb_reset_binlogs()
 
   return err;
 }
+
+
+/*
+  The low-level function handling binlog purge.
+
+  How much to purge is determined by:
+
+  1. Lowest file_no that should not be purged. This is determined as the
+  minimum of:
+    1a. active_binlog_file_no
+    1b. first_open_binlog_file_no
+    1c. Any file_no in use by an active dump thread
+    1d. Any file_no containing oob data referenced by file_no from (1c)
+    1e. User specified file_no (from PURGE BINARY LOGS TO, if any).
+    1f. (ToDo): Any file_no that was still active at the last checkpoint.
+
+  2. Unix timestamp specifying the minimal value that should not be purged,
+  optional (used by PURGE BINARY LOGS BEFORE and --binlog-expire-log-seconds).
+
+  3. Maximum total size of binlogs, optional (from --max-binlog-total-size).
+
+  Sets out_file_no to the earliest binlog file not purged.
+  Additionally returns:
+
+     0  Purged all files as requested.
+     1  Some files were not purged due to being currently in-use (by binlog
+        writing or active dump threads).
+*/
+static int
+innodb_binlog_purge_low(uint64_t limit_file_no,
+                        bool by_date, time_t limit_date,
+                        bool by_size, ulonglong limit_size,
+                        bool by_name, uint64_t limit_name_file_no,
+                        uint64_t *out_file_no)
+{
+  ut_ad(by_date || by_size || by_name);
+  ut_a(limit_file_no <= active_binlog_file_no.load(std::memory_order_relaxed));
+  ut_a(limit_file_no <= first_open_binlog_file_no);
+
+  mysql_mutex_assert_owner(&purge_binlog_mutex);
+  size_t loc_total_size= total_binlog_used_size;
+  uint64_t file_no;
+  bool want_purge;
+
+  for (file_no= earliest_binlog_file_no; ; ++file_no)
+  {
+    want_purge= false;
+
+    char filename[OS_FILE_MAX_PATH];
+    binlog_name_make(filename, file_no);
+    MY_STAT stat_buf;
+    if (!my_stat(filename, &stat_buf, MYF(0)))
+    {
+      if (my_errno == ENOENT)
+        sql_print_information("InnoDB: File already gone when purging binlog "
+                              "file '%s'", filename);
+      else
+        sql_print_warning("InnoDB: Failed to stat() when trying to purge "
+                          "binlog file '%' (errno: %d)", filename, my_errno);
+      continue;
+    }
+
+    if (by_date && stat_buf.st_mtime < limit_date)
+      want_purge= true;
+    if (by_size && loc_total_size > limit_size)
+      want_purge= true;
+    if (by_name && file_no < limit_name_file_no)
+      want_purge= true;
+    if (file_no >= limit_file_no || !want_purge)
+      break;
+    earliest_binlog_file_no= file_no + 1;
+    if (loc_total_size < (size_t)stat_buf.st_size)
+    {
+      /*
+        Somehow we miscounted size, files changed from outside server or
+        possibly bug. We will handle not underflowing the total. If this
+        assertion becomes a problem for testing, it can just be removed.
+      */
+      ut_ad(0);
+    }
+    else
+      loc_total_size-= stat_buf.st_size;
+    if (my_delete(filename, MYF(0)))
+    {
+      if (my_errno == ENOENT)
+      {
+        /*
+          File already gone, just ignore the error.
+          (This should be somewhat unusual to happen as stat() succeeded).
+        */
+      }
+      else
+      {
+        sql_print_warning("InnoDB: Delete failed while trying to purge binlog "
+                          "file '%s' (errno: %d)", filename, my_error);
+        continue;
+      }
+    }
+  }
+  total_binlog_used_size= loc_total_size;
+  *out_file_no= file_no;
+  return (want_purge ? 1 : 0);
+}
+
+
+static void
+innodb_binlog_autopurge(uint64_t first_open_file_no)
+{
+  handler_binlog_purge_info UNINIT_VAR(purge_info);
+#ifdef HAVE_REPLICATION
+  extern bool ha_binlog_purge_info(handler_binlog_purge_info *out_info);
+  bool can_purge= ha_binlog_purge_info(&purge_info);
+#else
+  bool can_purge= false;
+#endif
+  if (!can_purge ||
+      !(purge_info.purge_by_size || purge_info.purge_by_date))
+    return;
+
+  /*
+    ToDo: Here, we need to move back the purge_info.limit_file_no to the
+    earliest file containing any oob data referenced from the supplied
+    purge_info.limit_file_no.
+  */
+
+  /* Don't purge any actively open tablespace files. */
+  uint64_t limit_file_no= purge_info.limit_file_no;
+  if (limit_file_no == ~(uint64_t)0 || limit_file_no > first_open_file_no)
+    limit_file_no= first_open_file_no;
+  uint64_t active= active_binlog_file_no.load(std::memory_order_relaxed);
+  if (limit_file_no > active)
+    limit_file_no= active;
+
+  uint64_t file_no;
+  int res=
+    innodb_binlog_purge_low(limit_file_no,
+                            purge_info.purge_by_date, purge_info.limit_date,
+                            purge_info.purge_by_size, purge_info.limit_size,
+                            false, 0,
+                            &file_no);
+  if (res)
+  {
+    if (!purge_warning_given)
+    {
+      char filename[BINLOG_NAME_MAX_LEN];
+      binlog_name_make_short(filename, file_no);
+      if (purge_info.nonpurge_reason)
+        sql_print_information("InnoDB: Binlog file %s could not be purged "
+                              "because %s",
+                              filename, purge_info.nonpurge_reason);
+      else if (purge_info.limit_file_no == file_no)
+        sql_print_information("InnoDB: Binlog file %s could not be purged "
+                              "because it is in use by a binlog dump thread "
+                              "(connected slave)", filename);
+      else if (limit_file_no == file_no)
+        sql_print_information("InnoDB: Binlog file %s could not be purged "
+                              "because it is in active use", filename);
+      else
+        sql_print_information("InnoDB: Binlog file %s could not be purged "
+                              "because it might still be needed", filename);
+      purge_warning_given= true;
+    }
+  }
+  else
+    purge_warning_given= false;
+}
+
+
+int
+innodb_binlog_purge(handler_binlog_purge_info *purge_info)
+{
+  /*
+    Let us check that we do not get an attempt to purge by file, date, and/or
+    size at the same time.
+    (If we do, it is not necesarily a problem, but this cannot happen in
+    current server code).
+  */
+  ut_ad(1 == (!!purge_info->purge_by_name +
+              !!purge_info->purge_by_date +
+              !!purge_info->purge_by_size));
+
+  if (!purge_info->purge_by_name && !purge_info->purge_by_date &&
+      !purge_info->purge_by_size)
+    return 0;
+
+  mysql_mutex_lock(&active_binlog_mutex);
+  uint64_t limit_file_no=
+    std::min(active_binlog_file_no.load(std::memory_order_relaxed),
+             first_open_binlog_file_no);
+  uint64_t last_created= last_created_binlog_file_no;
+  mysql_mutex_unlock(&active_binlog_mutex);
+
+  uint64_t to_file_no= ~(uint64_t)0;
+  if (purge_info->purge_by_name)
+  {
+    if (!is_binlog_name(purge_info->limit_name, &to_file_no) ||
+        to_file_no > last_created)
+      return LOG_INFO_EOF;
+  }
+
+  mysql_mutex_lock(&purge_binlog_mutex);
+  uint64_t file_no;
+  int res= innodb_binlog_purge_low(
+        std::min(purge_info->limit_file_no, limit_file_no),
+        purge_info->purge_by_date, purge_info->limit_date,
+        purge_info->purge_by_size, purge_info->limit_size,
+                                   purge_info->purge_by_name, to_file_no,
+        &file_no);
+  mysql_mutex_unlock(&purge_binlog_mutex);
+  if (res == 1)
+  {
+    static_assert(sizeof(purge_info->nonpurge_filename) >= BINLOG_NAME_MAX_LEN,
+                  "No room to return filename");
+    binlog_name_make_short(purge_info->nonpurge_filename, file_no);
+    if (!purge_info->nonpurge_reason)
+    {
+      if (limit_file_no == file_no)
+        purge_info->nonpurge_reason= "the binlog file is in active use";
+      else if (purge_info->limit_file_no == file_no)
+        purge_info->nonpurge_reason= "it is in use by a binlog dump thread "
+          "(connected slave)";
+    }
+    res= LOG_INFO_IN_USE;
+  }
+  else
+    purge_warning_given= false;
+
+  return res;
+}
diff --git a/storage/innobase/include/fsp_binlog.h b/storage/innobase/include/fsp_binlog.h
index b6c4a393d7b..0d91e2dcb44 100644
--- a/storage/innobase/include/fsp_binlog.h
+++ b/storage/innobase/include/fsp_binlog.h
@@ -189,6 +189,10 @@ public:
   /* Release any buffer pool page latch. */
   void release(bool release_file_page= false);
   bool data_available();
+  uint64_t current_file_no() { return s.file_no; }
+  uint64_t current_pos() {
+    return (s.page_no << srv_page_size_shift) + s.in_page_offset;
+  }
 };
 
 
@@ -210,6 +214,7 @@ extern fil_space_t *fsp_binlog_open(const char *file_name, pfs_os_file_t fh,
                                     uint64_t file_no, size_t file_size,
                                     bool open_empty);
 extern dberr_t fsp_binlog_tablespace_create(uint64_t file_no,
+                                            uint32_t size_in_pages,
                                             fil_space_t **new_space);
 extern std::pair<uint64_t, uint64_t> fsp_binlog_write_rec(
   struct chunk_data_base *chunk_data, mtr_t *mtr, byte chunk_type);
diff --git a/storage/innobase/include/innodb_binlog.h b/storage/innobase/include/innodb_binlog.h
index a627f0c5d3e..00ed2fdd899 100644
--- a/storage/innobase/include/innodb_binlog.h
+++ b/storage/innobase/include/innodb_binlog.h
@@ -35,6 +35,7 @@ struct rpl_binlog_state_base;
 struct rpl_gtid;
 struct handler_binlog_event_group_info;
 class handler_binlog_reader;
+struct handler_binlog_purge_info;
 
 
 /*
@@ -72,6 +73,8 @@ extern uint32_t binlog_cur_page_no;
 extern uint32_t binlog_cur_page_offset;
 extern ulonglong innodb_binlog_state_interval;
 extern rpl_binlog_state_base binlog_diff_state;
+extern mysql_mutex_t purge_binlog_mutex;
+extern size_t total_binlog_used_size;
 
 
 static inline void
@@ -83,6 +86,13 @@ binlog_name_make(char name_buf[OS_FILE_MAX_PATH], uint64_t file_no)
 }
 
 
+static inline void
+binlog_name_make_short(char *name_buf, uint64_t file_no)
+{
+  sprintf(name_buf, BINLOG_NAME_BASE "%06" PRIu64 BINLOG_NAME_EXT, file_no);
+}
+
+
 extern void innodb_binlog_startup_init();
 extern bool innodb_binlog_init(size_t binlog_size, const char *directory);
 extern void innodb_binlog_close(bool shutdown);
@@ -99,5 +109,6 @@ extern bool innobase_binlog_write_direct
    const rpl_gtid *gtid);
 extern bool innodb_find_binlogs(uint64_t *out_first, uint64_t *out_last);
 extern bool innodb_reset_binlogs();
+extern int innodb_binlog_purge(handler_binlog_purge_info *purge_info);
 
 #endif /* innodb_binlog_h */
diff --git a/storage/innobase/include/univ.i b/storage/innobase/include/univ.i
index d346f99e7af..58e9f23ae52 100644
--- a/storage/innobase/include/univ.i
+++ b/storage/innobase/include/univ.i
@@ -482,6 +482,7 @@ extern mysql_pfs_key_t trx_pool_manager_mutex_key;
 extern mysql_pfs_key_t lock_wait_mutex_key;
 extern mysql_pfs_key_t srv_threads_mutex_key;
 extern mysql_pfs_key_t fsp_active_binlog_mutex_key;
+extern mysql_pfs_key_t fsp_purge_binlog_mutex_key;
 # endif /* UNIV_PFS_MUTEX */
 
 # ifdef UNIV_PFS_RWLOCK