merged (adf4e7a8) · Commits · Software / OSDI20 Artifacts / mariadb

mysql-test/r/flush_block_commit.result

0 → 100644

+23 −0

Original line number	Diff line number	Diff line
		drop table if exists t1;
		create table t1 (a int) type=innodb;
		begin;
		insert into t1 values(1);
		flush tables with read lock;
		select * from t1;
		a
		commit;
		select * from t1;
		a
		unlock tables;
		begin;
		select * from t1 for update;
		a
		1
		begin;
		select * from t1 for update;
		flush tables with read lock;
		commit;
		a
		1
		unlock tables;
		drop table t1;

mysql-test/t/flush_block_commit-master.opt

0 → 100644

+1 −0

Original line number	Diff line number	Diff line
		--innodb_lock_wait_timeout=5

mysql-test/t/flush_block_commit.test

0 → 100644

+49 −0

Original line number	Diff line number	Diff line
		# Let's see if FLUSH TABLES WITH READ LOCK blocks COMMIT of existing
		# transactions.
		# We verify that we did not introduce a deadlock.

		-- source include/have_innodb.inc

		connect (con1,localhost,root,,);
		connect (con2,localhost,root,,);
		connect (con3,localhost,root,,);
		connection con1;
		drop table if exists t1;
		create table t1 (a int) type=innodb;

		# blocks COMMIT ?

		begin;
		insert into t1 values(1);
		connection con2;
		flush tables with read lock;
		select * from t1;
		connection con1;
		send commit; # blocked by con2
		sleep 1;
		connection con2;
		select * from t1; # verify con1 was blocked and data did not move
		unlock tables;
		connection con1;
		reap;

		# No deadlock ?

		connection con1;
		begin;
		select * from t1 for update;
		connection con2;
		begin;
		send select * from t1 for update; # blocked by con1
		sleep 1;
		connection con3;
		send flush tables with read lock; # blocked by con2
		connection con1;
		commit; # should not be blocked by con3
		connection con2;
		reap;
		connection con3;
		reap;
		unlock tables;
		connection con1;
		drop table t1;

sql/handler.cc

+30 −14

Original line number	Diff line number	Diff line
		@@ -477,11 +477,24 @@ int ha_commit_trans(THD thd, THD_TRANS trans)
		if (opt_using_transactions)
		{
		bool operation_done= 0;
		bool transaction_commited= 0;
		bool operation_done= 0, need_start_waiters= 0;

		/* Update the binary log if we have cached some queries */
		if (trans == &thd->transaction.all && mysql_bin_log.is_open() &&
		/* If transaction has done some updates to tables */
		if (trans == &thd->transaction.all &&
		my_b_tell(&thd->transaction.trans_log))
		{
		if (error= wait_if_global_read_lock(thd, 0, 0))
		{
		/*
		Note that ROLLBACK [TO SAVEPOINT] does not have this test; it's
		because ROLLBACK never updates data, so needn't wait on the lock.
		*/
		my_error(ER_ERROR_DURING_COMMIT, MYF(0), error);
		error= 1;
		}
		else
		need_start_waiters= 1;
		if (mysql_bin_log.is_open())
		{
		mysql_bin_log.write(thd, &thd->transaction.trans_log, 1);
		statistic_increment(binlog_cache_use, &LOCK_status);
		@@ -498,6 +511,7 @@ int ha_commit_trans(THD thd, THD_TRANS trans)
		WRITE_CACHE, (my_off_t) 0, 0, 1);
		thd->transaction.trans_log.end_of_file= max_binlog_cache_size;
		}
		}
		#ifdef HAVE_NDBCLUSTER_DB
		if (trans->ndb_tid)
		{
		@@ -551,6 +565,8 @@ int ha_commit_trans(THD thd, THD_TRANS trans)
		statistic_increment(ha_commit_count,&LOCK_status);
		thd->transaction.cleanup();
		}
		if (need_start_waiters)
		start_waiting_global_read_lock(thd);
		}
		#endif // using transactions
		DBUG_RETURN(error);

sql/lock.cc

+47 −7

Original line number	Diff line number	Diff line
		@@ -99,7 +99,7 @@ MYSQL_LOCK mysql_lock_tables(THD thd,TABLE **tables,uint count)
		Someone has issued LOCK ALL TABLES FOR READ and we want a write lock
		Wait until the lock is gone
		*/
		if (wait_if_global_read_lock(thd, 1))
		if (wait_if_global_read_lock(thd, 1, 1))
		{
		my_free((gptr) sql_lock,MYF(0));
		sql_lock=0;
		@@ -474,7 +474,7 @@ int lock_and_wait_for_table_name(THD thd, TABLE_LIST table_list)
		int error= -1;
		DBUG_ENTER("lock_and_wait_for_table_name");

		if (wait_if_global_read_lock(thd,0))
		if (wait_if_global_read_lock(thd, 0, 1))
		DBUG_RETURN(1);
		VOID(pthread_mutex_lock(&LOCK_open));
		if ((lock_retcode = lock_table_name(thd, table_list)) < 0)
		@@ -707,14 +707,23 @@ static void print_lock_error(int error)

		The global locks are handled through the global variables:
		global_read_lock
		global_read_lock_blocks_commit
		waiting_for_read_lock
		protect_against_global_read_lock

		Taking the global read lock is TWO steps (2nd step is optional; without
		it, COMMIT of existing transactions will be allowed):
		lock_global_read_lock() THEN make_global_read_lock_block_commit().
		****************************************************************************/

		volatile uint global_read_lock=0;
		volatile uint global_read_lock_blocks_commit=0;
		static volatile uint protect_against_global_read_lock=0;
		static volatile uint waiting_for_read_lock=0;

		#define GOT_GLOBAL_READ_LOCK 1
		#define MADE_GLOBAL_READ_LOCK_BLOCK_COMMIT 2

		bool lock_global_read_lock(THD *thd)
		{
		DBUG_ENTER("lock_global_read_lock");
		@@ -737,27 +746,40 @@ bool lock_global_read_lock(THD *thd)
		thd->exit_cond(old_message);
		DBUG_RETURN(1);
		}
		thd->global_read_lock=1;
		thd->global_read_lock= GOT_GLOBAL_READ_LOCK;
		global_read_lock++;
		thd->exit_cond(old_message);
		}
		/*
		We DON'T set global_read_lock_blocks_commit now, it will be set after
		tables are flushed (as the present function serves for FLUSH TABLES WITH
		READ LOCK only). Doing things in this order is necessary to avoid
		deadlocks (we must allow COMMIT until all tables are closed; we should not
		forbid it before, or we can have a 3-thread deadlock if 2 do SELECT FOR
		UPDATE and one does FLUSH TABLES WITH READ LOCK).
		*/
		DBUG_RETURN(0);
		}

		void unlock_global_read_lock(THD *thd)
		{
		uint tmp;
		thd->global_read_lock=0;
		pthread_mutex_lock(&LOCK_open);
		tmp= --global_read_lock;
		if (thd->global_read_lock == MADE_GLOBAL_READ_LOCK_BLOCK_COMMIT)
		--global_read_lock_blocks_commit;
		pthread_mutex_unlock(&LOCK_open);
		/* Send the signal outside the mutex to avoid a context switch */
		if (!tmp)
		pthread_cond_broadcast(&COND_refresh);
		thd->global_read_lock= 0;
		}

		#define must_wait (global_read_lock && \
		(is_not_commit \|\| \
		global_read_lock_blocks_commit))

		bool wait_if_global_read_lock(THD *thd, bool abort_on_refresh)
		bool wait_if_global_read_lock(THD *thd, bool abort_on_refresh, bool is_not_commit)
		{
		const char *old_message;
		bool result= 0, need_exit_cond;
		@@ -765,7 +787,7 @@ bool wait_if_global_read_lock(THD *thd, bool abort_on_refresh)

		LINT_INIT(old_message);
		(void) pthread_mutex_lock(&LOCK_open);
		if (need_exit_cond= (bool)global_read_lock)
		if (need_exit_cond= must_wait)
		{
		if (thd->global_read_lock) // This thread had the read locks
		{
		@@ -775,7 +797,7 @@ bool wait_if_global_read_lock(THD *thd, bool abort_on_refresh)
		}
		old_message=thd->enter_cond(&COND_refresh, &LOCK_open,
		"Waiting for release of readlock");
		while (global_read_lock && ! thd->killed &&
		while (must_wait && ! thd->killed &&
		(!abort_on_refresh \|\| thd->version == refresh_version))
		(void) pthread_cond_wait(&COND_refresh,&LOCK_open);
		if (thd->killed)
		@@ -802,3 +824,21 @@ void start_waiting_global_read_lock(THD *thd)
		pthread_cond_broadcast(&COND_refresh);
		DBUG_VOID_RETURN;
		}


		void make_global_read_lock_block_commit(THD *thd)
		{
		/*
		If we didn't succeed lock_global_read_lock(), or if we already suceeded
		make_global_read_lock_block_commit(), do nothing.
		*/
		if (thd->global_read_lock != GOT_GLOBAL_READ_LOCK)
		return;
		pthread_mutex_lock(&LOCK_open);
		/* increment this BEFORE waiting on cond (otherwise race cond) */
		global_read_lock_blocks_commit++;
		while (protect_against_global_read_lock)
		pthread_cond_wait(&COND_refresh, &LOCK_open);
		pthread_mutex_unlock(&LOCK_open);
		thd->global_read_lock= MADE_GLOBAL_READ_LOCK_BLOCK_COMMIT;
		}