Commit cc95bc57 authored by sasha@mysql.sashanet.com's avatar sasha@mysql.sashanet.com
Browse files

manager clean-up ( added pid-file and kill in mysql-test-run when things go wrong) 

fixes for IO_CACHE
need to pull Monty's fixes - this is not final, will not be pushed
parent 4f106fe8

include/my_sys.h

+20 −11
Original line number Diff line number Diff line
@@ -311,13 +311,15 @@ typedef struct st_io_cache /* Used when cacheing files */ 
			     that will use a buffer allocated somewhere

			     else

			   */

  byte *append_buffer, *append_read_pos, *append_write_pos, *append_end;

  byte *append_buffer, *append_read_pos, *write_pos, *append_end,

    *write_end;

/* for append buffer used in READ_APPEND cache */

#ifdef THREAD

  pthread_mutex_t append_buffer_lock;

  /* need mutex copying from append buffer to read buffer */

#endif  

  int (*read_function)(struct st_io_cache *,byte *,uint);

  int (*write_function)(struct st_io_cache *,const byte *,uint);

  /* callbacks when the actual read I/O happens */

  IO_CACHE_CALLBACK pre_read;

  IO_CACHE_CALLBACK post_read;
@@ -352,21 +354,19 @@ typedef int (*qsort2_cmp)(const void *, const void *, const void *); 
    ((info)->rc_pos+=(Count)),0) :\

   (*(info)->read_function)((info),Buffer,Count))



#define my_b_write(info,Buffer,Count) \

 ((info)->write_pos + (Count) <=(info)->write_end ?\

  (memcpy((info)->write_pos, (Buffer), (size_t)(Count)),\

   ((info)->write_pos+=(Count)),0) : \

   (*(info)->write_function)((info),(Buffer),(Count)))

  

 



#define my_b_get(info) \

  ((info)->rc_pos != (info)->rc_end ?\

   ((info)->rc_pos++, (int) (uchar) (info)->rc_pos[-1]) :\

   _my_b_get(info))



#define my_b_write(info,Buffer,Count) \

  ((info)->type != SEQ_READ_APPEND) ? (\

  ((info)->rc_pos + (Count) <= (info)->rc_end ?\

   (memcpy((info)->rc_pos,Buffer,(size_t) (Count)), \

    ((info)->rc_pos+=(Count)),0) :\

   _my_b_write(info,Buffer,Count))) : \

   ((info)->append_write_pos + (Count) <= (info)->append_end ?\

   (memcpy((info)->append_write_pos,Buffer,(size_t)Count), \

   ((info)->append_write_pos+=(Count),0)) : \

   _my_b_append(info,Buffer,Count))



	/* my_b_write_byte dosn't have any err-check */

#define my_b_write_byte(info,chr) \
@@ -650,6 +650,7 @@ byte *my_compress_alloc(const byte *packet, ulong *len, ulong *complen); 
ulong checksum(const byte *mem, uint count);

uint my_bit_log2(ulong value);





#if defined(_MSC_VER) && !defined(__WIN__)

extern void sleep(int sec);

#endif
@@ -662,3 +663,11 @@ extern my_bool have_tcpip; /* Is set if tcpip is used */ 
#endif

#include "raid.h"

#endif /* _my_sys_h */

mysql-test/mysql-test-run.sh

+24 −3
Original line number Diff line number Diff line
@@ -164,7 +164,10 @@ while test $# -gt 0; do 
     --ssl-cert=$BASEDIR/SSL/server-cert.pem \

     --ssl-key=$BASEDIR/SSL/server-key.pem" ;;

    --no-manager | --skip-manager) USE_MANAGER=0 ;;

    --manager)                     USE_MANAGER=1 ;;

    --manager)

     USE_MANAGER=1

     USE_RUNNING_SERVER=

     ;;

    --skip-innobase)

     EXTRA_MASTER_MYSQLD_OPT="$EXTRA_MASTER_MYSQLD_OPT --skip-innobase"

     EXTRA_SLAVE_MYSQLD_OPT="$EXTRA_SLAVE_MYSQLD_OPT --skip-innobase" ;;
@@ -210,6 +213,7 @@ while test $# -gt 0; do 
    --gdb )

      START_WAIT_TIMEOUT=300

      STOP_WAIT_TIMEOUT=300

      USE_MANAGER=1

      if [ x$BINARY_DIST = x1 ] ; then

	$ECHO "Note: you will get more meaningful output on a source distribution compiled with debugging option when running tests with --gdb option"

      fi
@@ -255,6 +259,8 @@ done 
#--



MYRUN_DIR=$MYSQL_TEST_DIR/var/run

MANAGER_PID_FILE="$MYRUN_DIR/manager.pid"



MASTER_MYDDIR="$MYSQL_TEST_DIR/var/master-data"

MASTER_MYSOCK="$MYSQL_TMP_DIR/master.sock"

MASTER_MYPID="$MYRUN_DIR/mysqld.pid"
@@ -550,10 +556,20 @@ start_manager() 
  return

 fi

 $ECHO "Starting MySQL Manager"

 if [ -f "$MANAGER_PID_FILE" ] ; then

    kill `cat $MANAGER_PID_FILE`

    sleep 1

    if [ -f "$MANAGER_PID_FILE" ] ; then

     kill -9 `cat $MANAGER_PID_FILE`

     sleep 1

    fi

 fi



 rm -f $MANAGER_PID_FILE

 MYSQL_MANAGER_PW=`$MYSQL_MANAGER_PWGEN -u $MYSQL_MANAGER_USER \

 -o $MYSQL_MANAGER_PW_FILE`

 $MYSQL_MANAGER --log=$MYSQL_MANAGER_LOG --port=$MYSQL_MANAGER_PORT \

  --password-file=$MYSQL_MANAGER_PW_FILE

  --password-file=$MYSQL_MANAGER_PW_FILE --pid-file=$MANAGER_PID_FILE

  abort_if_failed "Could not start MySQL manager"

  mysqltest_manager_args="--manager-host=localhost \

  --manager-user=$MYSQL_MANAGER_USER \
@@ -562,7 +578,10 @@ start_manager() 
  --manager-wait-timeout=$START_WAIT_TIMEOUT"

  MYSQL_TEST="$MYSQL_TEST $mysqltest_manager_args"

  MYSQL_TEST_ARGS="$MYSQL_TEST_ARGS $mysqltest_manager_args"

  

  while [ ! -f $MANAGER_PID_FILE ] ; do

   sleep 1

  done

  echo "Manager started"

}



stop_manager()
@@ -574,6 +593,8 @@ stop_manager() 
  -p$MYSQL_MANAGER_PW -P $MYSQL_MANAGER_PORT <<EOF

shutdown

EOF

 echo "Manager terminated"



}



manager_launch()

mysys/mf_iocache.c

+77 −53
Original line number Diff line number Diff line
@@ -46,6 +46,7 @@ static void my_aiowait(my_aio_result *result); 
#endif



static void init_read_function(IO_CACHE* info, enum cache_type type);

static void init_write_function(IO_CACHE* info, enum cache_type type);



static void init_read_function(IO_CACHE* info, enum cache_type type)

{
@@ -69,6 +70,18 @@ static void init_read_function(IO_CACHE* info, enum cache_type type) 
  }

}



static void init_write_function(IO_CACHE* info, enum cache_type type)

{

  switch (type)

  {

  case SEQ_READ_APPEND:

    info->write_function = _my_b_append;

    break;

  default:

    info->write_function = _my_b_write;

  }

}



	/*

	** if cachesize == 0 then use default cachesize (from s-file)

	** if file == -1 then real_open_cached_file() will be called.
@@ -91,6 +104,7 @@ int init_io_cache(IO_CACHE *info, File file, uint cachesize, 
    if (! (cachesize= my_default_record_cache_size))

      DBUG_RETURN(1);				/* No cache requested */

  min_cache=use_async_io ? IO_SIZE*4 : IO_SIZE*2;

  info->alloced_buffer = 0;

  if (type == READ_CACHE || type == SEQ_READ_APPEND)

  {						/* Assume file isn't growing */

    if (cache_myflags & MY_DONT_CHECK_FILESIZE)
@@ -117,7 +131,6 @@ int init_io_cache(IO_CACHE *info, File file, uint cachesize, 
      }

    }

  }

  info->alloced_buffer = 0;

  if ((int) type < (int) READ_NET)

  {

    uint buffer_block;
@@ -156,8 +169,9 @@ int init_io_cache(IO_CACHE *info, File file, uint cachesize, 
  info->rc_request_pos=info->rc_pos=info->buffer;

  if (type == SEQ_READ_APPEND)

  {

    info->append_read_pos = info->append_write_pos = info->append_buffer;

    info->append_end = info->append_buffer + info->buffer_length;

    info->append_read_pos = info->write_pos = info->append_buffer;

    info->write_end = info->append_end =

      info->append_buffer + info->buffer_length;

#ifdef THREAD    

    pthread_mutex_init(&info->append_buffer_lock,MY_MUTEX_INIT_FAST);

#endif
@@ -170,7 +184,9 @@ int init_io_cache(IO_CACHE *info, File file, uint cachesize, 
  }

  else /* type == WRITE_CACHE */

  {

    info->rc_end=info->buffer+info->buffer_length- (seek_offset & (IO_SIZE-1));

    info->write_end=

      info->buffer+info->buffer_length- (seek_offset & (IO_SIZE-1));

    info->write_pos = info->buffer;

  }

  /* end_of_file may be changed by user later */

  info->end_of_file= ((type == READ_NET || type == READ_FIFO ) ? 0
@@ -178,6 +194,7 @@ int init_io_cache(IO_CACHE *info, File file, uint cachesize, 
  info->type=type;

  info->error=0;

  init_read_function(info,type);

  init_write_function(info,type);

#ifdef HAVE_AIOWAIT

  if (use_async_io && ! my_disable_async_io)

  {
@@ -238,15 +255,21 @@ my_bool reinit_io_cache(IO_CACHE *info, enum cache_type type, 
  {						/* use current buffer */

    if (info->type == WRITE_CACHE && type == READ_CACHE)

    {

      info->rc_end=info->rc_pos;

      info->rc_end=info->write_pos;

      info->end_of_file=my_b_tell(info);

    }

    else if (type == WRITE_CACHE)

    {

      if (info->type == READ_CACHE)

	info->rc_end=info->buffer+info->buffer_length;

      {

	info->write_end=info->buffer+info->buffer_length;

	info->write_pos=info->rc_pos;

      }

      info->end_of_file = ~(my_off_t) 0;

    }

    if (type == WRITE_CACHE)

      info->write_pos=info->rc_request_pos+(seek_offset-info->pos_in_file);

    else  

      info->rc_pos=info->rc_request_pos+(seek_offset-info->pos_in_file);

#ifdef HAVE_AIOWAIT

    my_aiowait(&info->aio_result);		/* Wait for outstanding req */
@@ -283,11 +306,12 @@ my_bool reinit_io_cache(IO_CACHE *info, enum cache_type type, 
  }

  if (info->type == SEQ_READ_APPEND)

  {

    info->append_read_pos = info->append_write_pos = info->append_buffer;

    info->append_read_pos = info->write_pos = info->append_buffer;

  }

  info->type=type;

  info->error=0;

  init_read_function(info,type);

  init_write_function(info,type);

#ifdef HAVE_AIOWAIT

  if (type != READ_NET)

  {
@@ -473,8 +497,8 @@ int _my_b_seq_read(register IO_CACHE *info, byte *Buffer, uint Count) 
  if (!Count) return 0;

  {

    uint copy_len = (uint)(info->append_read_pos -

			   info->append_write_pos);

    dbug_assert(info->append_read_pos <= info->append_write_pos);

			   info->write_pos);

    dbug_assert(info->append_read_pos <= info->write_pos);

    if (copy_len > Count)

      copy_len = Count;

    memcpy(Buffer, info->append_read_pos,
@@ -679,11 +703,11 @@ int _my_b_write(register IO_CACHE *info, const byte *Buffer, uint Count) 
{

  uint rest_length,length;



  rest_length=(uint) (info->rc_end - info->rc_pos);

  memcpy(info->rc_pos,Buffer,(size_t) rest_length);

  rest_length=(uint) (info->write_end - info->write_pos);

  memcpy(info->write_pos,Buffer,(size_t) rest_length);

  Buffer+=rest_length;

  Count-=rest_length;

  info->rc_pos+=rest_length;

  info->write_pos+=rest_length;

  if (info->pos_in_file+info->buffer_length > info->end_of_file)

  {

    my_errno=errno=EFBIG;
@@ -705,8 +729,8 @@ int _my_b_write(register IO_CACHE *info, const byte *Buffer, uint Count) 
    Buffer+=length;

    info->pos_in_file+=length;

  }

  memcpy(info->rc_pos,Buffer,(size_t) Count);

  info->rc_pos+=Count;

  memcpy(info->write_pos,Buffer,(size_t) Count);

  info->write_pos+=Count;

  return 0;

}
@@ -715,11 +739,11 @@ int _my_b_append(register IO_CACHE *info, const byte *Buffer, uint Count) 
  uint rest_length,length;



  rest_length=(uint) (info->append_end -

		      info->append_write_pos);

  memcpy(info->append_write_pos,Buffer,(size_t) rest_length);

		      info->write_pos);

  memcpy(info->write_pos,Buffer,(size_t) rest_length);

  Buffer+=rest_length;

  Count-=rest_length;

  info->append_write_pos+=rest_length;

  info->write_pos+=rest_length;

  if (flush_io_cache(info))

    return 1;

  if (Count >= IO_SIZE)
@@ -730,8 +754,8 @@ int _my_b_append(register IO_CACHE *info, const byte *Buffer, uint Count) 
    Count-=length;

    Buffer+=length;

  }

  memcpy(info->append_write_pos,Buffer,(size_t) Count);

  info->append_write_pos+=Count;

  memcpy(info->write_pos,Buffer,(size_t) Count);

  info->write_pos+=Count;

  return 0;

}
@@ -775,8 +799,8 @@ int my_block_write(register IO_CACHE *info, const byte *Buffer, uint Count, 
    Buffer+=length;

    Count-= length;

    /* Fix length of buffer if the new data was larger */

    if (info->buffer+length > info->rc_pos)

      info->rc_pos=info->buffer+length;

    if (info->buffer+length > info->write_pos)

      info->write_pos=info->buffer+length;

    if (!Count)

      return (error);

  }
@@ -786,62 +810,61 @@ int my_block_write(register IO_CACHE *info, const byte *Buffer, uint Count, 
  return error;

}



/* avoid warning about empty if body */

#ifdef THREAD

#define IF_APPEND_CACHE if (append_cache)

#else

#define IF_APPEND_CACHE

#endif



	/* Flush write cache */



int flush_io_cache(IO_CACHE *info)

{

  uint length;

  int append_cache;

  DBUG_ENTER("flush_io_cache");



  if (info->type == WRITE_CACHE)

  append_cache = (info->type == SEQ_READ_APPEND);

  if (info->type == WRITE_CACHE || append_cache)

  {

    if (info->file == -1)

    {

      if (real_open_cached_file(info))

	DBUG_RETURN((info->error= -1));

    }

    if (info->rc_pos != info->buffer)

    IF_APPEND_CACHE

      lock_append_buffer(info);

    if (info->write_pos != info->buffer)

    {

      length=(uint) (info->rc_pos - info->buffer);

      length=(uint) (info->write_pos - info->buffer);

      if (info->seek_not_done)

      {					/* File touched, do seek */

	if (my_seek(info->file,info->pos_in_file,MY_SEEK_SET,MYF(0)) ==

	    MY_FILEPOS_ERROR)

	{

	  IF_APPEND_CACHE

	    unlock_append_buffer(info);

	  DBUG_RETURN((info->error= -1));

	}

	info->seek_not_done=0;

      }

      info->rc_pos=info->buffer;

      info->write_pos=info->buffer;

      info->pos_in_file+=length;

      info->rc_end=(info->buffer+info->buffer_length-

      info->write_end=(info->buffer+info->buffer_length-

		    (info->pos_in_file & (IO_SIZE-1)));

      if (my_write(info->file,info->buffer,length,info->myflags | MY_NABP))

	DBUG_RETURN((info->error= -1));

      DBUG_RETURN(0);

    }

  }

  else if (info->type == SEQ_READ_APPEND)

      if (append_cache)

      {

    if (info->file == -1)

    {

      if (real_open_cached_file(info))

	DBUG_RETURN((info->error= -1));

	info->append_read_pos = info->buffer;

	info->append_end = info->write_end;

      }

    lock_append_buffer(info);

    if (info->append_write_pos != info->append_buffer)

    {

      length=(uint) (info->append_write_pos - info->append_buffer);

      info->append_read_pos=info->append_write_pos=info->append_buffer;

      info->append_end=(info->append_buffer+info->buffer_length-

		    (info->pos_in_file & (IO_SIZE-1)));

      if (my_write(info->file,info->buffer,length,info->myflags | MY_NABP))

      {

	unlock_append_buffer(info);

	DBUG_RETURN((info->error= -1));

      }

        info->error= -1;

      else

	info->error= 0;

      IF_APPEND_CACHE

        unlock_append_buffer(info);

      DBUG_RETURN(0);

      DBUG_RETURN(info->error);

    }

    unlock_append_buffer(info);

  }

#ifdef HAVE_AIOWAIT

  else if (info->type != READ_NET)
@@ -867,6 +890,7 @@ int end_io_cache(IO_CACHE *info) 
      error=flush_io_cache(info);

    my_free((gptr) info->buffer,MYF(MY_WME));

    info->buffer=info->rc_pos=(byte*) 0;

    info->alloced_buffer = 0;

  }

  DBUG_RETURN(error);

} /* end_io_cache */

sql/mysqld.cc

+5 −1
Original line number Diff line number Diff line
@@ -707,9 +707,11 @@ void unireg_end(int signal_number __attribute__((unused))) 


void unireg_abort(int exit_code)

{

  DBUG_ENTER("unireg_abort");

  if (exit_code)

    sql_print_error("Aborting\n");

  clean_up(); /* purecov: inspected */

  DBUG_PRINT("quit",("done with cleanup in unireg_abort"));

  my_thread_end();

  exit(exit_code); /* purecov: inspected */

}
@@ -760,13 +762,15 @@ void clean_up(bool print_message) 
  if (print_message && errmesg)

    sql_print_error(ER(ER_SHUTDOWN_COMPLETE),my_progname);

  x_free((gptr) my_errmsg[ERRMAPP]);	/* Free messages */



  DBUG_PRINT("quit", ("Error messages freed"));

  /* Tell main we are ready */

  (void) pthread_mutex_lock(&LOCK_thread_count);

  DBUG_PRINT("quit", ("got thread count lock"));

  ready_to_exit=1;

  /* do the broadcast inside the lock to ensure that my_end() is not called */

  (void) pthread_cond_broadcast(&COND_thread_count);

  (void) pthread_mutex_unlock(&LOCK_thread_count);

  DBUG_PRINT("quit", ("done with cleanup"));

} /* clean_up */

tools/mysqlmanager.c

+32 −1
Original line number Diff line number Diff line
@@ -189,6 +189,8 @@ static void run_launcher_loop(); 
int to_launcher_pipe[2],from_launcher_pipe[2];

pid_t launcher_pid;

int in_segfault=0;

const char* pid_file = "/var/run/mysqlmanager.pid";

int created_pid_file = 0;



struct manager_cmd

{
@@ -283,6 +285,7 @@ struct option long_options[] = 
  {"one-thread",no_argument,0,'d'},

  {"connect-retries",required_argument,0,'C'},

  {"password-file",required_argument,0,'p'},

  {"pid-file",required_argument,0,'f'},

  {"version", no_argument, 0, 'V'},

  {0, 0, 0, 0}

};
@@ -327,6 +330,17 @@ LOG_MSG_FUNC(log_debug,LOG_DEBUG) 
void log_debug(const char* __attribute__((unused)) fmt,...) {}

#endif



static void handle_sigterm(int sig)

{

  log_info("Got SIGTERM"); 

  if (!one_thread)

  {

    kill(launcher_pid,SIGTERM);

    pthread_kill(loop_th,SIGTERM);

  }

  clean_up();

  exit(0);

}



static void handle_segfault(int sig)

{
@@ -1250,6 +1264,8 @@ static void clean_up() 
  if (errfp != stderr)

    fclose(errfp);

  hash_free(&exec_hash);

  if (created_pid_file)

    my_delete(pid_file, MYF(0));

}



static void print_version(void)
@@ -1287,7 +1303,7 @@ static void usage() 
static int parse_args(int argc, char **argv)

{

  int c, option_index = 0;

  while ((c=getopt_long(argc,argv,"P:?#:Vl:b:B:g:m:dC:p:",

  while ((c=getopt_long(argc,argv,"P:?#:Vl:b:B:g:m:dC:p:f:",

			long_options,&option_index)) != EOF)

  {

    switch (c)
@@ -1301,6 +1317,9 @@ static int parse_args(int argc, char **argv) 
      case 'p':

	manager_pw_file=optarg;

	break;

      case 'f':

	pid_file=optarg;

	break;

      case 'C':

        manager_connect_retries=atoi(optarg);

	break;
@@ -1662,6 +1681,16 @@ static void init_user_hash() 
  fclose(f);

}



static void init_pid_file()

{

  FILE* fp = fopen(pid_file, "w");

  if (!fp)

    die("Could not open pid file %s", pid_file);

  created_pid_file=1;

  fprintf(fp, "%d\n", getpid());

  fclose(fp);

}



static void init_globals()

{

  pthread_attr_t thr_attr;
@@ -1680,8 +1709,10 @@ static void init_globals() 
    /* (void) pthread_attr_destroy(&thr_attr); */

  }

  init_user_hash();

  init_pid_file();

  loop_th=pthread_self();

  signal(SIGPIPE,handle_sigpipe);

  signal(SIGTERM,handle_sigterm);

}



static int open_and_dup(int fd,char* path)