merging.

client/mysqltest.cc

@@ -248,10 +248,14 @@ struct st_connection
   my_bool pending;
 
 #ifdef EMBEDDED_LIBRARY
+  pthread_t tid;
   const char *cur_query;
   int cur_query_len;
-  pthread_mutex_t mutex;
-  pthread_cond_t cond;
+  int command, result;
+  pthread_mutex_t query_mutex;
+  pthread_cond_t query_cond;
+  pthread_mutex_t result_mutex;
+  pthread_cond_t result_cond;
   int query_done;
 #endif /*EMBEDDED_LIBRARY*/
 };
@@ -703,67 +707,148 @@ void handle_no_error(struct st_command*);
 
 #ifdef EMBEDDED_LIBRARY
 
+#define EMB_SEND_QUERY 1
+#define EMB_READ_QUERY_RESULT 2
+#define EMB_END_CONNECTION 3
+
 /* attributes of the query thread */
 pthread_attr_t cn_thd_attrib;
 
+
 /*
-  send_one_query executes query in separate thread, which is
-  necessary in embedded library to run 'send' in proper way.
-  This implementation doesn't handle errors returned
-  by mysql_send_query. It's technically possible, though
-  I don't see where it is needed.
+  This procedure represents the connection and actually
+  runs queries when in the EMBEDDED-SERVER mode.
+  The run_query_normal() just sends request for running
+  mysql_send_query and mysql_read_query_result() here.
 */
-pthread_handler_t send_one_query(void *arg)
+
+pthread_handler_t connection_thread(void *arg)
 {
   struct st_connection *cn= (struct st_connection*)arg;
 
   mysql_thread_init();
-  (void) mysql_send_query(&cn->mysql, cn->cur_query, cn->cur_query_len);
+  while (cn->command != EMB_END_CONNECTION)
+  {
+    if (!cn->command)
+    {
+      pthread_mutex_lock(&cn->query_mutex);
+      while (!cn->command)
+        pthread_cond_wait(&cn->query_cond, &cn->query_mutex);
+      pthread_mutex_unlock(&cn->query_mutex);
+    }
+    switch (cn->command)
+    {
+      case EMB_END_CONNECTION:
+        goto end_thread;
+      case EMB_SEND_QUERY:
+        cn->result= mysql_send_query(&cn->mysql, cn->cur_query, cn->cur_query_len);
+        break;
+      case EMB_READ_QUERY_RESULT:
+        cn->result= mysql_read_query_result(&cn->mysql);
+        break;
+      default:
+        DBUG_ASSERT(0);
+    }
+    cn->command= 0;
+    pthread_mutex_lock(&cn->result_mutex);
+    cn->query_done= 1;
+    pthread_cond_signal(&cn->result_cond);
+    pthread_mutex_unlock(&cn->result_mutex);
+  }
 
-  mysql_thread_end();
-  pthread_mutex_lock(&cn->mutex);
+end_thread:
   cn->query_done= 1;
-  pthread_cond_signal(&cn->cond);
-  pthread_mutex_unlock(&cn->mutex);
+  mysql_thread_end();
   pthread_exit(0);
   return 0;
 }
 
-static int do_send_query(struct st_connection *cn, const char *q, int q_len,
-                         int flags)
-{
-  pthread_t tid;
-
-  if (flags & QUERY_REAP_FLAG)
-    return mysql_send_query(&cn->mysql, q, q_len);
-
-  if (pthread_mutex_init(&cn->mutex, NULL) ||
-      pthread_cond_init(&cn->cond, NULL))
-    die("Error in the thread library");
-
-  cn->cur_query= q;
-  cn->cur_query_len= q_len;
-  cn->query_done= 0;
-  if (pthread_create(&tid, &cn_thd_attrib, send_one_query, (void*)cn))
-    die("Cannot start new thread for query");
-
-  return 0;
-}
-
-static void wait_query_thread_end(struct st_connection *con)
+
+static void wait_query_thread_done(struct st_connection *con)
 {
+  DBUG_ASSERT(con->tid);
   if (!con->query_done)
   {
-    pthread_mutex_lock(&con->mutex);
+    pthread_mutex_lock(&con->result_mutex);
     while (!con->query_done)
-      pthread_cond_wait(&con->cond, &con->mutex);
-    pthread_mutex_unlock(&con->mutex);
+      pthread_cond_wait(&con->result_cond, &con->result_mutex);
+    pthread_mutex_unlock(&con->result_mutex);
   }
 }
 
+
+static void signal_connection_thd(struct st_connection *cn, int command)
+{
+  DBUG_ASSERT(cn->tid);
+  cn->query_done= 0;
+  cn->command= command;
+  pthread_mutex_lock(&cn->query_mutex);
+  pthread_cond_signal(&cn->query_cond);
+  pthread_mutex_unlock(&cn->query_mutex);
+}
+
+
+/*
+  Sometimes we try to execute queries when the connection is closed.
+  It's done to make sure it was closed completely.
+  So that if our connection is closed (cn->tid == 0), we just return
+  the mysql_send_query() result which is an error in this case.
+*/
+
+static int do_send_query(struct st_connection *cn, const char *q, int q_len)
+{
+  if (!cn->tid)
+    return mysql_send_query(&cn->mysql, q, q_len);
+  cn->cur_query= q;
+  cn->cur_query_len= q_len;
+  signal_connection_thd(cn, EMB_SEND_QUERY);
+  return 0;
+}
+
+
+static int do_read_query_result(struct st_connection *cn)
+{
+  DBUG_ASSERT(cn->tid);
+  wait_query_thread_done(cn);
+  signal_connection_thd(cn, EMB_READ_QUERY_RESULT);
+  wait_query_thread_done(cn);
+
+  return cn->result;
+}
+
+
+static void emb_close_connection(struct st_connection *cn)
+{
+  if (!cn->tid)
+    return;
+  wait_query_thread_done(cn);
+  signal_connection_thd(cn, EMB_END_CONNECTION);
+  pthread_join(cn->tid, NULL);
+  cn->tid= 0;
+  pthread_mutex_destroy(&cn->query_mutex);
+  pthread_cond_destroy(&cn->query_cond);
+  pthread_mutex_destroy(&cn->result_mutex);
+  pthread_cond_destroy(&cn->result_cond);
+}
+
+
+static void init_connection_thd(struct st_connection *cn)
+{
+  cn->query_done= 1;
+  cn->command= 0;
+  if (pthread_mutex_init(&cn->query_mutex, NULL) ||
+      pthread_cond_init(&cn->query_cond, NULL) ||
+      pthread_mutex_init(&cn->result_mutex, NULL) ||
+      pthread_cond_init(&cn->result_cond, NULL) ||
+      pthread_create(&cn->tid, &cn_thd_attrib, connection_thread, (void*)cn))
+    die("Error in the thread library");
+}
+
+
 #else /*EMBEDDED_LIBRARY*/
 
-#define do_send_query(cn,q,q_len,flags) mysql_send_query(&cn->mysql, q, q_len)
+#define do_send_query(cn,q,q_len) mysql_send_query(&cn->mysql, q, q_len)
+#define do_read_query_result(cn) mysql_read_query_result(&cn->mysql)
 
 #endif /*EMBEDDED_LIBRARY*/
 
@@ -1106,6 +1191,9 @@ void close_connections()
   DBUG_ENTER("close_connections");
   for (--next_con; next_con >= connections; --next_con)
   {
+#ifdef EMBEDDED_LIBRARY
+    emb_close_connection(next_con);
+#endif
     if (next_con->stmt)
       mysql_stmt_close(next_con->stmt);
     next_con->stmt= 0;
@@ -4866,7 +4954,7 @@ void do_close_connection(struct st_command *command)
     we need to check if the query's thread was finished and probably wait
     (embedded-server specific)
   */
-  wait_query_thread_end(con);
+  emb_close_connection(con);
 #endif /*EMBEDDED_LIBRARY*/
   if (con->stmt)
     mysql_stmt_close(con->stmt);
@@ -5216,8 +5304,9 @@ void do_connect(struct st_command *command)
   }
 
 #ifdef EMBEDDED_LIBRARY
-  con_slot->query_done= 1;
-#endif
+  init_connection_thd(con_slot);
+#endif /*EMBEDDED_LIBRARY*/
+
   if (!mysql_init(&con_slot->mysql))
     die("Failed on mysql_init()");
 
@@ -6768,21 +6857,13 @@ void run_query_normal(struct st_connection *cn, struct st_command *command,
     /*
       Send the query
     */
-    if (do_send_query(cn, query, query_len, flags))
+    if (do_send_query(cn, query, query_len))
     {
       handle_error(command, mysql_errno(mysql), mysql_error(mysql),
 		   mysql_sqlstate(mysql), ds);
       goto end;
     }
   }
-#ifdef EMBEDDED_LIBRARY
-  /*
-    Here we handle 'reap' command, so we need to check if the
-    query's thread was finished and probably wait
-  */
-  else if (flags & QUERY_REAP_FLAG)
-    wait_query_thread_end(cn);
-#endif /*EMBEDDED_LIBRARY*/
   if (!(flags & QUERY_REAP_FLAG))
   {
     cn->pending= TRUE;
@@ -6795,7 +6876,7 @@ void run_query_normal(struct st_connection *cn, struct st_command *command,
       When  on first result set, call mysql_read_query_result to retrieve
       answer to the query sent earlier
     */
-    if ((counter==0) && mysql_read_query_result(mysql))
+    if ((counter==0) && do_read_query_result(cn))
     {
       handle_error(command, mysql_errno(mysql), mysql_error(mysql),
 		   mysql_sqlstate(mysql), ds);
@@ -7970,6 +8051,9 @@ int main(int argc, char **argv)
     ps_protocol_enabled= 1;
 
   st_connection *con= connections;
+#ifdef EMBEDDED_LIBRARY
+  init_connection_thd(con);
+#endif /*EMBEDDED_LIBRARY*/
   if (!( mysql_init(&con->mysql)))
     die("Failed in mysql_init()");
   if (opt_connect_timeout)