APC: support waking up sleeping threads;

sql/my_apc.cc

@@ -133,9 +133,9 @@ int Apc_target::wait_for_completion(THD *caller_thd, Call_request *apc_request,
   int wait_res= 0;
   PSI_stage_info old_stage;
 
-  mysql_mutex_lock(&apc_request->LOCK_request);
   mysql_mutex_unlock(LOCK_thd_kill_ptr);
 
+  mysql_mutex_lock(&apc_request->LOCK_request);
   caller_thd->ENTER_COND(&apc_request->COND_request, &apc_request->LOCK_request,
                          &stage_show_explain, &old_stage);
   /* todo: how about processing other errors here? */
@@ -145,6 +145,13 @@ int Apc_target::wait_for_completion(THD *caller_thd, Call_request *apc_request,
                                    &apc_request->LOCK_request, &abstime);
     if (caller_thd->killed)
       break;
+
+    if (caller_thd->apc_target.have_apc_requests())
+    {
+      mysql_mutex_unlock(&apc_request->LOCK_request);
+      caller_thd->apc_target.process_apc_requests();
+      mysql_mutex_lock(&apc_request->LOCK_request);
+    }
   }
 
   if (!apc_request->processed)

sql/sql_class.cc

@@ -1910,13 +1910,13 @@ void THD::awake_no_mutex(killed_state state_to_set)
 }
 
 /* Broadcast a condition to kick the target if it is waiting on it. */
-void THD::abort_current_cond_wait(bool force)
+void THD::abort_current_cond_wait(bool force, bool mark_abort)
 {
   mysql_mutex_assert_owner(&LOCK_thd_kill);
   if (mysys_var)
   {
     mysql_mutex_lock(&mysys_var->mutex);
-    if (!system_thread || force)                 // Don't abort locks
+    if (mark_abort && (!system_thread || force))  // Don't abort locks
       mysys_var->abort=1;
 
     /*

sql/sql_class.h

@@ -3795,7 +3795,7 @@ class THD: public THD_count, /* this must be first */
     mysql_mutex_unlock(&LOCK_thd_data);
     mysql_mutex_unlock(&LOCK_thd_kill);
   }
-  void abort_current_cond_wait(bool force);
+  void abort_current_cond_wait(bool force, bool mark_abort= true);
  
   /** Disconnect the associated communication endpoint. */
   void disconnect();

storage/perfschema/pfs_variable.cc

@@ -397,6 +397,7 @@ int PFS_system_variable_cache::make_call(Request_func func, uint param)
     PFS_system_variable_cache_apc apc_call(this, func, param);
     auto *request= new Apc_target::Call_request;
     m_safe_thd->apc_target.enqueue_request(request, &apc_call);
+    m_safe_thd->abort_current_cond_wait(false, false);
     m_safe_thd->scheduler->notify_apc(m_safe_thd);
     DEBUG_SYNC(requestor_thd, "apc_after_notify");
     ret= m_safe_thd->apc_target.wait_for_completion(requestor_thd, request, 10);