diff --git a/include/mysql/thread_pool_priv.h b/include/mysql/thread_pool_priv.h
index efc264b713d..1da174e5f8b 100644
--- a/include/mysql/thread_pool_priv.h
+++ b/include/mysql/thread_pool_priv.h
@@ -1,5 +1,5 @@
 /*
-  Copyright (c) 2010, 2012, Oracle and/or its affiliates. All rights reserved.
+  Copyright (c) 2010, 2014, Oracle and/or its affiliates. All rights reserved.
 
   This program is free software; you can redistribute it and/or modify
   it under the terms of the GNU General Public License as published by
@@ -49,6 +49,8 @@ void thd_clear_errors(THD *thd);
 void thd_set_thread_stack(THD *thd, char *stack_start);
 void thd_lock_thread_count(THD *thd);
 void thd_unlock_thread_count(THD *thd);
+void thd_lock_thread_remove(THD *thd);
+void thd_unlock_thread_remove(THD *thd);
 void thd_close_connection(THD *thd);
 THD *thd_get_current_thd();
 void thd_new_connection_setup(THD *thd, char *stack_start);
diff --git a/mysql-test/r/show_processlist.result b/mysql-test/r/show_processlist.result
new file mode 100644
index 00000000000..f133e480e1a
--- /dev/null
+++ b/mysql-test/r/show_processlist.result
@@ -0,0 +1,47 @@
+Bug#17283409  4-WAY DEADLOCK: ZOMBIES, PURGING BINLOGS, SHOW PROCESSLIST,
+SHOW BINLOGS
+connect connection1,127.0.0.1,root,,test,$MASTER_MYPORT,;
+connection default;
+SET DEBUG_SYNC='before_one_element_read_from_threads_iterator SIGNAL parked1 WAIT_FOR go';
+SHOW PROCESSLIST;
+connection connection1;
+"Wait_for parked1"
+SET DEBUG_SYNC='now WAIT_FOR parked1';
+connect connection2,127.0.0.1,root,,test,$MASTER_MYPORT,;
+connect connection3,127.0.0.1,root,,test,$MASTER_MYPORT,;
+SET DEBUG_SYNC='now SIGNAL go';
+connection default;
+Id	User	Host	db	Command	Time	State	Info
+<Id>	root	<Host>	test	<Command>	<Time>	<State>	<Info>
+<Id>	root	<Host>	test	<Command>	<Time>	<State>	<Info>
+<Id>	root	<Host>	test	<Command>	<Time>	<State>	<Info>
+<Id>	root	<Host>	test	<Command>	<Time>	<State>	<Info>
+SET DEBUG_SYNC='after_one_element_read_from_threads_iterator SIGNAL parked1 WAIT_FOR go';
+SHOW PROCESSLIST;
+connection connection1;
+"Wait_for parked1"
+SET DEBUG_SYNC='now WAIT_FOR parked1';
+connect connection4,127.0.0.1,root,,test,$MASTER_MYPORT,;
+connect connection5,127.0.0.1,root,,test,$MASTER_MYPORT,;
+SET DEBUG_SYNC='now SIGNAL go';
+connection default;
+Id	User	Host	db	Command	Time	State	Info
+<Id>	root	<Host>	test	<Command>	<Time>	<State>	<Info>
+<Id>	root	<Host>	test	<Command>	<Time>	<State>	<Info>
+<Id>	root	<Host>	test	<Command>	<Time>	<State>	<Info>
+<Id>	root	<Host>	test	<Command>	<Time>	<State>	<Info>
+SHOW PROCESSLIST;
+Id	User	Host	db	Command	Time	State	Info
+<Id>	root	<Host>	test	<Command>	<Time>	<State>	<Info>
+<Id>	root	<Host>	test	<Command>	<Time>	<State>	<Info>
+<Id>	root	<Host>	test	<Command>	<Time>	<State>	<Info>
+<Id>	root	<Host>	test	<Command>	<Time>	<State>	<Info>
+<Id>	root	<Host>	test	<Command>	<Time>	<State>	<Info>
+<Id>	root	<Host>	test	<Command>	<Time>	<State>	<Info>
+"Cleanup"
+SET DEBUG_SYNC='RESET';
+disconnect connection1;
+disconnect connection2;
+disconnect connection3;
+disconnect connection4;
+disconnect connection5;
diff --git a/mysql-test/suite/rpl/r/rpl_4threads_deadlock.result b/mysql-test/suite/rpl/r/rpl_4threads_deadlock.result
new file mode 100644
index 00000000000..29161b0adf1
--- /dev/null
+++ b/mysql-test/suite/rpl/r/rpl_4threads_deadlock.result
@@ -0,0 +1,43 @@
+include/master-slave.inc
+[connection master]
+FLUSH LOGS;
+SET @save_debug=@@global.debug;
+SET GLOBAL DEBUG='+d,before_dump_thread_acquires_current_mutex,processlist_acquiring_dump_threads_LOCK_thd_data';
+include/stop_slave.inc
+include/start_slave.inc
+"Wait_for dump_thread_signal"
+SET DEBUG_SYNC='now WAIT_FOR dump_thread_signal';
+SET DEBUG_SYNC='show_binlogs_after_lock_log_before_lock_index SIGNAL parked1 WAIT_FOR go';
+SHOW BINARY LOGS;
+"Wait_for parked1"
+SET DEBUG_SYNC='now WAIT_FOR parked1';
+SET DEBUG_SYNC='purge_logs_after_lock_index_before_thread_count SIGNAL parked2 WAIT_FOR go';
+PURGE BINARY LOGS BEFORE '9999-12-31';
+"Wait_for parked2"
+SET DEBUG_SYNC='now WAIT_FOR parked2';
+SET DEBUG_SYNC='processlist_after_LOCK_thd_count_before_LOCK_thd_data SIGNAL parked3 WAIT_FOR go';
+SELECT * FROM information_schema.processlist;
+"Wait_for parked3"
+SET DEBUG_SYNC='now WAIT_FOR parked3';
+SET DEBUG_SYNC='now SIGNAL go';
+include/stop_slave.inc
+include/start_slave.inc
+"Wait_for dump_thread_signal"
+SET DEBUG_SYNC='now WAIT_FOR dump_thread_signal';
+SET DEBUG_SYNC='show_binlogs_after_lock_log_before_lock_index SIGNAL parked1 WAIT_FOR go';
+SHOW BINARY LOGS;
+"Wait_for parked1"
+SET DEBUG_SYNC='now WAIT_FOR parked1';
+SET DEBUG_SYNC='purge_logs_after_lock_index_before_thread_count SIGNAL parked2 WAIT_FOR go';
+PURGE BINARY LOGS BEFORE '9999-12-31';
+"Wait_for parked2"
+SET DEBUG_SYNC='now WAIT_FOR parked2';
+SET DEBUG_SYNC='processlist_after_LOCK_thd_count_before_LOCK_thd_data SIGNAL parked3 WAIT_FOR go';
+SHOW PROCESSLIST;
+"Wait_for parked3"
+SET DEBUG_SYNC='now WAIT_FOR parked3';
+SET DEBUG_SYNC='now SIGNAL go';
+"Cleanup"
+SET DEBUG_SYNC='RESET';
+SET GLOBAL DEBUG=@save_debug;
+include/rpl_end.inc
diff --git a/mysql-test/suite/rpl/t/rpl_4threads_deadlock.test b/mysql-test/suite/rpl/t/rpl_4threads_deadlock.test
new file mode 100644
index 00000000000..5349e0c92d8
--- /dev/null
+++ b/mysql-test/suite/rpl/t/rpl_4threads_deadlock.test
@@ -0,0 +1,117 @@
+###############################################################################
+# Bug#17283409  4-WAY DEADLOCK: ZOMBIES, PURGING BINLOGS, SHOW PROCESSLIST,
+#                        SHOW BINLOGS
+# Problem:  A deadlock was occuring when 4 threads were involved in acquiring
+#                locks
+#     Thread 1: Dump thread ( Slave is reconnecting, so on Master, a new dump
+#                    thread is trying kill zombie dump threads. It acquired
+#                    thread's LOCK_thd_data and it is about to acquire
+#                    mysys_var->current_mutex ( which LOCK_log)
+#     Thread 2: Application thread is executing show binlogs and acquired
+#                    LOCK_log and it is about to acquire LOCK_index.
+#     Thread 3: Applicaiton thread is executing Purge binary logs and acquired
+#                    LOCK_index and it is about to acquire LOCK_thread_count.
+#     Thread 4: Application thread is executing show processlist and acquired
+#                    LOCK_thread_count and it is about to acquire zombie dump
+#                    thread's LOCK_thd_data.
+#     Deadlock : Thread 1 -> Thread 2 -> Thread 3-> Thread 4 -> Thread 1
+# The same above deadlock was observed when thread 4 is replaced with another
+# thread which is executing 'SELECT * FROM information_schema.processlist'
+# command and acquired LOCK_thread_count and it is about to acquire zombie
+# dump thread's LOCK_thd_data.
+# Fix: Introuced a new lock (Lock_thread_remove) which will guard the thread
+# deletion (exit process) and the same lock is used in show processlist
+# execution instead of LOCK_thread_count. i.e., After the patch, Server allows
+# new threads to join the system but it will not any threads to execute its
+# 'cleanup' process.  Now the new lock order is: LOCK_thread_remove ->
+# LOCK_thd_data -> LOCK_log -> LOCK_index -> LOCK_thread_count Both the above
+# scenarios are tested with the below test code and it proves that there is no
+# deadlock after the fix.
+###############################################################################
+--source include/have_debug_sync.inc
+--source include/have_binlog_format_statement.inc
+--source include/master-slave.inc
+
+connect(master2,127.0.0.1,root,,test,$MASTER_MYPORT,);
+connect(master3,127.0.0.1,root,,test,$MASTER_MYPORT,);
+--connection master
+# Create atleast one binary log, otherwise purge logs command, show binary
+# logs command do not hit the DEBUG_SYNC points.
+FLUSH LOGS;
+SET @save_debug=@@global.debug;
+SET GLOBAL DEBUG='+d,before_dump_thread_acquires_current_mutex,processlist_acquiring_dump_threads_LOCK_thd_data';
+
+# Iteration 1 with "SHOW PROCESSLIST command"
+# Iteration 2 with "SELECT * FROM information_schema.processlist"
+--let $iter=2
+while ($iter)
+{
+  --connection slave
+  --source include/stop_slave.inc
+  --source include/start_slave.inc
+  # Thread 1
+  --connection master
+  --echo "Wait_for dump_thread_signal"
+  SET DEBUG_SYNC='now WAIT_FOR dump_thread_signal';
+
+  # Thread 2
+  --connection master1
+  SET DEBUG_SYNC='show_binlogs_after_lock_log_before_lock_index SIGNAL parked1 WAIT_FOR go';
+  --send SHOW BINARY LOGS
+
+  --connection master
+  echo "Wait_for parked1";
+  SET DEBUG_SYNC='now WAIT_FOR parked1';
+
+  # Thread 3
+  --connection master2
+  SET DEBUG_SYNC='purge_logs_after_lock_index_before_thread_count SIGNAL parked2 WAIT_FOR go';
+  --send PURGE BINARY LOGS BEFORE '9999-12-31'
+
+  --connection master
+  echo "Wait_for parked2";
+  SET DEBUG_SYNC='now WAIT_FOR parked2';
+
+  # Thread 4
+  --connection master3
+  SET DEBUG_SYNC='processlist_after_LOCK_thd_count_before_LOCK_thd_data SIGNAL parked3 WAIT_FOR go';
+  if ($iter == 1)
+  {
+    --let $query=SHOW PROCESSLIST
+  }
+  if ($iter == 2)
+  {
+    --let $query=SELECT * FROM information_schema.processlist
+  }
+  --send_eval $query
+
+  --connection master
+  echo "Wait_for parked3";
+  SET DEBUG_SYNC='now WAIT_FOR parked3';
+
+  --connection master
+  SET DEBUG_SYNC='now SIGNAL go';
+
+  # Output of each command vary depends on other tests ran in suite(output vary
+  # in Pb2 runs), so lets ignore output. Anyways, we are interested in seeing
+  # that there is no deadlock.
+  --disable_result_log
+  --connection master1
+  --reap
+  --connection master2
+  --reap
+  --connection master3
+  --reap
+  --enable_result_log
+
+  --dec $iter
+}
+
+--echo "Cleanup"
+--disconnect master2
+--disconnect master3
+
+--connection master
+SET DEBUG_SYNC='RESET';
+SET GLOBAL DEBUG=@save_debug;
+source include/rpl_end.inc;
diff --git a/mysql-test/t/show_processlist.test b/mysql-test/t/show_processlist.test
new file mode 100644
index 00000000000..478c079b95b
--- /dev/null
+++ b/mysql-test/t/show_processlist.test
@@ -0,0 +1,63 @@
+###############################################################################
+--echo Bug#17283409  4-WAY DEADLOCK: ZOMBIES, PURGING BINLOGS, SHOW PROCESSLIST,
+--echo SHOW BINLOGS
+# After the fix for bug 17283409, We allow new connections while processing
+# 'SHOW PROCESSLIST' command. This test script is to prove that even though we
+# allow new connections to come in, 'SHOW PROCESSLIST' will not consider them,
+# hence there is no worry of 'SHOW PROCESSLIST' going into a infinite loop if
+# new connections are coming in continously.
+###############################################################################
+--source include/have_debug_sync.inc
+--source include/count_sessions.inc
+--enable_connect_log
+connect(connection1,127.0.0.1,root,,test,$MASTER_MYPORT,);
+--connection default
+SET DEBUG_SYNC='before_one_element_read_from_threads_iterator SIGNAL parked1 WAIT_FOR go';
+--send SHOW PROCESSLIST
+--connection connection1
+echo "Wait_for parked1";
+SET DEBUG_SYNC='now WAIT_FOR parked1';
+connect(connection2,127.0.0.1,root,,test,$MASTER_MYPORT,);
+connect(connection3,127.0.0.1,root,,test,$MASTER_MYPORT,);
+SET DEBUG_SYNC='now SIGNAL go';
+
+--connection default
+# We are yet to read the first element from the iterator when connection 2 and
+# connection 3 joined the system. So this will show all 4 connections
+# ( default + connection 1,2,3).
+--replace_column 1 <Id> 3 <Host> 5 <Command> 6 <Time> 7 <State> 8 <Info>
+--reap
+
+# Test 2: To prove that connections which joined after reading the
+# first element from the thread iterator will not be entertained
+SET DEBUG_SYNC='after_one_element_read_from_threads_iterator SIGNAL parked1 WAIT_FOR go';
+--send SHOW PROCESSLIST
+
+--connection connection1
+echo "Wait_for parked1";
+SET DEBUG_SYNC='now WAIT_FOR parked1';
+connect(connection4,127.0.0.1,root,,test,$MASTER_MYPORT,);
+connect(connection5,127.0.0.1,root,,test,$MASTER_MYPORT,);
+SET DEBUG_SYNC='now SIGNAL go';
+--connection default
+# This will show 4 connections only ( default + connection 1,2,3 ).  Connections
+# (connection4, connection 5) which entered after reading the first element
+# from the iterator will not be processed by show processlist.
+--replace_column 1 <Id> 3 <Host> 5 <Command> 6 <Time> 7 <State> 8 <Info>
+--reap
+
+#Now try the command once again.
+# This will show 6 connections ( default + connection 1,2,3,4,5 )
+--replace_column 1 <Id> 3 <Host> 5 <Command> 6 <Time> 7 <State> 8 <Info>
+SHOW PROCESSLIST;
+
+--echo "Cleanup"
+# Reset DEBUG_SYNC
+SET DEBUG_SYNC='RESET';
+--disconnect connection1
+--disconnect connection2
+--disconnect connection3
+--disconnect connection4
+--disconnect connection5
+--disable_connect_log
+--source include/wait_until_count_sessions.inc
diff --git a/sql/event_scheduler.cc b/sql/event_scheduler.cc
index 52eae66b603..2c397f15357 100644
--- a/sql/event_scheduler.cc
+++ b/sql/event_scheduler.cc
@@ -1,4 +1,4 @@
-/* Copyright (c) 2006, 2013, Oracle and/or its affiliates. All rights reserved.
+/* Copyright (c) 2006, 2014, Oracle and/or its affiliates. All rights reserved.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -22,6 +22,7 @@
 #include "event_db_repository.h"
 #include "sql_connect.h"         // init_new_connection_handler_thread
 #include "sql_acl.h"             // SUPER_ACL
+extern void delete_thd(THD *);   // Used in deinit_event_thread()
 
 /**
   @addtogroup Event_Scheduler
@@ -156,12 +157,13 @@ deinit_event_thread(THD *thd)
   DBUG_ASSERT(thd->net.buff != 0);
   net_end(&thd->net);
   DBUG_PRINT("exit", ("Event thread finishing"));
+  mysql_mutex_lock(&LOCK_thd_remove);
   mysql_mutex_lock(&LOCK_thread_count);
-  thread_count--;
   dec_thread_running();
-  delete thd;
+  delete_thd(thd);
   mysql_cond_broadcast(&COND_thread_count);
   mysql_mutex_unlock(&LOCK_thread_count);
+  mysql_mutex_unlock(&LOCK_thd_remove);
 }
 
 
diff --git a/sql/log.cc b/sql/log.cc
index b318780ea41..7327015deb5 100644
--- a/sql/log.cc
+++ b/sql/log.cc
@@ -1,4 +1,4 @@
-/* Copyright (c) 2000, 2013, Oracle and/or its affiliates. All rights reserved.
+/* Copyright (c) 2000, 2014, Oracle and/or its affiliates. All rights reserved.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -3427,6 +3427,13 @@ bool MYSQL_BIN_LOG::reset_logs(THD* thd)
 
   ha_reset_logs(thd);
 
+  /*
+    We need to get both locks to be sure that no one is trying to
+    write to the index log file.
+  */
+  mysql_mutex_lock(&LOCK_log);
+  mysql_mutex_lock(&LOCK_index);
+
   /*
     The following mutex is needed to ensure that no threads call
     'delete thd' as we would then risk missing a 'rollback' from this
@@ -3435,13 +3442,6 @@ bool MYSQL_BIN_LOG::reset_logs(THD* thd)
   */
   mysql_mutex_lock(&LOCK_thread_count);
 
-  /*
-    We need to get both locks to be sure that no one is trying to
-    write to the index log file.
-  */
-  mysql_mutex_lock(&LOCK_log);
-  mysql_mutex_lock(&LOCK_index);
-
   /* Save variables so that we can reopen the log */
   save_name=name;
   name=0;					// Protect against free
diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index b61e64e627b..1489e3cdc31 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -642,7 +642,7 @@ SHOW_COMP_OPTION have_profiling;
 pthread_key(MEM_ROOT**,THR_MALLOC);
 pthread_key(THD*, THR_THD);
 mysql_mutex_t LOCK_thread_created;
-mysql_mutex_t LOCK_thread_count;
+mysql_mutex_t LOCK_thread_count, LOCK_thd_remove;
 mysql_mutex_t
   LOCK_status, LOCK_error_log, LOCK_uuid_generator,
   LOCK_delayed_insert, LOCK_delayed_status, LOCK_delayed_create,
@@ -1607,6 +1607,7 @@ static void clean_up_mutexes()
   mysql_mutex_destroy(&LOCK_prepared_stmt_count);
   mysql_mutex_destroy(&LOCK_error_messages);
   mysql_cond_destroy(&COND_thread_count);
+  mysql_mutex_destroy(&LOCK_thd_remove);
   mysql_cond_destroy(&COND_thread_cache);
   mysql_cond_destroy(&COND_flush_thread_cache);
   mysql_cond_destroy(&COND_manager);
@@ -2160,6 +2161,8 @@ void dec_connection_count()
 
 void delete_thd(THD *thd)
 {
+  mysql_mutex_assert_owner(&LOCK_thread_count);
+  mysql_mutex_assert_owner(&LOCK_thd_remove);
   thread_count--;
   delete thd;
 }
@@ -2173,7 +2176,7 @@ void delete_thd(THD *thd)
     thd		 Thread handler
 
   NOTES
-    LOCK_thread_count is locked and left locked
+    LOCK_thread_count, LOCK_thd_remove are locked and left locked
 */
 
 void unlink_thd(THD *thd)
@@ -2183,6 +2186,7 @@ void unlink_thd(THD *thd)
 
   thd_cleanup(thd);
   dec_connection_count();
+  mysql_mutex_lock(&LOCK_thd_remove);
   mysql_mutex_lock(&LOCK_thread_count);
   /*
     Used by binlog_reset_master.  It would be cleaner to use
@@ -2295,6 +2299,7 @@ bool one_thread_per_connection_end(THD *thd, bool put_in_cache)
 {
   DBUG_ENTER("one_thread_per_connection_end");
   unlink_thd(thd);
+  mysql_mutex_unlock(&LOCK_thd_remove);
   if (put_in_cache)
     put_in_cache= cache_thread();
   mysql_mutex_unlock(&LOCK_thread_count);
@@ -3589,6 +3594,8 @@ static int init_thread_environment()
   mysql_mutex_init(key_LOCK_thread_created, &LOCK_thread_created, MY_MUTEX_INIT_FAST);
   mysql_mutex_init(key_LOCK_thread_count, &LOCK_thread_count, MY_MUTEX_INIT_FAST);
   mysql_mutex_init(key_LOCK_status, &LOCK_status, MY_MUTEX_INIT_FAST);
+  mysql_mutex_init(key_LOCK_thd_remove,
+                   &LOCK_thd_remove, MY_MUTEX_INIT_FAST);
   mysql_mutex_init(key_LOCK_delayed_insert,
                    &LOCK_delayed_insert, MY_MUTEX_INIT_FAST);
   mysql_mutex_init(key_LOCK_delayed_status,
@@ -5080,9 +5087,11 @@ void create_thread_to_handle_connection(THD *thd)
                   ER_THD(thd, ER_CANT_CREATE_THREAD), error);
       net_send_error(thd, ER_CANT_CREATE_THREAD, error_message_buff, NULL);
       close_connection(thd);
+      mysql_mutex_lock(&LOCK_thd_remove);
       mysql_mutex_lock(&LOCK_thread_count);
       delete thd;
       mysql_mutex_unlock(&LOCK_thread_count);
+      mysql_mutex_unlock(&LOCK_thd_remove);
       return;
       /* purecov: end */
     }
@@ -7873,6 +7882,7 @@ PSI_mutex_key key_BINLOG_LOCK_index, key_BINLOG_LOCK_prep_xids,
   key_structure_guard_mutex, key_TABLE_SHARE_LOCK_ha_data,
   key_LOCK_error_messages, key_LOG_INFO_lock, key_LOCK_thread_count,
   key_PARTITION_LOCK_auto_inc;
+PSI_mutex_key key_LOCK_thd_remove;
 PSI_mutex_key key_RELAYLOG_LOCK_index;
 PSI_mutex_key key_LOCK_thread_created;
 
@@ -7928,7 +7938,8 @@ static PSI_mutex_info all_server_mutexes[]=
   { &key_LOG_INFO_lock, "LOG_INFO::lock", 0},
   { &key_LOCK_thread_count, "LOCK_thread_count", PSI_FLAG_GLOBAL},
   { &key_PARTITION_LOCK_auto_inc, "HA_DATA_PARTITION::LOCK_auto_inc", 0},
-  { &key_LOCK_thread_created, "LOCK_thread_created", PSI_FLAG_GLOBAL }
+  { &key_LOCK_thread_created, "LOCK_thread_created", PSI_FLAG_GLOBAL },
+  { &key_LOCK_thd_remove, "LOCK_thd_remove", PSI_FLAG_GLOBAL}
 };
 
 PSI_rwlock_key key_rwlock_LOCK_grant, key_rwlock_LOCK_logger,
diff --git a/sql/mysqld.h b/sql/mysqld.h
index f500a0abf3f..bd93264d50d 100644
--- a/sql/mysqld.h
+++ b/sql/mysqld.h
@@ -1,4 +1,4 @@
-/* Copyright (c) 2006, 2013, Oracle and/or its affiliates. All rights reserved.
+/* Copyright (c) 2006, 2014, Oracle and/or its affiliates. All rights reserved.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -252,7 +252,8 @@ extern PSI_mutex_key key_BINLOG_LOCK_index, key_BINLOG_LOCK_prep_xids,
   key_relay_log_info_log_space_lock, key_relay_log_info_run_lock,
   key_relay_log_info_sleep_lock,
   key_structure_guard_mutex, key_TABLE_SHARE_LOCK_ha_data,
-  key_LOCK_error_messages, key_LOCK_thread_count, key_PARTITION_LOCK_auto_inc;
+  key_LOCK_error_messages, key_LOCK_thread_count, key_PARTITION_LOCK_auto_inc,
+  key_LOCK_thd_remove;
 extern PSI_mutex_key key_RELAYLOG_LOCK_index;
 
 extern PSI_rwlock_key key_rwlock_LOCK_grant, key_rwlock_LOCK_logger,
@@ -346,6 +347,7 @@ extern mysql_mutex_t
        LOCK_global_system_variables, LOCK_user_conn,
        LOCK_prepared_stmt_count, LOCK_error_messages, LOCK_connection_count;
 extern MYSQL_PLUGIN_IMPORT mysql_mutex_t LOCK_thread_count;
+extern MYSQL_PLUGIN_IMPORT mysql_mutex_t LOCK_thd_remove;
 #ifdef HAVE_OPENSSL
 extern mysql_mutex_t LOCK_des_key_file;
 #endif
diff --git a/sql/scheduler.cc b/sql/scheduler.cc
index a2abc7c14bf..fa2c43c3b65 100644
--- a/sql/scheduler.cc
+++ b/sql/scheduler.cc
@@ -1,4 +1,4 @@
-/* Copyright (c) 2007, 2011, Oracle and/or its affiliates. All rights reserved.
+/* Copyright (c) 2007, 2014, Oracle and/or its affiliates. All rights reserved.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -36,6 +36,7 @@ static bool no_threads_end(THD *thd, bool put_in_cache)
 {
   unlink_thd(thd);
   mysql_mutex_unlock(&LOCK_thread_count);
+  mysql_mutex_unlock(&LOCK_thd_remove);
   return 1;                                     // Abort handle_one_connection
 }
 
diff --git a/sql/slave.cc b/sql/slave.cc
index 0e23b1af606..23460c1af63 100644
--- a/sql/slave.cc
+++ b/sql/slave.cc
@@ -1,4 +1,4 @@
-/* Copyright (c) 2000, 2013, Oracle and/or its affiliates. All rights reserved.
+/* Copyright (c) 2000, 2014, Oracle and/or its affiliates. All rights reserved.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -3119,10 +3119,12 @@ err:
   change_rpl_status(RPL_ACTIVE_SLAVE,RPL_IDLE_SLAVE);
   DBUG_ASSERT(thd->net.buff != 0);
   net_end(&thd->net); // destructor will not free it, because net.vio is 0
+  mysql_mutex_lock(&LOCK_thd_remove);
   mysql_mutex_lock(&LOCK_thread_count);
   THD_CHECK_SENTRY(thd);
   delete thd;
   mysql_mutex_unlock(&LOCK_thread_count);
+  mysql_mutex_unlock(&LOCK_thd_remove);
   mi->abort_slave= 0;
   mi->slave_running= 0;
   mi->io_thd= 0;
@@ -3518,10 +3520,12 @@ the slave SQL thread with \"SLAVE START\". We stopped at log \
   THD_CHECK_SENTRY(thd);
   rli->sql_thd= 0;
   set_thd_in_use_temporary_tables(rli);  // (re)set sql_thd in use for saved temp tables
+  mysql_mutex_lock(&LOCK_thd_remove);
   mysql_mutex_lock(&LOCK_thread_count);
   THD_CHECK_SENTRY(thd);
   delete thd;
   mysql_mutex_unlock(&LOCK_thread_count);
+  mysql_mutex_unlock(&LOCK_thd_remove);
  /*
   Note: the order of the broadcast and unlock calls below (first broadcast, then unlock)
   is important. Otherwise a killer_thread can execute between the calls and
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index 93f1d8eb3bb..136c6012514 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -1,5 +1,5 @@
 /*
-   Copyright (c) 2000, 2013, Oracle and/or its affiliates. All rights reserved.
+   Copyright (c) 2000, 2014, Oracle and/or its affiliates. All rights reserved.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -332,6 +332,26 @@ void thd_unlock_thread_count(THD *)
   mysql_mutex_unlock(&LOCK_thread_count);
 }
 
+/**
+  Lock thread removal mutex
+
+  @param thd                       THD object
+*/
+void thd_lock_thread_remove(THD *)
+{
+  mysql_mutex_lock(&LOCK_thd_remove);
+}
+
+/**
+  Unlock thread removal mutex
+
+  @param thd                       THD object
+*/
+void thd_unlock_thread_remove(THD *)
+{
+  mysql_mutex_unlock(&LOCK_thd_remove);
+}
+
 /**
   Close the socket used by this connection
 
@@ -1561,6 +1581,13 @@ void THD::awake(THD::killed_state state_to_set)
     */
     if (mysys_var->current_cond && mysys_var->current_mutex)
     {
+      DBUG_EXECUTE_IF("before_dump_thread_acquires_current_mutex",
+                      {
+                      const char act[]=
+                      "now signal dump_thread_signal wait_for go";
+                      DBUG_ASSERT(!debug_sync_set_action(current_thd,
+                                                         STRING_WITH_LEN(act)));
+                      };);
       mysql_mutex_lock(mysys_var->current_mutex);
       mysql_cond_broadcast(mysys_var->current_cond);
       mysql_mutex_unlock(mysys_var->current_mutex);
diff --git a/sql/sql_insert.cc b/sql/sql_insert.cc
index 0f432779305..d6a83b9c35e 100644
--- a/sql/sql_insert.cc
+++ b/sql/sql_insert.cc
@@ -1,5 +1,5 @@
 /*
-   Copyright (c) 2000, 2013, Oracle and/or its affiliates. All rights reserved.
+   Copyright (c) 2000, 2014, Oracle and/or its affiliates. All rights reserved.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -1933,6 +1933,7 @@ public:
       close_thread_tables(&thd);
       thd.mdl_context.release_transactional_locks();
     }
+    mysql_mutex_lock(&LOCK_thd_remove);
     mysql_mutex_lock(&LOCK_thread_count);
     mysql_mutex_destroy(&mutex);
     mysql_cond_destroy(&cond);
@@ -1944,6 +1945,7 @@ public:
     thread_count--;
     delayed_insert_threads--;
     mysql_mutex_unlock(&LOCK_thread_count);
+    mysql_mutex_unlock(&LOCK_thd_remove);
     mysql_cond_broadcast(&COND_thread_count); /* Tell main we are ready */
   }
 
diff --git a/sql/sql_repl.cc b/sql/sql_repl.cc
index ec1ef72cd73..343e128af7a 100644
--- a/sql/sql_repl.cc
+++ b/sql/sql_repl.cc
@@ -1,4 +1,4 @@
-/* Copyright (c) 2000, 2013, Oracle and/or its affiliates. All rights reserved.
+/* Copyright (c) 2000, 2014, Oracle and/or its affiliates. All rights reserved.
 
    This program is free software; you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
@@ -248,7 +248,7 @@ bool log_in_use(const char* log_name)
   size_t log_name_len = strlen(log_name) + 1;
   THD *tmp;
   bool result = 0;
-
+  DEBUG_SYNC(current_thd,"purge_logs_after_lock_index_before_thread_count");
   mysql_mutex_lock(&LOCK_thread_count);
   I_List_iterator<THD> it(threads);
 
@@ -1965,6 +1965,7 @@ bool show_binlogs(THD* thd)
     DBUG_RETURN(TRUE);
   
   mysql_mutex_lock(mysql_bin_log.get_log_lock());
+  DEBUG_SYNC(thd, "show_binlogs_after_lock_log_before_lock_index");
   mysql_bin_log.lock_index();
   index_file=mysql_bin_log.get_index_file();
   
diff --git a/sql/sql_show.cc b/sql/sql_show.cc
index ac63b2aaff9..f02ea6bead7 100644
--- a/sql/sql_show.cc
+++ b/sql/sql_show.cc
@@ -1814,13 +1814,23 @@ void mysqld_list_processes(THD *thd,const char *user, bool verbose)
                             Protocol::SEND_NUM_ROWS | Protocol::SEND_EOF))
     DBUG_VOID_RETURN;
 
-  mysql_mutex_lock(&LOCK_thread_count); // For unlink from list
+
   if (!thd->killed)
   {
+    /*
+      Acquire only LOCK_thd_remove and not LOCK_thread_count.
+      i.e., we allow new threads to be added to the list while processing
+      the list but we will not allow deletions from the list (Note that unlink
+      a thread is protected by both LOCK_thd_remove and LOCK_thread_count
+      mutexes).
+    */
+    mysql_mutex_lock(&LOCK_thd_remove);
     I_List_iterator<THD> it(threads);
     THD *tmp;
+    DEBUG_SYNC(thd,"before_one_element_read_from_threads_iterator");
     while ((tmp=it++))
     {
+      DEBUG_SYNC(thd,"after_one_element_read_from_threads_iterator");
       Security_context *tmp_sctx= tmp->security_ctx;
       struct st_my_thread_var *mysys_var;
       if ((tmp->vio_ok() || tmp->system_thread) &&
@@ -1845,6 +1855,11 @@ void mysqld_list_processes(THD *thd,const char *user, bool verbose)
                                       tmp_sctx->get_host()->length() ?
                                       tmp_sctx->get_host()->ptr() : "");
         thd_info->command=(int) tmp->command;
+        DBUG_EXECUTE_IF("processlist_acquiring_dump_threads_LOCK_thd_data",
+                        {
+                         if (tmp->command == COM_BINLOG_DUMP)
+                           DEBUG_SYNC(thd, "processlist_after_LOCK_thd_count_before_LOCK_thd_data");
+                        });
         mysql_mutex_lock(&tmp->LOCK_thd_data);
         if ((thd_info->db= tmp->db))             // Safe test
           thd_info->db= thd->strdup(thd_info->db);
@@ -1869,8 +1884,8 @@ void mysqld_list_processes(THD *thd,const char *user, bool verbose)
         thread_infos.append(thd_info);
       }
     }
+    mysql_mutex_unlock(&LOCK_thd_remove);
   }
-  mysql_mutex_unlock(&LOCK_thread_count);
 
   thread_info *thd_info;
   time_t now= my_time(0);
@@ -1910,10 +1925,15 @@ int fill_schema_processlist(THD* thd, TABLE_LIST* tables, COND* cond)
   user= thd->security_ctx->master_access & PROCESS_ACL ?
         NullS : thd->security_ctx->priv_user;
 
-  mysql_mutex_lock(&LOCK_thread_count);
-
   if (!thd->killed)
   {
+    /*
+      Acquire only LOCK_thd_remove and not LOCK_thread_count.
+      i.e., we allow new threads to be added to the list while processing
+      the list but we will not allow deletions from the list (Note that unlink
+      thread is also protected with LOCK_thd_remove mutex).
+    */
+    mysql_mutex_lock(&LOCK_thd_remove);
     I_List_iterator<THD> it(threads);
     THD* tmp;
 
@@ -1946,7 +1966,13 @@ int fill_schema_processlist(THD* thd, TABLE_LIST* tables, COND* cond)
       else
         table->field[2]->store(tmp_sctx->host_or_ip,
                                strlen(tmp_sctx->host_or_ip), cs);
+      DBUG_EXECUTE_IF("processlist_acquiring_dump_threads_LOCK_thd_data",
+                      {
+                       if (tmp->command == COM_BINLOG_DUMP)
+                         DEBUG_SYNC(thd, "processlist_after_LOCK_thd_count_before_LOCK_thd_data");
+                      });
       /* DB */
+      /* Lock THD mutex that protects its data when looking at it. */
       mysql_mutex_lock(&tmp->LOCK_thd_data);
       if ((db= tmp->db))
       {
@@ -1974,11 +2000,8 @@ int fill_schema_processlist(THD* thd, TABLE_LIST* tables, COND* cond)
 
       if (mysys_var)
         mysql_mutex_unlock(&mysys_var->mutex);
-      mysql_mutex_unlock(&tmp->LOCK_thd_data);
 
       /* INFO */
-      /* Lock THD mutex that protects its data when looking at it. */
-      mysql_mutex_lock(&tmp->LOCK_thd_data);
       if (tmp->query())
       {
         table->field[7]->store(tmp->query(),
@@ -1990,13 +2013,12 @@ int fill_schema_processlist(THD* thd, TABLE_LIST* tables, COND* cond)
 
       if (schema_table_store_record(thd, table))
       {
-        mysql_mutex_unlock(&LOCK_thread_count);
+        mysql_mutex_unlock(&LOCK_thd_remove);
         DBUG_RETURN(1);
       }
     }
+    mysql_mutex_unlock(&LOCK_thd_remove);
   }
-
-  mysql_mutex_unlock(&LOCK_thread_count);
   DBUG_RETURN(0);
 }