FLUSH TABLES WITH READ LOCK should block writes to binlog too

mysql-test/r/flush_block_commit.result:
  FLUSH TABLES WITH READ LOCK should block writes to binlog too
  it does not yet
mysql-test/t/flush_block_commit.test:
  FLUSH TABLES WITH READ LOCK should block writes to binlog too
  it does not yet

mysql-test/r/flush_block_commit.result

@@ -37,3 +37,18 @@ show create database test;
 Database	Create Database
 test	CREATE DATABASE `test` /*!40100 DEFAULT CHARACTER SET latin1 */
 drop table t1;
+create table t1 (a int) engine=innodb;
+reset master;
+set autocommit=0;
+insert t1 values (1);
+flush tables with read lock;
+show master status;
+File	Position	Binlog_Do_DB	Binlog_Ignore_DB
+master-bin.000001	98		
+ commit;
+show master status;
+File	Position	Binlog_Do_DB	Binlog_Ignore_DB
+master-bin.000001	276		
+unlock tables;
+drop table t1;
+set autocommit=1;

mysql-test/t/flush_block_commit.test

@@ -76,3 +76,24 @@ select * from t1;
 show create database test;
 
 drop table t1;
+
+# FLUSH TABLES WITH READ LOCK should block writes to binlog too
+connection con1;
+create table t1 (a int) engine=innodb;
+reset master;
+set autocommit=0;
+insert t1 values (1);
+connection con2;
+flush tables with read lock;
+show master status;
+connection con1;
+send commit;
+connection con2;
+sleep 1;
+show master status;
+unlock tables;
+connection con1;
+reap;
+drop table t1;
+set autocommit=1;
+

sql/handler.cc

@@ -586,6 +586,11 @@ int ha_commit_trans(THD *thd, bool all)
 #ifdef USING_TRANSACTIONS
   if (trans->nht)
   {
+    if (is_real_trans && wait_if_global_read_lock(thd, 0, 0))
+    {
+      ha_rollback_trans(thd, all);
+      DBUG_RETURN(1);
+    }
     DBUG_EXECUTE_IF("crash_commit_before", abort(););
     if (!trans->no_2pc && trans->nht > 1)
     {
@@ -604,7 +609,8 @@ int ha_commit_trans(THD *thd, bool all)
                     (error= !(cookie= tc_log->log(thd, xid)))))
       {
         ha_rollback_trans(thd, all);
-        return 1;
+        error= 1;
+        goto end;
       }
       DBUG_EXECUTE_IF("crash_commit_after_log", abort(););
     }
@@ -613,11 +619,18 @@ int ha_commit_trans(THD *thd, bool all)
     if (cookie)
       tc_log->unlog(cookie, xid);
     DBUG_EXECUTE_IF("crash_commit_after", abort(););
+end:
+    if (is_real_trans)
+      start_waiting_global_read_lock(thd);
   }
 #endif /* USING_TRANSACTIONS */
   DBUG_RETURN(error);
 }
 
+/*
+  NOTE - this function does not care about global read lock.
+  A caller should.
+*/
 int ha_commit_one_phase(THD *thd, bool all)
 {
   int error=0;
@@ -628,18 +641,6 @@ int ha_commit_one_phase(THD *thd, bool all)
 #ifdef USING_TRANSACTIONS
   if (trans->nht)
   {
-    bool need_start_waiters= 0;
-    if (is_real_trans)
-    {
-      if ((error= wait_if_global_read_lock(thd, 0, 0)))
-      {
-        my_error(ER_ERROR_DURING_COMMIT, MYF(0), error);
-        error= 1;
-      }
-      else
-        need_start_waiters= 1;
-    }
-
     for (ht=trans->ht; *ht; ht++)
     {
       int err;
@@ -664,8 +665,6 @@ int ha_commit_one_phase(THD *thd, bool all)
       thd->variables.tx_isolation=thd->session_tx_isolation;
       thd->transaction.cleanup();
     }
-    if (need_start_waiters)
-      start_waiting_global_read_lock(thd);
   }
 #endif /* USING_TRANSACTIONS */
   DBUG_RETURN(error);

sql/sql_parse.cc

@@ -4413,11 +4413,20 @@ unsent_create_error:
     }
     else
     if (thd->transaction.xa_state == XA_PREPARED && thd->lex->xa_opt == XA_NONE)
+    {
+      if (wait_if_global_read_lock(thd, 0, 0))
+      {
+        ha_rollback(thd);
+        my_error(ER_XAER_RMERR, MYF(0));
+      }
+      else
       {
         if (ha_commit_one_phase(thd, 1))
           my_error(ER_XAER_RMERR, MYF(0));
         else
           send_ok(thd);
+        start_waiting_global_read_lock(thd);
+      }
     }
     else
     {

sql/sql_yacc.yy

@@ -7343,7 +7343,6 @@ keyword:
 	| TIMESTAMP_ADD		{}
 	| TIMESTAMP_DIFF	{}
 	| TIME_SYM		{}
-	| TYPE_SYM		{}
 	| TYPES_SYM		{}
         | UDF_RETURNS_SYM       {}
 	| FUNCTION_SYM		{}