fixed slave to clean up load data infile temp files

instrumented the server with DBUG_SYNC_POINT() macro


mysql-test/t/rpl_log.test:
  no change
sql/item_func.cc:
  debug_sync_point()
sql/log_event.cc:
  clead load data file tmpdir on slave on start/stop events
sql/mysql_priv.h:
  DBUG_SYNC_POINT
sql/sql_repl.cc:
  DBUG_SYNC_POINT in Create_file_event

mysql-test/t/rpl_log.test

@@ -46,3 +46,4 @@ show new master for slave with master_log_file='master-bin.002' and
  master_log_pos=4 and master_log_seq=1 and master_server_id=1;
 show new master for slave with master_log_file='master-bin.002' and
  master_log_pos=137 and master_log_seq=3 and master_server_id=1;
+

sql/item_func.cc

@@ -1402,19 +1402,16 @@ void item_user_lock_release(ULL *ull)
   if (mysql_bin_log.is_open())
   {
     THD *thd = current_thd;
-    int save_errno;
     char buf[256];
     String tmp(buf,sizeof(buf));
     tmp.length(0);
     tmp.append("SELECT release_lock(\"");
     tmp.append(ull->key,ull->key_length);
     tmp.append("\")");
-    save_errno=thd->net.last_errno;
-    thd->net.last_errno=0;
     thd->query_length=tmp.length();
     Query_log_event qev(thd,tmp.ptr());
+    qev.error_code=0; // this query is always safe to run on slave
     mysql_bin_log.write(&qev);
-    thd->net.last_errno=save_errno;
   }
   if (--ull->count)
     pthread_cond_signal(&ull->cond);
@@ -1448,6 +1445,80 @@ longlong Item_master_pos_wait::val_int()
   return event_count;
 }
 
+#ifdef EXTRA_DEBUG
+void debug_sync_point(const char* lock_name, uint lock_timeout)
+{
+  THD* thd=current_thd;
+  ULL* ull;
+  struct timespec abstime;
+  int lock_name_len,error=0;
+  lock_name_len=strlen(lock_name);
+  pthread_mutex_lock(&LOCK_user_locks);
+
+  if (thd->ull)
+  {
+    item_user_lock_release(thd->ull);
+    thd->ull=0;
+  }
+
+  /* if the lock has not been aquired by some client, we do not want to
+     create an entry for it, since we immediately release the lock. In
+     this case, we will not be waiting, but rather, just waste CPU and
+     memory on the whole deal
+  */
+  if (!(ull= ((ULL*) hash_search(&hash_user_locks,lock_name,
+				 lock_name_len))))
+  {
+    pthread_mutex_unlock(&LOCK_user_locks);
+    return;
+  }
+  ull->count++;
+
+  /* structure is now initialized.  Try to get the lock */
+  /* Set up control struct to allow others to abort locks */
+  thd->proc_info="User lock";
+  thd->mysys_var->current_mutex= &LOCK_user_locks;
+  thd->mysys_var->current_cond=  &ull->cond;
+
+#ifdef HAVE_TIMESPEC_TS_SEC
+  abstime.ts_sec=time((time_t*) 0)+(time_t) lock_timeout;
+  abstime.ts_nsec=0;
+#else
+  abstime.tv_sec=time((time_t*) 0)+(time_t) lock_timeout;
+  abstime.tv_nsec=0;
+#endif
+
+  while (!thd->killed &&
+	 (error=pthread_cond_timedwait(&ull->cond,&LOCK_user_locks,&abstime))
+	 != ETIME && error != ETIMEDOUT && ull->locked) ;
+  if (ull->locked)
+  {
+    if (!--ull->count)
+      delete ull;				// Should never happen
+  }
+  else
+  {
+    ull->locked=1;
+    ull->thread=thd->real_id;
+    thd->ull=ull;
+  }
+  pthread_mutex_unlock(&LOCK_user_locks);
+  pthread_mutex_lock(&thd->mysys_var->mutex);
+  thd->proc_info=0;
+  thd->mysys_var->current_mutex= 0;
+  thd->mysys_var->current_cond=  0;
+  pthread_mutex_unlock(&thd->mysys_var->mutex);
+  pthread_mutex_lock(&LOCK_user_locks);
+  if (thd->ull)
+  {
+    item_user_lock_release(thd->ull);
+    thd->ull=0;
+  }
+  pthread_mutex_unlock(&LOCK_user_locks);
+}
+
+#endif
+
 /*
   Get a user level lock. If the thread has an old lock this is first released.
   Returns 1:  Got lock

sql/log_event.cc

@@ -21,6 +21,7 @@
 #endif
 #include  "mysql_priv.h"
 #include "slave.h"
+#include <my_dir.h>
 #endif /* MYSQL_CLIENT */
 
 #ifdef MYSQL_CLIENT
@@ -127,6 +128,25 @@ Log_event::Log_event(THD* thd_arg, uint16 flags_arg):
     when = time(NULL);
   }
 }
+
+static void cleanup_load_tmpdir()
+{
+  MY_DIR *dirp;
+  FILEINFO *file;
+  uint i;
+  if (!(dirp=my_dir(slave_load_tmpdir,MYF(MY_WME))))
+    return;
+
+  for (i=0;i<(uint)dirp->number_off_files;i++)
+  {
+    file=dirp->dir_entry+i;
+    if (!memcmp(file->name,"SQL_LOAD-",9))
+      my_delete(file->name,MYF(MY_WME));
+  }
+
+  my_dirend(dirp);
+}
+
 #endif
 
 Log_event::Log_event(const char* buf):cached_event_len(0),temp_buf(0)
@@ -1638,6 +1658,7 @@ int Load_log_event::exec_event(NET* net, struct st_master_info* mi)
 int Start_log_event::exec_event(struct st_master_info* mi)
 {
   close_temporary_tables(thd);
+  cleanup_load_tmpdir();
   return Log_event::exec_event(mi);
 }
 
@@ -1646,7 +1667,7 @@ int Stop_log_event::exec_event(struct st_master_info* mi)
   if(mi->pos > 4) // stop event should be ignored after rotate event
   {
     close_temporary_tables(thd);
-    //clean_up_load_tmp_dir();
+    cleanup_load_tmpdir();
     mi->inc_pos(get_event_len(), log_seq);
     flush_master_info(mi);
   }

sql/mysql_priv.h

@@ -190,6 +190,19 @@ char* query_table_status(THD *thd,const char *db,const char *table_name);
 
 #define RAID_BLOCK_SIZE 1024
 
+// Sync points allow us to force the server to reach a certain line of code
+// and block there until the client tells the server it is ok to go on.
+// The client tells the server to block with SELECT GET_LOCK()
+// and unblocks it with SELECT RELEASE_LOCK(). Used for debugging difficult
+// concurrency problems
+#ifdef EXTRA_DEBUG
+#define DBUG_SYNC_POINT(lock_name,lock_timeout) \
+ debug_sync_point(lock_name,lock_timeout)
+void debug_sync_point(const char* lock_name, uint lock_timeout);
+#else
+#define DBUG_SYNC_POINT(lock_name,lock_timeout)
+#endif
+
 /* BINLOG_DUMP options */
 
 #define BINLOG_DUMP_NON_BLOCK   1

sql/sql_repl.cc

@@ -1623,6 +1623,7 @@ int log_loaded_block(IO_CACHE* file)
 			    block_len);
     mysql_bin_log.write(&c);
     lf_info->wrote_create_file = 1;
+    DBUG_SYNC_POINT("debug_lock.created_file_event",10);
   }
   return 0;
 }