WL#1265: Fix proper ALTER/DROP support in the SP cache.

New sp_cache C API. When an SP is dropped, old caches (in other threads)
become invalid and are cleared.
Also, the caches in THD are only created on demand.


Docs/sp-imp-spec.txt:
  Brough the SP cache docs up-to-date.
sql/mysqld.cc:
  Initialize SP cache.
sql/sp.cc:
  New C API for SP cache.
sql/sp_cache.cc:
  New C API for sp_cache.
  The class sp_cache is still used, but not directly. The C functions makes takes
  care of updating caches when SPs are dropped. (This is done in the simplest
  possible way, by simply detecting drops and then clear all old caches.)
  The API is also designed so that the sp_cache is created on demand.
sql/sp_cache.h:
  New C API for sp_cache.
  The class sp_cache is still used, but not directly. The C functions makes takes
  care of updating caches when SPs are dropped.
  The API is also designed so that the sp_cache is created on demand.
sql/sql_class.cc:
  The new sp_cache API creates the caches on demand, to avoid allocating it
  when it's not needed.

5 files changed, 145 insertions, 18 deletions

diff --git a/sql/mysqld.cc b/sql/mysqld.cc
index 33964feac52..cb7c1fcf211 100644
--- a/sql/mysqld.cc
+++ b/sql/mysqld.cc
@@ -38,6 +38,7 @@
 #include <ft_global.h>
 #include <errmsg.h>
 #include "sp_rcontext.h"
+#include "sp_cache.h"
 
 #define mysqld_charset &my_charset_latin1
 
@@ -2140,6 +2141,7 @@ static int init_thread_environment()
   (void) pthread_mutex_init(&LOCK_rpl_status, MY_MUTEX_INIT_FAST);
   (void) pthread_cond_init(&COND_rpl_status, NULL);
 #endif
+  sp_cache_init();
   /* Parameter for threads created for connections */
   (void) pthread_attr_init(&connection_attrib);
   (void) pthread_attr_setdetachstate(&connection_attrib,
diff --git a/sql/sp.cc b/sql/sp.cc
index b4f9f85cd69..b1cdad4f619 100644
--- a/sql/sp.cc
+++ b/sql/sp.cc
@@ -251,13 +251,13 @@ sp_find_procedure(THD *thd, LEX_STRING *name)
 
   DBUG_PRINT("enter", ("name: %*s", name->length, name->str));
 
-  sp= thd->sp_proc_cache->lookup(name->str, name->length);
+  sp= sp_cache_lookup(&thd->sp_proc_cache, name->str, name->length);
   if (! sp)
   {
     if (db_find_routine(thd, TYPE_ENUM_PROCEDURE,
 			name->str, name->length, &sp) == SP_OK)
     {
-      thd->sp_proc_cache->insert(sp);
+      sp_cache_insert(&thd->sp_proc_cache, sp);
     }
   }
 
@@ -286,10 +286,10 @@ sp_drop_procedure(THD *thd, char *name, uint namelen)
   sp_head *sp;
   int ret;
 
-  sp= thd->sp_proc_cache->lookup(name, namelen);
+  sp= sp_cache_lookup(&thd->sp_proc_cache, name, namelen);
   if (sp)
   {
-    thd->sp_proc_cache->remove(sp);
+    sp_cache_remove(&thd->sp_proc_cache, sp);
     delete sp;
   }
   ret= db_drop_routine(thd, TYPE_ENUM_PROCEDURE, name, namelen);
@@ -312,7 +312,7 @@ sp_find_function(THD *thd, LEX_STRING *name)
 
   DBUG_PRINT("enter", ("name: %*s", name->length, name->str));
 
-  sp= thd->sp_func_cache->lookup(name->str, name->length);
+  sp= sp_cache_lookup(&thd->sp_func_cache, name->str, name->length);
   if (! sp)
   {
     if (db_find_routine(thd, TYPE_ENUM_FUNCTION,
@@ -344,10 +344,10 @@ sp_drop_function(THD *thd, char *name, uint namelen)
   sp_head *sp;
   int ret;
 
-  sp= thd->sp_func_cache->lookup(name, namelen);
+  sp= sp_cache_lookup(&thd->sp_func_cache, name, namelen);
   if (sp)
   {
-    thd->sp_func_cache->remove(sp);
+    sp_cache_remove(&thd->sp_func_cache, sp);
     delete sp;
   }
   ret= db_drop_routine(thd, TYPE_ENUM_FUNCTION, name, namelen);
@@ -363,7 +363,7 @@ sp_function_exists(THD *thd, LEX_STRING *name)
   bool ret= FALSE;
   bool opened= FALSE;
 
-  if (thd->sp_func_cache->lookup(name->str, name->length) ||
+  if (sp_cache_lookup(&thd->sp_func_cache, name->str, name->length) ||
       db_find_routine_aux(thd, TYPE_ENUM_FUNCTION,
 			  name->str, name->length, TL_READ,
 			  &table, &opened) == SP_OK)
@@ -419,7 +419,7 @@ sp_cache_functions(THD *thd, LEX *lex)
   {
     LEX_STRING *ls= (LEX_STRING *)hash_element(h, i);
 
-    if (! thd->sp_func_cache->lookup(ls->str, ls->length))
+    if (! sp_cache_lookup(&thd->sp_func_cache, ls->str, ls->length))
     {
       sp_head *sp;
       LEX *oldlex= thd->lex;
@@ -434,7 +434,7 @@ sp_cache_functions(THD *thd, LEX *lex)
 	thd->lex= oldlex;
 	if (ret)
 	  break;
-	thd->sp_func_cache->insert(sp);
+	sp_cache_insert(&thd->sp_func_cache, sp);
       }
       else
       {
diff --git a/sql/sp_cache.cc b/sql/sp_cache.cc
index 1b8816ee0f4..84e3565d543 100644
--- a/sql/sp_cache.cc
+++ b/sql/sp_cache.cc
@@ -22,6 +22,98 @@
 #include "sp_cache.h"
 #include "sp_head.h"
 
+static pthread_mutex_t Cversion_lock;
+static ulong Cversion = 0;
+
+void
+sp_cache_init()
+{
+  pthread_mutex_init(&Cversion_lock, MY_MUTEX_INIT_FAST);
+}
+
+void
+sp_cache_clear(sp_cache **cp)
+{
+  sp_cache *c= *cp;
+
+  if (c)
+  {
+    delete c;
+    *cp= NULL;
+  }
+}
+
+void
+sp_cache_insert(sp_cache **cp, sp_head *sp)
+{
+  sp_cache *c= *cp;
+
+  if (! c)
+    c= new sp_cache();
+  if (c)
+  {
+    ulong v;
+
+    pthread_mutex_lock(&Cversion_lock); // LOCK
+    v= Cversion;
+    pthread_mutex_unlock(&Cversion_lock); // UNLOCK
+
+    if (c->version < v)
+    {
+      if (*cp)
+	c->remove_all();
+      c->version= v;
+    }
+    c->insert(sp);
+    if (*cp == NULL)
+      *cp= c;
+  }
+}
+
+sp_head *
+sp_cache_lookup(sp_cache **cp, char *name, uint namelen)
+{
+  ulong v;
+  sp_cache *c= *cp;
+
+  if (! c)
+    return NULL;
+
+  pthread_mutex_lock(&Cversion_lock); // LOCK
+  v= Cversion;
+  pthread_mutex_unlock(&Cversion_lock); // UNLOCK
+
+  if (c->version < v)
+  {
+    c->remove_all();
+    c->version= v;
+    return NULL;
+  }
+  return c->lookup(name, namelen);
+}
+
+void
+sp_cache_remove(sp_cache **cp, sp_head *sp)
+{
+  sp_cache *c= *cp;
+
+  if (c)
+  {
+    ulong v;
+
+    pthread_mutex_lock(&Cversion_lock); // LOCK
+    v= Cversion++;
+    pthread_mutex_unlock(&Cversion_lock); // UNLOCK
+
+    if (c->version < v)
+      c->remove_all();
+    else
+      c->remove(sp);
+    c->version= v+1;
+  }
+}
+
+
 static byte *
 hash_get_key_for_sp_head(const byte *ptr, uint *plen,
 			       my_bool first)
@@ -44,6 +136,7 @@ sp_cache::init()
 {
   hash_init(&m_hashtable, system_charset_info, 0, 0, 0,
 	    hash_get_key_for_sp_head, 0, 0);
+  version= 0;
 }
 
 void
diff --git a/sql/sp_cache.h b/sql/sp_cache.h
index 5e7825d70fc..f5b330f6755 100644
--- a/sql/sp_cache.h
+++ b/sql/sp_cache.h
@@ -23,11 +23,36 @@
 #endif
 
 class sp_head;
+class sp_cache;
+
+/* Initialize the SP caching once at startup */
+void sp_cache_init();
+
+/* Clear the cache *cp and set *cp to NULL */
+void sp_cache_clear(sp_cache **cp);
+
+/* Insert an SP to cache. If 'cp' points to NULL, it's set to a new cache */
+void sp_cache_insert(sp_cache **cp, sp_head *sp);
+
+/* Lookup an SP in cache */
+sp_head *sp_cache_lookup(sp_cache **cp, char *name, uint namelen);
+
+/* Remove an SP from cache */
+void sp_cache_remove(sp_cache **cp, sp_head *sp);
+
+
+/*
+ *
+ * The cache class. Don't use this directly, use the C API above
+ *
+ */
 
 class sp_cache
 {
 public:
 
+  ulong version;
+
   sp_cache();
 
   ~sp_cache();
@@ -56,6 +81,13 @@ public:
     hash_delete(&m_hashtable, (byte *)sp);
   }
 
+  inline void
+  remove_all()
+  {
+    cleanup();
+    init();
+  }
+
 private:
 
   HASH m_hashtable;
diff --git a/sql/sql_class.cc b/sql/sql_class.cc
index 487ae14f4cd..025b1d7429d 100644
--- a/sql/sql_class.cc
+++ b/sql/sql_class.cc
@@ -156,8 +156,8 @@ THD::THD():user_time(0), is_fatal_error(0),
 	    (hash_get_key) get_var_key,
 	    (hash_free_key) free_user_var, 0);
 
-  sp_proc_cache= new sp_cache();
-  sp_func_cache= new sp_cache();
+  sp_proc_cache= NULL;
+  sp_func_cache= NULL;
 
   /* For user vars replication*/
   if (opt_bin_log)
@@ -260,8 +260,8 @@ void THD::change_user(void)
   hash_init(&user_vars, system_charset_info, USER_VARS_HASH_SIZE, 0, 0,
 	    (hash_get_key) get_var_key,
 	    (hash_free_key) free_user_var, 0);
-  sp_proc_cache->init();
-  sp_func_cache->init();
+  sp_cache_clear(&sp_proc_cache);
+  sp_cache_clear(&sp_func_cache);
 }
 
 
@@ -285,8 +285,8 @@ void THD::cleanup(void)
   close_temporary_tables(this);
   delete_dynamic(&user_var_events);
   hash_free(&user_vars);
-  sp_proc_cache->cleanup();
-  sp_func_cache->cleanup();
+  sp_cache_clear(&sp_proc_cache);
+  sp_cache_clear(&sp_func_cache);
   if (global_read_lock)
     unlock_global_read_lock(this);
   if (ull)
@@ -328,8 +328,8 @@ THD::~THD()
   }
 #endif
 
-  delete sp_proc_cache;
-  delete sp_func_cache;
+  sp_cache_clear(&sp_proc_cache);
+  sp_cache_clear(&sp_func_cache);
 
   DBUG_PRINT("info", ("freeing host"));
   if (host != localhost)			// If not pointer to constant